diff --git a/server/packages/kca-core/src/kca/core/requests.py b/server/packages/kca-core/src/kca/core/requests.py index 8d1f35c..cc0e87d 100644 --- a/server/packages/kca-core/src/kca/core/requests.py +++ b/server/packages/kca-core/src/kca/core/requests.py @@ -7,6 +7,7 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional from cryptography import x509 from cryptography.hazmat.primitives import serialization +from cryptography.x509.name import _ASN1Type from cryptography.x509.oid import ObjectIdentifier 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) for oid, value in (attributes or {}).items(): 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) 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) +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: - # cryptography returns the DER-encoded value bytes. Try UTF-8 first, - # then strip a simple ASN.1 string tag/length header. + # DER tag and length bytes can themselves be valid UTF-8, so decode the + # 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: return value.decode("utf-8") except UnicodeDecodeError: - pass - 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() + return value.hex() def attribute_rows(csr: x509.CertificateSigningRequest) -> List[Dict[str, Any]]: diff --git a/server/packages/kca-core/tests/test_core.py b/server/packages/kca-core/tests/test_core.py index b3e1531..359faf2 100644 --- a/server/packages/kca-core/tests/test_core.py +++ b/server/packages/kca-core/tests/test_core.py @@ -353,12 +353,34 @@ class RequestTest(unittest.TestCase): subject=_name("CSR"), private_key=key, 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)) loaded = requests.parse_csr(requests.csr_der(csr)) 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):