# SPDX-FileCopyrightText: 2022 FIT-Connect contributors
#
# SPDX-License-Identifier: EUPL-1.2
import os
from OpenSSL import crypto as openssl_crypto
from cryptography.hazmat.primitives import hashes
from cryptography import x509
# list of directories which contain trusted root CA certificates
TRUSTED_ROOT_CAS = {
"prod": "trusted-root-certificates/v-pki-prod",
"test": "trusted-root-certificates/v-pki-test",
"unit-test": "trusted-root-certificates/unit-tests",
}
REQUIRED_KEY_LENGTH = 4096
def check_key_length(certificate):
if certificate.get_pubkey().bits() < REQUIRED_KEY_LENGTH:
return False
return True
def check_cert_key_usage(cert):
"""check the key usages from the certificates allow for this"""
found_ext = False
allow_signing = False
allow_encryption = False
for i in range(0, cert.get_extension_count()):
if cert.get_extension(i).get_short_name() == b"keyUsage":
found_ext = True
usage = str(cert.get_extension(i))
print("Info: KeyUsage:", usage)
if "Digital Signature" in usage:
allow_signing = True
if "Key Encipherment" in usage:
allow_encryption = True
if not found_ext:
print("Error: KeyUsage Extension not found!")
return False
if not allow_signing:
print("Error: KeyUsage Digital Signature is missing!")
return False
if not allow_encryption:
print("Error: KeyUsage Key Encipherment is missing!")
return False
return True
def x5c_to_cert(x5c):
# RFC coded certificates
# check that the delimiters: "------BEGIN CERTIFICATE------" exist in the string.
if "-----BEGIN CERTIFICATE-----" not in x5c:
x5c = "-----BEGIN CERTIFICATE-----\n" + x5c + "\n-----END CERTIFICATE-----"
return x509.load_pem_x509_certificate(x5c.encode())
def validate_jwk_x5c_chain(jwks, base_cert):
# the generated jwks have to contain the entire certificate chain
# also, the leaf certificate in the x5c field has to be the deposited certificate!
for tmp_jwk in jwks:
is_leaf_certificate = True
for x5c_cert in tmp_jwk.get("x5c"):
cert = x5c_to_cert(x5c_cert)
if is_leaf_certificate:
# the leaf certificate in the x5c chain has to be the certificate with the key!
is_leaf_certificate = False
issuer = cert.issuer
if cert != base_cert:
print('Error! Leaf certificate in jwk("x5c") field is incorrect.')
return False
else:
# each subsequent certificate has to be the one used to certify the previous
if issuer != cert.subject:
print(
"Error, each subsequent certificate in jwk x5c chain "
"has to be the one issuing the previous certificate."
)
print(
f"Issuer {issuer.rfc4514_string()} and subject {cert.subject.rfc4514_string()} don't match!"
)
return False
else:
issuer = cert.issuer
return True
def verify_certificate_algorithms(certificate):
x509_certificate = certificate.to_cryptography()
# TODO - VPKI hat falsche algorithmen
return True
if not x509_certificate.signature_hash_algorithm.__class__ == hashes.SHA512:
print(
"Certificate is using hash algorithm {} but needs {}".format(
x509_certificate.signature_hash_algorithm, hashes.SHA512
)
)
return False
# base_certificate = openssl_crypto.X509.from_cryptography(base_certificate)
if certificate.get_signature_algorithm() != b"sha512WithRSAEncryption":
print(
"Certificate is using signature algorithm {} but needs {}".format(
str(certificate.get_signature_algorithm()), "RSASSA-PSS"
)
)
return False
return True
def check_jwk_key_length(jwks):
for tmp_jwk in jwks:
b64encoded_string = tmp_jwk.get("n")
# fill padding if necessary
b64encoded_string += "=" * ((4 - len(b64encoded_string) % 4) % 4)
key_length = len(b64encoded_string) * 8
if key_length < 4096:
print(
"JWK with id {} has wrong key length. Is {} but should be 4096 at least.".format(
tmp_jwk.get("kid"), len(b64encoded_string)
)
)
return False
return True
def verify_certificate_chain(cert, certificate_chain, environment):
"""Verify that certificates are from DOI CA and verify certificate chain"""
# create list of file paths to trusted CA certificates in PEM format
trusted_root_certs = [
os.path.join(TRUSTED_ROOT_CAS[environment], filename)
for filename in os.listdir(TRUSTED_ROOT_CAS[environment])
]
try:
# Create a certificate store and add trusted certs
store = openssl_crypto.X509Store()
# Add root certs from trusted_root_certs to certificate store
for root_cert_path in trusted_root_certs:
# skip all files other than pem files (e.g. .license files)
extension = os.path.splitext(root_cert_path)[1]
if not extension or extension != ".pem":
continue
# read certificate
root_cert_file = open(root_cert_path, "r")
root_cert_data = root_cert_file.read()
root_cert = openssl_crypto.load_certificate(
openssl_crypto.FILETYPE_PEM, root_cert_data
)
# Add root certificate to X509Store
# WARNING: only add root certificates here!
# `add_cert()` will treat the certificate as ultimately trusted!
# see https://duo.com/labs/research/chain-of-fools#section2
store.add_cert(root_cert)
# Create a certificate context using the store, the certificate and the certificate chain to verify
store_ctx = openssl_crypto.X509StoreContext(
store, cert, chain=certificate_chain
)
# Verify the certificate, returns None if the certificate cannot be
# validated.
# WARNING: `verify_certificate()` will succeed if any of the
# certificates added to the X509Store signed the leaf certificate, even
# if the root didn't sign the intermediate!
if store_ctx.verify_certificate() is None:
return True
except Exception as e:
print("Error", e)
print("Certificate chain: ")
for cert in certificate_chain:
print(get_certificate_issuer_and_subject_string(cert))
return False
# `store_ctx.verify_certificate()` will either return None or raise an exception
raise RuntimeError("Unexpected Error: This code should never be reached")
def get_certificate_issuer_and_subject_string(cert):
subject = str(cert.get_subject()).replace("<X509Name object '", "Subject: ")
subject = "Subject: " + subject.replace("'>", " ")
issuer = str(cert.get_issuer()).replace("<X509Name object '", "Subject: ")
issuer = "Issuer: " + issuer.replace("'>", "")
return subject + issuer