Add x509 asserts

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"(?<!\[)\.(?![^\[]*\])", path_str)
-
-    for token in tokens:
-        token = token.strip()
-        if not token:
-            continue
-
-        # Check for attr[index] pattern
-        match = re.match(r"^(.+?)\[([^\]]+)\]$", token)
-        if match:
-            attr_name = match.group(1).strip()
-            index_str = match.group(2).strip()
-
-            # Parse index (support integers, slices, etc.)
-            if index_str.isdigit():
-                index = int(index_str)
-            elif "-" in index_str:
-                # Handle negative indices like [-1]
-                index = int(index_str)
-            elif ":" in index_str:
-                # Handle slices like [0:10]
-                index = slice(
-                    *map(int, [x.strip() for x in index_str.split(":") if x.strip()])
-                )
-            else:
-                # Treat as string key
-                index = index_str
-
-            parts.extend([attr_name, index])
-        else:
-            # Plain attribute/key
-            parts.append(token)
-
-    return glom.Path(*parts)
-
-
-def eval_var(context: Context, var_path: str, as_json: bool = True):
-    parts = var_path.split(".", 1)
-    value = context.vars[parts[0]]
-    if len(parts) == 2:
-        value = glom.glom(value, parse_glom_path(parts[1]))
-    if as_json:
-        if isinstance(value, JSONObjectWithFields):
-            value = value.to_json()
-        elif isinstance(value, requests.Response):
-            value = value.json()
-        elif isinstance(value, requests.structures.CaseInsensitiveDict):
-            value = dict(value.lower_items())
-        elif isinstance(value, httpx.Response):
-            value = value.json()
-    return value
-
-
-def prepare_headers(context: Context) -> 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

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"(?<!\[)\.(?![^\[]*\])", path_str)
+
+    for token in tokens:
+        token = token.strip()
+        if not token:
+            continue
+
+        # Check for attr[index] pattern
+        match = re.match(r"^(.+?)\[([^\]]+)\]$", token)
+        if match:
+            attr_name = match.group(1).strip()
+            index_str = match.group(2).strip()
+
+            # Parse index (support integers, slices, etc.)
+            if index_str.isdigit():
+                index = int(index_str)
+            elif "-" in index_str:
+                # Handle negative indices like [-1]
+                index = int(index_str)
+            elif ":" in index_str:
+                # Handle slices like [0:10]
+                index = slice(
+                    *map(int, [x.strip() for x in index_str.split(":") if x.strip()])
+                )
+            else:
+                # Treat as string key
+                index = index_str
+
+            parts.extend([attr_name, index])
+        else:
+            # Plain attribute/key
+            parts.append(token)
+
+    return glom.Path(*parts)
+
+
+def eval_var(context: Context, var_path: str, as_json: bool = True):
+    parts = var_path.split(".", 1)
+    value = context.vars[parts[0]]
+    if len(parts) == 2:
+        value = glom.glom(value, parse_glom_path(parts[1]))
+    if as_json:
+        if isinstance(value, JSONObjectWithFields):
+            value = value.to_json()
+        elif isinstance(value, requests.Response):
+            value = value.json()
+        elif isinstance(value, requests.structures.CaseInsensitiveDict):
+            value = dict(value.lower_items())
+        elif isinstance(value, httpx.Response):
+            value = value.json()
+        elif isinstance(value, x509.Certificate):
+            value = x509_cert_to_dict(value)
+    return value
+
+
+def prepare_headers(context: Context) -> 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