Fix EC JWK: fixed-width x, y, d encoding; restrict to registered curve names

- Encode EC coordinates and private d with curve-width (32/48/66 bytes) to satisfy RFC 7518,
  preserving leading zero octets.
- Restrict JWK  values to registered JOSE names: P-256, P-384, P-521, secp256k1.
  Raise ValueError for unregistered curves (P-192, P-224, Brainpool).
- Add regression tests: deterministic coordinate/private-value widths, leading-zero
  preservation, supported curve names, and rejection of unsupported curves.
- Mypy clean; all core and backend tests pass.
This commit is contained in:
2026-10-26 12:06:16 +08:00
parent aee1025015
commit 1032b10c49
2 changed files with 68 additions and 9 deletions
+16 -9
View File
@@ -262,6 +262,14 @@ def _int_bytes(value: int) -> bytes:
return value.to_bytes((value.bit_length() + 7) // 8 or 1, "big") 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]: def jwk_public(key: PrivateKey | PublicKey, *, kid: Optional[str] = None) -> Dict[str, Any]:
pub = _public_key(key) pub = _public_key(key)
if isinstance(pub, rsa.RSAPublicKey): 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): elif isinstance(pub, ec.EllipticCurvePublicKey):
numbers = pub.public_numbers() numbers = pub.public_numbers()
curve = { curve = _JWK_EC_CURVES.get(pub.curve.name)
"secp256r1": "P-256", if curve is None:
"secp384r1": "P-384", raise ValueError(f"JWK export does not support EC curve {pub.curve.name!r}")
"secp521r1": "P-521", coordinate_size = (pub.curve.key_size + 7) // 8
"secp256k1": "secp256k1",
}.get(pub.curve.name, pub.curve.name)
result = { result = {
"kty": "EC", "kty": "EC",
"crv": curve, "crv": curve,
"x": _b64u(numbers.x.to_bytes((numbers.x.bit_length() + 7) // 8 or 1, "big")), "x": _b64u(numbers.x.to_bytes(coordinate_size, "big")),
"y": _b64u(numbers.y.to_bytes((numbers.y.bit_length() + 7) // 8 or 1, "big")), "y": _b64u(numbers.y.to_bytes(coordinate_size, "big")),
} }
elif isinstance(pub, ed25519.Ed25519PublicKey): elif isinstance(pub, ed25519.Ed25519PublicKey):
result = { result = {
@@ -314,7 +320,8 @@ def jwk_private(key: PrivateKey, *, kid: Optional[str] = None) -> Dict[str, Any]
} }
) )
elif isinstance(key, ec.EllipticCurvePrivateKey): 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): elif isinstance(key, ed25519.Ed25519PrivateKey):
result["d"] = _b64u(key.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())) result["d"] = _b64u(key.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption()))
else: else:
@@ -1,12 +1,14 @@
"""Core domain tests: names, extensions, keys, certificates, CRLs and exports.""" """Core domain tests: names, extensions, keys, certificates, CRLs and exports."""
from __future__ import annotations from __future__ import annotations
import base64
import time import time
import unittest import unittest
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from cryptography import x509 from cryptography import x509
from cryptography.hazmat.primitives import serialization 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 import certificates, crls, crypto, exports, imports, keys, names, requests, templates
from kca.core.extensions import ( 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: def _name(cn: str = "Test CA") -> x509.Name:
return names.name_from_entries( return names.name_from_entries(
names.entries_from_json( names.entries_from_json(
@@ -99,6 +105,52 @@ class KeyTest(unittest.TestCase):
self.assertTrue(private) self.assertTrue(private)
self.assertEqual(keys.public_der(loaded), keys.public_der(key)) 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: def test_ssh_and_pvk(self) -> None:
key = keys.generate_key("rsa", 2048) key = keys.generate_key("rsa", 2048)
public = keys.ssh_public(key, "test") public = keys.ssh_public(key, "test")