cacert
/
cacert-webdb
9
0
Fork 0
Code Issues Pull Requests 2 Projects Releases 4 Wiki Activity

Refactor and improve statistic script

Browse Source
add-quality-check-script
Jan Dittberner 4 months ago
parent ef68be8b60
commit 52992aad46
1 changed files with 205 additions and 103 deletions
Show Stats Download Patch File Download Diff File

308
scripts/check_db_certificates.py
Unescape Escape View File

@ -1,4 +1,5 @@
#!/usr/bin/env python3
import logging
import os
import re
import typing
@ -18,6 +19,12 @@ class CheckResult(NamedTuple):
def __str__(self):
return f"[{self.code:04d}: {self.name}]"
def __lt__(self, other):
return self.code < other.code
def __gt__(self, other):
return self.code > other.code
CODE_OK = CheckResult("OK", 0)
CODE_FILE_MISSING = CheckResult("file missing", 1000)
@ -40,110 +47,216 @@ SUPPORTED_SIGNATURE_ALGORITHMS = [
]
def check_csr(csr_name: str) -> [CheckResult, typing.Any]:
if not csr_name:
return CODE_EMPTY, None
def load_ca_certificates(root_ca_cert, sub_ca_cert) -> dict[int, x509.Certificate]:
with open(root_ca_cert, "rb") as f:
root_cert = x509.load_pem_x509_certificate(f.read())
if not os.path.isfile(csr_name):
return CODE_FILE_MISSING, None
with open(sub_ca_cert, "rb") as f:
class3_cert = x509.load_pem_x509_certificate(f.read())
with open(csr_name, "rb") as f:
csr_data = f.read()
return {
1: root_cert,
2: class3_cert,
}
if re.search(r"SPKAC = ", csr_data.decode("iso-8859-1")):
return CODE_DEPRECATED_SPKAC, None
try:
csr = x509.load_pem_x509_csr(csr_data)
except Exception as e:
print(e, csr_data)
return CODE_UNSUPPORTED_FORMAT, None
class Counters:
fail = 0
good = 0
good_csr = 0
good_crt = 0
missing_crt = 0
expired_crt = 0
if csr.signature_algorithm_oid not in SUPPORTED_SIGNATURE_ALGORITHMS:
return CODE_UNSUPPORTED_SIGNATURE_ALGORITHM, None
csr_codes: dict[CheckResult, int] = {}
crt_codes: dict[CheckResult, int] = {}
try:
if not csr.is_signature_valid:
return CODE_INVALID_SIGNATURE, None
except Exception as e:
print(e, csr_data)
return CODE_INVALID_SIGNATURE, None
def count_fail(self):
self.fail += 1
public_key = csr.public_key()
def count_good(self):
self.good += 1
if isinstance(public_key, rsa.RSAPublicKey):
if public_key.key_size < 2048:
return CODE_PUBLIC_KEY_TOO_WEAK, None
elif isinstance(public_key, ec.EllipticCurvePublicKey):
if public_key.key_size < 256:
return CODE_PUBLIC_KEY_TOO_WEAK, None
else:
return CODE_UNSUPPORTED_PUBLIC_KEY, None
def count_good_csr(self):
self.good_csr += 1
return CODE_OK, public_key.public_numbers()
def count_good_crt(self):
self.good_crt += 1
def count_missing_crt(self):
self.missing_crt += 1
def check_crt(crt_name: str, ca_certificate: x509.Certificate, public_numbers: typing.Any) -> CheckResult:
if not crt_name:
return CODE_EMPTY
def count_expired_crt(self):
self.expired_crt += 1
if public_numbers is None:
return CODE_CERTIFICATE_FOR_INVALID_CSR
def count_csr(self, csr_code: CheckResult):
self.csr_codes.setdefault(csr_code, 0)
self.csr_codes[csr_code] += 1
if not os.path.isfile(crt_name):
return CODE_FILE_MISSING
def count_crt(self, crt_code: CheckResult):
self.crt_codes.setdefault(crt_code, 0)
self.crt_codes[crt_code] += 1
with open(crt_name, "rb") as f:
crt_data = f.read()
def __str__(self):
return (
f"good CSR and certificate: {self.good}\n"
f"good CSR, issue with certificate: {self.good_csr}\n"
f"good certificate, issue with CSR: {self.good_crt}\n"
f"failed CSR and certificate: {self.fail}\n"
f"missing certificate: {self.missing_crt}\n"
f"expired certificate: {self.expired_crt}\n\nCSR results:\n"
) + "\n".join([f"{code}: {count:d}" for code, count in sorted(self.csr_codes.items())]) + (
"\n\nCertificate results:\n"
) + (
"\n".join([f"{code}: {count:d}" for code, count in sorted(self.crt_codes.items())])
)
class Analyzer:
def __init__(self, logger: logging.Logger, dsn: str, ca_certificates: dict[int, x509.Certificate]):
self.logger = logger
self.engine = create_engine(dsn)
self.ca_certificates = ca_certificates
self.c = Counters()
def analyze(self):
metadata_obj = MetaData()
for table in ("emailcerts", "domaincerts", "orgemailcerts", "orgdomaincerts"):
certs_table = Table(table, metadata_obj, autoload_with=self.engine)
stmt = select(certs_table)
with self.engine.connect() as conn:
for row in conn.execute(stmt):
self.analyze_row(table, row)
def analyze_row(self, table: str, row) -> None:
ca_cert = self.ca_certificates[row.rootcert]
csr_code, public_numbers = self.check_csr(row.csr_name)
self.c.count_csr(csr_code)
crt_code = self.check_crt(row.crt_name, ca_cert, public_numbers)
self.c.count_crt(crt_code)
if csr_code != CODE_OK and crt_code not in (CODE_OK, CODE_CERTIFICATE_IS_EXPIRED):
self.logger.debug(
"%06d %s: %06d -> csr_code: %s, crt_code: %s",
self.c.fail, table, row.id, csr_code, crt_code
)
self.c.count_fail()
return
if csr_code != CODE_OK:
self.c.count_good_crt()
if crt_code == CODE_CERTIFICATE_IS_EXPIRED:
self.c.count_expired_crt()
return
if crt_code == CODE_EMPTY:
self.c.count_missing_crt()
return
if csr_code == CODE_OK and crt_code not in (CODE_OK, CODE_CERTIFICATE_IS_EXPIRED):
self.c.count_good_csr()
return
if crt_code == CODE_CERTIFICATE_IS_EXPIRED:
self.c.count_expired_crt()
self.c.count_good()
def check_csr(self, csr_name: str) -> [CheckResult, typing.Any]:
if not csr_name:
return CODE_EMPTY, None
if not os.path.isfile(csr_name):
return CODE_FILE_MISSING, None
with open(csr_name, "rb") as f:
csr_data = f.read()
if re.search(r"SPKAC = ", csr_data.decode("iso-8859-1")):
return CODE_DEPRECATED_SPKAC, None
try:
csr = x509.load_pem_x509_csr(csr_data)
except Exception as e:
self.logger.debug("unsupported CSR format: %s for\n%s", e, csr_data)
return CODE_UNSUPPORTED_FORMAT, None
if csr.signature_algorithm_oid not in SUPPORTED_SIGNATURE_ALGORITHMS:
return CODE_UNSUPPORTED_SIGNATURE_ALGORITHM, None
try:
if not csr.is_signature_valid:
return CODE_INVALID_SIGNATURE, None
except Exception as e:
self.logger.debug("CSR signature check failed: %s for \n%s", e, csr_data)
return CODE_INVALID_SIGNATURE, None
try:
crt = x509.load_pem_x509_certificate(crt_data)
except ValueError:
return CODE_UNSUPPORTED_FORMAT
public_key = csr.public_key()
if crt.signature_algorithm_oid not in SUPPORTED_SIGNATURE_ALGORITHMS:
return CODE_UNSUPPORTED_SIGNATURE_ALGORITHM
if isinstance(public_key, rsa.RSAPublicKey):
if public_key.key_size < 2048:
return CODE_PUBLIC_KEY_TOO_WEAK, None
elif isinstance(public_key, ec.EllipticCurvePublicKey):
if public_key.key_size < 256:
return CODE_PUBLIC_KEY_TOO_WEAK, None
else:
return CODE_UNSUPPORTED_PUBLIC_KEY, None
try:
public_key = crt.public_key()
except UnsupportedAlgorithm:
return CODE_UNSUPPORTED_PUBLIC_KEY
return CODE_OK, public_key.public_numbers()
if isinstance(public_key, rsa.RSAPublicKey):
if public_key.key_size < 2048:
return CODE_PUBLIC_KEY_TOO_WEAK
elif isinstance(public_key, ec.EllipticCurvePublicKey):
if public_key.key_size < 256:
return CODE_PUBLIC_KEY_TOO_WEAK
else:
return CODE_UNSUPPORTED_PUBLIC_KEY
def check_crt(self, crt_name: str, ca_certificate: x509.Certificate, public_numbers: typing.Any) -> CheckResult:
if not crt_name:
return CODE_EMPTY
if public_key.public_numbers() != public_numbers:
return CODE_CSR_AND_CRT_PUBLIC_KEY_MISMATCH
if not os.path.isfile(crt_name):
return CODE_FILE_MISSING
if crt.not_valid_after_utc < datetime.now(timezone.utc):
return CODE_CERTIFICATE_IS_EXPIRED
with open(crt_name, "rb") as f:
crt_data = f.read()
try:
crt.verify_directly_issued_by(ca_certificate)
except Exception as e:
print(e, crt.issuer, ca_certificate.subject)
return CODE_NOT_SIGNED_BY_EXPECTED_CA_CERTIFICATE
try:
crt = x509.load_pem_x509_certificate(crt_data)
except ValueError:
return CODE_UNSUPPORTED_FORMAT
return CODE_OK
if crt.signature_algorithm_oid not in SUPPORTED_SIGNATURE_ALGORITHMS:
return CODE_UNSUPPORTED_SIGNATURE_ALGORITHM
try:
public_key = crt.public_key()
except UnsupportedAlgorithm:
return CODE_UNSUPPORTED_PUBLIC_KEY
def load_ca_certificates() -> dict[int, x509.Certificate]:
with open("../certs/root_X0F.crt", "rb") as f:
root_cert = x509.load_pem_x509_certificate(f.read())
if isinstance(public_key, rsa.RSAPublicKey):
if public_key.key_size < 2048:
return CODE_PUBLIC_KEY_TOO_WEAK
elif isinstance(public_key, ec.EllipticCurvePublicKey):
if public_key.key_size < 256:
return CODE_PUBLIC_KEY_TOO_WEAK
else:
return CODE_UNSUPPORTED_PUBLIC_KEY
with open("../certs/CAcert_Class3Root_x14E228.crt", "rb") as f:
class3_cert = x509.load_pem_x509_certificate(f.read())
if public_numbers and public_key.public_numbers() != public_numbers:
return CODE_CSR_AND_CRT_PUBLIC_KEY_MISMATCH
return {
1: root_cert,
2: class3_cert,
}
if crt.not_valid_after_utc < datetime.now(timezone.utc):
return CODE_CERTIFICATE_IS_EXPIRED
try:
crt.verify_directly_issued_by(ca_certificate)
except Exception as e:
self.logger.debug("certificate verification failed: %s\n issuer of certificate: %s\n"
" CA certificate: %s", e, crt.issuer,
ca_certificate.subject)
return CODE_NOT_SIGNED_BY_EXPECTED_CA_CERTIFICATE
return CODE_OK
def get_statistics(self):
self.logger.info("Statistics:\n%s", self.c)
def main():
@ -152,36 +265,25 @@ def main():
db_host = os.getenv("DB_HOST", "localhost")
db_port = os.getenv("DB_PORT", "3306")
db_name = os.getenv("DB_NAME", "cacert")
root_ca_cert: str = os.getenv("ROOT_CA_CERTIFICATE", "../www/certs/root_X0F.crt")
sub_ca_cert = os.getenv("SUB_CA_CERTIFICATE", "../www/certs/CAcert_Class3Root_x14E228.crt")
debug = bool(os.getenv("DEBUG", "false"))
ca_certificates = load_ca_certificates()
dsn = (
f"mariadb+mariadbconnector://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"
)
engine = create_engine(dsn)
metadata_obj = MetaData()
fail_counter = 0
good_counter = 0
for table in ("emailcerts", "domaincerts", "orgemailcerts", "orgdomaincerts"):
certs_table = Table(table, metadata_obj, autoload_with=engine)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
stmt = select(certs_table)
if debug:
logger.level = logging.DEBUG
with engine.connect() as conn:
for row in conn.execute(stmt):
csr_code, public_numbers = check_csr(row.csr_name)
ca_cert = ca_certificates[row.rootcert]
crt_code = check_crt(row.crt_name, ca_cert, public_numbers)
analyzer = Analyzer(
logger=logger,
dsn=f"mariadb+mariadbconnector://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}",
ca_certificates=load_ca_certificates(root_ca_cert, sub_ca_cert),
)
if csr_code != CODE_OK or crt_code != CODE_OK:
fail_counter += 1
print(f"{fail_counter:03d} {table}: {row.id:06d} -> csr_code: {csr_code}, crt_code: {crt_code}")
else:
good_counter += 1
analyzer.analyze()
print(f"Good: {good_counter}, Failed: {fail_counter}")
analyzer.get_statistics()
if __name__ == "__main__":

Write Preview
Loading…
Cancel
Save