diff --git a/server/packages/kca-core/src/kca/core/keys.py b/server/packages/kca-core/src/kca/core/keys.py index 0a2560e..3b524f3 100644 --- a/server/packages/kca-core/src/kca/core/keys.py +++ b/server/packages/kca-core/src/kca/core/keys.py @@ -262,6 +262,14 @@ def _int_bytes(value: int) -> bytes: return value.to_bytes((value.bit_length() + 7) // 8 or 1, "big") +_JWK_EC_CURVES = { + "secp256r1": "P-256", + "secp384r1": "P-384", + "secp521r1": "P-521", + "secp256k1": "secp256k1", +} + + def jwk_public(key: PrivateKey | PublicKey, *, kid: Optional[str] = None) -> Dict[str, Any]: pub = _public_key(key) if isinstance(pub, rsa.RSAPublicKey): @@ -273,17 +281,15 @@ def jwk_public(key: PrivateKey | PublicKey, *, kid: Optional[str] = None) -> Dic } elif isinstance(pub, ec.EllipticCurvePublicKey): numbers = pub.public_numbers() - curve = { - "secp256r1": "P-256", - "secp384r1": "P-384", - "secp521r1": "P-521", - "secp256k1": "secp256k1", - }.get(pub.curve.name, pub.curve.name) + curve = _JWK_EC_CURVES.get(pub.curve.name) + if curve is None: + raise ValueError(f"JWK export does not support EC curve {pub.curve.name!r}") + coordinate_size = (pub.curve.key_size + 7) // 8 result = { "kty": "EC", "crv": curve, - "x": _b64u(numbers.x.to_bytes((numbers.x.bit_length() + 7) // 8 or 1, "big")), - "y": _b64u(numbers.y.to_bytes((numbers.y.bit_length() + 7) // 8 or 1, "big")), + "x": _b64u(numbers.x.to_bytes(coordinate_size, "big")), + "y": _b64u(numbers.y.to_bytes(coordinate_size, "big")), } elif isinstance(pub, ed25519.Ed25519PublicKey): result = { @@ -314,7 +320,8 @@ def jwk_private(key: PrivateKey, *, kid: Optional[str] = None) -> Dict[str, Any] } ) elif isinstance(key, ec.EllipticCurvePrivateKey): - result["d"] = _b64u(_int_bytes(key.private_numbers().private_value)) + private_size = (key.curve.key_size + 7) // 8 + result["d"] = _b64u(key.private_numbers().private_value.to_bytes(private_size, "big")) elif isinstance(key, ed25519.Ed25519PrivateKey): result["d"] = _b64u(key.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())) else: diff --git a/server/packages/kca-core/tests/test_core.py b/server/packages/kca-core/tests/test_core.py index 0c72d72..eda4d6f 100644 --- a/server/packages/kca-core/tests/test_core.py +++ b/server/packages/kca-core/tests/test_core.py @@ -1,12 +1,14 @@ """Core domain tests: names, extensions, keys, certificates, CRLs and exports.""" from __future__ import annotations +import base64 import time import unittest from datetime import datetime, timedelta, timezone from cryptography import x509 from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec from kca.core import certificates, crls, crypto, exports, imports, keys, names, requests, templates from kca.core.extensions import ( @@ -16,6 +18,10 @@ from kca.core.extensions import ( ) +def _b64u_decode(value: str) -> bytes: + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + def _name(cn: str = "Test CA") -> x509.Name: return names.name_from_entries( names.entries_from_json( @@ -99,6 +105,52 @@ class KeyTest(unittest.TestCase): self.assertTrue(private) self.assertEqual(keys.public_der(loaded), keys.public_der(key)) + def test_ec_jwk_curve_names_and_private_width(self) -> None: + for curve, curve_name, size in [ + (ec.SECP256R1(), "P-256", 32), + (ec.SECP384R1(), "P-384", 48), + (ec.SECP521R1(), "P-521", 66), + (ec.SECP256K1(), "secp256k1", 32), + ]: + with self.subTest(curve=curve_name): + key = ec.derive_private_key(1, curve) + public = keys.jwk_public(key) + private = keys.jwk_private(key) + self.assertEqual(public["crv"], curve_name) + self.assertEqual(len(_b64u_decode(public["x"])), size) + self.assertEqual(len(_b64u_decode(public["y"])), size) + self.assertEqual(_b64u_decode(private["d"]), b"\0" * (size - 1) + b"\1") + + def test_ec_jwk_preserves_leading_zeros(self) -> None: + for private_value, leading_coordinate in ((43, "y"), (379, "x")): + with self.subTest(private_value=private_value): + key = ec.derive_private_key(private_value, ec.SECP256R1()) + numbers = key.private_numbers() + jwk = keys.jwk_private(key) + coordinate = getattr(numbers.public_numbers, leading_coordinate) + self.assertEqual(coordinate.to_bytes(32, "big")[0], 0) + for name, value in ( + ("x", numbers.public_numbers.x), + ("y", numbers.public_numbers.y), + ("d", numbers.private_value), + ): + self.assertEqual(_b64u_decode(jwk[name]), value.to_bytes(32, "big")) + + def test_ec_jwk_rejects_unregistered_curves(self) -> None: + for curve in ( + ec.SECP192R1(), + ec.SECP224R1(), + ec.BrainpoolP256R1(), + ec.BrainpoolP384R1(), + ec.BrainpoolP512R1(), + ): + with self.subTest(curve=curve.name): + key = ec.derive_private_key(1, curve) + with self.assertRaisesRegex(ValueError, "does not support EC curve"): + keys.jwk_public(key) + with self.assertRaisesRegex(ValueError, "does not support EC curve"): + keys.jwk_private(key) + def test_ssh_and_pvk(self) -> None: key = keys.generate_key("rsa", 2048) public = keys.ssh_public(key, "test")