mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
hakril
198 lines
7.6 KiB
Python
198 lines
7.6 KiB
Python
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)) |