Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2"""Certificate utilities for acme2certifier"""
4import base64
5import logging
6from typing import List, Tuple, Optional, Set
7from cryptography import x509
8from cryptography.hazmat.backends import default_backend
9from cryptography.hazmat.primitives import serialization, hashes
10from cryptography.hazmat.primitives.serialization.pkcs7 import (
11 load_pem_pkcs7_certificates,
12 load_der_pkcs7_certificates,
13)
14from cryptography.x509 import load_pem_x509_certificate, ocsp
15from cryptography.hazmat.primitives.asymmetric import (
16 dsa,
17 ec,
18 ed25519,
19 ed448,
20 padding,
21 rsa,
22)
23from .encoding import (
24 convert_string_to_byte,
25 convert_byte_to_string,
26 build_pem_file,
27 b64_url_recode,
28 b64_decode,
29)
30from .datetime_utils import date_to_uts_utc
31from .global_variables import CONFIGURATION_ERROR_DETAIL
32from pyasn1.codec.der import decoder
33from pyasn1.type import univ
34from pyasn1_modules import rfc5280
36_OID_SKI = "2.5.29.14"
37_OID_AKI = "2.5.29.35"
38# id-pe-acmeIdentifier (RFC 8737)
39ACME_TLS_ALPN_OID = x509.ObjectIdentifier("1.3.6.1.5.5.7.1.31")
40ACME_TLS_ALPN_PROTOCOL = "acme-tls/1"
43def _cert_pem_to_der(logger: logging.Logger, certificate: str) -> bytes:
44 """Convert certificate input to DER bytes without parsing extensions."""
45 pem_data = convert_string_to_byte(
46 build_pem_file(logger, None, b64_url_recode(logger, certificate), True)
47 )
48 # Prefer cryptography load for PEM framing only; public_bytes touches extensions
49 # on some versions, so strip PEM manually.
50 lines = convert_byte_to_string(pem_data).strip().splitlines()
51 b64 = "".join(line for line in lines if not line.startswith("-----"))
52 return base64.b64decode(b64)
55def _cert_extension_raw_get(
56 logger: logging.Logger, certificate: str, oid: str
57) -> Optional[bytes]:
58 """Return extnValue (OCTET STRING contents) for oid, or None."""
59 logger.debug("_cert_extension_raw_get(%s)", oid)
60 der = _cert_pem_to_der(logger, certificate)
61 cert_asn1, _ = decoder.decode(der, asn1Spec=rfc5280.Certificate())
62 extensions = cert_asn1["tbsCertificate"]["extensions"]
63 if extensions is None or not extensions.hasValue():
64 return None
65 for ext in extensions:
66 if str(ext["extnID"]) == oid:
67 return bytes(ext["extnValue"])
68 return None
71def _cert_aki_asn1_get(logger: logging.Logger, certificate: str) -> Optional[str]:
72 """Get AKI keyIdentifier as hex via ASN.1 (tolerates illegal BasicConstraints)."""
73 logger.debug("_cert_aki_asn1_get()")
74 extn_value = _cert_extension_raw_get(logger, certificate, _OID_AKI)
75 if not extn_value:
76 logger.warning("No AKI found in certificate")
77 return None
78 aki, _ = decoder.decode(extn_value, asn1Spec=rfc5280.AuthorityKeyIdentifier())
79 if not aki["keyIdentifier"].isValue:
80 logger.warning("AKI extension present but keyIdentifier missing")
81 return None
82 aki_hex = bytes(aki["keyIdentifier"]).hex()
83 logger.debug("_cert_aki_asn1_get() ended with: %s", aki_hex)
84 return aki_hex
87def cert_aki_get(logger: logging.Logger, certificate: str) -> str:
88 """get subject key identifier from certificate"""
89 logger.debug("Helper.cert_ski_get()")
91 cert = cert_load(logger, certificate, recode=True)
92 try:
93 aki = cert.extensions.get_extension_for_oid(x509.OID_AUTHORITY_KEY_IDENTIFIER)
94 aki_value = aki.value.key_identifier.hex()
95 except Exception as _err:
96 logger.error(
97 "Error while getting AKI from certificate: %s. Fallback to ASN.1 method",
98 _err,
99 )
100 aki_value = _cert_aki_asn1_get(logger, certificate)
102 logger.debug("cert_aki_get() ended with: %s", aki_value)
103 return aki_value
106def _cert_ski_asn1_get(logger: logging.Logger, certificate: str) -> Optional[str]:
107 """Get SKI as hex via ASN.1 (tolerates illegal BasicConstraints)."""
108 logger.debug("_cert_ski_asn1_get()")
109 extn_value = _cert_extension_raw_get(logger, certificate, _OID_SKI)
110 if not extn_value:
111 logger.warning("No SKI found in certificate")
112 return None
113 ski, _ = decoder.decode(extn_value, asn1Spec=univ.OctetString())
114 ski_hex = bytes(ski).hex()
115 logger.debug("_cert_ski_asn1_get() ended with: %s", ski_hex)
116 return ski_hex
119def cert_load(
120 logger: logging.Logger, certificate: str, recode: bool
121) -> x509.Certificate:
122 """load certificate object from pem _Format"""
123 logger.debug("Helper.cert_load(%s)", recode)
125 if recode:
126 pem_data = convert_string_to_byte(
127 build_pem_file(logger, None, b64_url_recode(logger, certificate), True)
128 )
129 else:
130 pem_data = convert_string_to_byte(certificate)
131 cert = x509.load_pem_x509_certificate(pem_data, default_backend())
133 return cert
136def cert_dates_get(logger: logging.Logger, certificate: str) -> Tuple[int, int]:
137 """get date number form certificate"""
138 logger.debug("Helper.cert_dates_get()")
140 issue_date = 0
141 expiration_date = 0
142 try:
143 cert = cert_load(logger, certificate, recode=True)
144 issue_date = date_to_uts_utc(
145 cert.not_valid_before_utc, _tformat="%Y-%m-%d %H:%M:%S"
146 )
147 expiration_date = date_to_uts_utc(
148 cert.not_valid_after_utc, _tformat="%Y-%m-%d %H:%M:%S"
149 )
150 except Exception as err:
151 logger.debug(
152 "Error while getting dates from certificate. Fallback to deprecated method: %s",
153 err,
154 )
155 try:
156 issue_date = date_to_uts_utc(
157 cert.not_valid_before, _tformat="%Y-%m-%d %H:%M:%S"
158 )
159 expiration_date = date_to_uts_utc(
160 cert.not_valid_after, _tformat="%Y-%m-%d %H:%M:%S"
161 )
162 except Exception:
163 logger.error("Error while getting dates from certificate: %s", err)
164 issue_date = 0
165 expiration_date = 0
167 logger.debug("cert_dates_get() ended with: %s/%s", issue_date, expiration_date)
168 return (issue_date, expiration_date)
171def cert_cn_get(logger: logging.Logger, certificate: str) -> str:
172 """get cn from certificate"""
173 logger.debug("Helper.cert_cn_get()")
175 cert = cert_load(logger, certificate, recode=True)
176 # get subject and look for common name
177 subject = cert.subject
178 result = None
179 for attr in subject:
180 if attr.oid == x509.NameOID.COMMON_NAME:
181 result = attr.value
182 break
183 logger.debug("Helper.cert_cn_get() ended with: %s", result)
184 return result
187def cert_der2pem(der_cert: bytes) -> str:
188 """convert certificate der to pem"""
189 cert = x509.load_der_x509_certificate(der_cert)
190 pem_cert = cert.public_bytes(serialization.Encoding.PEM)
191 return pem_cert
194def cert_issuer_get(logger: logging.Logger, certificate: str) -> str:
195 """get certificate issuer from certificate"""
196 logger.debug("Helper.cert_issuer_get()")
198 cert = cert_load(logger, certificate, recode=True)
199 result = cert.issuer.rfc4514_string()
200 logger.debug("Helper.cert_issuer_get() ended with: %s", result)
201 return result
204def cert_pem2der(pem_cert: str) -> bytes:
205 """convert certificate pem to der"""
206 cert = x509.load_pem_x509_certificate(pem_cert.encode(), default_backend())
207 der_cert = cert.public_bytes(serialization.Encoding.DER)
208 return der_cert
211def cert_pubkey_get(logger: logging.Logger, certificate=str) -> str:
212 """get public key from certificate"""
213 logger.debug("Helper.cert_pubkey_get()")
214 cert = cert_load(logger, certificate, recode=False)
215 public_key = cert.public_key()
216 pubkey_str = public_key.public_bytes(
217 encoding=serialization.Encoding.PEM,
218 format=serialization.PublicFormat.SubjectPublicKeyInfo,
219 )
220 logger.debug("Helper.cert_pubkey_get() ended with: %s", pubkey_str)
221 return convert_byte_to_string(pubkey_str)
224def cert_san_get(
225 logger: logging.Logger, certificate: str, recode: bool = True
226) -> List[str]:
227 """get subject alternate names from certificate"""
228 logger.debug("Helper.cert_san_get(%s)", recode)
230 cert = cert_load(logger, certificate, recode=recode)
231 sans = []
232 try:
233 ext = cert.extensions.get_extension_for_oid(x509.OID_SUBJECT_ALTERNATIVE_NAME)
234 sans_list = ext.value.get_values_for_type(x509.DNSName)
235 for san in sans_list:
236 sans.append(f"DNS:{san}")
237 sans_list = ext.value.get_values_for_type(x509.IPAddress)
238 for san in sans_list:
239 sans.append(f"IP:{san}")
240 sans_list = ext.value.get_values_for_type(x509.RFC822Name)
241 for san in sans_list:
242 sans.append(f"EMAIL:{san}")
243 except Exception as err:
244 logger.error("Error while getting SANs from certificate: %s", err)
246 logger.debug("Helper.cert_san_get() ended")
247 return sans
250def cert_bound_names_get(
251 logger: logging.Logger,
252 certificate: str,
253 recode: bool = True,
254 email_identifier_rewrite: bool = False,
255) -> Set[Tuple[str, str]]:
256 """Return normalized (type, value) pairs from certificate SANs and subject CN."""
257 from .csr import _cn_bound_type, _normalize_bound_name
259 logger.debug(
260 "Helper.cert_bound_names_get(email_identifier_rewrite=%s)",
261 email_identifier_rewrite,
262 )
263 names: Set[Tuple[str, str]] = set()
264 if not certificate:
265 logger.debug("Helper.cert_bound_names_get() ended with: %s", names)
266 return names
268 for san in cert_san_get(logger, certificate, recode=recode):
269 try:
270 san_type, san_value = san.split(":", 1)
271 except ValueError as err:
272 logger.error("Error while splitting SAN %s: %s", san, err)
273 continue
274 normalized = _normalize_bound_name(
275 san_type, san_value, email_identifier_rewrite=email_identifier_rewrite
276 )
277 if normalized:
278 names.add(normalized)
280 cn = cert_cn_get(logger, certificate)
281 if cn:
282 normalized = _normalize_bound_name(
283 _cn_bound_type(cn),
284 cn,
285 email_identifier_rewrite=email_identifier_rewrite,
286 )
287 if normalized:
288 names.add(normalized)
290 logger.debug("Helper.cert_bound_names_get() ended with: %s", names)
291 return names
294def cert_ski_get(logger: logging.Logger, certificate: str) -> str:
295 """get subject key identifier from certificate"""
296 logger.debug("Helper.cert_ski_get()")
298 cert = cert_load(logger, certificate, recode=True)
299 try:
300 ski = cert.extensions.get_extension_for_oid(x509.OID_SUBJECT_KEY_IDENTIFIER)
301 ski_value = ski.value.digest.hex()
302 except Exception as err:
303 logger.error("Error while getting the SKI: %s. Fallback to ASN.1 method", err)
304 ski_value = _cert_ski_asn1_get(logger, certificate)
306 logger.debug("Helper.cert_ski_get() ended with: %s", ski_value)
307 return ski_value
310def cert_extensions_get(logger: logging.Logger, certificate: str, recode: bool = True):
311 """get extenstions from certificate certificate"""
312 logger.debug("Helper.cert_extensions_get()")
314 cert = cert_load(logger, certificate, recode=recode)
315 extension_list = []
316 for extension in cert.extensions:
317 extension_list.append(
318 convert_byte_to_string(base64.b64encode(extension.value.public_bytes()))
319 )
321 logger.debug("Helper.cert_extensions_get() ended with: %s", extension_list)
322 return extension_list
325def cert_acme_tls_alpn_extension_ok(
326 logger: logging.Logger,
327 certificate: str,
328 sha256_digest_hex: str,
329 recode: bool = False,
330) -> bool:
331 """Return True when cert has a critical id-pe-acmeIdentifier with expected digest.
333 RFC 8737 §3: extension OID 1.3.6.1.5.5.7.1.31 MUST be critical and contain an
334 OCTET STRING of the SHA-256 digest of the key authorization.
335 """
336 logger.debug("Helper.cert_acme_tls_alpn_extension_ok()")
337 try:
338 cert = cert_load(logger, certificate, recode=recode)
339 extension = cert.extensions.get_extension_for_oid(ACME_TLS_ALPN_OID)
340 except x509.ExtensionNotFound:
341 logger.warning("tls-alpn-01: id-pe-acmeIdentifier extension not found")
342 return False
343 except Exception as err:
344 logger.warning(
345 "tls-alpn-01: failed to parse ACME identifier extension: %s", err
346 )
347 return False
349 if not extension.critical:
350 logger.warning("tls-alpn-01: id-pe-acmeIdentifier extension is not critical")
351 return False
353 try:
354 digest = bytes.fromhex(sha256_digest_hex)
355 except ValueError:
356 logger.warning("tls-alpn-01: invalid expected digest hex")
357 return False
358 if len(digest) != 32:
359 logger.warning("tls-alpn-01: expected digest must be 32 bytes")
360 return False
362 # DER OCTET STRING: tag 0x04, length 0x20, then digest
363 expected_der = b"\x04\x20" + digest
364 raw = getattr(extension.value, "value", None)
365 if raw is None:
366 try:
367 raw = extension.value.public_bytes()
368 except Exception as err:
369 logger.warning(
370 "tls-alpn-01: cannot read ACME identifier extension value: %s", err
371 )
372 return False
374 if raw not in (expected_der, digest):
375 logger.warning("tls-alpn-01: ACME identifier extension value mismatch")
376 return False
378 logger.debug("Helper.cert_acme_tls_alpn_extension_ok() ended with: True")
379 return True
382def cert_serial_get(logger: logging.Logger, certificate: str, hexformat: bool = False):
383 """get serial number form certificate"""
384 logger.debug("Helper.cert_serial_get()")
385 cert = cert_load(logger, certificate, recode=True)
386 if hexformat:
387 serial_number = f"{cert.serial_number:x}"
388 # add leading zero if needed
389 serial_number = serial_number.zfill(len(serial_number) + len(serial_number) % 2)
390 else:
391 serial_number = cert.serial_number
392 logger.debug("Helper.cert_serial_get() ended with: %s", serial_number)
393 return serial_number
396def pembundle_to_list(logger: logging.Logger, pem_bundle: str) -> List[str]:
397 """split pem bundle into a list of certificates"""
398 logger.debug("Helper.pembundle_to_list()")
399 cert_list = []
400 pem_data = ""
401 if "-----BEGIN CERTIFICATE-----" in pem_bundle:
402 for line in pem_bundle.splitlines():
403 line = line.strip()
404 if line.startswith("-----BEGIN CERTIFICATE-----") and pem_data:
405 cert_list.append(pem_data)
406 pem_data = ""
407 pem_data += line + "\n"
408 if pem_data:
409 cert_list.append(pem_data)
410 logger.debug("Helper.pembundle_to_list() returned %s certificates", cert_list)
411 return cert_list
414def _cert_sha256_fingerprint(cert: x509.Certificate) -> str:
415 """SHA-256 fingerprint as lowercase hex (no colons)."""
416 return cert.fingerprint(hashes.SHA256()).hex()
419def _pem_fingerprint(
420 logger: logging.Logger, pem_cert: str
421) -> Tuple[Optional[str], Optional[str]]:
422 """Return ``(error, fingerprint)`` for a single PEM certificate."""
423 try:
424 cert = cert_load(logger, pem_cert, recode=False)
425 except Exception as err_:
426 logger.error("Failed to parse certificate in chain: %s", err_)
427 return (
428 f"{CONFIGURATION_ERROR_DETAIL}: Failed to parse certificate chain",
429 None,
430 )
431 return None, _cert_sha256_fingerprint(cert)
434def cert_chain_skip(
435 logger: logging.Logger,
436 pem_bundle: Optional[str],
437 skip_list: Optional[List[str]],
438 link_check: bool = True,
439) -> Tuple[Optional[str], Optional[str]]:
440 """Drop certificates whose SHA-256 fingerprint is in *skip_list*."""
441 logger.debug("Helper.cert_chain_skip()")
442 if not pem_bundle:
443 logger.debug("Helper.cert_chain_skip() ended (empty bundle)")
444 return None, pem_bundle
445 if not skip_list:
446 logger.debug("Helper.cert_chain_skip() ended (skip list empty)")
447 return None, pem_bundle
449 pem_list = pembundle_to_list(logger, pem_bundle)
450 if not pem_list:
451 logger.error("cert_chain_skip_list is set but the bundle is not a PEM chain")
452 return (
453 f"{CONFIGURATION_ERROR_DETAIL}: Failed to parse certificate chain",
454 None,
455 )
457 skip_set = set(skip_list)
458 error, leaf_fp = _pem_fingerprint(logger, pem_list[0])
459 if error:
460 return error, None
461 if leaf_fp in skip_set:
462 logger.error("cert_chain_skip_list must not include the end-entity certificate")
463 return (
464 f"{CONFIGURATION_ERROR_DETAIL}: "
465 "cert_chain_skip_list includes the end-entity certificate",
466 None,
467 )
469 kept = [pem_list[0]]
470 for pem_cert in pem_list[1:]:
471 error, fingerprint = _pem_fingerprint(logger, pem_cert)
472 if error:
473 return error, None
474 if fingerprint not in skip_set:
475 kept.append(pem_cert)
477 if len(kept) != len(pem_list):
478 error, certs = _certs_from_pems(logger, kept)
479 if error:
480 return error, None
481 error = _chain_links_error(logger, certs, 0, link_check=link_check)
482 if error:
483 return error, None
485 result = "".join(kept)
486 logger.debug(
487 "Helper.cert_chain_skip() ended with %d of %d certificates kept",
488 len(kept),
489 len(pem_list),
490 )
491 return None, result
494def _cert_certifies(issuer: x509.Certificate, subject: x509.Certificate) -> bool:
495 """Return True when *issuer* signed *subject* (RFC 8555 chain link)."""
496 if subject.issuer != issuer.subject:
497 return False
498 public_key = issuer.public_key()
499 try:
500 if isinstance(public_key, rsa.RSAPublicKey):
501 hash_alg = subject.signature_hash_algorithm
502 if hash_alg is None:
503 return False
504 public_key.verify(
505 subject.signature,
506 subject.tbs_certificate_bytes,
507 padding.PKCS1v15(),
508 hash_alg,
509 )
510 return True
511 if isinstance(public_key, ec.EllipticCurvePublicKey):
512 hash_alg = subject.signature_hash_algorithm
513 if hash_alg is None:
514 return False
515 public_key.verify(
516 subject.signature,
517 subject.tbs_certificate_bytes,
518 ec.ECDSA(hash_alg),
519 )
520 return True
521 if isinstance(public_key, dsa.DSAPublicKey):
522 hash_alg = subject.signature_hash_algorithm
523 if hash_alg is None:
524 return False
525 public_key.verify(
526 subject.signature, subject.tbs_certificate_bytes, hash_alg
527 )
528 return True
529 if isinstance(public_key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
530 public_key.verify(subject.signature, subject.tbs_certificate_bytes)
531 return True
532 except Exception:
533 return False
534 return False
537def _certs_from_pems(
538 logger: logging.Logger, pem_list: List[str]
539) -> Tuple[Optional[str], List[x509.Certificate]]:
540 """Load PEM strings with ``cert_load()``. Fail closed on parse errors."""
541 certs: List[x509.Certificate] = []
542 for pem_cert in pem_list:
543 try:
544 certs.append(cert_load(logger, pem_cert, recode=False))
545 except Exception as err_:
546 logger.error("Failed to parse certificate in chain: %s", err_)
547 return (
548 f"{CONFIGURATION_ERROR_DETAIL}: Failed to parse certificate chain",
549 [],
550 )
551 return None, certs
554def _parse_pem_bundle(
555 logger: logging.Logger, pem_bundle: str
556) -> Tuple[Optional[str], List[str], List[x509.Certificate]]:
557 """Split *pem_bundle* and load each PEM. Fail closed on an empty or invalid chain."""
558 pems = pembundle_to_list(logger, pem_bundle)
559 if not pems:
560 logger.error("cert_chain_append is set but the bundle is not a PEM chain")
561 return (
562 f"{CONFIGURATION_ERROR_DETAIL}: Failed to parse certificate chain",
563 [],
564 [],
565 )
566 error, certs = _certs_from_pems(logger, pems)
567 if error:
568 return error, [], []
569 return None, pems, certs
572def _append_entry_check(
573 logger: logging.Logger,
574 cert: x509.Certificate,
575 leaf_fp: str,
576 seen_fps: Set[str],
577) -> Tuple[Optional[str], Optional[str]]:
578 """Return ``(error, fingerprint)``. Fingerprint is set only when the cert may be appended."""
579 logger.debug("Helper._append_entry_check()")
580 fingerprint = _cert_sha256_fingerprint(cert)
581 if fingerprint == leaf_fp:
582 logger.error("cert_chain_append must not include the end-entity certificate")
583 return (
584 f"{CONFIGURATION_ERROR_DETAIL}: "
585 "cert_chain_append includes the end-entity certificate",
586 None,
587 )
588 if fingerprint in seen_fps:
589 logger.error("cert_chain_append includes a certificate already in the chain")
590 return (
591 f"{CONFIGURATION_ERROR_DETAIL}: "
592 "cert_chain_append includes a certificate already in the chain",
593 None,
594 )
595 logger.debug("Helper._append_entry_check() ended")
596 return None, fingerprint
599def _chain_links_error(
600 logger: logging.Logger,
601 certs: List[x509.Certificate],
602 start: int,
603 *,
604 link_check: bool = True,
605) -> Optional[str]:
606 """Fail if any certificate from *start* does not certify the previous one."""
607 logger.debug("Helper._chain_links_error()")
608 for idx in range(start, len(certs) - 1):
609 prev = certs[idx]
610 nxt = certs[idx + 1]
611 issuer_name_match = prev.issuer == nxt.subject
612 certifies = _cert_certifies(nxt, prev)
613 if not certifies:
614 if issuer_name_match:
615 reason = "issuer name matches but signature verification failed"
616 else:
617 reason = (
618 f"previous issuer {prev.issuer.rfc4514_string()} != "
619 f"following subject {nxt.subject.rfc4514_string()}"
620 )
621 if not link_check:
622 logger.warning(
623 "cert_chain_link_check is False; certificate "
624 "does not certify the previous one (%s)",
625 reason,
626 )
627 continue
628 logger.error(
629 "certificate %d does not certify the previous one (%s)",
630 idx + 1,
631 reason,
632 )
633 return (
634 f"{CONFIGURATION_ERROR_DETAIL}: "
635 "certificate does not certify the previous one "
636 f"({reason})"
637 )
638 logger.debug("Helper._chain_links_error() ended")
639 return None
642def _extend_chain(
643 logger: logging.Logger,
644 pems: List[str],
645 certs: List[x509.Certificate],
646 pem_list: List[str],
647 *,
648 link_check: bool = True,
649) -> Optional[str]:
650 """Parse *pem_list*, reject leaf/duplicates, append in place, check RFC 8555 links."""
651 logger.debug("Helper._extend_chain()")
653 join_at = len(certs) - 1
654 leaf_fp = _cert_sha256_fingerprint(certs[0])
655 seen_fps = {_cert_sha256_fingerprint(cert) for cert in certs}
656 error, appended = _certs_from_pems(logger, pem_list)
657 if error:
658 return error
659 for pem_cert, parsed in zip(pem_list, appended):
660 error, fingerprint = _append_entry_check(logger, parsed, leaf_fp, seen_fps)
661 if error:
662 return error
663 seen_fps.add(fingerprint)
664 pems.append(pem_cert)
665 certs.append(parsed)
667 logger.debug("Helper._extend_chain() ended with: %s", pems)
668 return _chain_links_error(logger, certs, join_at, link_check=link_check)
671def cert_chain_append(
672 logger: logging.Logger,
673 pem_bundle: Optional[str],
674 pem_list: Optional[List[str]],
675 *,
676 link_check: bool = True,
677) -> Tuple[Optional[str], Optional[str]]:
678 """Append PEM certificates to *pem_bundle* and check RFC 8555 chain links."""
679 logger.debug("Helper.cert_chain_append()")
681 if not pem_bundle:
682 logger.debug("Helper.cert_chain_append() ended (empty bundle)")
683 return None, pem_bundle
684 if not pem_list:
685 logger.debug("Helper.cert_chain_append() ended (append list empty)")
686 return None, pem_bundle
688 error, pems, certs = _parse_pem_bundle(logger, pem_bundle)
689 if error:
690 return error, None
692 error = _extend_chain(logger, pems, certs, pem_list, link_check=link_check)
693 if error:
694 return error, None
696 result = "".join(pems)
697 logger.debug(
698 "Helper.cert_chain_append() ended with %d certificates",
699 len(pems),
700 )
701 logger.debug("Helper.cert_chain_append() ended with: %s", result)
702 return None, result
705def certid_asn1_get(logger: logging.Logger, cert_pem: str, issuer_pem: str) -> str:
706 """get renewal information from certificate"""
707 logger.debug("Helper.certid_asn1_get()")
709 cert = load_pem_x509_certificate(convert_string_to_byte(cert_pem))
710 issuer = load_pem_x509_certificate(convert_string_to_byte(issuer_pem))
712 builder = ocsp.OCSPRequestBuilder()
713 builder = builder.add_certificate(cert, issuer, hashes.SHA256())
714 ocsprequest = builder.build()
715 ocsprequest_hex = ocsprequest.public_bytes(serialization.Encoding.DER).hex()
717 # this is ugly but i did not find a better way to do this
718 _header, certid_hex = ocsprequest_hex.split("0420", 1)
720 return certid_hex
723def certid_hex_get(logger: logging.Logger, renewal_info: str) -> Tuple[str, str]:
724 """get certid in hex from renewal_info field"""
725 logger.debug("Helper.certid_hex_get()")
727 renewal_info_b64 = b64_url_recode(logger, renewal_info)
728 renewal_info_hex = b64_decode(logger, renewal_info_b64).hex()
730 # this is ugly but i did not find a better way to do this
731 mda, certid_renewal = renewal_info_hex.split("0420", 1)
732 mda = mda[4:]
734 logger.debug("Helper.certid_hex_get() endet with %s", certid_renewal)
735 return mda, certid_renewal
738def certid_check(
739 logger: logging.Logger, renewal_info: str, certid_database: str
740) -> str:
741 """compare certid with renewal info"""
742 logger.debug("Helper.certid_check()")
744 renewal_info_b64 = b64_url_recode(logger, renewal_info)
745 renewal_info_hex = b64_decode(logger, renewal_info_b64).hex()
747 # this is ugly but i did not find a better way to do this
748 _header, certid_renewal = renewal_info_hex.split("0420", 1)
749 result = certid_renewal == certid_database
751 logger.debug("Helper.certid_check() ended with: %s", result)
752 return result
755def pkcs7_to_pem(logger, pkcs7_content: str, outform: str = "string") -> List[str]:
756 """convert pkcs7 to pem"""
757 logger.debug("CAhandler._pkcs7_to_pem()")
759 # Define loading strategies in order of preference
760 loading_strategies = [
761 # Strategy 1: Load as PEM directly
762 lambda content: load_pem_pkcs7_certificates(convert_string_to_byte(content)),
763 # Strategy 2: Replace CERTIFICATE with PKCS7 tag and load as PEM
764 lambda content: load_pem_pkcs7_certificates(
765 convert_string_to_byte(content.replace("CERTIFICATE", "PKCS7"))
766 ),
767 # Strategy 3: Load as DER
768 lambda content: load_der_pkcs7_certificates(content),
769 ]
771 pkcs7_obj = None
772 last_error = None
774 for i, strategy in enumerate(loading_strategies):
775 try:
776 pkcs7_obj = strategy(pkcs7_content)
777 if i == 1: # Log only for the tag replacement strategy
778 logger.error("PKCS7-TAG not found, updated content successfully")
779 break
780 except Exception as err:
781 last_error = err
782 if i == 0:
783 logger.error("PKCS7-TAG not found updating content...")
784 elif i == 1:
785 logger.debug("CAhandler._pkcs7_to_pem(): load pem failed. Try der...")
787 if pkcs7_obj is None:
788 logger.error("All PKCS7 loading strategies failed. Last error: %s", last_error)
789 raise last_error
791 # Convert certificates to PEM format
792 cert_pem_list = [
793 convert_byte_to_string(cert.public_bytes(serialization.Encoding.PEM))
794 for cert in pkcs7_obj
795 ]
797 # Define output format
798 output_formats = {
799 "string": lambda certs: "".join(certs),
800 "list": lambda certs: certs,
801 }
803 result = output_formats.get(outform, lambda _: None)(cert_pem_list)
805 logger.debug("Certificate._pkcs7_to_pem() ended")
806 return result