woggioni
4f3a632f4b
- kaya backend with OIDC sessions, API tokens, Postgres/Tortoise models - AES-256-GCM private key storage and full certificate/CSR/CRL domain logic - Sycamore/WASM SPA for vaults, keys, requests, certificates, CRLs and templates - imports/exports, XCA .xdb migration, Docker Compose stack and tests
127 lines
5.1 KiB
Python
127 lines
5.1 KiB
Python
"""End-to-end smoke tests for the core PKI workflow."""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from tests.helpers.api import api_client
|
|
from tests.helpers.asynctest import async_test
|
|
from tests.helpers.oidc import oidc_user
|
|
|
|
|
|
def _subject(cn: str = "Test CA") -> list[dict]:
|
|
return [
|
|
{"name": "C", "value": "DE"},
|
|
{"name": "O", "value": "KCA"},
|
|
{"name": "CN", "value": cn},
|
|
]
|
|
|
|
|
|
def _extensions() -> dict:
|
|
return {
|
|
"basic_constraints": {"critical": True, "ca": True, "path_length": None},
|
|
"key_usage": {"critical": True, "usages": ["key_cert_sign", "crl_sign"]},
|
|
"subject_key_identifier": {"critical": False},
|
|
"authority_key_identifier": {"critical": False},
|
|
}
|
|
|
|
|
|
class SmokeTests(unittest.TestCase):
|
|
@async_test
|
|
async def test_health_and_auth(self) -> None:
|
|
async with api_client() as client:
|
|
response = await client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
response = await client.get("/api/me")
|
|
assert response.status_code == 401
|
|
|
|
@async_test
|
|
async def test_core_workflow(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
async with api_client() as client:
|
|
with oidc_user("alice", "Alice"):
|
|
response = await client.get("/api/me")
|
|
assert response.status_code == 200
|
|
assert response.json()["display_name"] == "Alice"
|
|
|
|
response = await client.post("/api/vaults", json={"name": "Test Vault"})
|
|
assert response.status_code == 201, response.text
|
|
vault = response.json()
|
|
vault_id = vault["id"]
|
|
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/keys",
|
|
json={"name": "CA key", "key_type": "rsa", "size": 2048},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
key = response.json()
|
|
assert key["has_private"] is True
|
|
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates",
|
|
json={
|
|
"name": "Test CA",
|
|
"comment": "",
|
|
"key_id": key["id"],
|
|
"subject": _subject(),
|
|
"not_before": now.isoformat(),
|
|
"not_after": (now + timedelta(days=3650)).isoformat(),
|
|
"signature_hash": "sha256",
|
|
"extensions": _extensions(),
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
cert = response.json()
|
|
assert cert["is_ca"] is True
|
|
assert cert["subject_rfc4514"].startswith("CN=Test CA")
|
|
assert cert["validation_errors"] == []
|
|
|
|
response = await client.get(f"/api/vaults/{vault_id}/certificates")
|
|
assert response.status_code == 200
|
|
assert len(response.json()["results"]) == 1
|
|
|
|
response = await client.get(f"/api/vaults/{vault_id}/search?q=test")
|
|
assert response.status_code == 200
|
|
assert len(response.json()["results"]) >= 1
|
|
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{cert['id']}/crl",
|
|
json={"days": 30},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
crl = response.json()
|
|
assert crl["crl_number"] == 1
|
|
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{cert['id']}/export",
|
|
json={"format": "pem"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
assert base64.b64decode(payload["content_base64"]).startswith(b"-----BEGIN CERTIFICATE")
|
|
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/import",
|
|
json={
|
|
"filename": "ca.crt",
|
|
"content_base64": payload["content_base64"],
|
|
"dry_run": True,
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["items"][0]["duplicate"] is True
|
|
|
|
@async_test
|
|
async def test_ownership_isolation(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice", "Alice"):
|
|
response = await client.post("/api/vaults", json={"name": "Alice Vault"})
|
|
vault_id = response.json()["id"]
|
|
with oidc_user("bob", "Bob"):
|
|
response = await client.get(f"/api/vaults/{vault_id}")
|
|
assert response.status_code == 404
|
|
response = await client.get("/api/vaults")
|
|
assert response.json()["results"] == []
|