"""Route-level coverage for XCA curve and signature-hash parity.""" 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": "CN", "value": cn}] def _extensions(ca: bool = True) -> 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": "Algorithm parity"}) assert response.status_code == 201, response.text return response.json()["id"] class AlgorithmRouteTests(unittest.TestCase): @async_test async def test_options_and_default_curve_settings(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}/options") assert response.status_code == 200, response.text options = response.json() self.assertEqual(len(options["curves"]), 79) self.assertEqual(len(options["curve_options"]), 79) self.assertEqual(len(options["hash_algorithms"]), 11) self.assertEqual(len(options["hash_options"]), 11) by_type = {item["type"]: item for item in options["key_types"]} self.assertIn("ripemd160", by_type["rsa"]["hashes"]) self.assertNotIn("ripemd160", by_type["ec"]["hashes"]) self.assertEqual(by_type["dsa"]["hashes"], ["sha1", "sha224", "sha256"]) self.assertEqual(by_type["ed25519"]["hashes"], []) response = await client.get(f"/api/vaults/{vault_id}/settings") version = response.json()["version"] response = await client.patch( f"/api/vaults/{vault_id}/settings", json={ "settings": { "default_hash": "ripemd160", "default_key": {"type": "ec", "size": 2048, "curve": "prime192v2"}, }, "version": version, }, ) assert response.status_code == 200, response.text settings = response.json()["settings"] self.assertEqual(settings["default_hash"], "ripemd160") self.assertEqual(settings["default_key"]["curve"], "prime192v2") @async_test async def test_compat_curve_complete_api_workflow(self) -> None: now = datetime.now(timezone.utc) async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) response = await client.post( f"/api/vaults/{vault_id}/keys", json={"name": "Binary CA key", "key_type": "ec", "curve": "sect163k1"}, ) assert response.status_code == 201, response.text key = response.json() self.assertEqual(key["curve"], "sect163k1") self.assertIn("sha3-512", key["possible_hashes"]) self.assertNotIn("md5", key["possible_hashes"]) response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "Binary request", "key_id": key["id"], "subject": _subject("binary.example.com"), "extensions": _extensions(False), "signature_hash": "sha3-256", }, ) assert response.status_code == 201, response.text request = response.json() self.assertTrue(request["signature_valid"]) self.assertEqual(request["signature_algorithm"], "ecdsa_with_SHA3-256") response = await client.post( f"/api/vaults/{vault_id}/certificates", json={ "name": "Binary CA", "key_id": key["id"], "subject": _subject("Binary CA"), "not_before": now.isoformat(), "not_after": (now + timedelta(days=30)).isoformat(), "extensions": _extensions(True), "signature_hash": "sha1", }, ) assert response.status_code == 201, response.text cert = response.json() self.assertEqual(cert["signature_algorithm"], "ecdsa-with-SHA1") self.assertEqual(cert["key_type"], "EC") self.assertEqual(cert["cert_spec"]["signature_hash"], "sha1") response = await client.post( f"/api/vaults/{vault_id}/certificates/{cert['id']}/crl", json={"days": 7, "signature_hash": "sha3-512"}, ) assert response.status_code == 201, response.text self.assertEqual(response.json()["signature_algorithm"], "ecdsa_with_SHA3-512") response = await client.post( f"/api/vaults/{vault_id}/certificates/{cert['id']}/renew", json={"validity": {"amount": 10, "unit": "days"}}, ) assert response.status_code == 201, response.text self.assertEqual(response.json()["signature_algorithm"], "ecdsa-with-SHA1") response = await client.post( f"/api/vaults/{vault_id}/certificates/{cert['id']}/export", json={"format": "pkcs12", "password": "route-p12-secret"}, ) assert response.status_code == 200, response.text self.assertGreater(len(base64.b64decode(response.json()["content_base64"])), 100) @async_test async def test_ed25519_and_invalid_nested_hashes(self) -> None: from kca.models import Key async with api_client() as client: with oidc_user("alice"): vault_id = await _vault(client) response = await client.post( f"/api/vaults/{vault_id}/keys", json={"name": "Ed key", "key_type": "ed25519"}, ) assert response.status_code == 201, response.text key = response.json() response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "Ed request", "key_id": key["id"], "subject": _subject("ed.example.com"), "extensions": _extensions(False), }, ) assert response.status_code == 201, response.text self.assertEqual(response.json()["signature_algorithm"], "Ed25519") self.assertIsNone(response.json()["csr_spec"]["signature_hash"]) response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "Invalid Ed request", "key_id": key["id"], "subject": _subject("invalid-ed.example.com"), "extensions": _extensions(False), "signature_hash": "sha256", }, ) assert response.status_code == 400, response.text before = await Key.filter().count() response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "Invalid nested EC request", "new_key": {"key_type": "ec", "curve": "secp112r1"}, "subject": _subject("invalid-ec.example.com"), "extensions": _extensions(False), "signature_hash": "md5", }, ) assert response.status_code == 400, response.text self.assertEqual(await Key.filter().count(), before) response = await client.post( f"/api/vaults/{vault_id}/requests", json={ "name": "Valid nested EC request", "new_key": {"key_type": "ec", "curve": "secp112r1"}, "subject": _subject("valid-ec.example.com"), "extensions": _extensions(False), }, ) assert response.status_code == 201, response.text self.assertEqual(response.json()["signature_algorithm"], "ecdsa-with-SHA256") if __name__ == "__main__": unittest.main()