"""Opt-in contract test for a dedicated non-production Infisical project. It has no product imports and cannot make a request without an explicit, external non-production confirmation. """ from __future__ import annotations import os import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any import httpx import pytest from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID pytestmark = pytest.mark.integration _REQUIRED = ( "INFISICAL_TEST_HOST", "INFISICAL_TEST_CLIENT_ID", "INFISICAL_TEST_CLIENT_SECRET", "INFISICAL_TEST_PROJECT_ID", ) _TIMEOUT = httpx.Timeout(connect=5.0, read=20.0, write=20.0, pool=5.0) @dataclass(frozen=True) class IntegrationConfig: host: str client_id: str client_secret: str project_id: str application_id: str | None profile_id: str | None pki_collection_id: str | None def _config_or_skip() -> IntegrationConfig: if os.getenv("INFISICAL_TEST_CONFIRM_NON_PRODUCTION") != "true": pytest.skip("INFISICAL_TEST_CONFIRM_NON_PRODUCTION=true is required") missing = [name for name in _REQUIRED if not os.getenv(name)] if missing: pytest.skip("missing non-production Infisical test configuration") return IntegrationConfig( host=os.environ["INFISICAL_TEST_HOST"].rstrip("/"), client_id=os.environ["INFISICAL_TEST_CLIENT_ID"], client_secret=os.environ["INFISICAL_TEST_CLIENT_SECRET"], project_id=os.environ["INFISICAL_TEST_PROJECT_ID"], application_id=os.getenv("INFISICAL_TEST_APPLICATION_ID"), profile_id=os.getenv("INFISICAL_TEST_PROFILE_ID"), pki_collection_id=os.getenv("INFISICAL_TEST_PKI_COLLECTION_ID"), ) @dataclass(frozen=True) class CreatedCertificate: certificate_id: str friendly_name: str fingerprint: str serial_number: str purpose: str @dataclass(frozen=True) class GeneratedMaterial: leaf_pem: str key_pem: str chain_pem: str leaf: x509.Certificate intermediate: x509.Certificate root: x509.Certificate def _name(common_name: str) -> x509.Name: if len(common_name.encode("utf-8")) > 64: raise ValueError( f"Test certificate commonName exceeds X.509 64-byte limit: {common_name!r}" ) return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]) def _issue_certificate( *, subject: x509.Name, issuer: x509.Name, public_key: Any, issuer_key: Any, is_ca: bool, serial_number: int, ) -> x509.Certificate: now = datetime.now(UTC) return ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(issuer) .public_key(public_key) .serial_number(serial_number) .not_valid_before(now - timedelta(minutes=1)) .not_valid_after(now + timedelta(days=2)) .add_extension(x509.BasicConstraints(ca=is_ca, path_length=None), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, content_commitment=False, key_encipherment=not is_ca, data_encipherment=False, key_agreement=False, key_cert_sign=is_ca, crl_sign=is_ca, encipher_only=None, decipher_only=None, ), critical=True, ) .sign(issuer_key, hashes.SHA256()) ) def _generate_material() -> GeneratedMaterial: marker = uuid.uuid4().hex[:12] root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) root_subject = _name(f"Infisical integration TEST root {marker}") root = _issue_certificate( subject=root_subject, issuer=root_subject, public_key=root_key.public_key(), issuer_key=root_key, is_ca=True, serial_number=x509.random_serial_number(), ) intermediate_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) intermediate_subject = _name(f"Infisical integration TEST intermediate {marker}") intermediate = _issue_certificate( subject=intermediate_subject, issuer=root.subject, public_key=intermediate_key.public_key(), issuer_key=root_key, is_ca=True, serial_number=x509.random_serial_number(), ) leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) leaf = _issue_certificate( subject=_name(f"Infisical integration TEST leaf {marker}"), issuer=intermediate.subject, public_key=leaf_key.public_key(), issuer_key=intermediate_key, is_ca=False, serial_number=x509.random_serial_number(), ) return GeneratedMaterial( leaf_pem=leaf.public_bytes(serialization.Encoding.PEM).decode(), key_pem=leaf_key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ).decode(), # The Certificate Manager import contract describes chainPem as intermediate # certificates; the root signs the intermediate but is not sent as chain material. chain_pem=intermediate.public_bytes(serialization.Encoding.PEM).decode(), leaf=leaf, intermediate=intermediate, root=root, ) def _fingerprint(certificate: x509.Certificate) -> str: return certificate.fingerprint(hashes.SHA256()).hex() def _certificates_from_pem(bundle: str) -> list[x509.Certificate]: marker = "-----END CERTIFICATE-----" return [ x509.load_pem_x509_certificate((part + marker).encode()) for part in bundle.split(marker) if "-----BEGIN CERTIFICATE-----" in part ] def _safe_shape(response: httpx.Response) -> str: try: body = response.json() except ValueError: return "non-json" return ( "object:" + ",".join(sorted(body.keys())) if isinstance(body, dict) else type(body).__name__ ) def _payload( config: IntegrationConfig, material: GeneratedMaterial, *, chain: bool ) -> dict[str, str]: payload = { "certificatePem": material.leaf_pem, "privateKeyPem": material.key_pem, "friendlyName": f"integration-test-{uuid.uuid4().hex}", } if chain: payload["chainPem"] = material.chain_pem if config.application_id: payload["applicationId"] = config.application_id if config.profile_id: payload["profileId"] = config.profile_id if config.pki_collection_id: payload["pkiCollectionId"] = config.pki_collection_id return payload def _login(client: httpx.Client, config: IntegrationConfig) -> tuple[str, int | None]: response = client.post( "/api/v1/auth/universal-auth/login", json={"clientId": config.client_id, "clientSecret": config.client_secret}, ) assert response.status_code == 200 body = response.json() token = body.get("accessToken") assert isinstance(token, str) and token ttl = body.get("expiresIn") return token, ttl if isinstance(ttl, int) else None def _import( client: httpx.Client, payload: dict[str, str] ) -> tuple[httpx.Response, str | None]: response = client.post( "/api/v1/cert-manager/certificates/import-certificate", json=payload ) certificate_id = ( response.json().get("certificateId") if response.status_code < 300 else None ) return response, certificate_id def test_certificate_manager_minimal_contract() -> None: created_certificates: list[CreatedCertificate] = [] config = _config_or_skip() material = _generate_material() with httpx.Client( base_url=config.host, verify=True, timeout=_TIMEOUT ) as unauthenticated: token, ttl = _login(unauthenticated, config) assert ttl is None or ttl > 0 invalid_token = unauthenticated.get( "/api/v1/cert-manager/certificates/00000000-0000-0000-0000-000000000000", headers={"Authorization": "Bearer invalid"}, ) assert invalid_token.status_code in {401, 403} headers = {"Authorization": f"Bearer {token}"} with httpx.Client( base_url=config.host, headers=headers, verify=True, timeout=_TIMEOUT ) as client: minimal = _payload(config, material, chain=False) imported, certificate_id = _import(client, minimal) import_body = imported.json() if imported.headers.get("content-type", "").startswith("application/json") else {} import_error = import_body.get("error") import_message = str(import_body.get("message", ""))[:300] assert imported.status_code in {200, 201}, ( f"import failed: status={imported.status_code} error={import_error!r} " f"message={import_message!r}" ) assert isinstance(certificate_id, str) and certificate_id created_certificates.append( CreatedCertificate( certificate_id=certificate_id, friendly_name=minimal["friendlyName"], fingerprint=_fingerprint(material.leaf), serial_number=format(material.leaf.serial_number, "x"), purpose="minimal-contract", ) ) metadata = client.get(f"/api/v1/cert-manager/certificates/{certificate_id}") certificate = client.get( f"/api/v1/cert-manager/certificates/{certificate_id}/certificate" ) private_key = client.get( f"/api/v1/cert-manager/certificates/{certificate_id}/private-key" ) assert ( metadata.status_code == certificate.status_code == private_key.status_code == 200 ) metadata_body = metadata.json().get("certificate", {}) observed_serial = metadata_body.get("serialNumber") assert isinstance(observed_serial, str) # Infisical returns a byte-aligned hexadecimal serial, which may preserve # a leading zero. Compare the certificate serial numerically. assert int(observed_serial, 16) == material.leaf.serial_number returned_leaf = x509.load_pem_x509_certificate( certificate.json()["certificate"].encode() ) assert _fingerprint(returned_leaf) == _fingerprint(material.leaf) returned_key = serialization.load_pem_private_key( private_key.content, password=None, ) assert ( returned_key.public_key().public_numbers() == returned_leaf.public_key().public_numbers() ) duplicate, duplicate_id = _import(client, minimal) assert duplicate.status_code in {200, 201, 400, 409, 422} duplicate_body = duplicate.json() if duplicate.headers.get("content-type", "").startswith("application/json") else {} duplicate_error = duplicate_body.get("error") duplicate_message = str(duplicate_body.get("message", ""))[:300] if duplicate.status_code < 300: assert isinstance(duplicate_id, str) and duplicate_id duplicate_classification = ( "DEDUPLICATED" if duplicate.status_code < 300 and duplicate_id == certificate_id else "DUPLICATE_CREATED" if duplicate.status_code < 300 else "DUPLICATE_REJECTED" if duplicate.status_code in {400, 409, 422} else "NO_DEMOSTRADO" ) assert duplicate_classification != "NO_DEMOSTRADO", ( f"duplicate status={duplicate.status_code} error={duplicate_error!r} " f"message={duplicate_message!r} id={(duplicate_id or chr(45))[:6]}…" ) chain_material = _generate_material() chained, chained_id = _import( client, _payload(config, chain_material, chain=True) ) chain_body = ( chained.json() if chained.headers.get("content-type", "").startswith("application/json") else {} ) chain_error = chain_body.get("error") chain_message = str(chain_body.get("message", ""))[:300] assert chained.status_code in {200, 201}, ( f"chain import failed: status={chained.status_code} error={chain_error!r} message={chain_message!r} " f"shape={_safe_shape(chained)}" ) assert isinstance(chained_id, str) and chained_id chain_response = client.get( f"/api/v1/cert-manager/certificates/{chained_id}/certificate" ) assert chain_response.status_code == 200 returned_chain = _certificates_from_pem( chain_response.json().get("certificateChain") or "" ) assert returned_chain assert [_fingerprint(item) for item in returned_chain] == [ _fingerprint(chain_material.intermediate) ] assert _fingerprint(chain_material.root) not in { _fingerprint(item) for item in returned_chain } missing = client.get("/api/v1/cert-manager/certificates/00000000-0000-0000-0000-000000000000") assert missing.status_code in {400, 404} malformed = client.post( "/api/v1/cert-manager/certificates/import-certificate", json={"certificatePem": "invalid", "privateKeyPem": "invalid"}, ) assert malformed.status_code == 500 # Keep only a stable, non-sensitive shape available in assertion output. assert _safe_shape(missing)