fix: normalize AT%KEYGEN CSRs

src/nrfcloud_utils/modem_credentials_parser.py

@@ -10,6 +10,7 @@
 import base64
 import hashlib
 import logging
+import textwrap
 from cryptography.hazmat.primitives import serialization
 from cryptography import x509
 from nrfcloud_utils.cli_helpers import write_file, setup_logging
@@ -70,10 +71,124 @@ def base64_decode(string):
     """
     add padding before decoding.
     """
-    padding = 4 - (len(string) % 4)
+    # Base64url data can omit padding; only add what is required.
+    padding = (-len(string)) % 4
     string = string + ("=" * padding)
     return base64.urlsafe_b64decode(string)
 
+def _decode_tlv_length(data, offset):
+    """Decode DER length and return (value_length, header_length)."""
+    first = data[offset]
+    if first < 0x80:
+        return first, 1
+
+    nbytes = first & 0x7F
+    if nbytes == 0:
+        raise ValueError("Indefinite DER length is not supported")
+
+    value_len = 0
+    for i in range(nbytes):
+        value_len = (value_len << 8) | data[offset + 1 + i]
+    return value_len, 1 + nbytes
+
+def _encode_tlv_length(value_len):
+    """Encode DER length bytes for value_len."""
+    if value_len < 0x80:
+        return bytes([value_len])
+
+    out = []
+    value = value_len
+    while value > 0:
+        out.append(value & 0xFF)
+        value >>= 8
+    out.reverse()
+    return bytes([0x80 | len(out), *out])
+
+def _normalize_ecdsa_csr_der(der_bytes):
+    """
+    Normalize CSR DER that encodes ecdsa-with-SHA256 with NULL params.
+
+    Older modem output may include signatureAlgorithm = SEQUENCE(OID, NULL).
+    For ECDSA-with-SHA* this must be encoded without parameters.
+    """
+    # ecdsa-with-SHA256 OID (1.2.840.10045.4.3.2)
+    ecdsa_sha256_oid_tlv = b"\x06\x08\x2A\x86\x48\xCE\x3D\x04\x03\x02"
+
+    if len(der_bytes) < 8 or der_bytes[0] != 0x30:
+        return None
+
+    try:
+        outer_len, outer_len_hdr = _decode_tlv_length(der_bytes, 1)
+    except (IndexError, ValueError):
+        return None
+
+    outer_start = 1 + outer_len_hdr
+    outer_end = outer_start + outer_len
+    if outer_end != len(der_bytes):
+        return None
+
+    # certificationRequestInfo
+    cri_tag_idx = outer_start
+    if der_bytes[cri_tag_idx] != 0x30:
+        return None
+    try:
+        cri_len, cri_len_hdr = _decode_tlv_length(der_bytes, cri_tag_idx + 1)
+    except (IndexError, ValueError):
+        return None
+    cri_start = cri_tag_idx
+    cri_value_start = cri_tag_idx + 1 + cri_len_hdr
+    cri_end = cri_value_start + cri_len
+
+    # signatureAlgorithm
+    sig_alg_tag_idx = cri_end
+    if sig_alg_tag_idx >= outer_end or der_bytes[sig_alg_tag_idx] != 0x30:
+        return None
+    try:
+        sig_alg_len, sig_alg_len_hdr = _decode_tlv_length(der_bytes, sig_alg_tag_idx + 1)
+    except (IndexError, ValueError):
+        return None
+    sig_alg_value_start = sig_alg_tag_idx + 1 + sig_alg_len_hdr
+    sig_alg_end = sig_alg_value_start + sig_alg_len
+    if sig_alg_end > outer_end:
+        return None
+
+    sig_alg_value = der_bytes[sig_alg_value_start:sig_alg_end]
+    # Only normalize the exact legacy ECDSA+NULL form.
+    if not (sig_alg_value.startswith(ecdsa_sha256_oid_tlv) and sig_alg_value.endswith(b"\x05\x00")):
+        return None
+    if len(sig_alg_value) != len(ecdsa_sha256_oid_tlv) + 2:
+        return None
+
+    # signature BIT STRING (keep as-is)
+    sig_tag_idx = sig_alg_end
+    if sig_tag_idx >= outer_end or der_bytes[sig_tag_idx] != 0x03:
+        return None
+    try:
+        sig_len, sig_len_hdr = _decode_tlv_length(der_bytes, sig_tag_idx + 1)
+    except (IndexError, ValueError):
+        return None
+    sig_value_start = sig_tag_idx + 1 + sig_len_hdr
+    sig_end = sig_value_start + sig_len
+    if sig_end != outer_end:
+        return None
+
+    cri_tlv = der_bytes[cri_start:cri_end]
+    normalized_sig_alg = b"\x30" + _encode_tlv_length(len(ecdsa_sha256_oid_tlv)) + ecdsa_sha256_oid_tlv
+    sig_tlv = der_bytes[sig_tag_idx:sig_end]
+
+    new_outer_value = cri_tlv + normalized_sig_alg + sig_tlv
+    return b"\x30" + _encode_tlv_length(len(new_outer_value)) + new_outer_value
+
+def _csr_der_to_pem(der_bytes):
+    """Convert CSR DER bytes to PEM without changing the original DER payload."""
+    b64 = base64.b64encode(der_bytes).decode("ascii")
+    wrapped = "\n".join(textwrap.wrap(b64, 64))
+    return (
+        "-----BEGIN CERTIFICATE REQUEST-----\n"
+        + wrapped
+        + "\n-----END CERTIFICATE REQUEST-----\n"
+    ).encode("ascii")
+
 def format_uuid(hex_str):
     return '{0}-{1}-{2}-{3}-{4}'.format(hex_str[:8],    hex_str[8:12],
                                         hex_str[12:16], hex_str[16:20],
@@ -179,30 +294,26 @@ def parse_keygen_output(keygen_str):
     body = body_cose[0]
 
     # Decode base64url to binary
-    body_bytes = base64_decode(body)
+    payload_body_bytes = base64_decode(body)
+    body_bytes = payload_body_bytes
 
-    # This can be either a CSR or device public key
     try:
-        # Try to load CSR, if it fails, assume public key
         csr = x509.load_der_x509_csr(body_bytes)
+    except:
+        logger.warning("normalizing CSR")
+        normalized = _normalize_ecdsa_csr_der(body_bytes)
+        logger.warning(f"keygen_str: {keygen_str}, normalized: {base64.urlsafe_b64encode(normalized)}")
+        csr = x509.load_der_x509_csr(normalized)
 
-    except ValueError:
-        # Handle public key only
-        pub_key = serialization.load_der_public_key(body_bytes)
-        pub_key_bytes = pub_key.public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-
-    else:
-        # CSR loaded, logger.info it
-        csr_pem_bytes = csr.public_bytes(serialization.Encoding.PEM)
-        csr_pem_list = str(csr_pem_bytes.decode()).split('\n')
-        logger.info(csr_pem_bytes.decode().replace('\n', '\\n'))
+    csr_pem_bytes = csr.public_bytes(serialization.Encoding.PEM)
+    logger.info(csr_pem_bytes.decode().replace('\n', '\\n'))
 
-        # Extract public key
-        pub_key_bytes = csr.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+    # Extract public key
+    pub_key_bytes = csr.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
     logger.info("Device public key: {}".format(pub_key_bytes.decode().replace('\n', '\\n')))
 
-    payload_digest = hashlib.sha256(body_bytes).hexdigest()
+    payload_digest = hashlib.sha256(payload_body_bytes).hexdigest()
     logger.info(f"SHA256 Digest: {payload_digest}")
 
     # Get optional cose

tests/test_modem_credentials_parser.py

@@ -8,12 +8,21 @@
 
 KEYGEN_CSR = "MIIBCjCBrwIBADAvMS0wKwYDVQQDDCQ1MDM2MzE1NC0zOTMxLTQ0ZjAtODAyMi0xMjFiNjQwMTYyN2QwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQqD6pNfa29o_EXnw62bnQWr8-JqsNh_HZxS3k3bMD4KZ8-qxnvgeoiqQ5zAycEP_Wcmzqypvwyf3qWMrZ2VB5aoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA-gwDAYIKoZIzj0EAwIFAANIADBFAiEAv7OLZ_dXbszfhhjcLMUT72wTmw-z6GlgWxVhyWgR27ACIAvY_lPu3yfYZY5AL6uYTkUFp4GQkbSOUC_lsHyCxOuG.0oRDoQEmoQRBIVhL2dn3hQlQUDYxVDkxRPCAIhIbZAFifUERWCBwKj1W8FsvclMdZQgl4gBB4unZMYw0toU6uQZuXHLoDFAbhyLuHetYFWbiyxNZsnzSWEDUiTl7wwFt0hEsCiEQsxj-hCtpBk8Za8UXfdAycpx2faCOPJIrkfmiSS8-Y6_2tTAoAMN1BiWiTOimY1wZE3Ud"
 KEYGEN_CSR_DATA = {
-    "CSR_PEM": b'-----BEGIN CERTIFICATE REQUEST-----\nMIIBCjCBrwIBADAvMS0wKwYDVQQDDCQ1MDM2MzE1NC0zOTMxLTQ0ZjAtODAyMi0x\nMjFiNjQwMTYyN2QwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQqD6pNfa29o/EX\nnw62bnQWr8+JqsNh/HZxS3k3bMD4KZ8+qxnvgeoiqQ5zAycEP/Wcmzqypvwyf3qW\nMrZ2VB5aoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA+gwDAYIKoZIzj0E\nAwIFAANIADBFAiEAv7OLZ/dXbszfhhjcLMUT72wTmw+z6GlgWxVhyWgR27ACIAvY\n/lPu3yfYZY5AL6uYTkUFp4GQkbSOUC/lsHyCxOuG\n-----END CERTIFICATE REQUEST-----\n',
+    "CSR_PEM": b'-----BEGIN CERTIFICATE REQUEST-----\nMIIBCDCBrwIBADAvMS0wKwYDVQQDDCQ1MDM2MzE1NC0zOTMxLTQ0ZjAtODAyMi0x\nMjFiNjQwMTYyN2QwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQqD6pNfa29o/EX\nnw62bnQWr8+JqsNh/HZxS3k3bMD4KZ8+qxnvgeoiqQ5zAycEP/Wcmzqypvwyf3qW\nMrZ2VB5aoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA+gwCgYIKoZIzj0E\nAwIDSAAwRQIhAL+zi2f3V27M34YY3CzFE+9sE5sPs+hpYFsVYcloEduwAiAL2P5T\n7t8n2GWOQC+rmE5FBaeBkJG0jlAv5bB8gsTrhg==\n-----END CERTIFICATE REQUEST-----\n',
     "PUB_PEM" : b'-----BEGIN PUBLIC KEY-----\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEKg+qTX2tvaPxF58Otm50Fq/PiarD\nYfx2cUt5N2zA+CmfPqsZ74HqIqkOcwMnBD/1nJs6sqb8Mn96ljK2dlQeWg==\n-----END PUBLIC KEY-----\n',
     "UUID" : "50363154-3931-44f0-8022-121b6401627d",
     "SEC_TAG" : "17",
 }
 
+KEYGEN_CSR_NORMALIZED = "MIIBBzCBrwIBADAvMS0wKwYDVQQDDCQ1MDM0Mzk1Ni0zMDM3LTQ3MzgtODAwZC0xZjI5ZTBhMGY3YmYwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAR0N0OGqO91QKMOJdJpSsAQbotqXNto_B-3rAggRKfvLNjQxhwnNoBAZ-b8ecoNLAR3kZnx5G5K3PjNCSj3KD8hoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA-gwCgYIKoZIzj0EAwIDRwAwRAIgV0WWIwPLZCMXsl8a-Umc7yTL3V4FuA3a9IEB0I1zjLECIHkrki8nS8X27nfYSLnbz_5lBvOZQjMn1ttZChLW31WO.0oRDoQEmoQRBIVhN2dn3hQlQUDQ5VjA3RziADR8p4KD3v0MZBNNYIGn-n_c55m-bB8f3kplNZ51kSw1bFY6PN31dlRwicMIEUCcaM3CUJG_LHnzHUntrtTJYQFttq5NPZ1TWYvpAUNyulC71-ZJdtCdgaT3ejcV_GGFuD-7Z2A163d1uVjddQ7VukItM2jFWi2i5C4gnNAPmzeM"
+KEYGEN_CSR_NORMALIZED_DATA = {
+    "CSR_PEM": b'-----BEGIN CERTIFICATE REQUEST-----\nMIIBBzCBrwIBADAvMS0wKwYDVQQDDCQ1MDM0Mzk1Ni0zMDM3LTQ3MzgtODAwZC0x\nZjI5ZTBhMGY3YmYwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAR0N0OGqO91QKMO\nJdJpSsAQbotqXNto/B+3rAggRKfvLNjQxhwnNoBAZ+b8ecoNLAR3kZnx5G5K3PjN\nCSj3KD8hoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA+gwCgYIKoZIzj0E\nAwIDRwAwRAIgV0WWIwPLZCMXsl8a+Umc7yTL3V4FuA3a9IEB0I1zjLECIHkrki8n\nS8X27nfYSLnbz/5lBvOZQjMn1ttZChLW31WO\n-----END CERTIFICATE REQUEST-----\n',
+    "PUB_PEM" : b'-----BEGIN PUBLIC KEY-----\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEdDdDhqjvdUCjDiXSaUrAEG6Lalzb\naPwft6wIIESn7yzY0MYcJzaAQGfm/HnKDSwEd5GZ8eRuStz4zQko9yg/IQ==\n-----END PUBLIC KEY-----\n',
+    "UUID" : "50343956-3037-4738-800d-1f29e0a0f7bf",
+    "SEC_TAG" : "1235",
+}
+
+
 ATTESTTOKEN = "2dn3hQFQUDYxVDkxRPCAIhIbZAFifQNQGv86y_GmR2SiY0wmRsHGVFDT791_BPH8YOWFiyCHND1q.0oRDoQEmoQRBIfZYQGuXwJliinHc6xDPruiyjsaXyXZbZVpUuOhHG9YS8L05VuglCcJhMN4EUhWVGpaHgNnHHno6ahi-d5tOeZmAcNY"
 ATTESTTOKEN_DATA = {
     "UUID" : "50363154-3931-44f0-8022-121b6401627d",
@@ -28,6 +37,14 @@ def test_parse_keygen_output_csr(self):
         assert dev_uuid_hex_str == KEYGEN_CSR_DATA["UUID"]
         assert sec_tag_str == KEYGEN_CSR_DATA["SEC_TAG"]
 
+    def test_parse_keygen_output_csr_normalized(self):
+        csr_pem_bytes, pub_key_bytes, dev_uuid_hex_str, sec_tag_str = \
+            modem_credentials_parser.parse_keygen_output(KEYGEN_CSR_NORMALIZED)
+        assert csr_pem_bytes == KEYGEN_CSR_NORMALIZED_DATA["CSR_PEM"]
+        assert pub_key_bytes == KEYGEN_CSR_NORMALIZED_DATA["PUB_PEM"]
+        assert dev_uuid_hex_str == KEYGEN_CSR_NORMALIZED_DATA["UUID"]
+        assert sec_tag_str == KEYGEN_CSR_NORMALIZED_DATA["SEC_TAG"]
+
     def test_attesttoken(self):
         _, _ = modem_credentials_parser.parse_attesttoken_output(ATTESTTOKEN)
         uuid = modem_credentials_parser.get_device_uuid(ATTESTTOKEN)