diff --git a/pyproject.toml b/pyproject.toml index 66ce841..65d2f7b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,4 +32,9 @@ dev = [ "httpx", "ruff", "mypy", -] \ No newline at end of file +] +[tool.pytest.ini_options] +addopts = "-m 'not integration'" +markers = [ + "integration: opt-in tests that contact a separately confirmed non-production service", +] diff --git a/tests/integration/infisical/README.md b/tests/integration/infisical/README.md new file mode 100644 index 0000000..7a73667 --- /dev/null +++ b/tests/integration/infisical/README.md @@ -0,0 +1,13 @@ +# Infisical Certificate Manager integration test + +This test is opt-in and makes real requests only when all required environment +variables are present and `INFISICAL_TEST_CONFIRM_NON_PRODUCTION=true`. + +```bash +pytest -m integration tests/integration/infisical +``` + +Use only a disposable Certificate Manager project and Machine Identity. The test +generates ephemeral RSA test material in memory. It never prints credentials, +PEM material, private keys, or Bearer tokens. Certificates created remotely may +remain if the API does not provide an explicitly verified delete endpoint. diff --git a/tests/integration/infisical/test_certificate_manager.py b/tests/integration/infisical/test_certificate_manager.py new file mode 100644 index 0000000..e920e7c --- /dev/null +++ b/tests/integration/infisical/test_certificate_manager.py @@ -0,0 +1,340 @@ +"""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 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_rest_contract() -> None: + 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 + + 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} + if duplicate.status_code < 300: + assert isinstance(duplicate_id, str) and duplicate_id + + 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 in {400, 422} + # Keep only a stable, non-sensitive shape available in assertion output. + assert _safe_shape(missing)