woggioni/kca
2
0
Fork 0
Code Actions Packages Releases Activity

Add full XCA curve and digest parity
CI / Build and push docker image (push) Successful in 37s
Details
CI / Build and push OCSP image (push) Successful in 38s
Details

Browse Source
This commit is contained in:
woggioni
2026-10-26 12:06:17 +08:00
parent e83dc42d3d
commit 1a7b4a10e8
36 changed files with 4326 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/packages/kca-core/tests/test_algorithm_workflows.py
+339
View File
@@ -0,0 +1,339 @@
"""End-to-end core workflows for XCA curve and signature-hash parity."""
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from kca.core import algorithms, certificates, crls, exports, imports, keys, requests
from kca.core.extensions import default_extension_spec
from kca.core.names import entries_from_json, name_from_entries
def _name(cn: str = "Algorithm parity CA") -> x509.Name:
return name_from_entries(entries_from_json([{"name": "CN", "value": cn}]))
def _ca_spec() -> dict:
spec = default_extension_spec()
spec["basic_constraints"] = {"critical": True, "ca": True, "path_length": None}
spec["key_usage"] = {"critical": True, "usages": ["digital_signature", "key_cert_sign", "crl_sign"]}
spec["subject_key_identifier"] = {"critical": False}
spec["authority_key_identifier"] = {"critical": False}
return spec
class AlgorithmWorkflowTest(unittest.TestCase):
maxDiff = None
def assert_workflow(self, key: object, signature_hash: str | None, curve: str | None = None) -> None:
now = datetime.now(timezone.utc)
public = keys.public_part(key)
public_der = keys.public_der(public)
csr = requests.build_csr(
subject=_name("Algorithm parity CSR"),
private_key=key,
extension_spec=_ca_spec(),
signature_hash=signature_hash,
attributes={requests.CHALLENGE_PASSWORD_OID: "challenge"},
)
self.assertTrue(requests.verify_csr(csr))
request_info = requests.request_info(csr)
self.assertEqual(request_info.public_key_der, public_der)
self.assertEqual(requests.csr_spec(csr)["signature_hash"], signature_hash)
self.assertEqual(requests.attribute_values(csr)[requests.CHALLENGE_PASSWORD_OID], "challenge")
cert = certificates.build_self_signed(
subject=_name(),
private_key=key,
serial=1,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
signature_hash=signature_hash,
)
self.assertTrue(certificates.is_self_signed(cert))
self.assertTrue(certificates.verify_directly_issued_by(cert, cert))
cert_info = certificates.certificate_info(cert)
self.assertEqual(cert_info.public_key_der, public_der)
self.assertEqual(certificates.signature_hash_name(cert), signature_hash)
ski = cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value
self.assertEqual(ski.digest, keys.public_key_identifier(public))
aki = cert.extensions.get_extension_for_class(x509.AuthorityKeyIdentifier).value
self.assertEqual(aki.key_identifier, ski.digest)
crl = crls.build_crl(
issuer_cert=cert,
issuer_private_key=key,
revocations=[],
last_update=now,
next_update=now + timedelta(days=7),
crl_number=1,
signature_hash=signature_hash,
)
self.assertTrue(crls.verify_crl(crl, cert))
self.assertEqual(crl.signature_algorithm_oid, cert.signature_algorithm_oid)
expected_curve = curve if curve is not None else keys.curve_name(key)
reloaded_private = keys.private_key_from_der(keys.private_der(key))
self.assertEqual(keys.public_der(reloaded_private), public_der)
self.assertEqual(keys.curve_name(reloaded_private), expected_curve)
reloaded_public = keys.public_key_from_der(public_der)
self.assertEqual(keys.public_der(reloaded_public), public_der)
self.assertEqual(keys.curve_name(reloaded_public), expected_curve)
def test_every_xca_curve(self) -> None:
self.assertEqual(len(algorithms.CURVES), 79)
for curve in algorithms.CURVES:
with self.subTest(curve=curve.name):
key = keys.generate_key("ec", curve=curve.name)
info = keys.key_info(key)
self.assertEqual(info.key_type, "EC")
self.assertEqual(info.curve, curve.name)
self.assertTrue(info.is_private)
self.assert_workflow(key, "sha256", curve.name)
def test_mixed_native_and_compat_issuers(self) -> None:
now = datetime.now(timezone.utc)
for ca_curve, leaf_curve in (("secp112r1", "secp256r1"), ("secp256r1", "sect163k1")):
with self.subTest(ca_curve=ca_curve, leaf_curve=leaf_curve):
ca_key = keys.generate_key("ec", curve=ca_curve)
leaf_key = keys.generate_key("ec", curve=leaf_curve)
ca = certificates.build_self_signed(
subject=_name("Mixed CA"),
private_key=ca_key,
serial=10,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
leaf_spec = default_extension_spec()
leaf_spec["subject_key_identifier"] = {"critical": False}
leaf_spec["authority_key_identifier"] = {"critical": False}
leaf = certificates.build_signed(
subject=_name("Mixed leaf"),
public_key=keys.public_part(leaf_key),
issuer_cert=ca,
issuer_private_key=ca_key,
serial=11,
not_before=now,
not_after=now + timedelta(days=10),
extension_spec=leaf_spec,
)
self.assertTrue(certificates.verify_directly_issued_by(leaf, ca))
self.assertFalse(certificates.verify_directly_issued_by(leaf, leaf))
self.assertEqual(certificates.certificate_info(leaf).public_key_der, keys.public_der(leaf_key))
ca_ski = ca.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value
leaf_aki = leaf.extensions.get_extension_for_class(x509.AuthorityKeyIdentifier).value
self.assertEqual(leaf_aki.key_identifier, ca_ski.digest)
def test_ed25519_issuer_with_compat_subject(self) -> None:
now = datetime.now(timezone.utc)
ca_key = keys.generate_key("ed25519")
leaf_key = keys.generate_key("ec", curve="sect163k1")
ca = certificates.build_self_signed(
subject=_name("Ed25519 CA"),
private_key=ca_key,
serial=20,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
signature_hash=None,
)
leaf = certificates.build_signed(
subject=_name("Ed25519 signed compat leaf"),
public_key=keys.public_part(leaf_key),
issuer_cert=ca,
issuer_private_key=ca_key,
serial=21,
not_before=now,
not_after=now + timedelta(days=10),
extension_spec=default_extension_spec(),
signature_hash=None,
)
self.assertTrue(certificates.verify_directly_issued_by(leaf, ca))
self.assertEqual(certificates.certificate_info(leaf).public_key_der, keys.public_der(leaf_key))
self.assertIsNone(certificates.signature_hash_name(leaf))
def test_issuer_name_is_checked_in_openssl_fallback(self) -> None:
now = datetime.now(timezone.utc)
ca_key = keys.generate_key("ec", curve="secp112r1")
first = certificates.build_self_signed(
subject=_name("First issuer"),
private_key=ca_key,
serial=30,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
other_with_same_key = certificates.build_self_signed(
subject=_name("Other issuer with the same key"),
private_key=ca_key,
serial=31,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
leaf = certificates.build_signed(
subject=_name("Leaf"),
public_key=keys.public_part(ca_key),
issuer_cert=first,
issuer_private_key=ca_key,
serial=32,
not_before=now,
not_after=now + timedelta(days=10),
extension_spec=default_extension_spec(),
)
self.assertTrue(certificates.verify_directly_issued_by(leaf, first))
self.assertFalse(certificates.verify_directly_issued_by(leaf, other_with_same_key))
def test_xca_signature_hash_matrix(self) -> None:
cases = [
("rsa", {"size": 1024}, algorithms.possible_hashes("rsa")),
("ec", {"curve": "secp256r1"}, algorithms.possible_hashes("ec")),
("dsa", {"size": 1024}, algorithms.possible_hashes("dsa")),
]
for key_type, kwargs, hashes in cases:
key = keys.generate_key(key_type, **kwargs)
for signature_hash in hashes:
with self.subTest(key_type=key_type, signature_hash=signature_hash):
self.assert_workflow(key, signature_hash)
def test_ed25519_has_no_separate_hash(self) -> None:
key = keys.generate_key("ed25519")
self.assert_workflow(key, None)
with self.assertRaises(ValueError):
requests.build_csr(subject=_name(), private_key=key, signature_hash="sha256")
def test_invalid_key_hash_combinations(self) -> None:
ec_key = keys.generate_key("ec", curve="secp256r1")
dsa_key = keys.generate_key("dsa", size=1024)
with self.assertRaises(ValueError):
requests.build_csr(subject=_name(), private_key=ec_key, signature_hash="md5")
with self.assertRaises(ValueError):
requests.build_csr(subject=_name(), private_key=dsa_key, signature_hash="sha3-256")
def test_representative_encrypted_compat_roundtrips(self) -> None:
for curve in ("secp112r1", "prime192v2", "sect163k1", "brainpoolP160t1", "wap-wsg-idm-ecid-wtls6"):
with self.subTest(curve=curve):
key = keys.generate_key("ec", curve=curve)
for traditional in (False, True):
for password in (None, "secret"):
encoded = keys.private_pem(key, password=password, traditional=traditional)
loaded, private = keys.load_key_bytes(encoded, password=password)
self.assertTrue(private)
self.assertEqual(keys.public_der(loaded), keys.public_der(key))
encrypted_der = keys.private_der_bytes(key, password="secret")
loaded = keys.private_key_from_der(encrypted_der, password="secret")
self.assertEqual(keys.public_der(loaded), keys.public_der(key))
public, private = keys.load_key_bytes(keys.public_pem(key).encode("ascii"))
self.assertFalse(private)
self.assertEqual(keys.public_der(public), keys.public_der(key))
def test_pkcs12_compat_export_import(self) -> None:
now = datetime.now(timezone.utc)
key = keys.generate_key("ec", curve="secp112r1")
cert = certificates.build_self_signed(
subject=_name("PKCS12 compat"),
private_key=key,
serial=40,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
encoded = exports.export_certificate(
cert,
"pkcs12",
name="compat",
private_key=key,
password="p12-secret",
).data
items = imports.parse_pkcs12(encoded, password="p12-secret")
by_kind = {}
for item in items:
by_kind.setdefault(item.kind, []).append(item)
self.assertEqual(len(by_kind["certificate"]), 1)
self.assertEqual(len(by_kind["key"]), 1)
self.assertEqual(
certificates.certificate_der(by_kind["certificate"][0].obj),
certificates.certificate_der(cert),
)
self.assertTrue(by_kind["key"][0].is_private)
self.assertEqual(keys.public_der(by_kind["key"][0].obj), keys.public_der(key))
def test_non_ec_keys_survive_openssl_fallback_import(self) -> None:
key = keys.generate_key("rsa", size=1024)
der = keys.private_der(key)
original = serialization.load_der_private_key
calls = 0
def legacy_only(data: bytes, password: object = None) -> object:
nonlocal calls
calls += 1
if calls == 1:
raise ValueError("simulated legacy-only encoding")
return original(data, password=password)
with patch("kca.core.keys.serialization.load_der_private_key", new=legacy_only):
loaded = keys.private_key_from_der(der)
self.assertIsInstance(loaded, rsa.RSAPrivateKey)
self.assertEqual(keys.public_der(loaded), keys.public_der(key))
self.assertEqual(calls, 2)
def test_blank_pkcs12_passwords_remain_unencrypted(self) -> None:
now = datetime.now(timezone.utc)
compat_key = keys.generate_key("ec", curve="secp112r1")
compat_ca = certificates.build_self_signed(
subject=_name("Unencrypted compat CA"),
private_key=compat_key,
serial=50,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
leaf_key = keys.generate_key("ec", curve="secp256r1")
leaf = certificates.build_signed(
subject=_name("Mixed-chain leaf"),
public_key=keys.public_part(leaf_key),
issuer_cert=compat_ca,
issuer_private_key=compat_key,
serial=51,
not_before=now,
not_after=now + timedelta(days=10),
extension_spec=default_extension_spec(),
)
pbes2_oid = bytes.fromhex("06092a864886f70d01050d")
for password in (None, ""):
with self.subTest(password=password):
encoded = exports.export_certificate(
leaf,
"pkcs12-chain",
name="unencrypted",
chain=[leaf, compat_ca],
private_key=leaf_key,
password=password,
).data
self.assertNotIn(pbes2_oid, encoded)
items = imports.parse_pkcs12(encoded, password=password)
imported_keys = [item for item in items if item.kind == "key"]
self.assertEqual(len(imported_keys), 1)
self.assertEqual(keys.public_der(imported_keys[0].obj), keys.public_der(leaf_key))
def test_option_metadata_is_complete(self) -> None:
options = keys.key_type_options()
self.assertEqual(len(options), 4)
by_type = {item["type"]: item for item in options}
self.assertEqual(len(by_type["ec"]["curves"]), 79)
for key_type, option in by_type.items():
self.assertEqual(option["hashes"], list(algorithms.possible_hashes(key_type)))
self.assertEqual(len(keys.curve_options()), 79)
self.assertEqual(len(keys.hash_options()), 11)
if __name__ == "__main__":
unittest.main()
server/packages/kca-core/tests/test_algorithms.py
+92
View File
@@ -0,0 +1,92 @@
"""XCA algorithm catalogue and signature metadata tests."""
from __future__ import annotations
import unittest
from kca.core import algorithms, signatures
EXPECTED_CURVES = {
"secp192r1", "secp224r1", "secp256r1", "secp384r1", "secp521r1",
"prime192v2", "prime192v3", "prime239v1", "prime239v2", "prime239v3",
"c2pnb163v1", "c2pnb163v2", "c2pnb163v3", "c2pnb176v1",
"c2tnb191v1", "c2tnb191v2", "c2tnb191v3", "c2pnb208w1",
"c2tnb239v1", "c2tnb239v2", "c2tnb239v3", "c2pnb272w1",
"c2pnb304w1", "c2tnb359v1", "c2pnb368w1", "c2tnb431r1",
"sect113r1", "sect113r2", "sect131r1", "sect131r2", "sect163k1",
"sect163r1", "sect163r2", "sect193r1", "sect193r2", "sect233k1",
"sect233r1", "sect239k1", "sect283k1", "sect283r1", "sect409k1",
"sect409r1", "sect571k1", "sect571r1",
"secp112r1", "secp112r2", "secp128r1", "secp128r2", "secp160k1",
"secp160r1", "secp160r2", "secp192k1", "secp224k1", "secp256k1",
"wap-wsg-idm-ecid-wtls1", "wap-wsg-idm-ecid-wtls3",
"wap-wsg-idm-ecid-wtls4", "wap-wsg-idm-ecid-wtls5",
"wap-wsg-idm-ecid-wtls6", "wap-wsg-idm-ecid-wtls7",
"wap-wsg-idm-ecid-wtls8", "wap-wsg-idm-ecid-wtls9",
"wap-wsg-idm-ecid-wtls10", "wap-wsg-idm-ecid-wtls11",
"wap-wsg-idm-ecid-wtls12",
"brainpoolP160r1", "brainpoolP160t1", "brainpoolP192r1",
"brainpoolP192t1", "brainpoolP224r1", "brainpoolP224t1",
"brainpoolP256r1", "brainpoolP256t1", "brainpoolP320r1",
"brainpoolP320t1", "brainpoolP384r1", "brainpoolP384t1",
"brainpoolP512r1", "brainpoolP512t1",
}
class AlgorithmCatalogueTest(unittest.TestCase):
def test_exact_xca_curve_set(self) -> None:
self.assertEqual({curve.name for curve in algorithms.CURVES}, EXPECTED_CURVES)
self.assertEqual(len(algorithms.CURVES), 79)
self.assertEqual(len({curve.openssl_name for curve in algorithms.CURVES}), 79)
def test_aliases_preserve_existing_kca_names(self) -> None:
self.assertEqual(algorithms.canonical_curve_name("prime192v1"), "secp192r1")
self.assertEqual(algorithms.canonical_curve_name("prime256v1"), "secp256r1")
self.assertEqual(algorithms.curve_info("BRAINPOOLP256R1").name, "brainpoolP256r1")
with self.assertRaises(ValueError):
algorithms.curve_info("SM2")
def test_xca_hash_matrix(self) -> None:
self.assertEqual(
algorithms.possible_hashes("RSA"),
("md5", "ripemd160", "sha1", "sha224", "sha256", "sha384", "sha512", "sha3-224", "sha3-256", "sha3-384", "sha3-512"),
)
self.assertEqual(
algorithms.possible_hashes("ec"),
("sha1", "sha224", "sha256", "sha384", "sha512", "sha3-224", "sha3-256", "sha3-384", "sha3-512"),
)
self.assertEqual(algorithms.possible_hashes("dsa"), ("sha1", "sha224", "sha256"))
self.assertEqual(algorithms.possible_hashes("ed25519"), ())
self.assertEqual(algorithms.default_signature_hash("ec", "md5"), "sha256")
self.assertIsNone(algorithms.default_signature_hash("ed25519", "md5"))
with self.assertRaises(ValueError):
algorithms.validate_signature_hash("ec", "md5")
with self.assertRaises(ValueError):
algorithms.validate_signature_hash("dsa", "sha3-256")
with self.assertRaises(ValueError):
algorithms.validate_signature_hash("ed25519", "sha256")
def test_signature_oid_names_and_hashes(self) -> None:
class OID:
def __init__(self, dotted: str) -> None:
self.dotted_string = dotted
self._name = "Unknown OID"
cases = {
"1.2.840.113549.1.1.4": ("md5WithRSAEncryption", "md5"),
"1.3.36.3.3.1.2": ("ripemd160WithRSA", "ripemd160"),
"2.16.840.1.101.3.4.3.14": ("RSA-SHA3-256", "sha3-256"),
"2.16.840.1.101.3.4.3.10": ("ecdsa_with_SHA3-256", "sha3-256"),
"1.2.840.10040.4.3": ("dsa_with_SHA1", "sha1"),
}
for dotted, (name, hash_name) in cases.items():
with self.subTest(oid=dotted):
oid = OID(dotted)
self.assertEqual(signatures.signature_algorithm_name(oid), name)
self.assertEqual(signatures.signature_hash_name(oid), hash_name)
self.assertEqual(signatures.signature_algorithm_name(OID("1.3.101.112")), "Ed25519")
self.assertIsNone(signatures.signature_hash_name(OID("1.3.101.112")))
if __name__ == "__main__":
unittest.main()
server/packages/kca-core/tests/test_openssl_compat.py
+449
View File
@@ -0,0 +1,449 @@
from __future__ import annotations
import hashlib
import unittest
from datetime import datetime, timedelta, timezone
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509.oid import NameOID
from kca.core import openssl_compat as compat
UNSUPPORTED_CURVES = (
"secp112r1",
"prime192v2",
"sect163k1",
"brainpoolP160t1",
"wap-wsg-idm-ecid-wtls6",
)
def _require_curve(name: str) -> None:
available = compat.available_ec_curves()
canonical = "prime256v1" if name == "secp256r1" else name
if canonical not in available:
raise unittest.SkipTest(f"system OpenSSL does not provide EC curve {name}")
def _self_signed_certificate(
private_key: rsa.RSAPrivateKey,
common_name: str,
serial: int,
) -> x509.Certificate:
now = datetime(2024, 1, 1, tzinfo=timezone.utc)
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
return (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(private_key.public_key())
.serial_number(serial)
.not_valid_before(now)
.not_valid_after(now + timedelta(days=3650))
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
.add_extension(
x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
critical=False,
)
.sign(private_key, hashes.SHA256())
)
def _build_native_templates() -> dict[str, object]:
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
certificate = _self_signed_certificate(private_key, "OpenSSL compatibility", 1000)
extra_ca = _self_signed_certificate(private_key, "OpenSSL compatibility extra CA", 1001)
now = datetime(2024, 2, 1, tzinfo=timezone.utc)
csr = (
x509.CertificateSigningRequestBuilder()
.subject_name(certificate.subject)
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
.sign(private_key, hashes.SHA256())
)
revoked = (
x509.RevokedCertificateBuilder()
.serial_number(42)
.revocation_date(now)
.build()
)
crl = (
x509.CertificateRevocationListBuilder()
.issuer_name(certificate.subject)
.last_update(now)
.next_update(now + timedelta(days=30))
.add_revoked_certificate(revoked)
.add_extension(x509.CRLNumber(7), critical=False)
.sign(private_key, hashes.SHA256())
)
return {
"key_object": private_key,
"private_der": private_key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
),
"public_der": private_key.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
),
"cert_object": certificate,
"cert_der": certificate.public_bytes(serialization.Encoding.DER),
"csr_der": csr.public_bytes(serialization.Encoding.DER),
"crl_der": crl.public_bytes(serialization.Encoding.DER),
"extra_ca_object": extra_ca,
"extra_ca_der": extra_ca.public_bytes(serialization.Encoding.DER),
}
def _resign_all(
templates: dict[str, object], key: compat.CompatKey, digest: str
) -> tuple[bytes, bytes, bytes]:
cert = compat.resign_certificate(
templates["cert_der"], key.public_der, key.private_der, digest
)
csr = compat.resign_csr(
templates["csr_der"], key.public_der, key.private_der, digest
)
crl = compat.resign_crl(templates["crl_der"], key.private_der, digest)
return cert, csr, crl
class OpenSSLCompatTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.native_templates = _build_native_templates()
def test_backend_and_curve_discovery(self) -> None:
self.assertTrue(compat.backend_available())
self.assertTrue(compat.openssl_version().startswith("OpenSSL 3."))
self.assertTrue(compat.library_path())
curves = compat.available_ec_curves()
self.assertTrue(curves)
self.assertEqual(len(curves), len(set(curves)))
self.assertIn("prime256v1", curves)
def test_ec_key_size_is_curve_field_degree(self) -> None:
for curve, key_size in (("c2pnb176v1", 176), ("secp160k1", 160)):
with self.subTest(curve=curve):
_require_curve(curve)
self.assertEqual(compat.generate_ec(curve).key_size, key_size)
def test_native_ec_interoperates_with_cryptography(self) -> None:
_require_curve("prime256v1")
key = compat.generate_ec("prime256v1")
self.assertTrue(key.is_private)
self.assertEqual(key.curve, "prime256v1")
self.assertEqual(key.key_size, 256)
# Native keys still cross the API boundary only as DER, never as pointers.
native_private = serialization.load_der_private_key(key.private_der, password=None)
native_public = serialization.load_der_public_key(key.public_der)
self.assertEqual(
native_private.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
),
key.private_der,
)
self.assertEqual(
native_public.public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo,
),
key.public_der,
)
def test_arbitrary_curve_and_key_round_trips(self) -> None:
for curve in UNSUPPORTED_CURVES:
with self.subTest(curve=curve):
_require_curve(curve)
key = compat.generate_ec(curve)
self.assertEqual(key.curve, curve)
self.assertTrue(key.is_private)
self.assertGreater(key.key_size, 0)
self.assertIsNotNone(key.private_der)
public = key.public_key()
self.assertFalse(public.is_private)
self.assertIsNone(public.private_der)
self.assertIs(public.public_key(), public)
self.assertEqual(public.public_der, key.public_der)
self.assertEqual(compat.public_key_from_der(key.public_der).public_der, key.public_der)
self.assertEqual(compat.load_key_bytes(key.public_der), public)
self.assertEqual(compat.private_key_from_der(key.private_der), key)
self.assertEqual(compat.load_key_bytes(key.private_der), key)
public_pem = compat.public_pem(key)
self.assertIsInstance(public_pem, str)
self.assertTrue(public_pem.startswith("-----BEGIN PUBLIC KEY-----"))
self.assertEqual(compat.load_key_bytes(public_pem.encode("ascii")), public)
pkcs8_pem = compat.private_pem(key)
self.assertTrue(pkcs8_pem.startswith(b"-----BEGIN PRIVATE KEY-----"))
self.assertEqual(compat.load_key_bytes(pkcs8_pem), key)
traditional_pem = compat.private_pem(key, traditional=True)
self.assertTrue(traditional_pem.startswith(b"-----BEGIN EC PRIVATE KEY-----"))
self.assertEqual(compat.load_key_bytes(traditional_pem), key)
encrypted_pem = compat.private_pem(key, password="correct horse")
self.assertTrue(encrypted_pem.startswith(b"-----BEGIN ENCRYPTED PRIVATE KEY-----"))
self.assertEqual(compat.load_key_bytes(encrypted_pem, password="correct horse"), key)
encrypted_traditional = compat.private_pem(
key, password="correct horse", traditional=True
)
self.assertTrue(encrypted_traditional.startswith(b"-----BEGIN EC PRIVATE KEY-----"))
self.assertIn(b"DEK-Info:", encrypted_traditional)
self.assertEqual(
compat.load_key_bytes(encrypted_traditional, password="correct horse"), key
)
self.assertEqual(compat.private_der_bytes(key), key.private_der)
encrypted_der = compat.private_der_bytes(key, password="correct horse")
self.assertNotEqual(encrypted_der, key.private_der)
self.assertEqual(
compat.private_key_from_der(encrypted_der, password="correct horse"), key
)
self.assertEqual(compat.load_key_bytes(encrypted_der, password="correct horse"), key)
def test_wrong_password_errors_do_not_disclose_password(self) -> None:
_require_curve("secp112r1")
key = compat.generate_ec("secp112r1")
password = "not-for-error-output"
encrypted = compat.private_der_bytes(key, password=password)
attempted_password = "definitely-wrong-secret"
with self.assertRaises(ValueError) as caught:
compat.private_key_from_der(encrypted, password=attempted_password)
self.assertNotIn(password, str(caught.exception))
self.assertNotIn(attempted_password, str(caught.exception))
with self.assertRaises(ValueError):
compat.private_pem(key.public_key())
self.assertEqual(compat.private_pem(key, password=""), compat.private_pem(key))
self.assertEqual(compat.private_der_bytes(key, password=""), key.private_der)
def test_native_rsa_traditional_and_pkcs1_der(self) -> None:
templates = self.native_templates
private_key = templates["key_object"]
traditional_der = private_key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption(),
)
pkcs1_public_der = private_key.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.PKCS1,
)
for encoded in (templates["private_der"], traditional_der):
loaded = compat.load_key_bytes(encoded)
self.assertTrue(loaded.is_private)
self.assertIsNone(loaded.curve)
self.assertEqual(loaded.key_size, 2048)
self.assertEqual(loaded.public_der, templates["public_der"])
public = compat.public_key_from_der(pkcs1_public_der)
self.assertEqual(public.public_der, templates["public_der"])
pkcs1_public_pem = private_key.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.PKCS1,
)
self.assertEqual(compat.load_key_bytes(pkcs1_public_pem).public_der, templates["public_der"])
encrypted_by_cryptography = private_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.BestAvailableEncryption(b"import-password"),
)
self.assertEqual(
compat.load_key_bytes(encrypted_by_cryptography, "import-password").public_der,
templates["public_der"],
)
def test_spki_key_identifier_known_answer(self) -> None:
# RFC 8032 test 1 Ed25519 public key in SubjectPublicKeyInfo. The expected
# value is SHA-1 of only the 32 bytes in the BIT STRING, independently of
# cryptography's SubjectKeyIdentifier implementation.
spki = bytes.fromhex(
"302a300506032b6570032100"
"d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a"
)
self.assertEqual(
compat.spki_key_identifier(spki).hex(),
"5b27aa5589179770e47575b162a1ded97b8bfc6d",
)
self.assertNotEqual(compat.spki_key_identifier(spki), hashlib.sha1(spki).digest())
def test_unsupported_curve_cert_csr_crl_sha256(self) -> None:
for curve in UNSUPPORTED_CURVES:
with self.subTest(curve=curve):
_require_curve(curve)
key = compat.generate_ec(curve)
cert, csr, crl = _resign_all(self.native_templates, key, "sha256")
self.assertEqual(compat.certificate_public_der(cert), key.public_der)
self.assertEqual(compat.csr_public_der(csr), key.public_der)
self.assertTrue(compat.verify_certificate(cert, cert))
self.assertTrue(compat.verify_csr(csr))
self.assertTrue(compat.verify_crl(crl, cert))
# No call to cryptography.Certificate.public_key() or CSR.public_key()
# is needed (or valid) for these unsupported curves.
def test_unsupported_subject_with_native_rsa_issuer(self) -> None:
_require_curve("sect163k1")
subject_key = compat.generate_ec("sect163k1")
cert = compat.resign_certificate(
self.native_templates["cert_der"],
subject_key.public_der,
self.native_templates["private_der"],
"sha256",
)
self.assertEqual(compat.certificate_public_der(cert), subject_key.public_der)
self.assertTrue(compat.verify_certificate(cert, self.native_templates["cert_der"]))
self.assertFalse(compat.verify_certificate(cert, cert))
def test_rsa_legacy_digest_cert_csr_crl(self) -> None:
for digest in ("sha256", "sha1", "md5", "ripemd160"):
with self.subTest(digest=digest):
key = compat.private_key_from_der(self.native_templates["private_der"])
cert, csr, crl = _resign_all(self.native_templates, key, digest)
self.assertTrue(compat.verify_certificate(cert, cert))
self.assertTrue(compat.verify_csr(csr))
self.assertTrue(compat.verify_crl(crl, cert))
# The signature algorithm must actually change, not merely verify
# under a cached template signature.
parsed_cert = x509.load_der_x509_certificate(cert)
parsed_csr = x509.load_der_x509_csr(csr)
parsed_crl = x509.load_der_x509_crl(crl)
expected_oid = {
"sha256": "1.2.840.113549.1.1.11",
"sha1": "1.2.840.113549.1.1.5",
"md5": "1.2.840.113549.1.1.4",
"ripemd160": "1.3.36.3.3.1.2",
}[digest]
self.assertEqual(parsed_cert.signature_algorithm_oid.dotted_string, expected_oid)
self.assertEqual(parsed_csr.signature_algorithm_oid.dotted_string, expected_oid)
self.assertEqual(parsed_crl.signature_algorithm_oid.dotted_string, expected_oid)
def test_legacy_and_sha3_ec_cert_csr_crl(self) -> None:
for digest in ("sha1", "sha3-256"):
with self.subTest(digest=digest):
_require_curve("prime256v1")
key = compat.generate_ec("prime256v1")
cert, csr, crl = _resign_all(self.native_templates, key, digest)
self.assertTrue(compat.verify_certificate(cert, cert))
self.assertTrue(compat.verify_csr(csr))
self.assertTrue(compat.verify_crl(crl, cert))
self.assertEqual(
x509.load_der_x509_certificate(cert).signature_hash_algorithm.name, digest
)
self.assertEqual(x509.load_der_x509_csr(csr).signature_hash_algorithm.name, digest)
self.assertEqual(x509.load_der_x509_crl(crl).signature_hash_algorithm.name, digest)
def test_bad_signatures_return_false(self) -> None:
key = compat.private_key_from_der(self.native_templates["private_der"])
cert, csr, crl = _resign_all(self.native_templates, key, "sha256")
other_native = rsa.generate_private_key(public_exponent=65537, key_size=2048)
other_cert = _self_signed_certificate(other_native, "Wrong issuer", 2000)
other_der = other_cert.public_bytes(serialization.Encoding.DER)
self.assertFalse(compat.verify_certificate(cert, other_der))
self.assertFalse(compat.verify_crl(crl, other_der))
damaged_csr = bytearray(csr)
damaged_csr[-1] ^= 1
self.assertFalse(compat.verify_csr(bytes(damaged_csr)))
def test_malformed_and_unsupported_inputs_raise_value_error(self) -> None:
with self.assertRaises(ValueError):
compat.public_key_from_der(b"not DER")
with self.assertRaises(ValueError):
compat.private_key_from_der(b"not DER")
with self.assertRaises(ValueError):
compat.certificate_public_der(b"not DER")
with self.assertRaises(ValueError):
compat.csr_public_der(b"not DER")
with self.assertRaises(ValueError):
compat.spki_key_identifier(b"not DER")
with self.assertRaises(ValueError):
compat.verify_certificate(b"not DER", self.native_templates["cert_der"])
with self.assertRaises(ValueError):
compat.verify_csr(b"not DER")
with self.assertRaises(ValueError):
compat.verify_crl(b"not DER", self.native_templates["cert_der"])
with self.assertRaisesRegex(ValueError, "digest"):
compat.resign_crl(
self.native_templates["crl_der"],
self.native_templates["private_der"],
"not-a-digest",
)
def test_pkcs12_modern_serialize_and_load(self) -> None:
encoded = compat.serialize_pkcs12(
"compat friendly name",
self.native_templates["private_der"],
self.native_templates["cert_der"],
[self.native_templates["extra_ca_der"]],
"p12-password",
)
# id-pbes2, aes256-CBC, and id-sha256: modern PBKDF2 encryption and MAC,
# not legacy RC2/3DES PBE with SHA-1.
self.assertIn(bytes.fromhex("06092a864886f70d01050d"), encoded)
self.assertIn(bytes.fromhex("060960864801650304012a"), encoded)
self.assertIn(bytes.fromhex("0609608648016503040201"), encoded)
private_der, cert_der, ca_ders = compat.load_pkcs12(encoded, "p12-password")
self.assertIsNotNone(private_der)
self.assertEqual(
compat.private_key_from_der(private_der).public_der,
self.native_templates["public_der"],
)
self.assertEqual(cert_der, self.native_templates["cert_der"])
self.assertEqual(ca_ders, [self.native_templates["extra_ca_der"]])
native_key, native_cert, native_cas = pkcs12.load_key_and_certificates(
encoded, b"p12-password"
)
self.assertIsNotNone(native_key)
self.assertIsNotNone(native_cert)
self.assertIsNotNone(native_cas)
self.assertEqual(len(native_cas), 1)
with self.assertRaises(ValueError) as caught:
compat.load_pkcs12(encoded, "wrong-password")
self.assertNotIn("wrong-password", str(caught.exception))
def test_pkcs12_loads_cryptography_container(self) -> None:
encoded = pkcs12.serialize_key_and_certificates(
b"native friendly name",
self.native_templates["key_object"],
self.native_templates["cert_object"],
[self.native_templates["extra_ca_object"]],
serialization.BestAvailableEncryption(b"native-p12-password"),
)
private_der, cert_der, ca_ders = compat.load_pkcs12(encoded, "native-p12-password")
self.assertIsNotNone(private_der)
self.assertEqual(
compat.private_key_from_der(private_der).public_der,
self.native_templates["public_der"],
)
self.assertEqual(cert_der, self.native_templates["cert_der"])
self.assertEqual(ca_ders, [self.native_templates["extra_ca_der"]])
def test_pkcs12_supports_unsupported_ec_curve(self) -> None:
_require_curve("secp112r1")
key = compat.generate_ec("secp112r1")
cert = compat.resign_certificate(
self.native_templates["cert_der"], key.public_der, key.private_der, "sha256"
)
encoded = compat.serialize_pkcs12("ec compat", key.private_der, cert, [], "ec-p12")
private_der, loaded_cert, ca_ders = compat.load_pkcs12(encoded, "ec-p12")
self.assertIsNotNone(private_der)
self.assertEqual(compat.private_key_from_der(private_der).public_der, key.public_der)
self.assertEqual(loaded_cert, cert)
self.assertEqual(ca_ders, [])
self.assertTrue(compat.verify_certificate(loaded_cert, loaded_cert))
if __name__ == "__main__":
unittest.main()
server/packages/kca-core/tests/test_xca_db.py
+85
View File
@@ -130,6 +130,91 @@ class XcaDbTest(unittest.TestCase):
self.assertEqual(items[3].name, "Synthetic CRL")
self.assertEqual(items[4].obj["name"], "Synthetic template")
def test_import_openssl_only_curve_database(self) -> None:
key = keys.generate_key("ec", curve="secp112r1")
now = datetime.now(timezone.utc)
spec = default_extension_spec()
spec["basic_constraints"] = {"critical": True, "ca": True, "path_length": None}
cert = certificates.build_self_signed(
subject=_name("Compat XCA CA"),
private_key=key,
serial=43,
not_before=now,
not_after=now + timedelta(days=365),
extension_spec=spec,
)
csr = requests.build_csr(subject=_name("Compat XCA CSR"), private_key=key, extension_spec=spec)
crl = crls.build_crl(
issuer_cert=cert,
issuer_private_key=key,
revocations=[],
last_update=now,
next_update=now + timedelta(days=30),
crl_number=4,
)
encrypted = keys.private_der_bytes(key, password="dbpass")
date = now.strftime("%Y%m%d%H%M%SZ")
with tempfile.NamedTemporaryFile(suffix=".xdb", delete=False) as handle:
path = handle.name
try:
db = sqlite3.connect(path)
db.executescript(
"""
CREATE TABLE settings (key_ CHAR(20) UNIQUE, value TEXT);
CREATE TABLE items (id INTEGER PRIMARY KEY, name VARCHAR(128), type INTEGER, source INTEGER, date CHAR(15), comment VARCHAR(2048), stamp INTEGER DEFAULT 0, del SMALLINT DEFAULT 0);
CREATE TABLE public_keys (item INTEGER, type CHAR(4), hash INTEGER, len INTEGER, "public" TEXT);
CREATE TABLE private_keys (item INTEGER, ownPass INTEGER, private TEXT);
CREATE TABLE x509super (item INTEGER, subj_hash INTEGER, pkey INTEGER, key_hash INTEGER);
CREATE TABLE requests (item INTEGER, hash INTEGER, signed INTEGER, request TEXT);
CREATE TABLE certs (item INTEGER, hash INTEGER, iss_hash INTEGER, serial VARCHAR(64), issuer INTEGER, ca INTEGER, cert TEXT);
CREATE TABLE authority (item INTEGER, template INTEGER, crlExpire CHAR(15), crlNo INTEGER, crlDays INTEGER);
CREATE TABLE crls (item INTEGER, hash INTEGER, num INTEGER, iss_hash INTEGER, issuer INTEGER, crl TEXT);
CREATE TABLE revocations (caId INTEGER, serial VARCHAR(64), date CHAR(15), invaldate CHAR(15), crlNo INTEGER, reasonBit INTEGER);
CREATE TABLE templates (item INTEGER, version INTEGER, template TEXT);
CREATE TABLE takeys (item INTEGER UNIQUE, value TEXT);
"""
)
db.execute("INSERT INTO settings VALUES ('schema', '8')")
for item_id, name, item_type in ((1, "Compat key", 1), (2, "Compat CA", 3), (3, "Compat CSR", 2), (4, "Compat CRL", 4)):
db.execute("INSERT INTO items VALUES (?, ?, ?, 2, ?, '', 0, 0)", (item_id, name, item_type, date))
db.execute(
"INSERT INTO public_keys VALUES (1, 'EC', 0, 112, ?)",
(base64.b64encode(keys.public_der(key)).decode(),),
)
db.execute("INSERT INTO private_keys VALUES (1, 0, ?)", (base64.b64encode(encrypted).decode(),))
db.execute("INSERT INTO x509super VALUES (2, 0, 1, 0)")
db.execute("INSERT INTO x509super VALUES (3, 0, 1, 0)")
db.execute(
"INSERT INTO certs VALUES (2, 0, 0, '2b', NULL, 1, ?)",
(base64.b64encode(certificates.certificate_der(cert)).decode(),),
)
db.execute("INSERT INTO authority VALUES (2, NULL, NULL, 0, 30)")
db.execute(
"INSERT INTO requests VALUES (3, 0, 0, ?)",
(base64.b64encode(requests.csr_der(csr)).decode(),),
)
db.execute(
"INSERT INTO crls VALUES (4, 0, 1, 0, 2, ?)",
(base64.b64encode(crls.crl_der(crl)).decode(),),
)
db.commit()
db.close()
with open(path, "rb") as handle:
items = import_xca_database(handle.read(), "compat.xdb", "dbpass")
finally:
import os
os.unlink(path)
self.assertEqual([item.kind for item in items], ["key", "certificate", "request", "crl"])
self.assertTrue(items[0].is_private)
self.assertEqual(keys.curve_name(items[0].obj), "secp112r1")
self.assertEqual(keys.public_der(items[0].obj), keys.public_der(key))
self.assertEqual(certificates.certificate_info(items[1].obj).public_key_der, keys.public_der(key))
self.assertTrue(requests.verify_csr(items[2].obj))
self.assertTrue(crls.verify_crl(items[3].obj, items[1].obj))
if __name__ == "__main__":
unittest.main()