Fixed broken basestring for py3 compatibility (see issue #15)

tests/test_crypto.py

@@ -222,3 +222,19 @@ def test_sign_verify_fail(rawcert, rawpfx):
     assert excinfo.value.winerror == gdef.STATUS_INVALID_SIGNATURE
 
 
+# str(windows.crypto.encrypt(TEST_CERT, "Hello crypto")).encode("base64")
+# Target serial == TEST_CERT.Serial == 1b 8e 94 cb 0b 3e eb b6 41 39 f3 c9 09 b1 6b 46
+TEST_CRYPTMSG = b"""MIIBJAYJKoZIhvcNAQcDoIIBFTCCARECAQAxgc0wgcoCAQAwMzAfMR0wGwYDVQQDExRQeXRob25G
+b3JXaW5kb3dzVGVzdAIQG46Uyws+67ZBOfPJCbFrRjANBgkqhkiG9w0BAQcwAASBgA1fwFY8w4Bb
+fOMer94JhazbJxaUnV305QzF27w4GwNQ2UIpl9KWJoJJaF7azU3nVhP33agAxlxmr9fP48B6DeE1
+pbu1jX9tEWlTJC6O0TmKcRPjblEaU6VJXXlpKlKZCmwCUuHR9VtcXGnxEU1Hy7FmHM96lvDRmYQT
+Y0MnRJLyMDwGCSqGSIb3DQEHATAdBglghkgBZQMEASoEEEdEGEzKBrDO/zC8z6q6HLaAEGbjGCay
+s6u32YhUxQ4/QhI="""
+
+def test_cryptmsg_from_data():
+    rawdata = b64decode(TEST_CRYPTMSG)
+    cryptmsg = windows.crypto.CryptMessage.from_buffer(rawdata)
+    rawtarget = b"\x1b\x8e\x94\xcb\x0b>\xeb\xb6A9\xf3\xc9\t\xb1kF"
+    assert cryptmsg.get_recipient_data(0).SerialNumber.data[::-1] == rawtarget
+
+

tests/test_event_trace.py

@@ -0,0 +1,34 @@
+import collections
+import os
+
+import windows
+import windows.generated_def as gdef
+
+
+TASK_SCHEDULER_PROVIDER = "047311A9-FA52-4A68-A1E4-4E289FBB8D17"
+
+EVENT_ID_COUNT = collections.Counter()
+
+def show(event):
+    EVENT_ID_COUNT.update([event.id])
+
+
+def test_etw_trace_open_with_guid():
+    trace = windows.system.etw.open_trace("PFW_test_etw_1", guid="42424242-4242-4242-4242-000000001234")
+
+def test_etw_trace_registration_and_processing():
+    EVENT_ID_COUNT.clear()
+    # RealTime Test
+    trace = windows.system.etw.open_trace("PFW_test_etw_2", logfile="pfw_test_trace.etl")
+    trace.start()
+    trace.enable(TASK_SCHEDULER_PROVIDER, 0xff, 0xff)
+    # Scheduler code that generate event
+    windows.system.task_scheduler(r"\Microsoft\Windows\Chkdsk")["SyspartRepair"]
+    # End of scheduler code
+    trace.stop()
+    trace.process(show)
+    # Task scheduler generate event id 10,11,12
+    assert EVENT_ID_COUNT[10] >= 1
+    assert EVENT_ID_COUNT[11] >= 1
+    assert EVENT_ID_COUNT[12] >= 1
+    os.unlink("pfw_test_trace.etl")

tests/test_evtlog.py

@@ -127,3 +127,12 @@ def test_evtlog_query_seek():
     events = query.all()
     assert len(events) == 1
     assert events[0].data["TaskName"] == taskpath
+
+
+@pytest.mark.parametrize("value", [
+    12,
+    "HELLO WORLD",
+    b"BYTES HELLO WORLD",
+])
+def test_evtlog_improved_variant_from_value(value):
+    assert evtl.ImprovedEVT_VARIANT.from_value(value)

tests/test_improved_buffer.py

@@ -36,3 +36,8 @@ def test_partial_buffer_size_guess(c_type, buffer, expected_size):
     assert len(buf) == expected_size
 
 
+def test_partial_buffer_string_call():
+    buffer = windows.utils.BUFFER(gdef.WCHAR)("LOL")
+    assert buffer[:] == "LOL"
+    assert len(buffer) == 3
+

tests/test_midl.py

@@ -0,0 +1,11 @@
+import windows.generated_def as gdef
+
+def test_format_charactere_values():
+    assert gdef.FC_ZERO == 0
+    assert gdef.FC_PAD == 0x5c
+    assert gdef.FC_PAD == 0x5c
+    assert gdef.FC_SPLIT_DEREFERENCE == 0x74
+    assert gdef. FC_SPLIT_DIV_2 == 0x75
+    assert gdef.FC_HARD_STRUCT  == 0xb1
+    assert gdef.FC_TRANSMIT_AS_PTR   == 0xb2
+    assert gdef.FC_END_OF_UNIVERSE   == 0xba

tests/test_process.py

@@ -443,3 +443,14 @@ class TestProcessWithCheckGarbage(object):
         proc32.execute_python("assert 1 == 1")
         with pytest.raises(windows.injection.RemotePythonError):
             proc32.execute_python("assert 1 == 2")
+
+
+    def test_process_set_security_descriptor(self, proc32_64):
+        current_user_sid = str(windows.current_process.token.user)
+        # Same Owner/Group -> ALL acces to all
+        SSDL_GR_EVERYONE = "O:{user}G:{user}D:(A;;0x1fffff;;;WD)".format(user=current_user_sid)
+        SD_GR_EVERYONE = windows.security.SecurityDescriptor.from_string(SSDL_GR_EVERYONE)
+        # Via string
+        proc32_64.security_descriptor = SSDL_GR_EVERYONE
+        # Via SD obj
+        proc32_64.security_descriptor = SD_GR_EVERYONE

tests/test_remotectypes.py

@@ -0,0 +1,47 @@
+import ctypes
+import windows
+import windows.generated_def as gdef
+import windows.remotectypes as rctypes
+
+
+def test_remote_struct_same_bitness():
+    target = windows.current_process
+    struct = gdef.OSVERSIONINFOEXA()
+    struct.dwMajorVersion = 42 # DWORD
+    struct.dwMinorVersion = 43 # DWORD
+    struct.dwPlatformId = 0x11223344 # DWORD
+    struct.szCSDVersion = b"LOL" # CHAR * (128)
+    struct.wProductType = 0x21 # Byte
+
+    # Create a remote-ctypes-struct that use our process as target
+    # Logic will be the same
+    remtype = rctypes.transform_type_to_remote(gdef.OSVERSIONINFOEXA)
+    remstruct = remtype(ctypes.addressof(struct), target)
+
+    assert struct.dwMajorVersion == remstruct.dwMajorVersion
+    assert struct.dwMinorVersion == remstruct.dwMinorVersion
+    assert struct.dwPlatformId == remstruct.dwPlatformId
+    assert struct.szCSDVersion == remstruct.szCSDVersion
+    assert struct.wProductType == remstruct.wProductType
+
+# This test fails for now.
+# Should I improve remote ctypes to handel this ?
+def test_remote_long_ptr():
+    # Bug thatwas in retrieving of NtCreateFile arguments
+    target = windows.current_process
+
+    large_int = gdef.LARGE_INTEGER(0x1122334455667788)
+    large_int_ptr = gdef.PLARGE_INTEGER(large_int)
+    assert large_int_ptr[0] == 0x1122334455667788
+
+    # A remote large_int POINTER
+    remtype = rctypes.transform_type_to_remote(gdef.PLARGE_INTEGER)
+    remstruct = remtype(ctypes.addressof(large_int_ptr), target)
+
+    assert remstruct.value == ctypes.addressof(large_int_ptr)
+    assert remstruct.contents == 0x1122334455667788
+
+
+    import pdb;pdb.set_trace()
+    print("LOL")
+

tests/test_security.py

@@ -21,6 +21,16 @@ TEST_SDDL = [
 def test_security_descriptor_from_string(sddl):
     sd = SecurityDescriptor.from_string(sddl)
 
+TEST_BIN_SDDL = [
+# TapiSrv security descriptor
+b'\x01\x00\x14\x80\x8c\x00\x00\x00\x98\x00\x00\x00\x14\x00\x00\x000\x00\x00\x00\x02\x00\x1c\x00\x01\x00\x00\x00\x02\x80\x14\x00\xff\x01\x0f\x00\x01\x01\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x02\x00\\\x00\x04\x00\x00\x00\x00\x00\x14\x00\xfd\x01\x02\x00\x01\x01\x00\x00\x00\x00\x00\x05\x12\x00\x00\x00\x00\x00\x18\x00\xff\x01\x0f\x00\x01\x02\x00\x00\x00\x00\x00\x05 \x00\x00\x00 \x02\x00\x00\x00\x00\x14\x00\x9d\x01\x02\x00\x01\x01\x00\x00\x00\x00\x00\x05\x04\x00\x00\x00\x00\x00\x14\x00\x9d\x01\x02\x00\x01\x01\x00\x00\x00\x00\x00\x05\x06\x00\x00\x00\x01\x01\x00\x00\x00\x00\x00\x05\x12\x00\x00\x00\x01\x01\x00\x00\x00\x00\x00\x05\x12\x00\x00\x00',
+]
+
+@pytest.mark.parametrize("binsd", TEST_BIN_SDDL)
+def test_security_descriptor_from_binary(binsd):
+    sd = SecurityDescriptor.from_binary(binsd)
+
+
 def test_empty_security_descriptor():
     esd = SecurityDescriptor.from_string("")
     assert esd.owner is None # Should NOT be NULL PSID but None
@@ -34,7 +44,6 @@ def test_security_descriptor__str__():
     assert str(sd) == sddl
 
 
-
 def test_pacl_object():
     SDDL = "O:ANG:S-1-2-3D:(A;;;;;S-1-42-42)(A;;;;;S-1-42-43)(A;;;;;S-1-42-44)"
     dacl = SecurityDescriptor.from_string(SDDL).dacl
@@ -77,6 +86,7 @@ def test_mask_sid_ace():
     assert ace.sid.to_string() == "S-1-42-42"
 
 
+
 SGUID = gdef.GUID.from_string
 
 COMPLEXE_SDDL_GUID = [

tests/test_service.py

@@ -12,10 +12,16 @@ def test_services_process():
 
 
 def test_service_appinfo():
-    appinfos = [x for x in windows.system.services if x.name == b"Appinfo"]
-    assert len(appinfos) == 1
-    appinfo = appinfos[0]
+    appinfo = windows.system.services[b"Appinfo"]
     assert appinfo.status.type & gdef.SERVICE_WIN32_OWN_PROCESS
     # Check other fields
     assert appinfo.name == b"Appinfo"
-    assert appinfo.description == b"Application Information"
+    assert appinfo.description == b"Application Information"
+
+
+def test_service_start():
+    faxservice = windows.system.services[b"Fax"]
+    # Just start a random serivce with a string
+    # Used to check string compat in py2/py3
+    faxservice.start(b"TEST STRING")
+

tests/test_symbol.py

@@ -0,0 +1,21 @@
+import pytest
+
+import windows.generated_def as gdef
+import windows.debug.symbols as symbols
+from .pfwtest import *
+
+
+@pytest.fixture()
+def symctx():
+    yield symbols.VirtualSymbolHandler()
+
+# Disable defered loading
+# symbols.engine.options = gdef.SYMOPT_UNDNAME
+
+def test_symbols_loadfile(symctx):
+    mod = symctx.load_file(path=r"c:\windows\system32\ntdll.dll", addr=0x42000)
+    assert mod.addr == 0x42000
+    # Resolve by name
+    createfile = symctx[b"ntdll!NtCreateFile"]
+    # Resolve by addr
+    assert symctx[createfile.addr].name == b"NtCreateFile"

tests/test_winutils.py

@@ -123,6 +123,23 @@ def test_unix_timestamp_from_filetime():
     # Because round(0.5) == 0 (vs 1 in py2)
     assert windows.utils.unix_timestamp_from_filetime(131492395680727305) == 1504765968.072731
 
+# Test values from https://docs.microsoft.com/en-us/cpp/atl-mfc-shared/date-type?view=vs-2019
+@pytest.mark.parametrize("comtime, date", [
+    (0, datetime(1899, 12, 30)),
+    (2, datetime(1900, 1, 1)),
+    (5, datetime(1900, 1, 4)),
+    (5.25, datetime(1900, 1, 4, hour=6)),
+    (5.5, datetime(1900, 1, 4, hour=12)),
+    (5.875, datetime(1900, 1, 4, hour=21)),
+    (-0.25, datetime(1899, 12, 30, hour=6)),
+    (-0.5, datetime(1899, 12, 30, hour=12)),
+    (-2, datetime(1899, 12, 28)),
+    (-2.5, datetime(1899, 12, 28, hour=12)),
+    (-2.75, datetime(1899, 12, 28, hour=18)),
+])
+def test_datetime_from_comtime(comtime, date):
+    assert windows.utils.datetime_from_comdate(comtime) == date
+
 @pytest.mark.parametrize("prefix", [
     ("long_ascii_prefix"),
     (u'\u4e2d\u56fd\u94f6\u884c\u7f51\u94f6\u52a9\u624b'),
@@ -136,4 +153,22 @@ def test_long_short_path_str_unicode(prefix):
         assert short_name != basename
         full_name = windows.utils.get_long_path(short_name).lower()
         assert isinstance(full_name, unicode)
-        assert full_name == basename
+        assert full_name == basename
+
+TEST_CERT = b"""
+MIIBwTCCASqgAwIBAgIQG46Uyws+67ZBOfPJCbFrRjANBgkqhkiG9w0BAQsFADAfMR0wGwYDVQQD
+ExRQeXRob25Gb3JXaW5kb3dzVGVzdDAeFw0xNzA0MTIxNDM5MjNaFw0xODA0MTIyMDM5MjNaMB8x
+HTAbBgNVBAMTFFB5dGhvbkZvcldpbmRvd3NUZXN0MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKB
+gQCRHwC/sRfXh5pc4poc85aidrudbPdya+0OeonQlf1JQ1ekf7KSfADV5FLkSQu2BzgBK9DIWTGX
+XknBJIzZF03UZsVg5D67V2mnSClXucc0cGFcK4pDDt0tHeabA2GPinVe7Z6qDT4ZxPR8lKaXDdV2
+Pg2hTdcGSpqaltHxph7G/QIDAQABMA0GCSqGSIb3DQEBCwUAA4GBACcQFdOlVjYICOIyAXowQaEN
+qcLpN1iWoL9UijNhTY37+U5+ycFT8QksT3Xmh9lEIqXMh121uViy2P/3p+Ek31AN9bB+BhWIM6PQ
+gy+ApYDdSwTtWFARSrMqk7rRHUveYEfMw72yaOWDxCzcopEuADKrrYEute4CzZuXF9PbbgK6"""
+
+
+def test_sprint_certificate():
+    cert = windows.crypto.Certificate.from_buffer(b64decode(TEST_CERT))
+    # Certificate is quite a complexe Windows structure
+    # With Sub-struct / Pointer & string
+    # It was broken on py3 -> ense this test
+    windows.utils.sprint(cert)

windows/crypto/cryptmsg.py

@@ -115,14 +115,15 @@ class CryptMessage(gdef.HCRYPTMSG):
 
     def update(self, blob, final):
         # Test isinstance string ?
-        if isinstance(blob, (basestring, bytearray)):
+        if isinstance(blob, (windows.pycompat.anybuff, bytearray)):
+            blob = windows.pycompat.raw_encode(blob)
             buffer = windows.utils.BUFFER(gdef.BYTE).from_buffer_copy(blob)
             return winproxy.CryptMsgUpdate(self, buffer, len(blob), final)
         return winproxy.CryptMsgUpdate(self, blob.pbData, blob.cbData, final)
 
     # constructor
     @classmethod
-    def from_data(self, data):
+    def from_buffer(self, data):
         hmsg = winproxy.CryptMsgOpenToDecode(windows.crypto.DEFAULT_ENCODING, 0, 0, None, None, None)
         newmsg = CryptMessage(hmsg)
         newmsg.update(data, final=True)

windows/debug/breakpoints.py

@@ -4,6 +4,7 @@ import windows
 from windows.generated_def.winstructs import *
 from windows.generated_def import windef
 from windows.winobject.process import WinProcess, WinThread
+from windows.pycompat import basestring
 
 
 STANDARD_BP = "BP"
@@ -222,10 +223,15 @@ class FunctionCallBP(Breakpoint):
     """A Breakpoint that allow to trigger at the return of a function"""
     def break_on_ret(self, dbg, exception):
         """Setup a breakpoint at the return address of the function, this breakpoint will call :func:`ret_trigger`"""
-        cproc = dbg.current_process
-        return_addr = dbg.current_process.read_ptr(dbg.current_thread.context.sp)
+        return_addr = self.get_ret_addr(dbg, exception)
         dbg.add_bp(FunctionRetBP(return_addr, self), target=dbg.current_process)
 
+    def get_ret_addr(self, dbg, exception):
+        """Get the return address of the current target, only valid in the trigger() function."""
+        cproc = dbg.current_process
+        return dbg.current_process.read_ptr(dbg.current_thread.context.sp)
+
+
     def ret_trigger(self, dbg, exception):
         """Called at the return of the function if :func:`break_on_ret` was called"""
         raise NotImplementedError("ret_trigger")

windows/debug/symbols.py

@@ -7,9 +7,11 @@ from collections import namedtuple
 import windows
 import windows.generated_def as gdef
 from windows import winproxy
+from windows.pycompat import basestring
 
 DEFAULT_DBG_OPTION = gdef.SYMOPT_DEFERRED_LOADS + gdef.SYMOPT_UNDNAME
 
+
 def set_dbghelp_path(path):
     loaded_modules =  [m.name.lower() for m in windows.current_process.peb.modules]
     if os.path.isdir(path):
@@ -234,6 +236,8 @@ class SymbolHandler(object):
                 path = name
             except Exception as e:
                 pass
+        # Expect a-string
+        path = windows.pycompat.raw_encode(path)
         try:
             load_addr = winproxy.SymLoadModuleEx(self.handle, file_handle, path, name, addr, size, data, flags)
         except WindowsError as e:
@@ -294,6 +298,8 @@ class SymbolHandler(object):
         sym = buff[0]
         sym.SizeOfStruct = ctypes.sizeof(SymbolInfo)
         sym.MaxNameLen  = max_len_size
+        # Expect a-string
+        name = windows.pycompat.raw_encode(name)
         windows.winproxy.SymFromName(self.handle, name, buff)
         sym.resolver = self
         sym.displacement = 0
@@ -301,7 +307,7 @@ class SymbolHandler(object):
 
     def resolve(self, name_or_addr):
         # Only returns None if symbol is not Found ?
-        if isinstance(name_or_addr, basestring):
+        if isinstance(name_or_addr, windows.pycompat.anybuff):
             return self.symbol_from_name(name_or_addr)
         try:
             return self.symbol_and_displacement_from_address(name_or_addr)

windows/security.py

@@ -4,6 +4,7 @@ import sys
 import windows
 import windows.generated_def as gdef
 from windows import winproxy
+from windows.pycompat import basestring
 
 from windows.winobject.token import Token, KNOW_INTEGRITY_LEVEL
 

windows/utils/improved_buffer.py

@@ -2,6 +2,7 @@ import sys
 import ctypes
 import _ctypes
 import windows.generated_def as gdef
+from windows.pycompat import basestring
 
 ## TESTING Improved Buffer code ###
 ## This code is not stable and WILL CHANGE ##

windows/utils/pythonutils.py

@@ -4,8 +4,9 @@ import ctypes
 import _ctypes
 import windows.generated_def as gdef
 
-from windows.dbgprint import dbgprint
 from windows import winproxy
+from windows.dbgprint import dbgprint
+from windows.pycompat import basestring
 
 
 

windows/winobject/event_log.py

@@ -6,7 +6,7 @@ from contextlib import contextmanager
 import windows
 import windows.generated_def as gdef
 from windows import winproxy
-from windows.pycompat import int_types
+from windows.pycompat import int_types, basestring
 
 
 # Helpers
@@ -333,6 +333,10 @@ class ImprovedEVT_VARIANT(gdef.EVT_VARIANT):
                 vtype = gdef.EvtVarTypeUInt64
             elif isinstance(value, basestring):
                 vtype = gdef.EvtVarTypeString
+            elif isinstance(value, bytes):
+                # not basestring and bytes -> py3 bytes
+                vtype = gdef.EvtVarTypeBinary
+                value = windows.utils.BUFFER(gdef.BYTE).from_buffer_copy(value)
             else:
                 raise NotImplementedError("LATER")
         self = cls()

windows/winobject/event_trace.py

@@ -1,6 +1,7 @@
 import ctypes
 import windows
 import windows.generated_def as gdef
+from windows.pycompat import basestring
 
 # Renommer le fichier etw ?
 
@@ -154,7 +155,7 @@ class CtxProcess(object):
 
 class EtwTrace(object):
     def __init__(self, name, logfile=None, guid=None):
-        self.name = name
+        self.name = windows.pycompat.raw_encode(name)
         self.logfile = logfile
         if guid and isinstance(guid, basestring):
             guid = gdef.GUID.from_string(guid)
@@ -221,14 +222,14 @@ class EtwTrace(object):
         return windows.winproxy.EnableTraceEx2(self.handle, guid, EVENT_CONTROL_CODE_ENABLE_PROVIDER, level , any_keyword, all_keyword, 0, None)
 
 
-    def process(self, callback, begin=None, end=None, context = None):
+    def process(self, callback, begin=None, end=None, context=None):
         if end == "now":
             end = gdef.FILETIME()
             windows.winproxy.GetSystemTimeAsFileTime(end)
             windows.utils.sprint(end)
 
         logfile = gdef.EVENT_TRACE_LOGFILEW()
-        logfile.LoggerName = self.name
+        logfile.LoggerName = windows.pycompat.raw_decode(self.name)
         # logfile.ProcessTraceMode = gdef.PROCESS_TRACE_MODE_EVENT_RECORD | gdef.PROCESS_TRACE_MODE_RAW_TIMESTAMP
         logfile.ProcessTraceMode = gdef.PROCESS_TRACE_MODE_EVENT_RECORD
         if not self.logfile:

windows/winobject/process.py

@@ -29,7 +29,7 @@ from windows.winobject import apisetmap
 from windows.winobject import token
 from windows import security
 
-from windows.pycompat import raw_encode, raw_decode
+from windows.pycompat import raw_encode, raw_decode, basestring
 
 TimeInfo = namedtuple("TimeInfo", ["creation", "exit", "kernel", "user"])
 """Time information about a process"""

windows/winobject/service.py

@@ -8,6 +8,7 @@ from windows import utils
 import windows.generated_def as gdef
 from windows.generated_def import *
 from windows import security
+from windows.pycompat import basestring
 
 # TODO: RM :)
 ServiceStatus = namedtuple("ServiceStatus", ["type", "state", "control_accepted", "flags"])
@@ -138,7 +139,7 @@ class NewService(gdef.SC_HANDLE):
     def start(self, args=None):
         nbelt = 0
         if args is not None:
-            if isinstance(args, basestring):
+            if isinstance(args, windows.pycompat.anybuff):
                 args = [args]
             nbelt = len(args)
             args = (gdef.LPCSTR * (nbelt))(*args)

windows/winproxy/apis/ntdll.py

@@ -364,8 +364,6 @@ def NtEnumerateValueKey(KeyHandle, Index, KeyValueInformationClass, KeyValueInfo
 
 @NtdllProxy()
 def NtDeleteValueKey(KeyHandle, ValueName):
-    if isinstance(ValueName, basestring):
-        ValueName = gdef.UNICODE_STRING.from_string(ValueName)
     return NtDeleteValueKey.ctypes_function(KeyHandle, ValueName)
 
 @NtdllProxy()