kca-core: fix CSR attribute DER encoding and conformant decoding

- Generated attributes are now explicitly tagged as UTF8String via
  CertificateSigningRequestBuilder.add_attribute(_tag=_ASN1Type.UTF8String),
  so values include proper DER TLV instead of raw bytes.
- The attribute decoder first attempts DER string parsing (UTF8String, PrintableString,
  IA5String, BMPString, long-form lengths), falling back to legacy raw UTF-8 decode for
  previously generated nonconformant CSRs.
- Regression tests verify DER encoding, Unicode round-tripping, supported string types,
  long values (>127 bytes), and legacy raw UTF-8 attributes.
- Mypy passes cleanly (no issues in 42 source files).
This commit is contained in:
2026-10-26 12:06:17 +08:00
parent c263f967dc
commit 4adf7c4882
2 changed files with 54 additions and 18 deletions
@@ -7,6 +7,7 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional
from cryptography import x509 from cryptography import x509
from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives import serialization
from cryptography.x509.name import _ASN1Type
from cryptography.x509.oid import ObjectIdentifier from cryptography.x509.oid import ObjectIdentifier
from .extensions import build_extensions, extension_config_lines, extension_rows, extension_specs from .extensions import build_extensions, extension_config_lines, extension_rows, extension_specs
@@ -62,7 +63,11 @@ def build_csr(
builder = builder.add_extension(extension.value, extension.critical) builder = builder.add_extension(extension.value, extension.critical)
for oid, value in (attributes or {}).items(): for oid, value in (attributes or {}).items():
if value: if value:
builder = builder.add_attribute(ObjectIdentifier(oid), str(value).encode("utf-8")) builder = builder.add_attribute(
ObjectIdentifier(oid),
str(value).encode("utf-8"),
_tag=_ASN1Type.UTF8String,
)
algorithm = hash_algorithm(signature_hash, private_key) algorithm = hash_algorithm(signature_hash, private_key)
return builder.sign(private_key, algorithm) return builder.sign(private_key, algorithm)
@@ -113,25 +118,34 @@ def _signature_name(csr: x509.CertificateSigningRequest) -> str:
return _SIGNATURE_OIDS.get(oid.dotted_string, getattr(oid, "name", None) or oid.dotted_string) return _SIGNATURE_OIDS.get(oid.dotted_string, getattr(oid, "name", None) or oid.dotted_string)
def _decode_der_string(value: bytes) -> Optional[str]:
if len(value) < 2 or value[0] not in (0x0C, 0x13, 0x16, 0x1E):
return None
length = value[1]
offset = 2
if length & 0x80:
count = length & 0x7F
if count == 0 or len(value) < offset + count:
return None
length = int.from_bytes(value[offset : offset + count], "big")
offset += count
if offset + length != len(value):
return None
raw = value[offset:]
encoding = "utf-16-be" if value[0] == 0x1E else "utf-8"
return raw.decode(encoding, "replace")
def _decode_attribute_value(value: bytes) -> str: def _decode_attribute_value(value: bytes) -> str:
# cryptography returns the DER-encoded value bytes. Try UTF-8 first, # DER tag and length bytes can themselves be valid UTF-8, so decode the
# then strip a simple ASN.1 string tag/length header. # ASN.1 string first. Raw UTF-8 remains supported for legacy KCA CSRs.
decoded = _decode_der_string(value)
if decoded is not None:
return decoded
try: try:
return value.decode("utf-8") return value.decode("utf-8")
except UnicodeDecodeError: except UnicodeDecodeError:
pass return value.hex()
if len(value) >= 2 and value[0] in (0x0C, 0x13, 0x16, 0x1E):
length = value[1]
offset = 2
if length & 0x80:
count = length & 0x7F
length = int.from_bytes(value[2 : 2 + count], "big")
offset = 2 + count
raw = value[offset : offset + length]
if value[0] == 0x1E:
return raw.decode("utf-16-be", "replace")
return raw.decode("utf-8", "replace")
return value.hex()
def attribute_rows(csr: x509.CertificateSigningRequest) -> List[Dict[str, Any]]: def attribute_rows(csr: x509.CertificateSigningRequest) -> List[Dict[str, Any]]:
+24 -2
View File
@@ -353,12 +353,34 @@ class RequestTest(unittest.TestCase):
subject=_name("CSR"), subject=_name("CSR"),
private_key=key, private_key=key,
extension_spec=_ca_spec(), extension_spec=_ca_spec(),
attributes={requests.CHALLENGE_PASSWORD_OID: "secret"}, attributes={
requests.CHALLENGE_PASSWORD_OID: "secret",
requests.UNSTRUCTURED_NAME_OID: "KCA ☃",
},
) )
self.assertTrue(requests.verify_csr(csr)) self.assertTrue(requests.verify_csr(csr))
loaded = requests.parse_csr(requests.csr_der(csr)) loaded = requests.parse_csr(requests.csr_der(csr))
self.assertEqual(requests.request_info(loaded).subject, csr.subject) self.assertEqual(requests.request_info(loaded).subject, csr.subject)
self.assertEqual(requests.attribute_values(loaded)[requests.CHALLENGE_PASSWORD_OID], "secret") der = requests.csr_der(loaded)
self.assertIn(b"\x0c\x06secret", der)
self.assertIn(b"\x0c\x07KCA \xe2\x98\x83", der)
values = requests.attribute_values(loaded)
self.assertEqual(values[requests.CHALLENGE_PASSWORD_OID], "secret")
self.assertEqual(values[requests.UNSTRUCTURED_NAME_OID], "KCA ☃")
def test_decode_attribute_der_strings(self) -> None:
long_value = "x" * 128
cases = [
(b"\x0c\x06secret", "secret"),
(b"\x13\x04name", "name"),
(b"\x16\x0bexample.com", "example.com"),
(b"\x1e\x04\x00a\x00b", "ab"),
(b"\x0c\x81\x80" + long_value.encode("ascii"), long_value),
(b"legacy", "legacy"),
]
for encoded, expected in cases:
with self.subTest(encoded=encoded):
self.assertEqual(requests._decode_attribute_value(encoded), expected)
class CrlTest(unittest.TestCase): class CrlTest(unittest.TestCase):