mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
class CertificateContext(PCERT_CONTEXT) replaced by Certificate(CERT_CONTEXT) + multi-received encrypt/decrypt test
This commit is contained in:
hakril
@@ -33,7 +33,7 @@ l2ec6CyjDQc6HcQBNCsbJVq6qGtQbYNE+ih+KhIU4tO5jf25xthf2g==
|
||||
|
||||
|
||||
raw_cert = ("".join(windowscert.split("\n")[1:-1])).decode('base64')
|
||||
cert = windows.crypto.CertificateContext.from_buffer(raw_cert)
|
||||
cert = windows.crypto.Certificate.from_buffer(raw_cert)
|
||||
|
||||
print("Analysing certificate: {0}".format(cert))
|
||||
print(" * name: <{0}>".format(cert.name))
|
||||
@@ -55,7 +55,7 @@ for i, chain in enumerate(chains):
|
||||
print ""
|
||||
cert_to_verif = ccert
|
||||
print("Looking for <{0}> in trusted certificates".format(cert_to_verif.name))
|
||||
root_store = windows.crypto.EHCERTSTORE.from_system_store("Root")
|
||||
root_store = windows.crypto.CertificateStore.from_system_store("Root")
|
||||
# This is not the correct way verify the validity of a certificate chain.
|
||||
# I would say that if the goal is to verify the signature of the certificate: use wintrust.
|
||||
# (or maybe CertVerifyCertificateChainPolicy : https://msdn.microsoft.com/en-us/library/windows/desktop/aa377163(v=vs.85).aspx)
|
||||
|
||||
@@ -42,7 +42,7 @@ PFW_TMP_KEY_CONTAINER = "PythonForWindowsTMPContainer"
|
||||
def genkeys(common_name, pfxpassword, outname, keysize=2048, **kwargs):
|
||||
"""Generate a SHA256/RSA key pair. A self-signed certificate with 'common_name' is stored as 'outname'.cer.
|
||||
The private key is stored in 'outname'.pfx protected with 'pfxpassword'"""
|
||||
cert_store = crypto.EHCERTSTORE.new_in_memory()
|
||||
cert_store = crypto.CertificateStore.new_in_memory()
|
||||
# Create a TMP context that will hold our newly generated key-pair
|
||||
with crypto.CryptContext(PFW_TMP_KEY_CONTAINER, None, PROV_RSA_FULL, 0, retrycreate=True) as ctx:
|
||||
key = HCRYPTKEY()
|
||||
|
||||
+87
-5
@@ -2,6 +2,7 @@ import pytest
|
||||
|
||||
import windows.crypto
|
||||
import windows.generated_def as gdef
|
||||
import windows.crypto.generation
|
||||
|
||||
from pfwtest import *
|
||||
|
||||
@@ -63,17 +64,71 @@ def rawcert():
|
||||
def rawpfx():
|
||||
return TEST_PFX.decode("base64")
|
||||
|
||||
PFW_TEST_TMP_KEY_CONTAINER = "PythonForWindowsTMPContainerTest"
|
||||
RANDOM_CERTIF_NAME = "PythonForWindowsGeneratedRandomCertifTest"
|
||||
RANDOM_PFX_PASSWORD = "PythonForWindowsGeneratedRandomPFXPassword"
|
||||
|
||||
@pytest.fixture()
|
||||
def randomkeypair(keysize=1024):
|
||||
"""Generate a cert / pfx. Based on samples\crypto\encryption_demo.py"""
|
||||
cert_store = windows.crypto.CertificateStore.new_in_memory()
|
||||
# Create a TMP context that will hold our newly generated key-pair
|
||||
with windows.crypto.CryptContext(PFW_TEST_TMP_KEY_CONTAINER, None, gdef.PROV_RSA_FULL, 0, retrycreate=True) as ctx:
|
||||
key = gdef.HCRYPTKEY()
|
||||
keysize_flags = keysize << 16
|
||||
# Generate a key-pair that is exportable
|
||||
windows.winproxy.CryptGenKey(ctx, gdef.AT_KEYEXCHANGE, gdef.CRYPT_EXPORTABLE | keysize_flags, key)
|
||||
# It does NOT destroy the key-pair from the container,
|
||||
# It only release the key handle
|
||||
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa379918(v=vs.85).aspx
|
||||
windows.winproxy.CryptDestroyKey(key)
|
||||
|
||||
# Descrption of the key-container that will be used to generate the certificate
|
||||
KeyProvInfo = gdef.CRYPT_KEY_PROV_INFO()
|
||||
KeyProvInfo.pwszContainerName = PFW_TEST_TMP_KEY_CONTAINER
|
||||
KeyProvInfo.pwszProvName = None
|
||||
KeyProvInfo.dwProvType = gdef.PROV_RSA_FULL
|
||||
KeyProvInfo.dwFlags = 0
|
||||
KeyProvInfo.cProvParam = 0
|
||||
KeyProvInfo.rgProvParam = None
|
||||
#KeyProvInfo.dwKeySpec = AT_SIGNATURE
|
||||
KeyProvInfo.dwKeySpec = gdef.AT_KEYEXCHANGE
|
||||
|
||||
crypt_algo = gdef.CRYPT_ALGORITHM_IDENTIFIER()
|
||||
crypt_algo.pszObjId = gdef.szOID_RSA_SHA256RSA
|
||||
|
||||
certif_name = "CN={0}".format(RANDOM_CERTIF_NAME)
|
||||
# Generate a self-signed certificate based on the given key-container and signature algorithme
|
||||
certif = windows.crypto.generation.generate_selfsigned_certificate(certif_name, key_info=KeyProvInfo, signature_algo=crypt_algo)
|
||||
# Add the newly created certificate to our TMP cert-store
|
||||
cert_store.add_certificate(certif)
|
||||
# Generate a pfx from the TMP cert-store
|
||||
pfx = windows.crypto.generation.generate_pfx(cert_store, RANDOM_PFX_PASSWORD)
|
||||
yield certif, pfx
|
||||
# Destroy the TMP key container
|
||||
prov = gdef.HCRYPTPROV()
|
||||
windows.winproxy.CryptAcquireContextW(prov, PFW_TEST_TMP_KEY_CONTAINER, None, gdef.PROV_RSA_FULL, gdef.CRYPT_DELETEKEYSET)
|
||||
|
||||
|
||||
|
||||
def test_certificate(rawcert):
|
||||
cert = windows.crypto.CertificateContext.from_buffer(rawcert)
|
||||
cert = windows.crypto.Certificate.from_buffer(rawcert)
|
||||
assert cert.serial == '1b 8e 94 cb 0b 3e eb b6 41 39 f3 c9 09 b1 6b 46'
|
||||
assert cert.name == 'PythonForWindowsTest'
|
||||
assert cert.issuer == 'PythonForWindowsTest'
|
||||
assert cert.thumprint == 'EF 0C A8 C9 F9 E0 96 AF 74 18 56 8B C1 C9 57 27 A0 89 29 6A'
|
||||
assert cert.encoded == rawcert
|
||||
assert cert.version == 2
|
||||
assert cert == cert
|
||||
assert cert is cert.duplicate()
|
||||
cert.chains # TODO: craft a certificate with a chain for test purpose
|
||||
cert.store.certs
|
||||
cert.properties
|
||||
|
||||
|
||||
def test_pfx(rawcert, rawpfx):
|
||||
pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD)
|
||||
orig_cert = windows.crypto.CertificateContext.from_buffer(rawcert)
|
||||
orig_cert = windows.crypto.Certificate.from_buffer(rawcert)
|
||||
certs = pfx.certs
|
||||
assert len(certs) == 1
|
||||
# Test cert comparaison
|
||||
@@ -87,10 +142,10 @@ def test_open_pfx_bad_password(rawpfx):
|
||||
|
||||
def test_encrypt_decrypt(rawcert, rawpfx):
|
||||
message_to_encrypt = "Testing message \xff\x01"
|
||||
cert = windows.crypto.CertificateContext.from_buffer(rawcert)
|
||||
cert = windows.crypto.Certificate.from_buffer(rawcert)
|
||||
# encrypt should accept a cert or iterable of cert
|
||||
res = windows.crypto.encrypt(cert, message_to_encrypt)
|
||||
res2 = windows.crypto.encrypt([cert], message_to_encrypt)
|
||||
res2 = windows.crypto.encrypt([cert, cert], message_to_encrypt)
|
||||
del cert
|
||||
assert message_to_encrypt not in res
|
||||
|
||||
@@ -102,6 +157,33 @@ def test_encrypt_decrypt(rawcert, rawpfx):
|
||||
assert message_to_encrypt == decrypt
|
||||
assert decrypt == decrypt2
|
||||
|
||||
|
||||
|
||||
def test_randomkeypair(randomkeypair):
|
||||
randcert, randrawpfx = randomkeypair
|
||||
assert randcert.name == RANDOM_CERTIF_NAME
|
||||
randpfx = windows.crypto.import_pfx(randrawpfx, RANDOM_PFX_PASSWORD) # Check password is good too
|
||||
|
||||
|
||||
def test_encrypt_decrypt_multiple_receivers(rawcert, rawpfx, randomkeypair):
|
||||
message_to_encrypt = "\xff\x00 Testing message \xff\x01"
|
||||
# Receiver 1: random key pair
|
||||
randcert, randrawpfx = randomkeypair
|
||||
randpfx = windows.crypto.import_pfx(randrawpfx, RANDOM_PFX_PASSWORD)
|
||||
# Receiver 1: PFW-test-keypair
|
||||
pfx = windows.crypto.import_pfx(rawpfx, TEST_PFX_PASSWORD)
|
||||
cert = windows.crypto.Certificate.from_buffer(rawcert)
|
||||
assert cert.name != randcert.name
|
||||
assert cert.encoded != randcert.encoded
|
||||
# Encrypt the message with 2 differents certificates
|
||||
encrypted = windows.crypto.encrypt([cert, randcert], message_to_encrypt)
|
||||
# Decrypt with each PFX and check the result is valid/the same
|
||||
decrypted = windows.crypto.decrypt(pfx, encrypted)
|
||||
decrypted2 = windows.crypto.decrypt(randpfx, encrypted)
|
||||
assert decrypted == decrypted2 == message_to_encrypt
|
||||
|
||||
|
||||
|
||||
def test_crypt_obj():
|
||||
path = r"C:\windows\system32\kernel32.dll"
|
||||
x = windows.crypto.CryptObject(path)
|
||||
@@ -111,6 +193,6 @@ def test_crypt_obj():
|
||||
# TODO: Need some better ideas
|
||||
|
||||
def test_certificate_from_store():
|
||||
return windows.crypto.EHCERTSTORE.from_system_store("Root")
|
||||
return windows.crypto.CertificateStore.from_system_store("Root")
|
||||
|
||||
|
||||
|
||||
+114
-50
@@ -49,7 +49,7 @@ class CryptObject(object):
|
||||
dwEncoding = gdef.DWORD()
|
||||
dwContentType = gdef.DWORD()
|
||||
dwFormatType = gdef.DWORD()
|
||||
hStore = EHCERTSTORE()
|
||||
hStore = CertificateStore()
|
||||
hMsg = windows.crypto.cryptmsg.CryptMessage()
|
||||
|
||||
winproxy.CryptQueryObject(gdef.CERT_QUERY_OBJECT_FILE,
|
||||
@@ -71,6 +71,8 @@ class CryptObject(object):
|
||||
self.content_type = CRYPT_OBJECT_FORMAT_TYPE_DICT[dwContentType.value]
|
||||
|
||||
def _signers_and_certs_generator(self):
|
||||
if self.crypt_msg is None:
|
||||
return
|
||||
for signer in self.crypt_msg.signers:
|
||||
cert = self.cert_store.find(signer.Issuer, signer.SerialNumber)
|
||||
yield signer, cert
|
||||
@@ -84,13 +86,13 @@ class CryptObject(object):
|
||||
|
||||
# TODO: rename to CertificateStore ?
|
||||
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa382037(v=vs.85).aspx
|
||||
class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
class CertificateStore(gdef.HCERTSTORE):
|
||||
"""A certificate store"""
|
||||
@property
|
||||
def certs(self):
|
||||
"""The certificates in the store
|
||||
|
||||
:type: [:class:`CertificateContext`] -- A list of Certificate
|
||||
:type: [:class:`Certificate`] -- A list of Certificate
|
||||
"""
|
||||
res = []
|
||||
last = None
|
||||
@@ -102,7 +104,7 @@ class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
return tuple(res)
|
||||
raise
|
||||
# Need to duplicate as CertEnumCertificatesInStore will free the context 'last'
|
||||
ecert = windows.crypto.CertificateContext(cert[0])
|
||||
ecert = windows.crypto.Certificate.from_pointer(cert)
|
||||
res.append(ecert.duplicate())
|
||||
last = ecert
|
||||
raise RuntimeError("Out of infinit loop")
|
||||
@@ -113,7 +115,7 @@ class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, filename):
|
||||
"""Create a new :class:`EHCERTSTORE` from ``filename``"""
|
||||
"""Create a new :class:`CertificateStore` from ``filename``"""
|
||||
res = winproxy.CertOpenStore(gdef.CERT_STORE_PROV_FILENAME_A, DEFAULT_ENCODING, None, gdef.CERT_STORE_OPEN_EXISTING_FLAG, filename)
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
@@ -128,7 +130,7 @@ class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
# See https://msdn.microsoft.com/en-us/library/windows/desktop/aa388136(v=vs.85).aspx
|
||||
@classmethod
|
||||
def from_system_store(cls, store_name):
|
||||
"""Create a new :class:`EHCERTSTORE` from system store``store_name``
|
||||
"""Create a new :class:`CertificateStore` from system store``store_name``
|
||||
(see https://msdn.microsoft.com/en-us/library/windows/desktop/aa388136(v=vs.85).aspx)
|
||||
"""
|
||||
res = winproxy.CertOpenStore(gdef.CERT_STORE_PROV_SYSTEM_A, DEFAULT_ENCODING, None, gdef.CERT_SYSTEM_STORE_LOCAL_MACHINE | gdef.CERT_STORE_READONLY_FLAG, store_name)
|
||||
@@ -136,7 +138,7 @@ class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
|
||||
@classmethod
|
||||
def new_in_memory(cls):
|
||||
"""Create a new temporary :class:`EHCERTSTORE` in memory"""
|
||||
"""Create a new temporary :class:`CertificateStore` in memory"""
|
||||
res = winproxy.CertOpenStore(gdef.CERT_STORE_PROV_MEMORY, DEFAULT_ENCODING, None, 0, None)
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
@@ -145,15 +147,14 @@ class EHCERTSTORE(gdef.HCERTSTORE):
|
||||
def find(self, issuer, serialnumber):
|
||||
"""Return the certificate that match `issuer` and `serialnumber`
|
||||
|
||||
:return: :class:`CertificateContext`
|
||||
:return: :class:`Certificate`
|
||||
"""
|
||||
# data = self.get_signer_data(index)
|
||||
cert_info = gdef.CERT_INFO()
|
||||
cert_info.Issuer = issuer
|
||||
cert_info.SerialNumber = serialnumber
|
||||
rawcertcontext = winproxy.CertFindCertificateInStore(self, DEFAULT_ENCODING, 0, gdef.CERT_FIND_SUBJECT_CERT, ctypes.byref(cert_info), None)
|
||||
# return rawcertcontext
|
||||
return CertificateContext(rawcertcontext[0])
|
||||
return Certificate.from_pointer(rawcertcontext)
|
||||
|
||||
|
||||
# PKCS12_NO_PERSIST_KEY -> do not save it in a key container on disk
|
||||
@@ -176,32 +177,23 @@ def import_pfx(pfx, password=None, flags=gdef.CRYPT_USER_KEYSET | gdef.PKCS12_NO
|
||||
|
||||
``PKCS12_NO_PERSIST_KEY`` tells ``CryptoAPI`` to NOT save the keys in a on-disk container.
|
||||
|
||||
:return: :class:`EHCERTSTORE`
|
||||
:return: :class:`CertificateStore`
|
||||
"""
|
||||
if isinstance(pfx, basestring):
|
||||
if isinstance(pfx, (basestring, bytearray)):
|
||||
pfx = gdef.CRYPT_DATA_BLOB.from_string(pfx)
|
||||
cert_store = winproxy.PFXImportCertStore(pfx, password, flags)
|
||||
return EHCERTSTORE(cert_store)
|
||||
return CertificateStore(cert_store)
|
||||
|
||||
|
||||
# Why PCCERT_CONTEXT (pointer type) and not _CERT_CONTEXT ?
|
||||
class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
"""Represent a Certificate.
|
||||
|
||||
note: It is a pointer ctypes structure (``PCCERT_CONTEXT``)
|
||||
"""
|
||||
_type_ = gdef.PCCERT_CONTEXT._type_ # Not herited from PCCERT_CONTEXT
|
||||
|
||||
|
||||
def __repr__(self):
|
||||
return '<{0} "{1}" serial="{2}">'.format(type(self).__name__, self.name, self.serial)
|
||||
class Certificate(gdef.CERT_CONTEXT):
|
||||
"""Represent a Certificate """
|
||||
|
||||
@property
|
||||
def raw_serial(self):
|
||||
"""The raw serial number of the certificate.
|
||||
|
||||
:type: [:class:`int`]: A list of int ``0 <= x <= 255``"""
|
||||
serial_number = self[0].pCertInfo[0].SerialNumber
|
||||
serial_number = self.pCertInfo[0].SerialNumber
|
||||
return [(c & 0xff) for c in serial_number.pbData[:serial_number.cbData][::-1]]
|
||||
|
||||
@property
|
||||
@@ -210,7 +202,6 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
|
||||
:type: :class:`str`
|
||||
"""
|
||||
serial_number = self[0].pCertInfo[0].SerialNumber
|
||||
serial_bytes = self.raw_serial
|
||||
return " ".join("{:02x}".format(x) for x in serial_bytes)
|
||||
|
||||
@@ -230,6 +221,21 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
|
||||
:type: :class:`str`"""
|
||||
|
||||
def raw_hash(self):
|
||||
size = gdef.DWORD(100)
|
||||
buffer = ctypes.c_buffer(size.value)
|
||||
winproxy.CryptHashCertificate(None, 0, 0, self.pbCertEncoded, self.cbCertEncoded, ctypes.cast(buffer, gdef.LPBYTE), size)
|
||||
return buffer[:size.value]
|
||||
|
||||
@property
|
||||
def thumprint(self):
|
||||
"""The thumprint of the certificate (which is the sha1 of the encoded cert) with the format:
|
||||
'XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX XX'
|
||||
|
||||
:type: :class:`str`
|
||||
"""
|
||||
return " ".join("{:02X}".format(x) for x in bytearray(self.raw_hash()))
|
||||
|
||||
@property
|
||||
def issuer(self):
|
||||
"""The name of the certificate's issuer.
|
||||
@@ -241,9 +247,9 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
def store(self):
|
||||
"""The certificate store that contains the certificate
|
||||
|
||||
:type: :class:`EHCERTSTORE`
|
||||
:type: :class:`CertificateStore`
|
||||
"""
|
||||
return EHCERTSTORE(self[0].hCertStore)
|
||||
return CertificateStore(self.hCertStore)
|
||||
|
||||
def get_raw_certificate_chains(self): # Rename to all_chains ?
|
||||
chain_context = EPCCERT_CHAIN_CONTEXT()
|
||||
@@ -260,16 +266,18 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
chain_para.cbSize = ctypes.sizeof(chain_para)
|
||||
chain_para.RequestedUsage = cert_usage
|
||||
|
||||
winproxy.CertGetCertificateChain(None, self, None, self[0].hCertStore, ctypes.byref(chain_para), 0, None, ctypes.byref(chain_context))
|
||||
winproxy.CertGetCertificateChain(None, self, None, self.hCertStore, ctypes.byref(chain_para), 0, None, ctypes.byref(chain_context))
|
||||
# Lower chains ?
|
||||
# winproxy.CertGetCertificateChain(None, self, None, self[0].hCertStore, ctypes.byref(chain_para), 0x80, None, ctypes.byref(chain_context))
|
||||
#return CertficateChain(chain_context)
|
||||
return chain_context
|
||||
|
||||
@property # fixedproperty ?
|
||||
def chains(self):
|
||||
"""The list of chain context available for this certificate. Each elements of this list is a list of ``CertificateContext`` that should
|
||||
"""The list of chain context available for this certificate. Each elements of this list is a list of ``Certificate`` that should
|
||||
go from the ``self`` certificate to a trusted certificate.
|
||||
|
||||
:type: [[:class:`CertificateContext`]] -- A list of chain (list) of :class:`CertificateContext`
|
||||
:type: [[:class:`Certificate`]] -- A list of chain (list) of :class:`Certificate`
|
||||
"""
|
||||
chain_context = self.get_raw_certificate_chains()
|
||||
res = []
|
||||
@@ -286,19 +294,62 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
|
||||
note: The object returned is ``self``
|
||||
|
||||
:return: :class:`CertificateContext`
|
||||
:return: :class:`Certificate`
|
||||
"""
|
||||
res = winproxy.CertDuplicateCertificateContext(self)
|
||||
# Check what the doc says: the pointer returned is actually the PCERT in parameter
|
||||
# Only the refcount is incremented
|
||||
# This postulate allow us to return 'self' directly
|
||||
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa376045(v=vs.85).aspx
|
||||
if not ctypes.cast(res, gdef.PVOID).value == ctypes.cast(self, gdef.PVOID).value:
|
||||
if not ctypes.addressof(res[0]) == ctypes.addressof(self):
|
||||
raise ValueError("CertDuplicateCertificateContext did not returned the argument (check doc)")
|
||||
return self
|
||||
|
||||
def view(self, title=None):
|
||||
return windows.winproxy.CryptUIDlgViewContext(gdef.CERT_STORE_CERTIFICATE_CONTEXT, self, None, title, 0, None)
|
||||
return windows.winproxy.CryptUIDlgViewContext(gdef.CERT_STORE_CERTIFICATE_CONTEXT, ctypes.byref(self), None, title, 0, None)
|
||||
|
||||
KNOWN_PROPERTIES_VALUES = gdef.FlagMapper(
|
||||
gdef.CERT_KEY_PROV_HANDLE_PROP_ID,
|
||||
gdef.CERT_KEY_PROV_INFO_PROP_ID,
|
||||
gdef.CERT_SHA1_HASH_PROP_ID,
|
||||
gdef.CERT_MD5_HASH_PROP_ID,
|
||||
gdef.CERT_HASH_PROP_ID,
|
||||
gdef.CERT_KEY_CONTEXT_PROP_ID,
|
||||
gdef.CERT_KEY_SPEC_PROP_ID,
|
||||
gdef.CERT_IE30_RESERVED_PROP_ID,
|
||||
gdef.CERT_PUBKEY_HASH_RESERVED_PROP_ID,
|
||||
gdef.CERT_ENHKEY_USAGE_PROP_ID,
|
||||
gdef.CERT_CTL_USAGE_PROP_ID,
|
||||
gdef.CERT_NEXT_UPDATE_LOCATION_PROP_ID,
|
||||
gdef.CERT_FRIENDLY_NAME_PROP_ID,
|
||||
gdef.CERT_PVK_FILE_PROP_ID,
|
||||
gdef.CERT_DESCRIPTION_PROP_ID,
|
||||
gdef.CERT_ACCESS_STATE_PROP_ID,
|
||||
gdef.CERT_SIGNATURE_HASH_PROP_ID,
|
||||
gdef.CERT_SMART_CARD_DATA_PROP_ID,
|
||||
gdef.CERT_EFS_PROP_ID,
|
||||
gdef.CERT_FORTEZZA_DATA_PROP_ID,
|
||||
gdef.CERT_ARCHIVED_PROP_ID,
|
||||
gdef.CERT_KEY_IDENTIFIER_PROP_ID,
|
||||
gdef.CERT_AUTO_ENROLL_PROP_ID,
|
||||
gdef.CERT_PUBKEY_ALG_PARA_PROP_ID,
|
||||
gdef.CERT_CROSS_CERT_DIST_POINTS_PROP_ID,
|
||||
gdef.CERT_ISSUER_PUBLIC_KEY_MD5_HASH_PROP_ID,
|
||||
gdef.CERT_SUBJECT_PUBLIC_KEY_MD5_HASH_PROP_ID,
|
||||
gdef.CERT_ENROLLMENT_PROP_ID,
|
||||
gdef.CERT_DATE_STAMP_PROP_ID,
|
||||
gdef.CERT_ISSUER_SERIAL_NUMBER_MD5_HASH_PROP_ID,
|
||||
gdef.CERT_SUBJECT_NAME_MD5_HASH_PROP_ID,
|
||||
gdef.CERT_EXTENDED_ERROR_INFO_PROP_ID,
|
||||
gdef.CERT_RENEWAL_PROP_ID,
|
||||
gdef.CERT_ARCHIVED_KEY_HASH_PROP_ID,
|
||||
gdef.CERT_AUTO_ENROLL_RETRY_PROP_ID,
|
||||
gdef.CERT_AIA_URL_RETRIEVED_PROP_ID,
|
||||
gdef.CERT_AUTHORITY_INFO_ACCESS_PROP_ID,
|
||||
gdef.CERT_BACKED_UP_PROP_ID,
|
||||
gdef.CERT_OCSP_RESPONSE_PROP_ID,
|
||||
gdef.CERT_REQUEST_ORIGINATOR_PROP_ID,
|
||||
gdef.CERT_SOURCE_LOCATION_PROP_ID)
|
||||
|
||||
def enum_properties(self):
|
||||
prop = 0
|
||||
@@ -307,7 +358,7 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
prop = winproxy.CertEnumCertificateContextProperties(self, prop)
|
||||
if not prop:
|
||||
return res
|
||||
res.append(prop)
|
||||
res.append(self.KNOWN_PROPERTIES_VALUES[prop])
|
||||
raise RuntimeError("Unreachable code")
|
||||
|
||||
properties = property(enum_properties)
|
||||
@@ -323,43 +374,49 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
"""The encoded certificate.
|
||||
|
||||
:type: :class:`bytearray`"""
|
||||
return bytearray(self[0].pbCertEncoded[:self[0].cbCertEncoded])
|
||||
return bytearray(self.pbCertEncoded[:self.cbCertEncoded])
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"TODO: doc"
|
||||
return self[0].pbCertInfo.dwVersion
|
||||
return self.pCertInfo[0].dwVersion
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, filename):
|
||||
"""Create a :class:`CertificateContext` from the file ``filename``
|
||||
"""Create a :class:`Certificate` from the file ``filename``
|
||||
|
||||
:return: :class:`CertificateContext`
|
||||
:return: :class:`Certificate`
|
||||
"""
|
||||
with open(filename, "rb") as f:
|
||||
data = f.read()
|
||||
buf = (ctypes.c_ubyte * len(data))(*bytearray(data))
|
||||
res = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return ctypes.cast(res, cls)
|
||||
pcert = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return cls.from_pointer(pcert)
|
||||
|
||||
@classmethod
|
||||
def from_buffer(cls, data):
|
||||
"""Create a :class:`CertificateContext` from the buffer ``data``
|
||||
"""Create a :class:`Certificate` from the buffer ``data``
|
||||
|
||||
:return: :class:`CertificateContext`
|
||||
:return: :class:`Certificate`
|
||||
"""
|
||||
buf = (ctypes.c_ubyte * len(data))(*bytearray(data))
|
||||
res = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return ctypes.cast(res, cls)
|
||||
pcert = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return cls.from_pointer(pcert)
|
||||
|
||||
@classmethod
|
||||
def from_pointer(self, ptr):
|
||||
return ctypes.cast(ptr, ctypes.POINTER(Certificate))[0]
|
||||
|
||||
|
||||
def __eq__(self, other):
|
||||
if not isinstance(other, CertificateContext):
|
||||
if not isinstance(other, Certificate):
|
||||
return NotImplemented
|
||||
return windows.winproxy.CertCompareCertificate(DEFAULT_ENCODING, self[0].pCertInfo, other[0].pCertInfo)
|
||||
return windows.winproxy.CertCompareCertificate(DEFAULT_ENCODING, self.pCertInfo, other.pCertInfo)
|
||||
|
||||
def __repr__(self):
|
||||
return '<{0} "{1}" serial="{2}">'.format(type(self).__name__, self.name, self.serial)
|
||||
|
||||
# CertCompareCertificate ?
|
||||
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa376027(v=vs.85).aspx
|
||||
|
||||
|
||||
# class CertficateChain(object):
|
||||
@@ -374,12 +431,19 @@ class CertificateContext(gdef.PCCERT_CONTEXT):
|
||||
|
||||
|
||||
# Those classes are more of a POC than anything else
|
||||
# Should be the struct itself (like Certificate ?)
|
||||
class EPCCERT_CHAIN_CONTEXT(gdef.PCCERT_CHAIN_CONTEXT):
|
||||
_type_ = gdef.PCCERT_CHAIN_CONTEXT._type_
|
||||
|
||||
@property
|
||||
def chains(self):
|
||||
res = []
|
||||
# if (self[0].cLowerQualityChainContext):
|
||||
# print("LOL")
|
||||
# import pdb;pdb.set_trace()
|
||||
# if self[0].cChain > 1:
|
||||
# print("HAAAAA")
|
||||
# import pdb;pdb.set_trace()
|
||||
for i in range(self[0].cChain):
|
||||
simple_chain = ctypes.cast(self[0].rgpChain[i], EPCCERT_SIMPLE_CHAIN)
|
||||
res.append(simple_chain)
|
||||
@@ -411,7 +475,7 @@ class EPCERT_CHAIN_ELEMENT(gdef.PCERT_CHAIN_ELEMENT):
|
||||
|
||||
@property
|
||||
def cert(self):
|
||||
return ctypes.cast(self[0].pCertContext, CertificateContext)
|
||||
return Certificate.from_pointer(self[0].pCertContext)
|
||||
|
||||
|
||||
# Move this in another .py ?
|
||||
|
||||
@@ -10,12 +10,14 @@ class CryptMessage(gdef.HCRYPTMSG):
|
||||
gdef.CMSG_CERT_COUNT_PARAM: gdef.DWORD}
|
||||
|
||||
|
||||
def get_param(self, param_type, index=0):
|
||||
def get_param(self, param_type, index=0, raw=False):
|
||||
data_size = gdef.DWORD()
|
||||
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa380227(v=vs.85).aspx
|
||||
winproxy.CryptMsgGetParam(self, param_type, index, None, data_size)
|
||||
buffer = ctypes.c_buffer(data_size.value)
|
||||
winproxy.CryptMsgGetParam(self, param_type, index, buffer, data_size)
|
||||
if raw:
|
||||
return (buffer, data_size)
|
||||
|
||||
if param_type in self.MSG_PARAM_KNOW_TYPES:
|
||||
buffer = self.MSG_PARAM_KNOW_TYPES[param_type].from_buffer(buffer)
|
||||
@@ -23,7 +25,6 @@ class CryptMessage(gdef.HCRYPTMSG):
|
||||
return buffer.value
|
||||
return buffer
|
||||
|
||||
|
||||
# Certificate accessors
|
||||
|
||||
@property
|
||||
@@ -42,9 +43,9 @@ class CryptMessage(gdef.HCRYPTMSG):
|
||||
|
||||
note: not all embded certificate are directly used to sign the :class:`CryptObject`.
|
||||
|
||||
:return: :class:`CertificateContext`
|
||||
:return: :class:`Certificate`
|
||||
"""
|
||||
return windows.crypto.CertificateContext.from_buffer(self.get_raw_cert(index))
|
||||
return windows.crypto.Certificate.from_buffer(self.get_raw_cert(index))
|
||||
|
||||
@property
|
||||
def certs(self):
|
||||
|
||||
@@ -4,6 +4,7 @@ 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
|
||||
|
||||
def encode_init_vector(data):
|
||||
blob = ECRYPT_DATA_BLOB.from_string(data)
|
||||
@@ -46,10 +47,14 @@ def encrypt(cert_or_certlist, msg, algo=szOID_NIST_AES256_CBC, initvector=genini
|
||||
"""
|
||||
alg_ident = CRYPT_ALGORITHM_IDENTIFIER()
|
||||
alg_ident.pszObjId = algo
|
||||
if isinstance(cert_or_certlist, PCERT_CONTEXT):
|
||||
certlist = (cert_or_certlist,)
|
||||
# We want to have automatique translation of Certificate -> PCERT_CONTEXT
|
||||
# In order to simple create the 'PCERT_CONTEXT[] certs'
|
||||
# For that we need a tuple of X * 1-item-tuple
|
||||
# as a (cert,) will be automaticly translatable to a PCERT_CONTEXT
|
||||
if isinstance(cert_or_certlist, CERT_CONTEXT):
|
||||
certlist = ((cert_or_certlist,),)
|
||||
else:
|
||||
certlist = tuple(cert_or_certlist)
|
||||
certlist = tuple((c,) for c in cert_or_certlist)
|
||||
|
||||
# Set (compute if needed) the IV
|
||||
if initvector is None:
|
||||
@@ -74,6 +79,7 @@ def encrypt(cert_or_certlist, msg, algo=szOID_NIST_AES256_CBC, initvector=genini
|
||||
param.dwFlags = 0
|
||||
param.dwInnerContentType = 0
|
||||
|
||||
|
||||
certs = (PCERT_CONTEXT * len(certlist))(*certlist)
|
||||
#Ask the output buffer size
|
||||
size = DWORD()
|
||||
|
||||
@@ -3,7 +3,7 @@ from windows import winproxy
|
||||
from windows.generated_def import *
|
||||
|
||||
from windows.crypto.helper import ECRYPT_DATA_BLOB
|
||||
from windows.crypto import DEFAULT_ENCODING, EHCERTSTORE
|
||||
from windows.crypto import DEFAULT_ENCODING, CertificateStore
|
||||
|
||||
|
||||
def generate_selfsigned_certificate(name="CN=DEFAULT", prov=None, key_info=None, flags=0, signature_algo=None):
|
||||
@@ -11,14 +11,14 @@ def generate_selfsigned_certificate(name="CN=DEFAULT", prov=None, key_info=None,
|
||||
|
||||
See https://msdn.microsoft.com/en-us/library/windows/desktop/aa376039(v=vs.85).aspx
|
||||
|
||||
:return: :class:`windows.crypto.CertificateContext`
|
||||
:return: :class:`windows.crypto.Certificate`
|
||||
"""
|
||||
size = ULONG(len(name) + 0x100)
|
||||
buffer = (ctypes.c_ubyte * size.value)()
|
||||
winproxy.CertStrToNameA(X509_ASN_ENCODING, name, CERT_OID_NAME_STR, None, buffer, size, None)
|
||||
blobname = ECRYPT_DATA_BLOB(size.value, buffer)
|
||||
cert = winproxy.CertCreateSelfSignCertificate(prov, blobname, flags, key_info, signature_algo, None, None, None)
|
||||
return windows.crypto.CertificateContext(cert[0])
|
||||
return windows.crypto.Certificate.from_pointer(cert)
|
||||
|
||||
|
||||
def generate_key(prov, keytype=AT_KEYEXCHANGE, flags=CRYPT_EXPORTABLE):
|
||||
|
||||
@@ -4,8 +4,6 @@ import _ctypes
|
||||
from windows.generated_def import Flag, LPCSTR, LPWSTR
|
||||
|
||||
|
||||
|
||||
|
||||
def buffer(size): # Test
|
||||
buf = ctypes.create_string_buffer(size)
|
||||
buf.size = size
|
||||
@@ -25,6 +23,7 @@ def buffer(size): # Test
|
||||
|
||||
return ImprovedCtypesBufferImpl()
|
||||
|
||||
|
||||
def fixedpropety(f):
|
||||
cache_name = "_" + f.__name__
|
||||
|
||||
@@ -73,5 +72,6 @@ def print_ctypes_struct(struct, name="", ident=0, hexa=False):
|
||||
continue
|
||||
print_ctypes_struct(value, "{0}.{1}".format(name, fname), hexa=hexa)
|
||||
|
||||
|
||||
def sprint(struct, name="struct", hexa=True):
|
||||
return print_ctypes_struct(struct, name=name, hexa=hexa)
|
||||
@@ -104,8 +104,6 @@ class Handle(SYSTEM_HANDLE):
|
||||
del thread._handle
|
||||
return res
|
||||
|
||||
|
||||
|
||||
def __repr__(self):
|
||||
return "<{0} value=<0x{1:x}> in process pid={2}>".format(type(self).__name__, self.wValue, self.dwProcessId)
|
||||
|
||||
|
||||
@@ -1,76 +0,0 @@
|
||||
import ctypes
|
||||
|
||||
import windows
|
||||
from windows import winproxy
|
||||
from windows.generated_def import *
|
||||
|
||||
|
||||
|
||||
callback_type = ctypes.WINFUNCTYPE(UINT, HWND, LPARAM)
|
||||
|
||||
|
||||
class Point(POINT):
|
||||
def __repr__(self):
|
||||
return "<{0} x={1} y={2}>".format(type(self).__name__, self.x, self.y)
|
||||
#return "<{0} x={1:#x} y={2:#x}>".format(type(self).__name__, self.x, self.y)
|
||||
|
||||
class Rect(RECT):
|
||||
def __repr__(self):
|
||||
return "<{0} left={1} top={2} right={3} bottom={4}>".format(type(self).__name__, self.left, self.top, self.right, self.bottom)
|
||||
#return "<{0} x={1:#x} y={2:#x}>".format(type(self).__name__, self.x, self.y)
|
||||
|
||||
|
||||
def get_cursor_pos():
|
||||
res = Point()
|
||||
winproxy.GetCursorPos(res)
|
||||
return res
|
||||
|
||||
class Window(object):
|
||||
def __init__(self, handle):
|
||||
self.handle = handle
|
||||
|
||||
def name(self):
|
||||
size = 0x1024
|
||||
buffer = ctypes.c_buffer(size)
|
||||
|
||||
res = windows.winproxy.GetWindowTextA(self.handle, buffer, size)
|
||||
return buffer[:res]
|
||||
|
||||
def rect(self):
|
||||
res = Rect()
|
||||
winproxy.GetWindowRect(self.handle, res)
|
||||
return res
|
||||
|
||||
def size(self):
|
||||
rect = self.rect()
|
||||
width = rect.right - rect.left
|
||||
heigth = rect.bottom - rect.top
|
||||
return width, heigth
|
||||
|
||||
@classmethod
|
||||
def at_point(cls, point):
|
||||
handle = winproxy.WindowFromPoint(point)
|
||||
return cls(handle)
|
||||
|
||||
|
||||
# I don't understand the interest:
|
||||
# Either return "" or C:\Python27\python.exe
|
||||
#def module(self):
|
||||
# size = 0x1024
|
||||
# buffer = ctypes.c_buffer(size)
|
||||
# res = windows.winproxy.GetWindowModuleFileNameA(self.handle, buffer, size)
|
||||
# return buffer[:res]
|
||||
|
||||
|
||||
def enumwindows():
|
||||
result = []
|
||||
def callback(handle, param):
|
||||
result.append(handle)
|
||||
return True
|
||||
|
||||
try:
|
||||
x = windows.winproxy.EnumWindows(callback_type(callback), 0)
|
||||
except WindowsError:
|
||||
if not result:
|
||||
raise
|
||||
return result
|
||||
Reference in New Issue
Block a user