From 1c21b77292cf4b0123da700d0586702f293011bb Mon Sep 17 00:00:00 2001 From: hakril Date: Fri, 10 Aug 2018 10:44:44 +0200 Subject: [PATCH] Add sign/verify methods to windows.crypto + POC/Play with improved ctypes buffer/array --- tests/test_crypto.py | 25 +++++++++++ windows/crypto/__init__.py | 1 + windows/crypto/encrypt_decrypt.py | 3 +- windows/crypto/sign_verify.py | 63 ++++++++++++++++++++++++++ windows/utils/pythonutils.py | 75 ++++++++++++++++++++++++------- windows/winproxy.py | 43 +++++++++++++++--- 6 files changed, 187 insertions(+), 23 deletions(-) create mode 100644 windows/crypto/sign_verify.py diff --git a/tests/test_crypto.py b/tests/test_crypto.py index 63b5092..38c66b8 100644 --- a/tests/test_crypto.py +++ b/tests/test_crypto.py @@ -196,3 +196,28 @@ def test_certificate_from_store(): return windows.crypto.CertificateStore.from_system_store("Root") +def test_sign_verify(rawcert, rawpfx): + message_to_sign = "Testing message \xff\x01" + # Load PFX (priv+pub key) & certif (pubkey only) + pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD) + cert = windows.crypto.Certificate.from_buffer(rawcert) + signed_blob = windows.crypto.sign(pfx.certs[0], message_to_sign) + assert message_to_sign in signed_blob + decoded_blob = windows.crypto.verify_signature(cert, signed_blob) + assert decoded_blob == message_to_sign + + +def test_sign_verify_fail(rawcert, rawpfx): + message_to_sign = "Testing message \xff\x01" + # Load PFX (priv+pub key) & certif (pubkey only) + pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD) + cert = windows.crypto.Certificate.from_buffer(rawcert) + signed_blob = windows.crypto.sign(pfx.certs[0], message_to_sign) + assert message_to_sign in signed_blob + # Tamper the signed mesasge content + signed_blob = signed_blob.replace("message", "massage") + with pytest.raises(windows.winproxy.Kernel32Error) as excinfo: + decoded_blob = windows.crypto.verify_signature(cert, signed_blob) + assert excinfo.value.winerror == gdef.STATUS_INVALID_SIGNATURE + + diff --git a/windows/crypto/__init__.py b/windows/crypto/__init__.py index df3f40e..377b925 100644 --- a/windows/crypto/__init__.py +++ b/windows/crypto/__init__.py @@ -4,4 +4,5 @@ DEFAULT_ENCODING = X509_ASN_ENCODING | PKCS_7_ASN_ENCODING # Keep other imports here so sub-crypto file can import windows.crypto.DEFAULT_ENCODING from windows.crypto.certificate import * from windows.crypto.encrypt_decrypt import * +from windows.crypto.sign_verify import * from windows.crypto.cryptmsg import CryptMessage diff --git a/windows/crypto/encrypt_decrypt.py b/windows/crypto/encrypt_decrypt.py index 6853187..5210873 100644 --- a/windows/crypto/encrypt_decrypt.py +++ b/windows/crypto/encrypt_decrypt.py @@ -4,7 +4,8 @@ from windows import winproxy from windows.crypto import DEFAULT_ENCODING from windows.crypto.helper import ECRYPT_DATA_BLOB from windows.generated_def import * -from windows.crypto import Certificate + +__all__ = ["encrypt", "decrypt"] def encode_init_vector(data): blob = ECRYPT_DATA_BLOB.from_string(data) diff --git a/windows/crypto/sign_verify.py b/windows/crypto/sign_verify.py new file mode 100644 index 0000000..126f911 --- /dev/null +++ b/windows/crypto/sign_verify.py @@ -0,0 +1,63 @@ +import windows +from windows import winproxy +from windows.crypto import DEFAULT_ENCODING +from windows.crypto.helper import ECRYPT_DATA_BLOB +import windows.generated_def as gdef + +import ctypes + +__all__ = ["sign", "verify_signature"] + +def sign(cert, msg, detached_signature=False): + # hash algorithm + alg_hash = gdef.CRYPT_ALGORITHM_IDENTIFIER() + alg_hash.pszObjId = gdef.szOID_RSA_SHA256RSA # Set as parameter ? + + # Signing parameters + sign_para = gdef.CRYPT_SIGN_MESSAGE_PARA() + sign_para.cbSize = ctypes.sizeof(sign_para) + sign_para.dwMsgEncodingType = DEFAULT_ENCODING + sign_para.pSigningCert = gdef.PCERT_CONTEXT(cert) + sign_para.HashAlgorithm = alg_hash + sign_para.pvHashAuxInfo = None + sign_para.cMsgCert = 0 + sign_para.rgpMsgCert = None + sign_para.cMsgCrl = 0 + sign_para.rgpMsgCrl = None + sign_para.cAuthAttr = 0 + sign_para.rgAuthAttr = None + sign_para.cUnauthAttr = 0 + sign_para.rgUnauthAttr = None + sign_para.dwFlags = 0 + sign_para.dwInnerContentType = 0 + sign_para.HashEncryptionAlgorithm = alg_hash + sign_para.pvHashEncryptionAuxInfo = None + + # TODO: Clean + result_buffer = windows.utils.buffer(gdef.BYTE, len(msg) + 0x2000)() + result_size = gdef.DWORD(len(result_buffer)) + buff_pr = windows.utils.buffer(gdef.LPBYTE)(windows.utils.CharBuffer.from_buffer_copy(msg).cast(gdef.LPBYTE)) + buff_size = gdef.DWORD(len(msg)) + windows.winproxy.CryptSignMessage(sign_para, False, 1, buff_pr, buff_size, result_buffer.cast(gdef.LPBYTE), result_size) + return bytearray(result_buffer[:result_size.value]) + + +def verify_signature(cert, encoded_blob): + # Verify parameters + verif_param = gdef.CRYPT_KEY_VERIFY_MESSAGE_PARA() + verif_param.cbSize = ctypes.sizeof(gdef.CRYPT_KEY_VERIFY_MESSAGE_PARA) + verif_param.dwMsgEncodingType = windows.crypto.DEFAULT_ENCODING + verif_param.hCryptProv = None + # The public key used + pubkey = cert.pCertInfo[0].SubjectPublicKeyInfo + # Preparing in/out buffer/size + signed_buffer = windows.utils.buffer(gdef.BYTE).from_buffer_copy(encoded_blob) + decoded_buffer = windows.utils.CharBuffer.from_buffer_copy(encoded_blob) + decoded_size = gdef.DWORD(len(decoded_buffer)) + winproxy.CryptVerifyMessageSignatureWithKey(verif_param, + pubkey, + signed_buffer.cast(gdef.LPBYTE), + len(encoded_blob), + decoded_buffer.cast(gdef.LPBYTE), + decoded_size) + return decoded_buffer[:decoded_size.value] \ No newline at end of file diff --git a/windows/utils/pythonutils.py b/windows/utils/pythonutils.py index 95fab93..0cd19f9 100644 --- a/windows/utils/pythonutils.py +++ b/windows/utils/pythonutils.py @@ -2,13 +2,29 @@ import sys import ctypes import _ctypes -from windows.generated_def import Flag, LPCSTR, LPWSTR, INFINITE +import windows.generated_def as gdef from windows.dbgprint import dbgprint from windows import winproxy -def buffer(size): # Test +## TESTING Improved Buffer code ### + +class ImprovedCtypesBufferBase(object): + def cast(self, type): + return ctypes.cast(self, type) + + def as_string(self): + return ctypes.cast(self, LPCSTR).value + + def as_wstring(self): + return ctypes.cast(self, LPWSTR).value + + def as_pvoid(self): + return self.cast(gdef.PVOID) + +def buffer(size): # Test: DONT USE + raise NotImplementedError("utils.buffer") buf = ctypes.create_string_buffer(size) buf.size = size buf.address = ctypes.addressof(buf) @@ -19,16 +35,11 @@ def buffer(size): # Test def lol(self): return "lol" - def as_string(self): - return ctypes.cast(self, LPCSTR).value - - def as_wstring(self): - return ctypes.cast(self, LPWSTR).value - return ImprovedCtypesBufferImpl() -def wbuffer(size): # Test +def wbuffer(size): # Test: DONT USE + raise NotImplementedError("utils.buffer") buf = ctypes.create_string_buffer(size) buf.size = size buf.address = ctypes.addressof(buf) @@ -39,14 +50,44 @@ def wbuffer(size): # Test def lol(self): return "lol" - def as_string(self): - return ctypes.cast(self, LPCSTR).value - - def as_wstring(self): - return ctypes.cast(self, LPWSTR).value - return ImprovedCtypesBufferImpl() +# Used in windows.crypto.sign_verify for test +class PartialBufferType(object): + def __init__(self, type, size=None): + self.type = type + self.size = None + + @staticmethod + def create_real_implem(item_type, size): + cls_name = "YOLO<{0}><{1}>".format(item_type.__name__, size) + + class TmpImplemArrayName(ctypes.Array, ImprovedCtypesBufferBase): + _type_ = item_type + _length_ = size + + TmpImplemArrayName.__name__ = cls_name + return TmpImplemArrayName + + def from_buffer(self, buffer): + return self.create_real_implem(self.type, len(buffer)).from_buffer(buffer) + + def from_buffer_copy(self, buffer): + return self.create_real_implem(self.type, len(buffer)).from_buffer_copy(buffer) + + def __call__(self, *args): + return self.create_real_implem(self.type, len(args))(*args) + +CharBuffer = PartialBufferType(gdef.CHAR) +# CharBuffer vs CharString that append the +1 ? + +def buffer(type, size=None): + if size is None: + return PartialBufferType(type) + return PartialBufferType.create_real_implem(type, size) + + +### Other utils ### def fixedpropety(f): cache_name = "_" + f.__name__ @@ -79,7 +120,7 @@ def print_ctypes_struct(struct, name="", ident=0, hexa=False): if isinstance(value, basestring): value = repr(value) - if hexa and not isinstance(value, Flag): + if hexa and not isinstance(value, gdef.Flag): try: print("{0} -> {1}".format(name, hex(value))) return @@ -124,7 +165,7 @@ class AutoHandle(object): dbgprint("Open handle {0} for {1}".format(hex(self._handle), self), "HANDLE") return self._handle - def wait(self, timeout=INFINITE): + def wait(self, timeout=gdef.INFINITE): """Wait for the object""" return winproxy.WaitForSingleObject(self.handle, timeout) diff --git a/windows/winproxy.py b/windows/winproxy.py index 3b82bd9..975ae76 100644 --- a/windows/winproxy.py +++ b/windows/winproxy.py @@ -44,7 +44,7 @@ class Kernel32Error(WindowsError): win_error = ctypes.WinError() api_error = super(Kernel32Error, cls).__new__(cls) api_error.api_name = func_name - api_error.winerror = win_error.winerror + api_error.winerror = win_error.winerror & 0xffffffff api_error.strerror = win_error.strerror api_error.args = (func_name, win_error.winerror, win_error.strerror) return api_error @@ -690,16 +690,16 @@ def GetMappedFileNameAWrapper(hProcess, lpv, lpFilename, nSize=None): return GetMappedFileNameAWrapper.ctypes_function(hProcess, lpv, lpFilename, nSize) GetMappedFileNameA = Kernel32Proxy("GetMappedFileNameA")(GetMappedFileNameAWrapper) + def QueryWorkingSetWrapper(hProcess, pv, cb): return QueryWorkingSet.ctypes_function(hProcess, pv, cb) QueryWorkingSet = Kernel32Proxy("QueryWorkingSet")(QueryWorkingSetWrapper) + def QueryWorkingSetExWrapper(hProcess, pv, cb): return QueryWorkingSetEx.ctypes_function(hProcess, pv, cb) QueryWorkingSetEx = Kernel32Proxy("QueryWorkingSetEx")(QueryWorkingSetExWrapper) - - if not is_implemented(GetMappedFileNameA): GetMappedFileNameW = PsapiProxy("GetMappedFileNameW")(GetMappedFileNameWWrapper) GetMappedFileNameA = PsapiProxy("GetMappedFileNameA")(GetMappedFileNameAWrapper) @@ -712,7 +712,6 @@ def GetModuleBaseNameAWrapper(hProcess, hModule, lpBaseName, nSize=None): return GetModuleBaseNameAWrapper.ctypes_function(hProcess, hModule, lpBaseName, nSize) GetModuleBaseNameA = Kernel32Proxy("GetMappedFileNameA")(GetModuleBaseNameAWrapper) - def GetModuleBaseNameWWrapper(hProcess, hModule, lpBaseName, nSize=None): if nSize is None: nSize = len(lpBaseName) @@ -723,7 +722,6 @@ if not is_implemented(GetModuleBaseNameA): GetModuleBaseNameA = PsapiProxy("GetModuleBaseNameA")(GetModuleBaseNameAWrapper) GetModuleBaseNameW = PsapiProxy("GetModuleBaseNameW")(GetModuleBaseNameWWrapper) - def GetProcessImageFileNameAWrapper(hProcess, lpImageFileName, nSize=None): if nSize is None: nSize = len(lpImageFileName) @@ -740,6 +738,14 @@ if not is_implemented(GetProcessImageFileNameA): GetProcessImageFileNameA = PsapiProxy("GetProcessImageFileNameA")(GetProcessImageFileNameAWrapper) GetProcessImageFileNameW = PsapiProxy("GetProcessImageFileNameW")(GetProcessImageFileNameWWrapper) + +def GetProcessMemoryInfoWrapper(Process, ppsmemCounters, cb): + return GetProcessMemoryInfo.ctypes_function(Process, ppsmemCounters, cb) +GetProcessMemoryInfo = Kernel32Proxy("GetProcessMemoryInfo")(QueryWorkingSetExWrapper) + +if not is_implemented(GetProcessMemoryInfo): + GetProcessMemoryInfo = PsapiProxy("GetProcessMemoryInfo")(GetProcessMemoryInfoWrapper) + # Debug API DebugBreak = TransparentKernel32Proxy("DebugBreak") @@ -1713,6 +1719,33 @@ def CryptMsgVerifyCountersignatureEncodedEx(hCryptProv, dwEncodingType, pbSigner def CryptHashCertificate(hCryptProv, Algid, dwFlags, pbEncoded, cbEncoded, pbComputedHash, pcbComputedHash): return CryptHashCertificate.ctypes_function(hCryptProv, Algid, dwFlags, pbEncoded, cbEncoded, pbComputedHash, pcbComputedHash) + +@Crypt32Proxy('CryptSignMessage') +def CryptSignMessage(pSignPara, fDetachedSignature, cToBeSigned, rgpbToBeSigned, rgcbToBeSigned, pbSignedBlob, pcbSignedBlob): + return CryptSignMessage.ctypes_function(pSignPara, fDetachedSignature, cToBeSigned, rgpbToBeSigned, rgcbToBeSigned, pbSignedBlob, pcbSignedBlob) + + +@Crypt32Proxy('CryptSignAndEncryptMessage') +def CryptSignAndEncryptMessage(pSignPara, pEncryptPara, cRecipientCert, rgpRecipientCert, pbToBeSignedAndEncrypted, cbToBeSignedAndEncrypted, pbSignedAndEncryptedBlob, pcbSignedAndEncryptedBlob): + return CryptSignAndEncryptMessage.ctypes_function(pSignPara, pEncryptPara, cRecipientCert, rgpRecipientCert, pbToBeSignedAndEncrypted, cbToBeSignedAndEncrypted, pbSignedAndEncryptedBlob, pcbSignedAndEncryptedBlob) + + +@Crypt32Proxy('CryptVerifyMessageSignature') +def CryptVerifyMessageSignature(pVerifyPara, dwSignerIndex, pbSignedBlob, cbSignedBlob, pbDecoded, pcbDecoded, ppSignerCert): + return CryptVerifyMessageSignature.ctypes_function(pVerifyPara, dwSignerIndex, pbSignedBlob, cbSignedBlob, pbDecoded, pcbDecoded, ppSignerCert) + + +@Crypt32Proxy('CryptVerifyMessageSignatureWithKey') +def CryptVerifyMessageSignatureWithKey(pVerifyPara, pPublicKeyInfo, pbSignedBlob, cbSignedBlob, pbDecoded, pcbDecoded): + return CryptVerifyMessageSignatureWithKey.ctypes_function(pVerifyPara, pPublicKeyInfo, pbSignedBlob, cbSignedBlob, pbDecoded, pcbDecoded) + + +@Crypt32Proxy('CryptVerifyMessageHash') +def CryptVerifyMessageHash(pHashPara, pbHashedBlob, cbHashedBlob, pbToBeHashed, pcbToBeHashed, pbComputedHash, pcbComputedHash): + return CryptVerifyMessageHash.ctypes_function(pHashPara, pbHashedBlob, cbHashedBlob, pbToBeHashed, pcbToBeHashed, pbComputedHash, pcbComputedHash) + + + # ## CryptUI ## # @CryptUIProxy('CryptUIDlgViewContext')