From bd5976dce2913276056c63e7a4c3290817de1c26 Mon Sep 17 00:00:00 2001 From: Clement Rouault Date: Mon, 18 Sep 2017 11:11:33 +0200 Subject: [PATCH] Added alpc + rpc tests files --- tests/test_alpc.py | 84 +++++++++++++++++++++++++++++++++++++ tests/test_rpc.py | 102 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+) create mode 100644 tests/test_alpc.py create mode 100644 tests/test_rpc.py diff --git a/tests/test_alpc.py b/tests/test_alpc.py new file mode 100644 index 0000000..393e485 --- /dev/null +++ b/tests/test_alpc.py @@ -0,0 +1,84 @@ +import pytest +import threading +import time + +import windows.alpc +import windows.generated_def as gdef + +from pfwtest import * + + +def generate_client_server_test(client_function, server_function): + def generated_test(): + th = threading.Thread(target=server_function, args=()) + th.start() + time.sleep(0.5) + client_function() + th.join() + return True + return generated_test + +PORT_NAME = r"\RPC Control\PythonForWindowsTestPort" +CLIENT_MESSAGE = "Message 1\x00\xffABCD" +SERVER_MESSAGE = "Message 2-" + "".join(chr(i) for i in range(256)) + +def alpc_simple_test_server(): + server = windows.alpc.AlpcServer(PORT_NAME) + msg = server.recv() + assert msg.type & 0xfff == gdef.LPC_CONNECTION_REQUEST + server.accept_connection(msg) + msg = server.recv() + assert msg.type & 0xfff == gdef.LPC_REQUEST + assert msg.data == CLIENT_MESSAGE + msg.data = SERVER_MESSAGE + server.send(msg) + +def alpc_simple_test_client(): + client = windows.alpc.AlpcClient(PORT_NAME) + response = client.send_receive(CLIENT_MESSAGE) + assert response.data == SERVER_MESSAGE + +test_simple_alpc = generate_client_server_test(alpc_simple_test_client, alpc_simple_test_server) + + +def send_message_with_view(client, message_data, view_data): + # Create View + section = client.create_port_section(0, 0, 0x4000) + view = client.map_section(section[0], 0x4000) + + # New message with a View + msg = windows.alpc.AlpcMessage(0x2000) + msg.attributes.ValidAttributes |= gdef.ALPC_MESSAGE_VIEW_ATTRIBUTE + msg.view_attribute.Flags = 0 + msg.view_attribute.ViewBase = view.ViewBase + msg.view_attribute.SectionHandle = view.SectionHandle + msg.view_attribute.ViewSize = 0x4000 + msg.data = message_data + windows.current_process.write_memory(view.ViewBase, view_data) + return client.send_receive(msg) + + +CLIENT_VIEW_MESSAGE = "Message 1\x00\xffABCD" +CLIENT_VIEW_DATA = "Message Data-view" + "".join(chr(i) for i in range(256)) + + +def alpc_view_test_server(): + server = windows.alpc.AlpcServer(PORT_NAME) + msg = server.recv() + assert msg.type & 0xfff == gdef.LPC_CONNECTION_REQUEST + server.accept_connection(msg) + msg = server.recv() + assert msg.type & 0xfff == gdef.LPC_REQUEST + assert msg.view_is_valid + view_data = windows.current_process.read_memory(msg.view_attribute.ViewBase, len(CLIENT_VIEW_DATA)) + assert view_data == CLIENT_VIEW_DATA + assert msg.data == CLIENT_VIEW_MESSAGE + msg.data = SERVER_MESSAGE + server.send(msg) + +def alpc_view_test_client(): + client = windows.alpc.AlpcClient(PORT_NAME) + response = send_message_with_view(client, CLIENT_VIEW_MESSAGE, CLIENT_VIEW_DATA) + assert response.data == SERVER_MESSAGE + +test_view_alpc = generate_client_server_test(alpc_view_test_client, alpc_view_test_server) diff --git a/tests/test_rpc.py b/tests/test_rpc.py new file mode 100644 index 0000000..6ecda3a --- /dev/null +++ b/tests/test_rpc.py @@ -0,0 +1,102 @@ +import pytest +import os.path + +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 test_rpc_epmapper(): + 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(): + client = windows.rpc.find_alpc_endpoint_and_connect(UAC_UIID) + iid = client.bind(UAC_UIID) + + python_path = 'C:\\Python27\\python.exe' + 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 + proc.exit(0) +