From d8ac02e6a2bd7f5ca4cebc5b9a7023333ef42f2c Mon Sep 17 00:00:00 2001 From: hakril Date: Mon, 15 Jun 2020 10:00:05 +0200 Subject: [PATCH] Improved NDR padding capabilities for primitive types --- tests/test_ndr.py | 127 +++++++++++++++++++++++++++++++++++++++++++++ windows/rpc/ndr.py | 109 +++++++++++++++++++++++++++++++++++--- 2 files changed, 229 insertions(+), 7 deletions(-) create mode 100644 tests/test_ndr.py diff --git a/tests/test_ndr.py b/tests/test_ndr.py new file mode 100644 index 0000000..748d0aa --- /dev/null +++ b/tests/test_ndr.py @@ -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 \ No newline at end of file diff --git a/windows/rpc/ndr.py b/windows/rpc/ndr.py index 382acff..2e092f6 100644 --- a/windows/rpc/ndr.py +++ b/windows/rpc/ndr.py @@ -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(" 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):