From fd2134753a37ad96dc091d5f8c7fcb297839b363 Mon Sep 17 00:00:00 2001 From: Kishi85 Date: Sat, 23 Mar 2019 09:46:36 +0100 Subject: [PATCH] tools: cleanup function names and add crypto wrappers --- acertmgr/__init__.py | 12 +++--- acertmgr/authority/v1.py | 31 +++++---------- acertmgr/authority/v2.py | 31 ++++++--------- acertmgr/modes/dns/abstract.py | 7 +--- acertmgr/tools.py | 69 +++++++++++++++++++++++++--------- 5 files changed, 80 insertions(+), 70 deletions(-) diff --git a/acertmgr/__init__.py b/acertmgr/__init__.py index 9d288dc..9abdd3a 100755 --- a/acertmgr/__init__.py +++ b/acertmgr/__init__.py @@ -37,8 +37,8 @@ def create_authority(settings): acc_file = settings['account_key'] if not os.path.isfile(acc_file): print("Account key not found at '{0}'. Creating RSA key.".format(acc_file)) - tools.new_rsa_key(acc_file) - acc_key = tools.read_key(acc_file) + tools.new_account_key(acc_file) + acc_key = tools.read_pem_key(acc_file) authority_module = importlib.import_module("acertmgr.authority.{0}".format(settings["api"])) authority_class = getattr(authority_module, "ACMEAuthority") @@ -67,7 +67,7 @@ def cert_get(settings): key_length = settings['key_length'] if not os.path.isfile(key_file): print("SSL key not found at '{0}'. Creating {1} bit RSA key.".format(key_file, key_length)) - tools.new_rsa_key(key_file, key_length) + tools.new_ssl_key(key_file, key_length) acme = create_authority(settings) @@ -82,13 +82,13 @@ def cert_get(settings): challenge_handlers[domain] = create_challenge_handler(settings['handlers'][domain]) try: - key = tools.read_key(key_file) + key = tools.read_pem_key(key_file) cr = tools.new_cert_request(settings['domainlist'], key) print("Reading account key...") acme.register_account() crt, ca = acme.get_crt_from_csr(cr, settings['domainlist'], challenge_handlers) with io.open(crt_file, "w") as crt_fd: - crt_fd.write(tools.convert_cert_to_pem(crt)) + crt_fd.write(tools.convert_cert_to_pem_str(crt)) # if resulting certificate is valid: store in final location if tools.is_cert_valid(crt_file, 60): @@ -97,7 +97,7 @@ def cert_get(settings): os.chmod(crt_final, stat.S_IREAD) if "static_ca" in settings and not settings['static_ca'] and ca is not None: with io.open(settings['ca_file'], "w") as ca_fd: - ca_fd.write(tools.convert_cert_to_pem(ca)) + ca_fd.write(tools.convert_cert_to_pem_str(ca)) finally: os.remove(csr_file) os.remove(crt_file) diff --git a/acertmgr/authority/v1.py b/acertmgr/authority/v1.py index 9e0da9a..f247e19 100644 --- a/acertmgr/authority/v1.py +++ b/acertmgr/authority/v1.py @@ -12,11 +12,6 @@ import json import re import time -from cryptography import x509 -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import padding - from acertmgr import tools from acertmgr.authority.acme import ACMEAuthority as AbstractACMEAuthority @@ -38,9 +33,9 @@ class ACMEAuthority(AbstractACMEAuthority): header = { "alg": "RS256", "jwk": { - "e": tools.to_json_base64(tools.byte_string_format(numbers.e)), + "e": tools.bytes_to_base64url(tools.number_to_byte_format(numbers.e)), "kty": "RSA", - "n": tools.to_json_base64(tools.byte_string_format(numbers.n)), + "n": tools.bytes_to_base64url(tools.number_to_byte_format(numbers.n)), }, } return header @@ -51,17 +46,14 @@ class ACMEAuthority(AbstractACMEAuthority): # @param payload the message # @return tuple of return code and request answer def _send_signed(self, url, header, payload): - payload64 = tools.to_json_base64(json.dumps(payload).encode('utf8')) + payload64 = tools.bytes_to_base64url(json.dumps(payload).encode('utf8')) protected = copy.deepcopy(header) protected["nonce"] = tools.get_url(self.ca + "/directory").headers['Replay-Nonce'] - protected64 = tools.to_json_base64(json.dumps(protected).encode('utf8')) - # @todo check why this padding is not working - # pad = padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH) - pad = padding.PKCS1v15() - out = self.key.sign('.'.join([protected64, payload64]).encode('utf8'), pad, hashes.SHA256()) + protected64 = tools.bytes_to_base64url(json.dumps(protected).encode('utf8')) + out = tools.signature_of_str(self.key, '.'.join([protected64, payload64])) data = json.dumps({ "header": header, "protected": protected64, - "payload": payload64, "signature": tools.to_json_base64(out), + "payload": payload64, "signature": tools.bytes_to_base64url(out), }) try: resp = tools.get_url(url, data.encode('utf8')) @@ -94,10 +86,8 @@ class ACMEAuthority(AbstractACMEAuthority): # @note algorithm and parts of the code are from acme-tiny def get_crt_from_csr(self, csr, domains, challenge_handlers): header = self._prepare_header() - accountkey_json = json.dumps(header['jwk'], sort_keys=True, separators=(',', ':')) - account_hash = hashes.Hash(hashes.SHA256(), backend=default_backend()) - account_hash.update(accountkey_json.encode('utf8')) - account_thumbprint = tools.to_json_base64(account_hash.finalize()) + account_thumbprint = tools.bytes_to_base64url( + tools.hash_of_str(json.dumps(header['jwk'], sort_keys=True, separators=(',', ':')))) challenges = dict() tokens = dict() @@ -173,15 +163,14 @@ class ACMEAuthority(AbstractACMEAuthority): # get the new certificate print("Signing certificate...") - csr_der = csr.public_bytes(serialization.Encoding.DER) code, result = self._send_signed(self.ca + "/acme/new-cert", header, { "resource": "new-cert", - "csr": tools.to_json_base64(csr_der), + "csr": tools.bytes_to_base64url(tools.convert_csr_to_der_bytes(csr)), }) if code != 201: raise ValueError("Error signing certificate: {0} {1}".format(code, result)) # return signed certificate! print("Certificate signed!") - cert = x509.load_der_x509_certificate(result, default_backend()) + cert = tools.convert_der_bytes_to_cert(result) return cert, tools.download_issuer_ca(cert) diff --git a/acertmgr/authority/v2.py b/acertmgr/authority/v2.py index b91a2b0..7e926c7 100644 --- a/acertmgr/authority/v2.py +++ b/acertmgr/authority/v2.py @@ -11,11 +11,6 @@ import json import re import time -from cryptography import x509 -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import padding - from acertmgr import tools from acertmgr.authority.acme import ACMEAuthority as AbstractACMEAuthority @@ -56,8 +51,8 @@ class ACMEAuthority(AbstractACMEAuthority): "alg": self.algorithm, "jwk": { "kty": "RSA", - "e": tools.to_json_base64(tools.byte_string_format(numbers.e)), - "n": tools.to_json_base64(tools.byte_string_format(numbers.n)), + "e": tools.bytes_to_base64url(tools.number_to_byte_format(numbers.e)), + "n": tools.bytes_to_base64url(tools.number_to_byte_format(numbers.n)), }, } self.account_id = None # will be updated to correct value during account registration @@ -92,7 +87,7 @@ class ACMEAuthority(AbstractACMEAuthority): payload = {} if not protected: protected = {} - payload64 = tools.to_json_base64(json.dumps(payload).encode('utf8')) + payload64 = tools.bytes_to_base64url(json.dumps(payload).encode('utf8')) # Request a new nonce if there is none in cache if not self.nonce: @@ -104,13 +99,12 @@ class ACMEAuthority(AbstractACMEAuthority): protected["alg"] = self.algorithm if self.account_id: protected["kid"] = self.account_id - protected64 = tools.to_json_base64(json.dumps(protected).encode('utf8')) - pad = padding.PKCS1v15() - out = self.key.sign('.'.join([protected64, payload64]).encode('utf8'), pad, hashes.SHA256()) + protected64 = tools.bytes_to_base64url(json.dumps(protected).encode('utf8')) + out = tools.signature_of_str(self.key, '.'.join([protected64, payload64])) data = json.dumps({ "protected": protected64, "payload": payload64, - "signature": tools.to_json_base64(out), + "signature": tools.bytes_to_base64url(out), }) try: return self._request_url(url, data, raw_result) @@ -153,10 +147,8 @@ class ACMEAuthority(AbstractACMEAuthority): # @return the certificate and corresponding ca as a tuple # @note algorithm and parts of the code are from acme-tiny def get_crt_from_csr(self, csr, domains, challenge_handlers): - accountkey_json = json.dumps(self.account_protected['jwk'], sort_keys=True, separators=(',', ':')) - account_hash = hashes.Hash(hashes.SHA256(), backend=default_backend()) - account_hash.update(accountkey_json.encode('utf8')) - account_thumbprint = tools.to_json_base64(account_hash.finalize()) + account_thumbprint = tools.bytes_to_base64url( + tools.hash_of_str(json.dumps(self.account_protected['jwk'], sort_keys=True, separators=(',', ':')))) print("Ordering certificate for {}".format(domains)) identifiers = [{'type': 'dns', 'value': domain} for domain in domains] @@ -242,9 +234,8 @@ class ACMEAuthority(AbstractACMEAuthority): # get the new certificate print("Finalizing certificate") - csr_der = csr.public_bytes(serialization.Encoding.DER) code, finalize, _ = self._request_acme_url(order['finalize'], { - "csr": tools.to_json_base64(csr_der), + "csr": tools.bytes_to_base64url(tools.convert_csr_to_der_bytes(csr)), }) while code < 400 and (finalize.get('status') == 'pending' or finalize.get('status') == 'processing'): time.sleep(5) @@ -261,10 +252,10 @@ class ACMEAuthority(AbstractACMEAuthority): cert_dict = re.match((r'(?P-----BEGIN CERTIFICATE-----[^\-]+-----END CERTIFICATE-----)\n\n' r'(?P-----BEGIN CERTIFICATE-----[^\-]+-----END CERTIFICATE-----)?'), certificate.decode('utf-8'), re.DOTALL).groupdict() - cert = x509.load_pem_x509_certificate(cert_dict['cert'].encode('utf-8'), default_backend()) + cert = tools.convert_pem_str_to_cert(cert_dict['cert']) if cert_dict['ca'] is None: ca = tools.download_issuer_ca(cert) else: - ca = x509.load_pem_x509_certificate(cert_dict['ca'].encode('utf-8'), default_backend()) + ca = tools.convert_pem_str_to_cert(cert_dict['ca']) return cert, ca diff --git a/acertmgr/modes/dns/abstract.py b/acertmgr/modes/dns/abstract.py index ce54d47..61ce2c3 100644 --- a/acertmgr/modes/dns/abstract.py +++ b/acertmgr/modes/dns/abstract.py @@ -9,8 +9,6 @@ import dns.query import dns.resolver import dns.tsigkeyring import dns.update -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes from acertmgr import tools from acertmgr.modes.abstract import AbstractChallengeHandler @@ -40,10 +38,7 @@ class DNSChallengeHandler(AbstractChallengeHandler): @staticmethod def _determine_txtvalue(thumbprint, token): - keyauthorization = "{0}.{1}".format(token, thumbprint) - digest = hashes.Hash(hashes.SHA256(), backend=default_backend()) - digest.update(keyauthorization.encode('utf8')) - return tools.to_json_base64(digest.finalize()) + return tools.bytes_to_base64url(tools.hash_of_str("{0}.{1}".format(token, thumbprint))) def create_challenge(self, domain, thumbprint, token): domain = self._determine_challenge_domain(domain) diff --git a/acertmgr/tools.py b/acertmgr/tools.py index d11c1b2..59b453c 100644 --- a/acertmgr/tools.py +++ b/acertmgr/tools.py @@ -16,7 +16,7 @@ import six from cryptography import x509 from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.x509.oid import NameOID, ExtensionOID try: @@ -39,8 +39,7 @@ def get_url(url, data=None, headers=None): # @return the tuple of dates: (notBefore, notAfter) def get_cert_valid_times(cert_file): with io.open(cert_file, 'r') as f: - cert_data = f.read().encode('utf-8') - cert = x509.load_pem_x509_certificate(cert_data, default_backend()) + cert = convert_pem_str_to_cert(f.read()) return cert.not_valid_before, cert.not_valid_after @@ -85,9 +84,15 @@ def new_cert_request(names, key): return req -# @brief generate a new rsa key +# @brief generate a new ssl key # @param path path where the new key file should be written -def new_rsa_key(path, key_size=4096): +def new_account_key(path, key_size=4096): + return new_ssl_key(path, key_size) + + +# @brief generate a new ssl key +# @param path path where the new key file should be written +def new_ssl_key(path, key_size=4096): private_key = rsa.generate_private_key( public_exponent=65537, key_size=key_size, @@ -106,6 +111,15 @@ def new_rsa_key(path, key_size=4096): print('Warning: Could not set file permissions on {0}!'.format(path)) +# @brief read a key from file +# @param path path to key file +# @return the key in pyopenssl format +def read_pem_key(path): + with io.open(path, 'r') as f: + key_data = f.read().encode('utf-8') + return serialization.load_pem_private_key(key_data, None, default_backend()) + + # @brief download the issuer ca for a given certificate # @param cert certificate data # @returns ca certificate data @@ -124,7 +138,7 @@ def download_issuer_ca(cert): print("Downloading CA certificate from {}".format(ca_issuers)) resp = get_url(ca_issuers) code = resp.getcode() - if code >= 400: + if code >= 400: print("Could not download issuer CA (error {}) for given certificate: {}".format(code, cert)) return None @@ -134,30 +148,51 @@ def download_issuer_ca(cert): # @brief convert certificate to PEM format # @param cert certificate object in pyopenssl format # @return the certificate in PEM format -def convert_cert_to_pem(cert): +def convert_cert_to_pem_str(cert): return cert.public_bytes(serialization.Encoding.PEM).decode('utf8') -# @brief read a key from file -# @param path path to key file -# @return the key in pyopenssl format -def read_key(path): - with io.open(path, 'r') as f: - key_data = f.read().encode('utf-8') - return serialization.load_pem_private_key(key_data, None, default_backend()) +# @brief load a PEM certificate from str +def convert_pem_str_to_cert(certdata): + return x509.load_pem_x509_certificate(certdata.encode('utf8'), default_backend()) + + +# @brief serialize CSR to DER bytes +def convert_csr_to_der_bytes(data): + return data.public_bytes(serialization.Encoding.DER) + + +# @brief load a DER certificate from str +def convert_der_bytes_to_cert(data): + return x509.load_der_x509_certificate(data, default_backend()) + + +# @brief sign string with key +def signature_of_str(key, string): + # @todo check why this padding is not working + # pad = padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH) + pad = padding.PKCS1v15() + return key.sign(string.encode('utf8'), pad, hashes.SHA256()) + + +# @brief hash a string +def hash_of_str(string): + account_hash = hashes.Hash(hashes.SHA256(), backend=default_backend()) + account_hash.update(string.encode('utf8')) + return account_hash.finalize() # @brief helper function to base64 encode for JSON objects -# @param b the string to encode +# @param b the byte-string to encode # @return the encoded string -def to_json_base64(b): +def bytes_to_base64url(b): return base64.urlsafe_b64encode(b).decode('utf8').replace("=", "") # @brief convert numbers to byte-string # @param num number to convert # @return byte-string containing the number -def byte_string_format(num): +def number_to_byte_format(num): n = format(num, 'x') n = "0{0}".format(n) if len(n) % 2 else n return binascii.unhexlify(n)