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:
@@ -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,24 +118,33 @@ def _signature_name(csr: x509.CertificateSigningRequest) -> str:
|
||||
return _SIGNATURE_OIDS.get(oid.dotted_string, getattr(oid, "name", None) or oid.dotted_string)
|
||||
|
||||
|
||||
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.
|
||||
try:
|
||||
return value.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
pass
|
||||
if len(value) >= 2 and value[0] in (0x0C, 0x13, 0x16, 0x1E):
|
||||
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
|
||||
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")
|
||||
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:
|
||||
# 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:
|
||||
return value.hex()
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user