Add sign/verify methods to windows.crypto + POC/Play with improved ctypes buffer/array

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
+
+

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

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)

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]

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<IMPLE><{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)
 

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')