diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..82d3e5f --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,167 @@ +import datetime + +import pytest +from pfwtest import * + +import windows + + +testbasekeypath = r"HKEY_CURRENT_USER\SOFTWARE\PythonForWindows\Test" +basekeytest = windows.system.registry(testbasekeypath, gdef.KEY_WOW64_64KEY | gdef.KEY_READ | gdef.KEY_WRITE) + +if not basekeytest.exists: + basekeytest.create() + +@pytest.fixture() +def empty_test_base_key(): + assert basekeytest.exists + for subkey in basekeytest.subkeys: + subkey.delete() + # Use of lowlevel_value_enum allow deleting value with NULL bytes + for value in basekeytest.values: + del basekeytest[value.name] + assert not basekeytest.subkeys + assert not basekeytest.values + +# Clean registry before everytest +pytestmark = pytest.mark.usefixtures("empty_test_base_key") + + +@pytest.mark.parametrize("value", [1, "LOL", 0x11223344, ""]) +def test_registry_set_get_simple_values(value): + basekeytest["tst1"] = value + assert basekeytest["tst1"].value == value + +# TODO: test with other registry type (the stranges ones) +@pytest.mark.parametrize("value, type", [ + (0x11223344, gdef.REG_DWORD), + (0x1122334455667788, gdef.REG_QWORD), + ("", gdef.REG_SZ), + (["AAAA", "BBBB", "CCCC"], gdef.REG_MULTI_SZ), + ("123\x00123" + "".join(chr(c) for c in range(256)), gdef.REG_BINARY), +]) +def test_registry_set_get_simple_values_with_types(value, type): + basekeytest["tst2"] = (value, type) + assert basekeytest["tst2"].value == value + +UNICODE_PATH_NAME = u'\u4e2d\u56fd\u94f6\u884c\u7f51\u94f6\u52a9\u624b' +UNICODE_RU_STRING = u"\u0441\u0443\u043a\u0430\u0020\u0431\u043b\u044f\u0442\u044c" # CYKA BLYAT in Cyrillic + +# Could be done in test_registry_set_get_simple_values +# But was the cause a special bug / reimplem due to _winreg using ANSI functions +# So create a special test with a very identifiable name / bug cause + +@pytest.mark.parametrize("unistr", [UNICODE_PATH_NAME, UNICODE_RU_STRING, u""]) +def test_registry_unicode_string_value(unistr): + basekeytest["tst3"] = unistr + assert basekeytest["tst3"].value == unistr + +def test_registry_unicode_multi_string(): + TST_MULTI = [UNICODE_PATH_NAME, "Hello World", UNICODE_RU_STRING] + basekeytest["tst4"] = (TST_MULTI, gdef.REG_MULTI_SZ) + assert basekeytest["tst4"].value == TST_MULTI + + +@pytest.mark.parametrize("unistr", [UNICODE_PATH_NAME, UNICODE_RU_STRING]) +def test_registry_unicode_value_name(unistr): + basekeytest[unistr] = 42 + assert basekeytest[unistr].value == 42 + # assert unistr in [v.name for v in basekeytest.values] + del basekeytest[unistr] + +def test_registry_subkeys_create_delete(): + subname = "MyTestSubKey" + subkey = basekeytest(subname) + assert not subkey.exists + subkey.create() + assert subkey.exists + subkey.delete() + assert not subkey.exists + +def test_registry_get_key_info(): + subname = "MyTestSubKeySizeInfo" + subkey = basekeytest(subname).create() + subkey["A"] = "12345" + subkey["AAAA"] = "1" + max_name_size, max_value_size = subkey.get_key_size_info() + assert max_name_size == 4 # AAAA + assert max_value_size == 6 * 2 # 12345\x00 -> 2 BYTE per char (utf-16) + other_info = subkey.info + assert other_info[0] == 0 # Nb subkeys + assert other_info[1] == 2 # Nb values + assert isinstance(other_info[2], (int, long)) # Last write + + +def test_registry_unicode_value_name_enumerate(): + name1 = u"enum_" + UNICODE_PATH_NAME + name2 = u"enum_" + UNICODE_RU_STRING + basekeytest[name1] = 1 + basekeytest[name2] = 2 + values_names = [v.name for v in basekeytest.values] + assert name1 in values_names + assert name2 in values_names + + +class CustomCountForRegistryTest(object): + TESTKEY = None + + def __init__(self): + self.value = 0 + + + def __iter__(self): + while True: + print("NEXT") + if self.value == 1: + print("ADDING HARDCODE KEY !") + self.TESTKEY[BIG_KEY_NAME] = BIG_KEY_VALUE + yield self.value + self.value += 1 + +BIG_KEY_NAME = "BIG" * 50 +BIG_KEY_VALUE = "BIG" * 0x2000 + +def test_registry_unicode_value_name_enumerate_with_race_condition(monkeypatch): + import itertools + # With itertools.count() to add a big key in the middle of the enumeration + # With a bigger name & data that the key currently existing + + # Create a new subkey so that the KeyInfos are "reset" + subkeyname = str(datetime.datetime.now()) + assert not basekeytest(subkeyname).exists + subkey = basekeytest(subkeyname).create() + try: + CustomCountForRegistryTest.TESTKEY = subkey + monkeypatch.setattr(itertools, "count", CustomCountForRegistryTest) + name1 = u"enum_" + UNICODE_PATH_NAME + name2 = u"enum_" + UNICODE_RU_STRING + subkey[name1] = 1 + subkey[name2] = 2 + values_names = [v.name for v in subkey.values] + assert name1 in values_names + assert name2 in values_names + assert BIG_KEY_NAME in values_names + finally: + subkey.delete() + +def test_registry_unicode_subkeys_create_delete(): + subname = UNICODE_RU_STRING + unicode(datetime.datetime.now()) + subkey = basekeytest(subname) + assert not subkey.exists + subkey.create() + assert subkey.exists + subkey.delete() + assert not subkey.exists + + +def test_registry_unicode_subkeys_enumerate(): + name1 = u"subkey" + UNICODE_PATH_NAME + name2 = u"subkey" + UNICODE_RU_STRING + basekeytest(name1).create() + basekeytest(name2).create() + subkey_names = [sk.name for sk in basekeytest.subkeys] + assert name1 in subkey_names + assert name2 in subkey_names + + + diff --git a/windows/winobject/registry.py b/windows/winobject/registry.py index e6cbb57..9423a34 100644 --- a/windows/winobject/registry.py +++ b/windows/winobject/registry.py @@ -1,10 +1,29 @@ -import _winreg +import sys +import ctypes import itertools import struct from collections import namedtuple import windows -from windows.generated_def.windef import KEY_READ, REG_QWORD +from windows.dbgprint import dbgprint +import windows.generated_def as gdef +from windows import winproxy + +WENCODING = "utf-16-le" + +# So _winreg does not handle unicode stuff in Py2 :( +# Need to rewrite some stuff manually :( +import _winreg + +class WinRegistryKey(gdef.HKEY): + _close_function = staticmethod(winproxy.RegCloseKey) + + def __del__(self): + if sys.path is None: # Late shutdown (not sur winproxy is still up + return + if self: # Not NULL handle ? + dbgprint("Closing registry key handle {0:#x}".format(self.value), 'REGISTRY') + self._close_function(self) @@ -16,7 +35,75 @@ class ExpectWindowsError(object): pass def __exit__(self, etype, e, tb): - return (etype == WindowsError and e.winerror == self.errornumber) + return (etype in (winproxy.WinproxyError, WindowsError) and e.winerror == self.errornumber) + +# Translation reg-buffer <-> python methodes +def Reg2Py_QWORD(buffer, size): + return buffer.cast(gdef.PULONG64)[0] + +def Py2Reg_QWORD(obj): + return struct.pack("'.format(self.fullname) + def _open_key(self, handle, name, sam): + result = WinRegistryKey() + winproxy.RegOpenKeyExW(handle, name, 0, sam, result) + return result + + def _create_key(self, parent, name, sam): + result = WinRegistryKey() + flags = 0 + winproxy.RegCreateKeyExW(parent, name, 0, None, flags, sam, None, result, None) + return result + @property def phkey(self): if self._phkey is not None: return self._phkey try: - self._phkey = _winreg.OpenKeyEx(self.surkey.phkey, self.name, 0, self.sam) + self._phkey = self._open_key(self.surkey.phkey, self.name, self.sam) except WindowsError as e: raise WindowsError(e.winerror, "Could not open registry key <{0}> ({1})".format(self.fullname, e.strerror)) return self._phkey @property def exists(self): - # Best way todo ? - # TODO: document - if self._phkey is not None: + # May have been deleted in between + # So tells use nothing + if self._phkey: # Not None + pointer not NULL + try: + self.get_key_size_info() + except WindowsError as e: + if e.winerror == gdef.ERROR_KEY_DELETED: + return False + raise return True try: - tmpphkey = _winreg.OpenKeyEx(self.surkey.phkey, self.name) + tmpphkey = self._open_key(self.surkey.phkey, self.name, gdef.KEY_READ) except WindowsError as e: return False - _winreg.CloseKey(tmpphkey) + winproxy.RegCloseKey(tmpphkey) return True @property @@ -66,8 +170,13 @@ class PyHKey(object): :type: [:class:`PyHKey`] - A list of keys""" res = [] with ExpectWindowsError(259): + default_name_size = 256 + 1 + name_size = gdef.DWORD(default_name_size) + name_buffer = ctypes.create_unicode_buffer(name_size.value) for i in itertools.count(): - res.append(_winreg.EnumKey(self.phkey, i)) + name_size.value = default_name_size + winproxy.RegEnumKeyExW(self.phkey, i, name_buffer, name_size, None, None, None, None) + res.append(name_buffer.value) return [PyHKey(self, n) for n in res] @property @@ -78,10 +187,11 @@ class PyHKey(object): res = [] with ExpectWindowsError(259): for i in itertools.count(): + # name_value_type = _winreg.EnumValue(self.phkey, i) # _winreg doest not support REG_QWORD in python2 # See http://bugs.python.org/issue23026 - if name_value_type[2] == REG_QWORD: + if name_value_type[2] == gdef.REG_QWORD: name = name_value_type[0] value = struct.unpack(" ({1})".format(self.fullname, e.strerror)) return self @@ -158,7 +335,7 @@ class PyHKey(object): def delete(self): """Delete the registry key""" try: - _winreg.DeleteKeyEx(self.surkey.phkey, self.name, self.sam, 0) + windows.winproxy.RegDeleteKeyExW(self.surkey.phkey, self.name, self.sam, 0) except WindowsError as e: raise WindowsError(e.winerror, "Could not delete registry key <{0}> ({1})".format(self.fullname, e.strerror)) return None @@ -204,7 +381,7 @@ class Registry(object): "HKEY_USERS" : HKEY_USERS } - def __init__(self, sam=KEY_READ): + def __init__(self, sam=gdef.KEY_READ): self.sam = sam @classmethod diff --git a/windows/winproxy/apis/advapi32.py b/windows/winproxy/apis/advapi32.py index 4e949ad..314e64d 100644 --- a/windows/winproxy/apis/advapi32.py +++ b/windows/winproxy/apis/advapi32.py @@ -167,19 +167,19 @@ def GetNamedSecurityInfoA(pObjectName, ObjectType, SecurityInfo, ppsidOwner=None def GetNamedSecurityInfoW(pObjectName, ObjectType, SecurityInfo, ppsidOwner=None, ppsidGroup=None, ppDacl=None, ppSacl=None, ppSecurityDescriptor=None): return GetNamedSecurityInfoW.ctypes_function(pObjectName, ObjectType, SecurityInfo, ppsidOwner, ppsidGroup, ppDacl, ppSacl, ppSecurityDescriptor) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def GetSecurityInfo(handle, ObjectType, SecurityInfo, ppsidOwner=None, ppsidGroup=None, ppDacl=None, ppSacl=None, ppSecurityDescriptor=None): return GetSecurityInfo.ctypes_function(handle, ObjectType, SecurityInfo, ppsidOwner, ppsidGroup, ppDacl, ppSacl, ppSecurityDescriptor) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def SetSecurityInfo(handle, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl): return SetSecurityInfo.ctypes_function(handle, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def SetNamedSecurityInfoA(pObjectName, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl): return SetNamedSecurityInfoA.ctypes_function(pObjectName, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def SetNamedSecurityInfoW(pObjectName, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl): return SetNamedSecurityInfoW.ctypes_function(pObjectName, ObjectType, SecurityInfo, psidOwner, psidGroup, pDacl, pSacl) @@ -240,34 +240,111 @@ def GetAce(pAcl, dwAceIndex, pAce): # Registry -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegOpenKeyExA(hKey, lpSubKey, ulOptions, samDesired, phkResult): return RegOpenKeyExA.ctypes_function(hKey, lpSubKey, ulOptions, samDesired, phkResult) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegOpenKeyExW(hKey, lpSubKey, ulOptions, samDesired, phkResult): return RegOpenKeyExW.ctypes_function(hKey, lpSubKey, ulOptions, samDesired, phkResult) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) +def RegCreateKeyExA(hKey, lpSubKey, Reserved, lpClass, dwOptions, samDesired, lpSecurityAttributes, phkResult, lpdwDisposition): + return RegCreateKeyExA.ctypes_function(hKey, lpSubKey, Reserved, lpClass, dwOptions, samDesired, lpSecurityAttributes, phkResult, lpdwDisposition) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegCreateKeyExW(hKey, lpSubKey, Reserved, lpClass, dwOptions, samDesired, lpSecurityAttributes, phkResult, lpdwDisposition): + return RegCreateKeyExW.ctypes_function(hKey, lpSubKey, Reserved, lpClass, dwOptions, samDesired, lpSecurityAttributes, phkResult, lpdwDisposition) + +@Advapi32Proxy(error_check=result_is_error_code) def RegGetValueA(hkey, lpSubKey, lpValue, dwFlags, pdwType, pvData, pcbData): return RegGetValueA.ctypes_function(hkey, lpSubKey, lpValue, dwFlags, pdwType, pvData, pcbData) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegGetValueW(hkey, lpSubKey=None, lpValue=NeededParameter, dwFlags=0, pdwType=None, pvData=None, pcbData=None): return RegGetValueW.ctypes_function(hkey, lpSubKey, lpValue, dwFlags, pdwType, pvData, pcbData) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegQueryValueExA(hKey, lpValueName, lpReserved, lpType, lpData, lpcbData): return RegQueryValueExA.ctypes_function(hKey, lpValueName, lpReserved, lpType, lpData, lpcbData) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegQueryValueExW(hKey, lpValueName, lpReserved, lpType, lpData, lpcbData): - return RegQueryValueExA.ctypes_function(hKey, lpValueName, lpReserved, lpType, lpData, lpcbData) + return RegQueryValueExW.ctypes_function(hKey, lpValueName, lpReserved, lpType, lpData, lpcbData) -@Advapi32Proxy(error_check=succeed_on_zero) +@Advapi32Proxy(error_check=result_is_error_code) def RegCloseKey(hKey): return RegCloseKey.ctypes_function(hKey) +@Advapi32Proxy(error_check=result_is_error_code) +def RegSetValueExW(hKey, lpValueName, Reserved, dwType, lpData, cbData): + return RegSetValueExW.ctypes_function(hKey, lpValueName, Reserved, dwType, lpData, cbData) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegSetValueExA(hKey, lpValueName, Reserved, dwType, lpData, cbData): + return RegSetValueExA.ctypes_function(hKey, lpValueName, Reserved, dwType, lpData, cbData) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegSetKeyValueA(hKey, lpSubKey, lpValueName, dwType, lpData, cbData): + return RegSetKeyValueA.ctypes_function(hKey, lpSubKey, lpValueName, dwType, lpData, cbData) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegSetKeyValueW(hKey, lpSubKey, lpValueName, dwType, lpData, cbData): + return RegSetKeyValueW.ctypes_function(hKey, lpSubKey, lpValueName, dwType, lpData, cbData) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegEnumKeyExA(hKey, dwIndex, lpName, lpcchName, lpReserved, lpClass, lpcchClass, lpftLastWriteTime): + return RegEnumKeyExA.ctypes_function(hKey, dwIndex, lpName, lpcchName, lpReserved, lpClass, lpcchClass, lpftLastWriteTime) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegEnumKeyExW(hKey, dwIndex, lpName, lpcchName, lpReserved, lpClass, lpcchClass, lpftLastWriteTime): + return RegEnumKeyExW.ctypes_function(hKey, dwIndex, lpName, lpcchName, lpReserved, lpClass, lpcchClass, lpftLastWriteTime) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegGetKeySecurity(hKey, SecurityInformation, pSecurityDescriptor, lpcbSecurityDescriptor): + return RegGetKeySecurity.ctypes_function(hKey, SecurityInformation, pSecurityDescriptor, lpcbSecurityDescriptor) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegQueryInfoKeyA(hKey, lpClass, lpcchClass, lpReserved, lpcSubKeys, lpcbMaxSubKeyLen, lpcbMaxClassLen, lpcValues, lpcbMaxValueNameLen, lpcbMaxValueLen, lpcbSecurityDescriptor, lpftLastWriteTime): + return RegQueryInfoKeyA.ctypes_function(hKey, lpClass, lpcchClass, lpReserved, lpcSubKeys, lpcbMaxSubKeyLen, lpcbMaxClassLen, lpcValues, lpcbMaxValueNameLen, lpcbMaxValueLen, lpcbSecurityDescriptor, lpftLastWriteTime) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegQueryInfoKeyW(hKey, lpClass, lpcchClass, lpReserved, lpcSubKeys, lpcbMaxSubKeyLen, lpcbMaxClassLen, lpcValues, lpcbMaxValueNameLen, lpcbMaxValueLen, lpcbSecurityDescriptor, lpftLastWriteTime): + return RegQueryInfoKeyW.ctypes_function(hKey, lpClass, lpcchClass, lpReserved, lpcSubKeys, lpcbMaxSubKeyLen, lpcbMaxClassLen, lpcValues, lpcbMaxValueNameLen, lpcbMaxValueLen, lpcbSecurityDescriptor, lpftLastWriteTime) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteKeyValueW(hKey, lpSubKey, lpValueName): + return RegDeleteKeyValueW.ctypes_function(hKey, lpSubKey, lpValueName) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteKeyValueA(hKey, lpSubKey, lpValueName): + return RegDeleteKeyValueA.ctypes_function(hKey, lpSubKey, lpValueName) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteKeyExA(hKey, lpSubKey, samDesired, Reserved): + return RegDeleteKeyExA.ctypes_function(hKey, lpSubKey, samDesired, Reserved) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteKeyExW(hKey, lpSubKey, samDesired, Reserved): + return RegDeleteKeyExW.ctypes_function(hKey, lpSubKey, samDesired, Reserved) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteValueA(hKey, lpValueName): + return RegDeleteValueA.ctypes_function(hKey, lpValueName) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegDeleteValueW(hKey, lpValueName): + return RegDeleteValueW.ctypes_function(hKey, lpValueName) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegEnumValueA(hKey, dwIndex, lpValueName, lpcchValueName, lpReserved, lpType, lpData, lpcbData): + return RegEnumValueA.ctypes_function(hKey, dwIndex, lpValueName, lpcchValueName, lpReserved, lpType, lpData, lpcbData) + +@Advapi32Proxy(error_check=result_is_error_code) +def RegEnumValueW(hKey, dwIndex, lpValueName, lpcchValueName, lpReserved, lpType, lpData, lpcbData): + return RegEnumValueW.ctypes_function(hKey, dwIndex, lpValueName, lpcchValueName, lpReserved, lpType, lpData, lpcbData) + + # Service @Advapi32Proxy()