import sys import pytest import os.path import time import windows.rpc as rpc from windows.rpc import ndr import windows.generated_def as gdef from .pfwtest import * UAC_UIID = "201ef99a-7fa0-444c-9399-19ba84f12a1a" def start_uac_service(): appinfo_service = windows.system.services["AppInfo"] if appinfo_service.status.state == gdef.SERVICE_RUNNING: return False if appinfo_service.status.state != gdef.SERVICE_START_PENDING: appinfo_service.start() time.sleep(1) # Wait if just started or not marked as running yet return True def test_rpc_epmapper(): start_uac_service() endpoints = windows.rpc.find_alpc_endpoints(UAC_UIID) assert endpoints endpoint = endpoints[0] assert endpoint.protseq == "ncalrpc" assert endpoint.endpoint assert endpoint.object.Uuid.to_string().lower() == UAC_UIID # NDR Descriptions class NDRPoint(ndr.NdrStructure): MEMBERS = [ndr.NdrLong, ndr.NdrLong] class NdrUACStartupInfo(ndr.NdrStructure): MEMBERS = [ndr.NdrUniquePTR(ndr.NdrWString), ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, NDRPoint] class UACParameters(ndr.NdrParameters): MEMBERS = [ndr.NdrUniquePTR(ndr.NdrWString), ndr.NdrUniquePTR(ndr.NdrWString), ndr.NdrLong, ndr.NdrLong, ndr.NdrWString, ndr.NdrWString, NdrUACStartupInfo, ndr.NdrLong, ndr.NdrLong] class NdrProcessInformation(ndr.NdrParameters): MEMBERS = [ndr.NdrLong] * 4 def test_rpc_uac_call(): start_uac_service() client = windows.rpc.find_alpc_endpoint_and_connect(UAC_UIID) iid = client.bind(UAC_UIID) python_path = sys.executable python_name = os.path.basename(python_path) # Marshalling parameters. parameters = UACParameters.pack([ python_path + "\x00", # Application Path python_path + "\x00", # Commandline 0, # UAC-Request Flag gdef.CREATE_UNICODE_ENVIRONMENT, # dwCreationFlags "\x00", # 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 result = client.call(iid, 0, parameters) stream = ndr.NdrStream(result) ph, th, pid, tid = NdrProcessInformation.unpack(stream) return_value = ndr.NdrLong.unpack(stream) assert ph assert pid assert th assert tid windows.winproxy.CloseHandle(th) # NoLeak proc = windows.WinProcess(handle=ph) assert proc.name == python_name assert proc.pid == pid proc.exit(0) class DbgRpcClient(windows.rpc.RPCClient): def __init__(self, *args, **kwargs): super(DbgRpcClient, self).__init__(*args, **kwargs) self.last_response_was_view = False def _get_response_effective_data(self, response): self.last_response_was_view = response.view_is_valid return super(DbgRpcClient, self)._get_response_effective_data(response) FIREWALL_RPC_IID = "2fb92682-6599-42dc-ae13-bd2ca89bd11c" Proc0_RPC_FWOpenPolicyStore = 0 Proc9_RPC_FWEnumFirewallRules = 9 def test_rpc_response_as_view(): """Check that parsing response as view in RPC Client works. Testing after a bug in 32b RPCCLient""" # We test what by using a RPC endpoint that returns a lot of info : forcing a response in a view # In this case we use the Firewall RPC and we list all Firerules. # We use a custom RPCClient subclasse to track if last response was a view client = windows.rpc.find_alpc_endpoint_and_connect(FIREWALL_RPC_IID, sid=gdef.WinLocalSid) client.__class__ = DbgRpcClient iid = client.bind(FIREWALL_RPC_IID) # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-fasp/230d1ae7-b42e-4d9c-b997-b1463aaa0ded # !\x02\x02\x00\x01\x00\x00\x00\x00\x00\x00\x00 # Binaryversion : 0x022f # FW_STORE_TYPE_LOCAL # FW_POLICY_ACCESS_RIGHT_READ # Flags = 0 resp1 = client.call(iid, Proc0_RPC_FWOpenPolicyStore, params=b"!\x02\x02\x00\x01\x00\x00\x00\x00\x00\x00\x00") rawpolstore = resp1[:20] assert not client.last_response_was_view # Proc9_RPC_FWEnumFirewallRules # \x00\x00\x03\x00\xff\xff\xff\x7f\x07\x00 # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-fasp/36cddff4-c427-4863-a58d-3d913a12b221 # FW_PROFILE_TYPE_ALL : 0x7FFFFFFF # FW_RULE_STATUS_CLASS_OK + FW_RULE_STATUS_PARTIALLY_IGNORED = 0x00010000 + 0x00020000 # Flags = 7 ? resp2 = client.call(iid, Proc9_RPC_FWEnumFirewallRules, params=rawpolstore + b"\x00\x00\x03\x00\xff\xff\xff\x7f\x07\x00") assert client.last_response_was_view # This cannot work as is: as juste calling NtAlpcDeleteSectionView does not works. # We need to disconnect from the port or (guess) reuse the message (based on ID ?) so that the serv known we are done with the message and can suppress it. # More alpc client/serer test need to be done # def test_rpc_response_as_view_is_unmapped(): # """Test that when a RCP client response as a view, it is unmapped and does not take place in memory ater the call # It's not private memory, but it take some place in the AS : which can be problematic in 32b python""" # # The logic we use : make a LOT of call to an API known to send response as a view. And monitor the evolution of WorkingSetSize # client = windows.rpc.find_alpc_endpoint_and_connect(FIREWALL_RPC_IID, sid=gdef.WinLocalSid) # client.__class__ = DbgRpcClient # iid = client.bind(FIREWALL_RPC_IID) # # # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-fasp/230d1ae7-b42e-4d9c-b997-b1463aaa0ded # # !\x02\x02\x00\x01\x00\x00\x00\x00\x00\x00\x00 # # Binaryversion : 0x022f # # FW_STORE_TYPE_LOCAL # # FW_POLICY_ACCESS_RIGHT_READ # # Flags = 0 # resp1 = client.call(iid, Proc0_RPC_FWOpenPolicyStore, params=b"!\x02\x02\x00\x01\x00\x00\x00\x00\x00\x00\x00") # rawpolstore = resp1[:20] # assert not client.last_response_was_view # initial_ws_size = windows.current_process.memory_info.WorkingSetSize # ws_sizes = [] # # for i in range(100): # # Proc9_RPC_FWEnumFirewallRules # # \x00\x00\x03\x00\xff\xff\xff\x7f\x07\x00 # # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-fasp/36cddff4-c427-4863-a58d-3d913a12b221 # # FW_PROFILE_TYPE_ALL : 0x7FFFFFFF # # FW_RULE_STATUS_CLASS_OK + FW_RULE_STATUS_PARTIALLY_IGNORED = 0x00010000 + 0x00020000 # # Flags = 7 ? # resp2 = client.call(iid, Proc9_RPC_FWEnumFirewallRules, params=rawpolstore + b"\x00\x00\x03\x00\xff\xff\xff\x7f\x07\x00") # assert client.last_response_was_view # ws_sizes.append(windows.current_process.memory_info.WorkingSetSize - initial_ws_size) # print("WorkingSetSize: {0}".format(windows.current_process.memory_info.WorkingSetSize)) # assert max(ws_sizes) - min(ws_sizes) < (1024 ** 2) # We should not vary more than 1 MO ? # # Check if not all ws_size is not 4k bigger that the last one # assert not all((ws_sizes[i+1] - ws_sizes[i] > 0x1000) for i in range(len(ws_sizes) - 1))