mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Improved NDR padding capabilities for primitive types
This commit is contained in:
hakril
@@ -0,0 +1,127 @@
|
||||
import pytest
|
||||
|
||||
from windows.rpc import ndr
|
||||
from .pfwtest import *
|
||||
|
||||
from tests.test_rpc import UACParameters
|
||||
|
||||
# Memo: Padding byte is 'P' in ndr.py
|
||||
|
||||
# 20 Bytes structures alignes on 4 butees
|
||||
DoubleDwordStructure = ndr.make_structure([ndr.NdrLong] * 5)
|
||||
|
||||
# Struct with some specificities
|
||||
# Start Must be aligned on 4 (even if it begin with a short)
|
||||
# full size == 13 -> usefull to test afterward padding as well
|
||||
# Pack format ah follow:
|
||||
# SSBPLLLLBPSSB
|
||||
InternalAlignementStructure = ndr.make_structure([ndr.NdrShort, ndr.NdrByte, ndr.NdrLong, ndr.NdrByte, ndr.NdrShort, ndr.NdrByte])
|
||||
# IDL code:
|
||||
# typedef struct InternalAlignementStructure
|
||||
# {
|
||||
#short sfield0;
|
||||
#byte bfield1;
|
||||
#long lfield2;
|
||||
#byte bfield3;
|
||||
#short sfield4;
|
||||
#byte bfield5;
|
||||
# }InternalAlignementStructure;
|
||||
|
||||
|
||||
# NdrObject, Values, result
|
||||
NDR_PACK_TEST_CASE = [
|
||||
# Simple case
|
||||
(ndr.make_structure([ndr.NdrLong, ndr.NdrLong]), (2, 2), b"\x02\x00\x00\x00\x02\x00\x00\x00"),
|
||||
# Check alignement on small native types (dword aligned)
|
||||
(ndr.make_structure([ndr.NdrShort, ndr.NdrByte]), (0x0101, 2), b"\x01\x01\x02"),
|
||||
# Check alignement on small native types (dword aligned)
|
||||
(ndr.make_structure([ndr.NdrShort, ndr.NdrShort]), (0x0101, 0x0202), b"\x01\x01\x02\x02"),
|
||||
# Same check on parameters
|
||||
(ndr.make_parameters([ndr.NdrShort, ndr.NdrByte]), (0x0101, 2), b"\x01\x01\x02"),
|
||||
# Test some Hyper
|
||||
(ndr.make_parameters([ndr.NdrByte, ndr.NdrHyper, ndr.NdrByte, ndr.NdrLong, ndr.NdrHyper]),
|
||||
(0x01, 0x4141414141414141, 0x42, 0x43434343, 0x4444444444444444 ),
|
||||
b"\x01PPPPPPPAAAAAAAABPPPCCCCDDDDDDDD"),
|
||||
# Complexe structure (with 4B alignement of 20B structure)
|
||||
(ndr.make_parameters([ndr.NdrShort, DoubleDwordStructure]),
|
||||
(0x0101, [0x41414141, 0x42424242, 0x43434343, 0x44444444, 0x45454545]),
|
||||
b"\x01\x01PPAAAABBBBCCCCDDDDEEEE"),
|
||||
# Same check on parameters
|
||||
(ndr.make_parameters([ndr.NdrShort, DoubleDwordStructure]),
|
||||
(0x0101, [0x41414141, 0x42424242, 0x43434343, 0x44444444, 0x45454545]),
|
||||
b"\x01\x01PPAAAABBBBCCCCDDDDEEEE"),
|
||||
|
||||
# Check on InternalAlignementStructure before any nested test
|
||||
(InternalAlignementStructure,
|
||||
[0x4141, 0x42, 0x43434343, 0x44, 0x4545, 0x46],
|
||||
b"AABPCCCCDPEEF"),
|
||||
|
||||
# Nested alignement
|
||||
# Alignement with a sub-structure that also have internal alignement
|
||||
(ndr.make_parameters([ndr.NdrShort, InternalAlignementStructure, ndr.NdrByte]),
|
||||
(0x0101, [0x4141, 0x42, 0x43434343, 0x44, 0x4545, 0x46], 0x47),
|
||||
# Verified with an actual RPC server
|
||||
b"\x01\x01PPAABPCCCCDPEEFG"),
|
||||
# Nested alignement
|
||||
# Alignement with a sub-structure that also have internal alignement
|
||||
# Afterward short should be aligned on 2
|
||||
(ndr.make_parameters([ndr.NdrShort, InternalAlignementStructure, ndr.NdrShort]),
|
||||
(0x0101, [0x4141, 0x42, 0x43434343, 0x44, 0x4545, 0x46], 0x4747),
|
||||
# Verified with an actual RPC server
|
||||
b"\x01\x01PPAABPCCCCDPEEFPGG"),
|
||||
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("ndrobj, values, result", NDR_PACK_TEST_CASE)
|
||||
def test_ndr_packing(ndrobj, values, result):
|
||||
assert ndrobj.pack(values) == result
|
||||
|
||||
# Check the result of serializing a fixed 'EptMapAuthParameters' call known to works
|
||||
# It allow to test for packing of real-complexe Parameter without relying on the whole ALPC/RPC stack
|
||||
def test_ndr_packing_complex_epmapper_call():
|
||||
# Param from a real UAC endpoint resolution
|
||||
|
||||
targetiid = gdef.GUID.from_string("201EF99A-7FA0-444C-9399-19BA84F12A1A")
|
||||
towerarray = bytearray(b'\x04\x00\x13\x00\r\x9a\xf9\x1e \xa0\x7fLD\x93\x99\x19\xba\x84\xf1*\x1a\x01\x00\x02\x00\x00\x00\x13\x00\r\x04]\x88\x8a\xeb\x1c\xc9\x11\x9f\xe8\x08\x00+\x10H`\x02\x00\x02\x00\x00\x00\x01\x00\x0c\x02\x00\x00\x00\x01\x00\x10\x00\x00')
|
||||
local_system_psid = gdef.PSID.from_string("S-1-5-18")
|
||||
context = (0, 0, 0, 0, 0)
|
||||
nb_response = 1
|
||||
|
||||
packed = windows.rpc.epmapper.EptMapAuthParameters.pack([bytearray(targetiid),
|
||||
(len(towerarray), towerarray),
|
||||
local_system_psid,
|
||||
context,
|
||||
nb_response])
|
||||
|
||||
expected_result = b'\x9a\xf9\x1e \xa0\x7fLD\x93\x99\x19\xba\x84\xf1*\x1a@\x00\x00\x00@\x00\x00\x00\x04\x00\x13\x00\r\x9a\xf9\x1e \xa0\x7fLD\x93\x99\x19\xba\x84\xf1*\x1a\x01\x00\x02\x00\x00\x00\x13\x00\r\x04]\x88\x8a\xeb\x1c\xc9\x11\x9f\xe8\x08\x00+\x10H`\x02\x00\x02\x00\x00\x00\x01\x00\x0c\x02\x00\x00\x00\x01\x00\x10\x00\x00\x02\x02\x02\x02\x01\x00\x00\x00\x01\x01\x00\x00\x00\x00\x00\x05\x12\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00'
|
||||
assert packed == packed
|
||||
|
||||
|
||||
# Check the result of serializing a fixed 'UAC' call known to works
|
||||
# It allow to test for packing of real-complexe Parameter without relying on the whole ALPC/RPC stack
|
||||
def test_ndr_packing_complex_uac_call():
|
||||
parameters = UACParameters.pack([
|
||||
r"c:\windows\system32\notepad.exe", # Application Path
|
||||
"NOT_ALIGNED_STRINGXXX", # Commandline
|
||||
17, # UAC-Request Flag
|
||||
gdef.CREATE_UNICODE_ENVIRONMENT, # dwCreationFlags
|
||||
"", # StartDirectory
|
||||
"WinSta0\\Default\x00", # Station
|
||||
# Startup Info
|
||||
(None, # Title
|
||||
0, # dwX
|
||||
0, # dwY
|
||||
0, # dwXSize
|
||||
0, # dwYSize
|
||||
0, # dwXCountChars
|
||||
0, # dwYCountChars
|
||||
0, # dwFillAttribute
|
||||
0, # dwFlags
|
||||
5, # wShowWindow
|
||||
# Point structure: Use MonitorFromPoint to setup StartupInfo.hStdOutput
|
||||
(0, 0)),
|
||||
0, # Window-Handle to know if UAC can steal focus
|
||||
0xffffffff]) # UAC Timeout
|
||||
|
||||
expected_result = b'\x02\x02\x02\x02 \x00\x00\x00\x00\x00\x00\x00 \x00\x00\x00c\x00:\x00\\\x00w\x00i\x00n\x00d\x00o\x00w\x00s\x00\\\x00s\x00y\x00s\x00t\x00e\x00m\x003\x002\x00\\\x00n\x00o\x00t\x00e\x00p\x00a\x00d\x00.\x00e\x00x\x00e\x00\x00\x00\x02\x02\x02\x02\x16\x00\x00\x00\x00\x00\x00\x00\x16\x00\x00\x00N\x00O\x00T\x00_\x00A\x00L\x00I\x00G\x00N\x00E\x00D\x00_\x00S\x00T\x00R\x00I\x00N\x00G\x00X\x00X\x00X\x00\x00\x00\x11\x00\x00\x00\x00\x04\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00PP\x10\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x00W\x00i\x00n\x00S\x00t\x00a\x000\x00\\\x00D\x00e\x00f\x00a\x00u\x00l\x00t\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff'
|
||||
assert parameters == expected_result
|
||||
+102
-7
@@ -64,6 +64,11 @@ class NdrUniquePTR(object):
|
||||
return None
|
||||
return self.subcls.parse(stream)
|
||||
|
||||
def get_alignment(self):
|
||||
# 14.3.2 Alignment of Constructed Types
|
||||
# Pointer alignment is always modulo 4.
|
||||
return 4
|
||||
|
||||
class NdrUnpackNone(object):
|
||||
@classmethod
|
||||
def unpack(cls, stream):
|
||||
@@ -94,6 +99,9 @@ class NdrFixedArray(object):
|
||||
def unpack(self, stream):
|
||||
return [self.subcls.unpack(stream) for i in range(self.size)]
|
||||
|
||||
def get_alignment(self):
|
||||
return self.subcls.get_alignment()
|
||||
|
||||
|
||||
class NdrSID(object):
|
||||
@classmethod
|
||||
@@ -113,6 +121,11 @@ class NdrSID(object):
|
||||
subcount = NdrLong.unpack(stream)
|
||||
return stream.read(8 + (subcount * 4))
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
# Not sur, but it seems to contain an array of long
|
||||
return 4
|
||||
|
||||
class NdrVaryingCString(object):
|
||||
@classmethod
|
||||
def pack(cls, data):
|
||||
@@ -126,6 +139,11 @@ class NdrVaryingCString(object):
|
||||
result += data
|
||||
return dword_pad(result)
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
# Not sur, but size is on 4 bytes so...
|
||||
return 4
|
||||
|
||||
class NdrWString(object):
|
||||
@classmethod
|
||||
def pack(cls, data):
|
||||
@@ -149,6 +167,11 @@ class NdrWString(object):
|
||||
s = stream.read(size1 * 2)
|
||||
return s.decode("utf-16-le")
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
# Not sur, but size is on 4 bytes so...
|
||||
return 4
|
||||
|
||||
class NdrCString(object):
|
||||
@classmethod
|
||||
def pack(cls, data):
|
||||
@@ -162,6 +185,11 @@ class NdrCString(object):
|
||||
result += data
|
||||
return dword_pad(result)
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
# Not sur, but size is on 4 bytes so...
|
||||
return 4
|
||||
|
||||
# @classmethod
|
||||
# def unpack(self, stream):
|
||||
# maxcount, offset, count = stream.partial_unpack("<3I")
|
||||
@@ -180,6 +208,10 @@ class NdrLong(object):
|
||||
stream.align(4)
|
||||
return stream.partial_unpack("<I")[0]
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
return 4
|
||||
|
||||
class NdrHyper(object):
|
||||
@classmethod
|
||||
def pack(cls, data):
|
||||
@@ -190,6 +222,10 @@ class NdrHyper(object):
|
||||
stream.align(8)
|
||||
return stream.partial_unpack("<Q")[0]
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
return 8
|
||||
|
||||
class NdrShort(object):
|
||||
@classmethod
|
||||
def pack(cls, data):
|
||||
@@ -199,6 +235,10 @@ class NdrShort(object):
|
||||
def unpack(self, stream):
|
||||
return stream.partial_unpack("<H")[0]
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
return 2
|
||||
|
||||
|
||||
class NdrByte(object):
|
||||
@classmethod
|
||||
@@ -209,6 +249,10 @@ class NdrByte(object):
|
||||
def unpack(self, stream):
|
||||
return stream.partial_unpack("<B")[0]
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
return 1
|
||||
|
||||
|
||||
class NdrGuid(object):
|
||||
@classmethod
|
||||
@@ -251,21 +295,30 @@ class NdrStructure(object):
|
||||
raise ValueError("NdrStructure packing number elements mismatch: structure has <{0}> members got <{1}>".format(len(cls.MEMBERS), len(data)))
|
||||
conformant_size = []
|
||||
res = []
|
||||
res_size = 0
|
||||
pointed = []
|
||||
outstream = NdrWriteStream()
|
||||
for i, (member, memberdata) in enumerate(zip(cls.MEMBERS, data)):
|
||||
if hasattr(member, "pack_in_struct"):
|
||||
x, y = member.pack_in_struct(memberdata, i)
|
||||
res.append(x)
|
||||
outstream.align(member.get_alignment())
|
||||
outstream.write(x)
|
||||
# res.append(x)
|
||||
# res_size += len(x)
|
||||
if y is not None:
|
||||
pointed.append(y)
|
||||
elif hasattr(member, "pack_conformant"):
|
||||
size, data = member.pack_conformant(memberdata)
|
||||
outstream.align(member.get_alignment())
|
||||
outstream.write(data)
|
||||
conformant_size.append(size)
|
||||
res.append(data)
|
||||
# res.append(data)
|
||||
# res_size += len(data)
|
||||
else:
|
||||
packed_member = member.pack(memberdata)
|
||||
res.append(packed_member)
|
||||
return dword_pad(b"".join(conformant_size)) + dword_pad(b"".join(res)) + dword_pad(b"".join(pointed))
|
||||
outstream.align(member.get_alignment())
|
||||
outstream.write(packed_member)
|
||||
return dword_pad(b"".join(conformant_size)) + outstream.get_data() + dword_pad(b"".join(pointed))
|
||||
|
||||
@classmethod
|
||||
def unpack(cls, stream):
|
||||
@@ -305,6 +358,10 @@ class NdrStructure(object):
|
||||
def post_unpack(cls, data):
|
||||
return data
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
return max([x.get_alignment() for x in self.MEMBERS])
|
||||
|
||||
|
||||
|
||||
class NdrParameters(object):
|
||||
@@ -320,11 +377,17 @@ class NdrParameters(object):
|
||||
print(" * data {0}".format(data))
|
||||
print(" * members = {0}".format(cls.MEMBERS))
|
||||
raise ValueError("NdrParameters packing number elements mismatch: structure has <{0}> members got <{1}>".format(len(cls.MEMBERS), len(data)))
|
||||
res = []
|
||||
|
||||
|
||||
outstream = NdrWriteStream()
|
||||
for (member, memberdata) in zip(cls.MEMBERS, data):
|
||||
alignment = member.get_alignment()
|
||||
outstream.align(alignment)
|
||||
packed_member = member.pack(memberdata)
|
||||
res.append(packed_member)
|
||||
return b"".join(dword_pad(elt) for elt in res)
|
||||
outstream.write(packed_member)
|
||||
return outstream.get_data()
|
||||
|
||||
|
||||
|
||||
@classmethod
|
||||
def unpack(cls, stream):
|
||||
@@ -334,6 +397,9 @@ class NdrParameters(object):
|
||||
res.append(unpacked_member)
|
||||
return res
|
||||
|
||||
def get_alignment(self):
|
||||
raise ValueError("NdrParameters should always be top type in NDR description")
|
||||
|
||||
|
||||
class NdrConformantArray(object):
|
||||
MEMBER_TYPE = None
|
||||
@@ -364,6 +430,11 @@ class NdrConformantArray(object):
|
||||
stream.align(4)
|
||||
return res
|
||||
|
||||
@classmethod
|
||||
def get_alignment(self):
|
||||
# TODO: test on array of Hyper
|
||||
return max(4, self.MEMBER_TYPE.get_alignment())
|
||||
|
||||
|
||||
class NdrConformantVaryingArrays(object):
|
||||
MEMBER_TYPE = None
|
||||
@@ -406,6 +477,9 @@ class NdrConformantVaryingArrays(object):
|
||||
def _post_unpack(cls, result):
|
||||
return result
|
||||
|
||||
def get_alignment(self):
|
||||
# TODO: test on array of Hyper
|
||||
return max(4, self.MEMBER_TYPE.get_alignment())
|
||||
|
||||
class NdrWcharConformantVaryingArrays(NdrConformantVaryingArrays):
|
||||
MEMBER_TYPE = NdrShort
|
||||
@@ -487,6 +561,27 @@ class NdrStream(object):
|
||||
# print("align {0}: 0".format(size))
|
||||
return 0
|
||||
|
||||
class NdrWriteStream(object):
|
||||
def __init__(self):
|
||||
self.data_parts = []
|
||||
self.data_size = 0
|
||||
|
||||
def get_data(self):
|
||||
data = b"".join(self.data_parts)
|
||||
assert len(data) == self.data_size
|
||||
return data
|
||||
|
||||
def write(self, data):
|
||||
self.data_parts.append(data)
|
||||
self.data_size += len(data)
|
||||
return None
|
||||
|
||||
def align(self, alignement):
|
||||
if self.data_size % alignement == 0:
|
||||
return
|
||||
topadsize = (alignement) - (self.data_size % alignement)
|
||||
self.write(b"P" * topadsize)
|
||||
return
|
||||
|
||||
def make_parameters(types, name=None):
|
||||
class NdrCustomParameters(NdrParameters):
|
||||
|
||||
Reference in New Issue
Block a user