309 lines
15 KiB
Python
309 lines
15 KiB
Python
"""Additional route-level tests: CSRs, tokens, import/export and templates."""
|
|
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) -> list[dict]:
|
|
return [{"name": "C", "value": "DE"}, {"name": "O", "value": "KCA"}, {"name": "CN", "value": cn}]
|
|
|
|
|
|
def _extensions(ca: bool = False) -> dict:
|
|
return {
|
|
"basic_constraints": {"critical": True, "ca": ca, "path_length": None},
|
|
"key_usage": {"critical": True, "usages": ["key_cert_sign", "crl_sign"] if ca else ["digital_signature"]},
|
|
"subject_key_identifier": {"critical": False},
|
|
"authority_key_identifier": {"critical": False},
|
|
}
|
|
|
|
|
|
async def _vault(client) -> str:
|
|
response = await client.post("/api/vaults", json={"name": "Test"})
|
|
assert response.status_code == 201, response.text
|
|
return response.json()["id"]
|
|
|
|
|
|
async def _key(client, vault_id: str, name: str = "key") -> dict:
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/keys", json={"name": name, "key_type": "rsa", "size": 2048}
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
async def _cert(client, vault_id: str, key_id: str, name: str, ca: bool = False, signer_id: str | None = None) -> dict:
|
|
now = datetime.now(timezone.utc)
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates",
|
|
json={
|
|
"name": name,
|
|
"key_id": key_id,
|
|
"signer_id": signer_id,
|
|
"subject": _subject(name),
|
|
"not_before": now.isoformat(),
|
|
"not_after": (now + timedelta(days=365)).isoformat(),
|
|
"extensions": _extensions(ca),
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
class RouteExtraTests(unittest.TestCase):
|
|
@async_test
|
|
async def test_csr_creation_and_signing(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
ca_key = await _key(client, vault_id, "ca key")
|
|
ca = await _cert(client, vault_id, ca_key["id"], "Root CA", ca=True)
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/requests",
|
|
json={
|
|
"name": "web request",
|
|
"subject": _subject("web.example.com"),
|
|
"new_key": {"key_type": "ec", "curve": "secp256r1", "name": "web key"},
|
|
"extensions": _extensions(False),
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
request = response.json()
|
|
assert request["signature_valid"] is True
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates",
|
|
json={
|
|
"name": "web cert",
|
|
"request_id": request["id"],
|
|
"signer_id": ca["id"],
|
|
"not_before": datetime.now(timezone.utc).isoformat(),
|
|
"not_after": (datetime.now(timezone.utc) + timedelta(days=30)).isoformat(),
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
cert = response.json()
|
|
assert cert["issuer_id"] == ca["id"]
|
|
response = await client.get(f"/api/vaults/{vault_id}/requests/{request['id']}")
|
|
assert response.json()["signed"] is True
|
|
|
|
@async_test
|
|
async def test_token_authentication(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
response = await client.post("/api/tokens", json={"name": "script"})
|
|
assert response.status_code == 201, response.text
|
|
secret = response.json()["token"]
|
|
assert secret.startswith("kca_")
|
|
# Session is no longer used: the bearer token authenticates.
|
|
response = await client.get("/api/vaults", headers={"authorization": f"Bearer {secret}"})
|
|
assert response.status_code == 200
|
|
response = await client.post("/api/tokens", headers={"authorization": f"Bearer {secret}"}, json={"name": "x"})
|
|
assert response.status_code == 401
|
|
|
|
@async_test
|
|
async def test_import_preview_and_commit(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
key = await _key(client, vault_id, "import key")
|
|
cert = await _cert(client, vault_id, key["id"], "Import Me", ca=True)
|
|
export = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{cert['id']}/export", json={"format": "pem"}
|
|
)
|
|
pem = base64.b64decode(export.json()["content_base64"])
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/import",
|
|
json={"filename": "test.crt", "content_base64": base64.b64encode(pem).decode(), "dry_run": True},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["items"][0]["duplicate"] is True
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/import",
|
|
json={"filename": "test.crt", "content_base64": base64.b64encode(pem).decode()},
|
|
)
|
|
assert response.status_code == 200
|
|
assert len(response.json()["skipped"]) == 1
|
|
|
|
@async_test
|
|
async def test_settings_and_templates(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
response = await client.get(f"/api/vaults/{vault_id}/settings")
|
|
assert response.status_code == 200
|
|
version = response.json()["version"]
|
|
response = await client.patch(
|
|
f"/api/vaults/{vault_id}/settings",
|
|
json={"settings": {"default_hash": "sha384", "serial_len_bits": 80}, "version": version},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["settings"]["default_hash"] == "sha384"
|
|
response = await client.get(f"/api/vaults/{vault_id}/templates")
|
|
templates = response.json()["results"]
|
|
assert any(t["id"] == "builtin:ca" for t in templates)
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/templates/builtin:ca/apply",
|
|
json={"target": {"subject": [], "extensions": {}}},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["extensions"]["basic_constraints"]["ca"] is True
|
|
|
|
@async_test
|
|
async def test_revocation_and_crl(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
ca_key = await _key(client, vault_id, "ca")
|
|
ca = await _cert(client, vault_id, ca_key["id"], "Root", ca=True)
|
|
leaf_key = await _key(client, vault_id, "leaf")
|
|
leaf = await _cert(
|
|
client, vault_id, leaf_key["id"], "Leaf", signer_id=ca["id"]
|
|
)
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke",
|
|
json={"reason": "keyCompromise"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
response = await client.get(
|
|
f"/api/vaults/{vault_id}/certificates/{leaf['id']}"
|
|
)
|
|
assert response.json()["revoked"] is True
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{ca['id']}/crl",
|
|
json={"days": 30},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
crl = response.json()
|
|
assert crl["revoked_count"] == 1
|
|
assert crl["crl_number"] == 1
|
|
|
|
@async_test
|
|
async def test_key_delete_unlinks(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
key = await _key(client, vault_id)
|
|
cert = await _cert(client, vault_id, key["id"], "Unlink")
|
|
response = await client.delete(f"/api/vaults/{vault_id}/keys/{key['id']}")
|
|
assert response.status_code == 204
|
|
response = await client.get(f"/api/vaults/{vault_id}/certificates/{cert['id']}")
|
|
assert response.json()["key_id"] is None
|
|
|
|
|
|
class ExtendedRouteTests(unittest.TestCase):
|
|
@async_test
|
|
async def test_preview_advanced_extensions(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
key = await _key(client, vault_id)
|
|
payload = {
|
|
"name": "preview",
|
|
"key_id": key["id"],
|
|
"subject": _subject("preview"),
|
|
"extensions": {
|
|
"basic_constraints": {"critical": True, "ca": False, "path_length": None},
|
|
"crl_distribution_points": {"critical": False, "urls": ["http://example.com/ca.crl"]},
|
|
"authority_information_access": {
|
|
"critical": False,
|
|
"descriptions": [{"method": "ocsp", "location": {"type": "uri", "value": "http://ocsp.example.com"}}],
|
|
},
|
|
"raw": [{"oid": "1.2.3.4", "critical": False, "der_hex": "0c03616263"}],
|
|
},
|
|
}
|
|
response = await client.post(f"/api/vaults/{vault_id}/certificates/preview", json=payload)
|
|
assert response.status_code == 200, response.text
|
|
names = [row["name"] for row in response.json()["extensions"]]
|
|
assert "CRL Distribution Points" in names
|
|
assert "Authority Information Access" in names
|
|
|
|
@async_test
|
|
async def test_mandatory_dn_and_no_expiration(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
settings = (await client.get(f"/api/vaults/{vault_id}/settings")).json()
|
|
await client.patch(
|
|
f"/api/vaults/{vault_id}/settings",
|
|
json={"settings": {"mandatory_dn": ["CN"]}, "version": settings["version"]},
|
|
)
|
|
key = await _key(client, vault_id)
|
|
now = datetime.now(timezone.utc)
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates",
|
|
json={
|
|
"name": "missing cn",
|
|
"key_id": key["id"],
|
|
"subject": [{"name": "O", "value": "KCA"}],
|
|
"validity": {"amount": 1, "unit": "days"},
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "mandatory" in response.json()["error"]
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates",
|
|
json={
|
|
"name": "forever",
|
|
"key_id": key["id"],
|
|
"subject": _subject("forever"),
|
|
"validity": {"no_expiration": True},
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
assert response.json()["no_expiration"] is True
|
|
|
|
@async_test
|
|
async def test_ca_template_and_revocation_edit(self) -> None:
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
ca_key = await _key(client, vault_id, "ca")
|
|
ca = await _cert(client, vault_id, ca_key["id"], "CA", ca=True)
|
|
response = await client.patch(
|
|
f"/api/vaults/{vault_id}/certificates/{ca['id']}/ca",
|
|
json={"crl_days": 45, "template_id": "builtin:tls-server"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["default_template"] == "builtin:tls-server"
|
|
assert response.json()["crl_days"] == 45
|
|
response = await client.post(
|
|
f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations",
|
|
json={"serial_hex": "1234", "reason": "keyCompromise", "invalidity_date": datetime.now(timezone.utc).isoformat()},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
rev = response.json()
|
|
response = await client.patch(
|
|
f"/api/vaults/{vault_id}/certificates/{ca['id']}/revocations/{rev['id']}",
|
|
json={"reason": "superseded", "invalidity_date": None},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["reason"] == "superseded"
|
|
|
|
|
|
class VaultExportTests(unittest.TestCase):
|
|
@async_test
|
|
async def test_vault_zip_export(self) -> None:
|
|
import io
|
|
import zipfile
|
|
async with api_client() as client:
|
|
with oidc_user("alice"):
|
|
vault_id = await _vault(client)
|
|
key = await _key(client, vault_id, "zip key")
|
|
cert = await _cert(client, vault_id, key["id"], "Zip Cert", ca=True)
|
|
response = await client.post(f"/api/vaults/{vault_id}/export", json={"password": "secret"})
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
data = base64.b64decode(payload["content_base64"])
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
names = archive.namelist()
|
|
assert "manifest.json" in names
|
|
assert any(name.startswith("certificates/") for name in names)
|
|
assert any(name.startswith("keys/") for name in names)
|
|
manifest = archive.read("manifest.json")
|
|
assert b"Zip Cert" in manifest
|