diff --git a/aia.py b/aia.py index a1a480e..f3923e3 100644 --- a/aia.py +++ b/aia.py @@ -18,6 +18,7 @@ from cryptography import x509 from cryptography.hazmat.primitives.serialization import Encoding from cryptography.hazmat.primitives.serialization import pkcs7 +from cryptography.hazmat.primitives import hashes import certifi @@ -122,6 +123,27 @@ def openssl_get_cert_info(cert_der): return cert_info + +def print_cert(cert, label=None): + if label: + print(label + ":") + if isinstance(cert, cryptography.x509.Certificate): + # cryptography cert + # https://cryptography.io/en/latest/x509/reference/ + print(f" subject: {cert.subject}") + print(f" issuer: {cert.issuer})") + print(f" fingerprint: {cert.fingerprint(hashes.SHA256())}") + return + if isinstance(cert, OpenSSL.crypto.X509): + # pyopenssl cert + print(f" subject: {cert.get_subject()}") + print(f" issuer: {cert.get_issuer()})") + print(f' fingerprint: {cert.digest("sha256")}') + return + raise ValueError("unknown cert type {type(cert)}") + + + class AIASession: def __init__( @@ -364,6 +386,7 @@ def add_trusted_root_cert(self, cert): cert = cert.to_cryptography() assert isinstance(cert, cryptography.x509.Certificate) """ + print_cert(cert, "add_trusted_root_cert cert") if isinstance(cert, cryptography.x509.Certificate): # convert cert from cryptography to pyopenssl # for OpenSSL.crypto.X509StoreContext @@ -387,7 +410,8 @@ def add_trusted_root_cert(self, cert): for c in self._trusted_root_certs: if c.digest("sha256") == cert_digest: return False # cert already was added - logger.debug(f"adding trusted root cert {cert}") + digest_hex = cert_digest # .decode("ascii").replace(":", "").lower() + logger.debug(f"adding trusted root cert {digest_hex}") self._trusted_root_certs.append(cert) return True # cert was added @@ -443,7 +467,7 @@ def print_chain(cert_chain): for (idx, cert) in enumerate(cert_chain): print(f" {idx} subject: {cert.get_subject()}") print(f" issuer: {cert.get_issuer()})") - print(f' fingerprint: {cert.digest("sha1")}') + print(f' fingerprint: {cert.digest("sha256")}') print("verified_cert_chain"); print_chain(verified_cert_chain) print("rest_cert_chain"); print_chain(rest_cert_chain) @@ -486,7 +510,6 @@ def print_chain(cert_chain): # add trusted root certs for cert in self._trusted_root_certs: - # FIXME invalid CA certificate @ ctx.verify_certificate() store.add_cert(cert) # verified_cert_chain[0:-1] certs are verified diff --git a/test/test.py b/test/test.py index d3e83b9..6c87b63 100755 --- a/test/test.py +++ b/test/test.py @@ -75,8 +75,10 @@ def create_cert(name, issuer_cert=None, issuer_key=None, issuer_cert_url=None, i print(f"creating cert {repr(name)}") + # https://cryptography.io/en/latest/x509/tutorial/ # https://cryptography.io/en/latest/x509/reference/#x-509-certificate-builder # https://stackoverflow.com/questions/56285000/python-cryptography-create-a-certificate-signed-by-an-existing-ca-and-export + # https://gist.github.com/major/8ac9f98ae8b07f46b208 is_root = issuer_cert is None @@ -99,6 +101,8 @@ def create_cert(name, issuer_cert=None, issuer_key=None, issuer_cert_url=None, i issuer_name = subject_name if is_root else issuer_cert.subject + issuer_key = key if is_root else issuer_key + cert = x509.CertificateBuilder() cert = cert.subject_name(subject_name) @@ -108,12 +112,46 @@ def create_cert(name, issuer_cert=None, issuer_key=None, issuer_cert_url=None, i cert = cert.not_valid_before(datetime.datetime.utcnow()) cert = cert.not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650)) - # FIXME invalid CA certificate @ ctx.verify_certificate() + cert = cert.add_extension( + x509.SubjectKeyIdentifier.from_public_key(key.public_key()), + critical=False, + ) # https://stackoverflow.com/a/72320618/10440128 - #if not is_leaf: # no. certificate signature failure - if is_root: + #if is_root: # no. invalid CA certificate @ cert1 + + if not is_leaf: cert = cert.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + cert = cert.add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + else: + cert = cert.add_extension( + x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier( + issuer_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value + ), + critical=False, + ) + + if is_leaf: + cert = cert.add_extension( + x509.ExtendedKeyUsage([ + x509.ExtendedKeyUsageOID.CLIENT_AUTH, + x509.ExtendedKeyUsageOID.SERVER_AUTH, + ]), + critical=False, + ) if issuer_cert_url: # add AIA extension @@ -126,7 +164,9 @@ def create_cert(name, issuer_cert=None, issuer_key=None, issuer_cert_url=None, i ), ]), critical=False) - cert = cert.sign(key, hashes.SHA256(), default_backend()) + # no. certificate signature failure + #cert = cert.sign(key, hashes.SHA256(), default_backend()) + cert = cert.sign(issuer_key, hashes.SHA256(), default_backend()) return cert, key @@ -178,8 +218,6 @@ def run_test(tmpdir): # create certs # TODO refactor ... create_cert_chain - # FIXME invalid CA certificate @ ctx.verify_certificate() - cert0, key0 = create_cert("root cert") with open(f"{server_root}/cert0", "wb") as f: # PEM format @@ -298,33 +336,31 @@ def handle_exit(): print("creating aia_session") aia_session = aia.AIASession() - def print_chain(cert_chain): - if not cert_chain: - print(" (empty)") - return - for (idx, cert) in enumerate(cert_chain): - print(f" {idx} subject: {cert.get_subject()}") - print(f" issuer: {cert.get_issuer()})") - print(f' fingerprint: {cert.digest("sha1")}') - - def print_cert(cert, label=None): + def print_cert(cert, label=None, indent=""): if label: - print(label + ":") + print(indent + label + ":") if isinstance(cert, cryptography.x509.Certificate): # cryptography cert # https://cryptography.io/en/latest/x509/reference/ - print(f" subject: {cert.subject}") - print(f" issuer: {cert.issuer})") - print(f" fingerprint: {cert.fingerprint(hashes.SHA256())}") + print(indent + f" subject: {cert.subject}") + print(indent + f" issuer: {cert.issuer})") + print(indent + f" fingerprint: {cert.fingerprint(hashes.SHA256())}") return if isinstance(cert, OpenSSL.crypto.X509): # pyopenssl cert - print(f" subject: {cert.get_subject()}") - print(f" issuer: {cert.get_issuer()})") - print(f' fingerprint: {cert.digest("sha256")}') + print(indent + f" subject: {cert.get_subject()}") + print(indent + f" issuer: {cert.get_issuer()})") + print(indent + f' fingerprint: {cert.digest("sha256")}') return raise ValueError("unknown cert type {type(cert)}") + def print_chain(cert_chain): + if not cert_chain: + print(" (empty)") + return + for (idx, cert) in enumerate(cert_chain): + print_cert(cert, f"cert {idx}", " ") + print("aia_session.aia_chase ...") print("-" * 80) @@ -365,17 +401,24 @@ def print_cert(cert, label=None): test_name = "aia_session.add_trusted_root_cert with non-root cert" print(f"{test_name} ...") + print_cert(cert1, "cert1") try: aia_session.add_trusted_root_cert(cert1) #raise ValueError("must be a CA cert") + #raise ValueError("must be a self-signed cert") except ValueError as exc: - assert str(exc) == "must be a CA cert" + expected_errors = [ + "must be a CA cert", + "must be a self-signed cert", + ] + assert str(exc) in expected_errors print(f"{test_name} ok") print("-" * 80) test_name = "aia_session.add_trusted_root_cert" print(f"{test_name} ...") + print_cert(cert0, "cert0") assert aia_session.add_trusted_root_cert(cert0) == True assert aia_session.add_trusted_root_cert(cert0) == False # already exists print(f"{test_name} ok") @@ -398,9 +441,15 @@ def print_cert(cert, label=None): print("FIXME got unexpected exception:") print("exc.args", exc.args) print("exc.certificate", exc.certificate) + print_cert(exc.certificate, "exc.certificate") print("exc.errors", exc.errors) print("exc str", str(exc)) print("exc dir", dir(exc)) + cert_path = f"/run/user/{os.getuid()}/python-aia-invalid-ca-cert.pem" + print("writing", cert_path) + cert_pem = OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, exc.certificate) + with open(cert_path, "wb") as f: + f.write(cert_pem) raise """ FIXME got unexpected exception: