diff --git a/backend/bdd/features/steps/pki_acme.py b/backend/bdd/features/steps/pki_acme.py index 604aa5e7a..aa8a110dd 100644 --- a/backend/bdd/features/steps/pki_acme.py +++ b/backend/bdd/features/steps/pki_acme.py @@ -4,11 +4,7 @@ import re import urllib.parse import acme.client -import httpx import jq -import requests -import requests.structures -import glom from faker import Faker from acme import client from acme import messages @@ -19,7 +15,6 @@ from behave import given from behave import when from behave import then from josepy.jwk import JWKRSA -from josepy import JSONObjectWithFields from josepy import json_util from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa @@ -27,6 +22,11 @@ from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes +from utils import replace_vars +from utils import eval_var +from utils import prepare_headers + + ACC_KEY_BITS = 2048 ACC_KEY_PUBLIC_EXPONENT = 65537 logger = logging.getLogger(__name__) @@ -40,98 +40,6 @@ class AcmeProfile: self.eab_secret = eab_secret -def replace_vars(payload: dict | list | int | float | str, vars: dict): - if isinstance(payload, dict): - return { - replace_vars(key, vars): replace_vars(value, vars) - for key, value in payload.items() - } - elif isinstance(payload, list): - return [replace_vars(item, vars) for item in payload] - elif isinstance(payload, str): - return payload.format(**vars) - else: - return payload - - -def parse_glom_path(path_str: str) -> glom.Path: - """ - Parse a glom path string with 'attr[index]' syntax into a Path object. - - Examples: - >>> parse_glom_path('authorizations[0]') == Path('authorizations', 0) - True - >>> parse_glom_path('data.items[1].name') == Path('data', 'items', 1, 'name') - True - >>> parse_glom_path('user.addresses[0].street') == Path('user', 'addresses', 0, 'street') - True - """ - parts = [] - - # Split by dots, but preserve bracketed content - tokens = re.split(r"(? dict | None: - headers = {} - auth_token = getattr(context, "auth_token", None) - if auth_token is not None: - headers["authorization"] = "Bearer {}".format(auth_token) - if not headers: - return None - return headers - - @given("I make a random {faker_type} as {var_name}") def step_impl(context: Context, faker_type: str, var_name: str): context.vars[var_name] = getattr(faker, faker_type)() @@ -746,3 +654,10 @@ def step_impl(context: Context, var_path: str, finalized_var: str): acme_client = context.acme_client finalized_order = acme_client.poll_and_finalize(order) context.vars[finalized_var] = finalized_order + + +@then("I parse the full-chain certificate from order {order_var_path} as {cert_var}") +def step_impl(context: Context, order_var_path: str, cert_var: str): + order = eval_var(context, order_var_path, as_json=False) + cert = x509.load_pem_x509_certificate(order.fullchain_pem.encode()) + context.vars[cert_var] = cert diff --git a/backend/bdd/features/steps/utils.py b/backend/bdd/features/steps/utils.py new file mode 100644 index 000000000..cd967267a --- /dev/null +++ b/backend/bdd/features/steps/utils.py @@ -0,0 +1,257 @@ +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.x509.oid import NameOID +import logging +import re + +import httpx +import requests +import requests.structures +import glom +from faker import Faker +from behave.runner import Context +from josepy import JSONObjectWithFields + +ACC_KEY_BITS = 2048 +ACC_KEY_PUBLIC_EXPONENT = 65537 +logger = logging.getLogger(__name__) +faker = Faker() + + +class AcmeProfile: + def __init__(self, id: str, eab_kid: str, eab_secret: str): + self.id = id + self.eab_kid = eab_kid + self.eab_secret = eab_secret + + +def replace_vars(payload: dict | list | int | float | str, vars: dict): + if isinstance(payload, dict): + return { + replace_vars(key, vars): replace_vars(value, vars) + for key, value in payload.items() + } + elif isinstance(payload, list): + return [replace_vars(item, vars) for item in payload] + elif isinstance(payload, str): + return payload.format(**vars) + else: + return payload + + +def parse_glom_path(path_str: str) -> glom.Path: + """ + Parse a glom path string with 'attr[index]' syntax into a Path object. + + Examples: + >>> parse_glom_path('authorizations[0]') == Path('authorizations', 0) + True + >>> parse_glom_path('data.items[1].name') == Path('data', 'items', 1, 'name') + True + >>> parse_glom_path('user.addresses[0].street') == Path('user', 'addresses', 0, 'street') + True + """ + parts = [] + + # Split by dots, but preserve bracketed content + tokens = re.split(r"(? dict | None: + headers = {} + auth_token = getattr(context, "auth_token", None) + if auth_token is not None: + headers["authorization"] = "Bearer {}".format(auth_token) + if not headers: + return None + return headers + + +def x509_cert_to_dict(cert: x509.Certificate) -> dict: + """ + Convert a cryptography.x509.Certificate to a JSON-serializable nested dict + with human-readable keys. + """ + + def oid_to_name(oid): + # Map known OIDs to human-readable names + mapping = { + NameOID.COMMON_NAME: "common_name", + NameOID.ORGANIZATION_NAME: "organization", + NameOID.ORGANIZATIONAL_UNIT_NAME: "organizational_unit", + NameOID.COUNTRY_NAME: "country", + NameOID.LOCALITY_NAME: "locality", + NameOID.STATE_OR_PROVINCE_NAME: "state_or_province", + NameOID.EMAIL_ADDRESS: "email_address", + NameOID.SERIAL_NUMBER: "serial_number", + NameOID.SURNAME: "surname", + NameOID.GIVEN_NAME: "given_name", + NameOID.TITLE: "title", + NameOID.JURISDICTION_COUNTRY_NAME: "jurisdiction_country", + NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME: "jurisdiction_state", + NameOID.JURISDICTION_LOCALITY_NAME: "jurisdiction_locality", + NameOID.BUSINESS_CATEGORY: "business_category", + NameOID.POSTAL_CODE: "postal_code", + NameOID.STREET_ADDRESS: "street_address", + NameOID.DOMAIN_COMPONENT: "domain_component", + NameOID.USER_ID: "user_id", + # Add more as needed + } + return mapping.get(oid, oid.dotted_string) + + def name_to_dict(name: x509.Name) -> dict: + return {oid_to_name(attr.oid): attr.value for attr in name} + + def extension_to_dict(ext): + if isinstance(ext.value, x509.SubjectAlternativeName): + return { + "critical": ext.critical, + "general_names": [str(gn) for gn in ext.value], + } + elif isinstance(ext.value, x509.BasicConstraints): + return { + "critical": ext.critical, + "ca": ext.value.ca, + "path_length": ext.value.path_length, + } + elif isinstance(ext.value, x509.KeyUsage): + return { + "critical": ext.critical, + **{ + field.lower(): getattr(ext.value, field) + for field in [ + "digital_signature", + "content_commitment", + "key_encipherment", + "data_encipherment", + "key_agreement", + "key_cert_sign", + "crl_sign", + "encipher_only", + "decipher_only", + ] + if getattr(ext.value, field) is not None + }, + } + elif isinstance(ext.value, x509.ExtendedKeyUsage): + return { + "critical": ext.critical, + "usages": [eku.dotted_string for eku in ext.value], + } + elif isinstance(ext.value, x509.CRLDistributionPoints): + return { + "critical": ext.critical, + "distribution_points": [ + { + "full_name": [str(uri) for uri in dp.full_name] + if dp.full_name + else None, + "crl_issuer": [str(issuer) for issuer in dp.crl_issuer] + if dp.crl_issuer + else None, + "reasons": [r.name for r in dp.reasons] if dp.reasons else None, + } + for dp in ext.value + ], + } + elif isinstance(ext.value, x509.AuthorityKeyIdentifier): + return { + "critical": ext.critical, + "key_identifier": ext.value.key_identifier.hex() + if ext.value.key_identifier + else None, + "authority_cert_issuer": [ + str(n) for n in ext.value.authority_cert_issuer + ] + if ext.value.authority_cert_issuer + else None, + "authority_cert_serial_number": ext.value.authority_cert_serial_number, + } + elif isinstance(ext.value, x509.SubjectKeyIdentifier): + return {"critical": ext.critical, "digest": ext.value.digest.hex()} + else: + return { + "critical": ext.critical, + "oid": ext.oid.dotted_string, + "value": str(ext.value), + } + + # Build the main dict + result = dict( + version=cert.version.name, + serial_number=cert.serial_number, + signature_algorithm=cert.signature_algorithm_oid._name, + issuer=name_to_dict(cert.issuer), + subject=name_to_dict(cert.subject), + validity={ + "not_valid_before": cert.not_valid_before.isoformat(), + "not_valid_after": cert.not_valid_after.isoformat(), + }, + public_key={ + "key_size": cert.public_key().key_size, + }, + extensions={ + ext.oid._name + if hasattr(ext.oid, "_name") and ext.oid._name + else ext.oid.dotted_string: extension_to_dict(ext) + for ext in cert.extensions + }, + fingerprint={ + "sha1": cert.fingerprint(hashes.SHA1()).hex(), + "sha256": cert.fingerprint(hashes.SHA256()).hex(), + }, + ) + + return result