mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
More work on windows.crypto and Add an encryption sample POC
This commit is contained in:
Clement Rouault
@@ -0,0 +1,103 @@
|
||||
import argparse
|
||||
import windows.crypto as crypto
|
||||
from windows import winproxy
|
||||
from windows.generated_def import *
|
||||
|
||||
import windows.crypto.generation as gencrypt
|
||||
|
||||
# http://stackoverflow.com/questions/1461272/basic-questions-on-microsoft-cryptoapi
|
||||
|
||||
def crypt(src, dst, certs, **kwargs):
|
||||
"""Encrypt the content of 'src' file with the certifacte in 'certs' into 'dst'"""
|
||||
# Open every certificate in the certs list
|
||||
certlist = [crypto.CertificatContext.from_file(x) for x in certs]
|
||||
# Encrypt the content of 'src' with all the public keys(certs)
|
||||
res = crypto.encrypt(certlist, src.read())
|
||||
print("Encryption done. Result:")
|
||||
print(repr(res))
|
||||
# Write the result in 'dst'
|
||||
dst.write(res)
|
||||
dst.close()
|
||||
src.close()
|
||||
|
||||
def decrypt(src, pfxfile, password, **kwargs):
|
||||
"""Decrypt the content of 'src' with the private key in 'pfxfile'. the 'pfxfile' is open using the 'password'"""
|
||||
# Open the 'pfx' with the given password
|
||||
pfx = crypto.import_pfx(pfxfile.read(), password)
|
||||
# Decrypt the content of the file
|
||||
decrypted = crypto.decrypt(pfx, src.read())
|
||||
print(u"Result = <{0}>".format(decrypted.decode("utf8")))
|
||||
|
||||
PFW_TMP_KEY_CONTAINER = "PythonForWindowsTMPContainer"
|
||||
|
||||
def genkeys(common_name, pfxpassword, outname, **kwargs):
|
||||
"""Generate a SHA256/RSA key pair. A self-signed certificate with 'common_name' if store at 'outname'.cer.
|
||||
The private key is stored in 'outname'.pfx protected with 'pfxpassword'"""
|
||||
cert_store = crypto.EHCERTSTORE.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()
|
||||
# Generate a key-pair that is exportable
|
||||
winproxy.CryptGenKey(ctx, AT_KEYEXCHANGE, CRYPT_EXPORTABLE, key)
|
||||
winproxy.CryptDestroyKey(key)
|
||||
|
||||
# Descrption of the key-container that will be used to generate the certificate
|
||||
KeyProvInfo = CRYPT_KEY_PROV_INFO()
|
||||
KeyProvInfo.pwszContainerName = PFW_TMP_KEY_CONTAINER
|
||||
KeyProvInfo.pwszProvName = None
|
||||
KeyProvInfo.dwProvType = PROV_RSA_FULL
|
||||
KeyProvInfo.dwFlags = 0
|
||||
KeyProvInfo.cProvParam = 0
|
||||
KeyProvInfo.rgProvParam = None
|
||||
#KeyProvInfo.dwKeySpec = AT_SIGNATURE
|
||||
KeyProvInfo.dwKeySpec = AT_KEYEXCHANGE
|
||||
|
||||
crypt_algo = CRYPT_ALGORITHM_IDENTIFIER()
|
||||
crypt_algo.pszObjId = szOID_RSA_SHA256RSA
|
||||
|
||||
certif_name = "CN={0}".format(common_name)
|
||||
# Generate a self-signed certificate based on the given key-container and signature algorithme
|
||||
certif = gencrypt.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 = gencrypt.generate_pfx(cert_store, pfxpassword)
|
||||
if outname is None:
|
||||
outname = common_name.lower()
|
||||
|
||||
# Dump the certif (public key) and pfx (public + private keys)
|
||||
with open(outname + ".cer", "wb") as f:
|
||||
# The encoded certif only contains the public key
|
||||
f.write(certif.encoded())
|
||||
with open(outname + ".pfx", "wb") as f:
|
||||
f.write(pfx)
|
||||
print(certif)
|
||||
# Destroy the TMP key container
|
||||
prov = HCRYPTPROV()
|
||||
winproxy.CryptAcquireContextW(prov, PFW_TMP_KEY_CONTAINER, None, PROV_RSA_FULL, CRYPT_DELETEKEYSET)
|
||||
|
||||
parser = argparse.ArgumentParser(prog='PROG')
|
||||
subparsers = parser.add_subparsers(description='valid subcommands',)
|
||||
|
||||
cryptparse = subparsers.add_parser('crypt')
|
||||
cryptparse.set_defaults(func=crypt)
|
||||
|
||||
cryptparse.add_argument('src', type=argparse.FileType('rb'), help='File to encrypt')
|
||||
cryptparse.add_argument('dst', type=argparse.FileType('wb'), help='The encrypted file')
|
||||
cryptparse.add_argument('certs', type=str, nargs='+',
|
||||
help='List of certfile used to encrypt the src')
|
||||
|
||||
decryptparse = subparsers.add_parser('decrypt')
|
||||
decryptparse.set_defaults(func=decrypt)
|
||||
decryptparse.add_argument('src', type=argparse.FileType('rb'), help='File to decrypt')
|
||||
decryptparse.add_argument('pfxfile', type=argparse.FileType('rb'), help='PFX file to use')
|
||||
decryptparse.add_argument('password', help='Password of the PFX')
|
||||
|
||||
genkeysparse = subparsers.add_parser('genkey')
|
||||
genkeysparse.set_defaults(func=genkeys)
|
||||
genkeysparse.add_argument('common_name', nargs='?', metavar='CommonName', default='DEFAULT', help='the common name of the certificate')
|
||||
genkeysparse.add_argument('outname', nargs='?',help='The filename base for the generated files')
|
||||
genkeysparse.add_argument('--pfxpassword', nargs='?', help='Password to protect the PFX')
|
||||
|
||||
res = parser.parse_args()
|
||||
res.func(**res.__dict__)
|
||||
@@ -35,7 +35,8 @@ CRYPT_OBJECT_FORMAT_TYPE_DICT = {x:x for x in CRYPT_OBJECT_FORMAT_TYPE}
|
||||
|
||||
class CryptObject(object):
|
||||
MSG_PARAM_KNOW_TYPES = {CMSG_SIGNER_INFO_PARAM: CMSG_SIGNER_INFO,
|
||||
CMSG_SIGNER_COUNT_PARAM: DWORD}
|
||||
CMSG_SIGNER_COUNT_PARAM: DWORD,
|
||||
CMSG_CERT_COUNT_PARAM: DWORD}
|
||||
|
||||
def __init__(self, filename, content_type=CERT_QUERY_CONTENT_FLAG_ALL):
|
||||
# No other API than filename for now..
|
||||
@@ -64,11 +65,11 @@ class CryptObject(object):
|
||||
self.encoding = dwEncoding
|
||||
self.content_type = CRYPT_OBJECT_FORMAT_TYPE_DICT.get(dwContentType.value, dwContentType)
|
||||
|
||||
def msg_get_param(self, param_type):
|
||||
def msg_get_param(self, param_type, index=0):
|
||||
signer_info = DWORD()
|
||||
winproxy.CryptMsgGetParam(self.hmsg, param_type, 0, None, signer_info)
|
||||
winproxy.CryptMsgGetParam(self.hmsg, param_type, index, None, signer_info)
|
||||
buffer = ctypes.c_buffer(signer_info.value)
|
||||
winproxy.CryptMsgGetParam(self.hmsg, param_type, 0, buffer, signer_info)
|
||||
winproxy.CryptMsgGetParam(self.hmsg, param_type, index, buffer, signer_info)
|
||||
|
||||
if param_type in self.MSG_PARAM_KNOW_TYPES:
|
||||
buffer = self.MSG_PARAM_KNOW_TYPES[param_type].from_buffer_copy(buffer)
|
||||
@@ -77,8 +78,8 @@ class CryptObject(object):
|
||||
def get_nb_signer(self):
|
||||
return self.msg_get_param(CMSG_SIGNER_COUNT_PARAM).value
|
||||
|
||||
def get_signer_data(self):
|
||||
return self.msg_get_param(CMSG_SIGNER_INFO_PARAM)
|
||||
def get_signer_data(self, index=0):
|
||||
return self.msg_get_param(CMSG_SIGNER_INFO_PARAM, index)
|
||||
|
||||
def get_signer_certificate(self):
|
||||
data = self.get_signer_data()
|
||||
@@ -89,6 +90,18 @@ class CryptObject(object):
|
||||
#return rawcertcontext
|
||||
return CertificatContext(rawcertcontext[0])
|
||||
|
||||
def get_cert(self, index=0):
|
||||
return self.msg_get_param(CMSG_CERT_PARAM, index)
|
||||
|
||||
def get_nb_cert(self):
|
||||
"TEST"
|
||||
return self.msg_get_param(CMSG_CERT_COUNT_PARAM).value
|
||||
|
||||
def test_all_certs(self):
|
||||
nb_cert = self.get_nb_cert()
|
||||
return [CertificatContext.from_buffer(self.get_cert(i)) for i in range(nb_cert)]
|
||||
|
||||
|
||||
def __repr__(self):
|
||||
return '<{0} "{1}" content_type={2}>'.format(type(self).__name__, self.filename, self.content_type)
|
||||
|
||||
@@ -122,13 +135,19 @@ class EHCERTSTORE(HCERTSTORE):
|
||||
res = winproxy.CertOpenStore(CERT_STORE_PROV_FILENAME_A, DEFAULT_ENCODING, None, CERT_STORE_OPEN_EXISTING_FLAG, filename)
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
@classmethod
|
||||
def from_system_store(cls, store_name):
|
||||
res = winproxy.CertOpenStore(CERT_STORE_PROV_SYSTEM_A, DEFAULT_ENCODING, None, CERT_SYSTEM_STORE_LOCAL_MACHINE | CERT_STORE_READONLY_FLAG, store_name)
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
@classmethod
|
||||
def new_in_memory(cls):
|
||||
res = winproxy.CertOpenStore(CERT_STORE_PROV_MEMORY, DEFAULT_ENCODING, None, 0, None)
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
|
||||
def import_pfx(pfx, password=None, flags=CRYPT_USER_KEYSET):
|
||||
# PKCS12_NO_PERSIST_KEY -> do not save it in a key container on disk
|
||||
# Without it a key container is created at 'C:\Users\USERNAME\AppData\Roaming\Microsoft\Crypto\RSA\S-1-5-21-3241049326-165485355-1070449050-1001'
|
||||
def import_pfx(pfx, password=None, flags=CRYPT_USER_KEYSET | PKCS12_NO_PERSIST_KEY):
|
||||
if isinstance(pfx, basestring):
|
||||
pfx = ECRYPT_DATA_BLOB.from_string(pfx)
|
||||
cert_store = winproxy.PFXImportCertStore(pfx, password, flags)
|
||||
@@ -211,6 +230,10 @@ class CertificatContext(PCCERT_CONTEXT):
|
||||
|
||||
properties = property(enum_properties)
|
||||
|
||||
|
||||
def encoded(self):
|
||||
return bytearray(self[0].pbCertEncoded[:self[0].cbCertEncoded])
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, filename):
|
||||
with open(filename, "rb") as f:
|
||||
@@ -219,6 +242,13 @@ class CertificatContext(PCCERT_CONTEXT):
|
||||
res = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
@classmethod
|
||||
def from_buffer(cls, data):
|
||||
buf = (ctypes.c_ubyte * len(data))(*bytearray(data))
|
||||
res = windows.winproxy.CertCreateCertificateContext(windows.crypto.DEFAULT_ENCODING, buf, len(data))
|
||||
return ctypes.cast(res, cls)
|
||||
|
||||
|
||||
|
||||
|
||||
class CertficateChain(object):
|
||||
@@ -229,4 +259,36 @@ class CertficateChain(object):
|
||||
res = []
|
||||
for i in range(self.chain.rgpChain[0][0].cElement):
|
||||
res.append(CertificatContext(self.chain.rgpChain[0][0].rgpElement[i][0].pCertContext[0]))
|
||||
return res
|
||||
return res
|
||||
|
||||
# Move this in another .py ?
|
||||
|
||||
class CryptContext(HCRYPTPROV):
|
||||
_type_ = HCRYPTPROV._type_
|
||||
|
||||
def __init__(self, pszContainer=None, pszProvider=None, dwProvType=0, dwFlags=0, retrycreate=False):
|
||||
self.pszContainer = pszContainer
|
||||
self.pszProvider = pszProvider
|
||||
self.dwProvType = dwProvType
|
||||
self.dwFlags = dwFlags
|
||||
self.retrycreate = True
|
||||
#self.value = HCRYPTPROV()
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
self.acquire()
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
self.release()
|
||||
|
||||
def acquire(self):
|
||||
try:
|
||||
return winproxy.CryptAcquireContextW(self, self.pszContainer, self.pszProvider, self.dwProvType, self.dwFlags)
|
||||
except WindowsError as e:
|
||||
if not self.retrycreate:
|
||||
raise
|
||||
return winproxy.CryptAcquireContextW(self, self.pszContainer, self.pszProvider, self.dwProvType, self.dwFlags | CRYPT_NEWKEYSET)
|
||||
|
||||
def release(self):
|
||||
return winproxy.CryptReleaseContext(self, False)
|
||||
@@ -28,16 +28,22 @@ class GenerateInitVector(object):
|
||||
geninitvector = GenerateInitVector()
|
||||
|
||||
|
||||
def encrypt(cert, msg, algo=szOID_RSA_DES_EDE3_CBC, initvector=geninitvector):
|
||||
def encrypt(cert_or_certlist, msg, algo=szOID_RSA_DES_EDE3_CBC, initvector=geninitvector):
|
||||
alg_ident = CRYPT_ALGORITHM_IDENTIFIER()
|
||||
alg_ident.pszObjId = algo
|
||||
# Is 'certs' an iterable ?
|
||||
try:
|
||||
certlist = tuple(cert_or_certlist)
|
||||
except TypeError as e:
|
||||
certlist = (cert_or_certlist,)
|
||||
|
||||
# Set (compute if needed) the IV
|
||||
if initvector is None:
|
||||
alg_ident.Parameters.cbData = 0
|
||||
elif initvector is geninitvector:
|
||||
initvector = initvector.generate_init_vector(algo)
|
||||
if initvector is None:
|
||||
raise ValueError("I Don't know how to generate an <initvector> for <{0}> please provide one (or None)".format(algo))
|
||||
raise ValueError("I don't know how to generate an <initvector> for <{0}> please provide one (or None)".format(algo))
|
||||
initvector_encoded = encode_init_vector(initvector)
|
||||
alg_ident.Parameters = ECRYPT_DATA_BLOB.from_string(initvector_encoded)
|
||||
else:
|
||||
@@ -54,7 +60,7 @@ def encrypt(cert, msg, algo=szOID_RSA_DES_EDE3_CBC, initvector=geninitvector):
|
||||
param.dwFlags = 0
|
||||
param.dwInnerContentType = 0
|
||||
|
||||
certs = (PCERT_CONTEXT * 1)(cert)
|
||||
certs = (PCERT_CONTEXT * len(certlist))(*certlist)
|
||||
#Ask the output buffer size
|
||||
size = DWORD()
|
||||
winproxy.CryptEncryptMessage(param, len(certs), certs, msg, len(msg), None, size)
|
||||
|
||||
@@ -5,7 +5,7 @@ from windows.crypto.helper import ECRYPT_DATA_BLOB
|
||||
from windows.crypto import DEFAULT_ENCODING, EHCERTSTORE
|
||||
|
||||
|
||||
def generate_selfsigned_certificate(name="CN=Testing", prov=None, key_info=None, flags=0, signature_algo=None):
|
||||
def generate_selfsigned_certificate(name="CN=DEFAULT", prov=None, key_info=None, flags=0, signature_algo=None):
|
||||
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)
|
||||
|
||||
@@ -1467,4 +1467,23 @@ REPORT_NO_PRIVATE_KEY = make_flag("REPORT_NO_PRIVATE_KEY", 0x0001)
|
||||
REPORT_NOT_ABLE_TO_EXPORT_PRIVATE_KEY = make_flag("REPORT_NOT_ABLE_TO_EXPORT_PRIVATE_KEY", 0x0002)
|
||||
EXPORT_PRIVATE_KEYS = make_flag("EXPORT_PRIVATE_KEYS", 0x0004)
|
||||
PKCS12_INCLUDE_EXTENDED_PROPERTIES = make_flag("PKCS12_INCLUDE_EXTENDED_PROPERTIES", 0x0010)
|
||||
PKCS12_EXPORT_RESERVED_MASK = make_flag("PKCS12_EXPORT_RESERVED_MASK", 0xffff0000)
|
||||
PKCS12_EXPORT_RESERVED_MASK = make_flag("PKCS12_EXPORT_RESERVED_MASK", 0xffff0000)
|
||||
CERT_SYSTEM_STORE_UNPROTECTED_FLAG = make_flag("CERT_SYSTEM_STORE_UNPROTECTED_FLAG", 0x40000000)
|
||||
CERT_SYSTEM_STORE_LOCATION_MASK = make_flag("CERT_SYSTEM_STORE_LOCATION_MASK", 0x00FF0000)
|
||||
CERT_SYSTEM_STORE_LOCATION_SHIFT = make_flag("CERT_SYSTEM_STORE_LOCATION_SHIFT", 16)
|
||||
CERT_SYSTEM_STORE_CURRENT_USER_ID = make_flag("CERT_SYSTEM_STORE_CURRENT_USER_ID", 1)
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE_ID = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE_ID", 2)
|
||||
CERT_SYSTEM_STORE_CURRENT_SERVICE_ID = make_flag("CERT_SYSTEM_STORE_CURRENT_SERVICE_ID", 4)
|
||||
CERT_SYSTEM_STORE_SERVICES_ID = make_flag("CERT_SYSTEM_STORE_SERVICES_ID", 5)
|
||||
CERT_SYSTEM_STORE_USERS_ID = make_flag("CERT_SYSTEM_STORE_USERS_ID", 6)
|
||||
CERT_SYSTEM_STORE_CURRENT_USER_GROUP_POLICY_ID = make_flag("CERT_SYSTEM_STORE_CURRENT_USER_GROUP_POLICY_ID", 7)
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE_GROUP_POLICY_ID = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE_GROUP_POLICY_ID", 8)
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE_ENTERPRISE_ID = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE_ENTERPRISE_ID", 9)
|
||||
CERT_SYSTEM_STORE_CURRENT_USER = make_flag("CERT_SYSTEM_STORE_CURRENT_USER", ( CERT_SYSTEM_STORE_CURRENT_USER_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE", ( CERT_SYSTEM_STORE_LOCAL_MACHINE_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_CURRENT_SERVICE = make_flag("CERT_SYSTEM_STORE_CURRENT_SERVICE", ( CERT_SYSTEM_STORE_CURRENT_SERVICE_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_SERVICES = make_flag("CERT_SYSTEM_STORE_SERVICES", ( CERT_SYSTEM_STORE_SERVICES_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_USERS = make_flag("CERT_SYSTEM_STORE_USERS", ( CERT_SYSTEM_STORE_USERS_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_CURRENT_USER_GROUP_POLICY = make_flag("CERT_SYSTEM_STORE_CURRENT_USER_GROUP_POLICY", ( CERT_SYSTEM_STORE_CURRENT_USER_GROUP_POLICY_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE_GROUP_POLICY = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE_GROUP_POLICY", ( CERT_SYSTEM_STORE_LOCAL_MACHINE_GROUP_POLICY_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
CERT_SYSTEM_STORE_LOCAL_MACHINE_ENTERPRISE = make_flag("CERT_SYSTEM_STORE_LOCAL_MACHINE_ENTERPRISE", ( CERT_SYSTEM_STORE_LOCAL_MACHINE_ENTERPRISE_ID << CERT_SYSTEM_STORE_LOCATION_SHIFT ))
|
||||
File diff suppressed because one or more lines are too long
@@ -1035,6 +1035,11 @@ def CryptGenKey(hProv, Algid, dwFlags, phKey):
|
||||
return CryptGenKey.ctypes_function(hProv, Algid, dwFlags, phKey)
|
||||
|
||||
|
||||
@Advapi32Proxy('CryptDestroyKey')
|
||||
def CryptDestroyKey(hKey):
|
||||
return CryptDestroyKey.ctypes_function(hKey)
|
||||
|
||||
|
||||
@Advapi32Proxy('CryptAcquireContextA')
|
||||
def CryptAcquireContextA(phProv, pszContainer, pszProvider, dwProvType, dwFlags):
|
||||
return CryptAcquireContextA.ctypes_function(phProv, pszContainer, pszProvider, dwProvType, dwFlags)
|
||||
@@ -1045,6 +1050,11 @@ def CryptAcquireContextW(phProv, pszContainer, pszProvider, dwProvType, dwFlags)
|
||||
return CryptAcquireContextW.ctypes_function(phProv, pszContainer, pszProvider, dwProvType, dwFlags)
|
||||
|
||||
|
||||
@Advapi32Proxy('CryptReleaseContext')
|
||||
def CryptReleaseContext(hProv, dwFlags):
|
||||
return CryptReleaseContext.ctypes_function(hProv, dwFlags)
|
||||
|
||||
|
||||
@Advapi32Proxy('CryptExportKey')
|
||||
def CryptExportKey(hKey, hExpKey, dwBlobType, dwFlags, pbData, pdwDataLen):
|
||||
return CryptExportKey.ctypes_function(hKey, hExpKey, dwBlobType, dwFlags, pbData, pdwDataLen)
|
||||
|
||||
Reference in New Issue
Block a user