From dfac37e7d87948034a582f7f96067a41d7dff4d6 Mon Sep 17 00:00:00 2001 From: Tomasz Swierszcz Date: Fri, 4 Sep 2026 15:59:28 +0200 Subject: [PATCH] fix(cloud): tolerate unsupported key sizes/curves in CIT policy parse NGTS zones backed by third-party CAs advertise key lengths the client cannot represent (e.g. DigiCert/ZTPKI CITs advertise RSA 1024). _parse_policy_response_to_object built KeyType() for every advertised size/curve with no guard, so KeyType.__init__ raised BadData and crashed read_zone_conf / get_policy / request_cert for the entire zone - making such CITs unusable via the SDK. Skip entries KeyType cannot represent (log + continue) in both the keyTypes loop and the recommended-settings key, matching the Go SDK which tolerates unknown key lengths (cloud.go passes them through unchecked). Request-side validation stays strict. Live-verified against an NGTS DigiCert CIT: read_zone_conf and enrollment now succeed (RSA 1024 dropped, RSA 2048/3072/4096 + EC p256/p384/p521/ed25519 retained). Adds an offline regression test; offline suite: 56 passed. --- tests/test_local_methods.py | 20 ++++++++++++++++++++ vcert/connection_cloud.py | 27 ++++++++++++++++++++------- 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/tests/test_local_methods.py b/tests/test_local_methods.py index 94c8066..cdc3929 100644 --- a/tests/test_local_methods.py +++ b/tests/test_local_methods.py @@ -189,6 +189,26 @@ def test_parse_cloud_zone1(self): self.assertEqual(p.key_types[1].option, 4096) self.assertTrue(len(p.key_types) == 2) + def test_parse_policy_skips_unsupported_key_entries(self): + # A CIT/policy may advertise key sizes/curves the client cannot represent (real example: + # DigiCert/ZTPKI zones on NGTS advertise RSA 1024). The parse must skip those entries, not + # raise BadData and brick read_zone_conf/get_policy/enrollment for the whole zone. + conn = CloudConnection(token="") + cit = { + "id": "cit-1", + "name": "RequestPolicyDC", + "certificateAuthority": "DIGICERT", + "keyTypes": [ + {"keyType": "RSA", "keyLengths": [2048, 1024, 4096]}, + {"keyType": "EC", "keyCurves": ["P256", "brainpoolP256r1", "P384"]}, + ], + } + p = conn._parse_policy_response_to_object(cit) # must not raise + rsa = [kt.option for kt in p.key_types if kt.key_type == KeyType.RSA] + ec = [kt.option for kt in p.key_types if kt.key_type == KeyType.ECDSA] + self.assertEqual(rsa, [2048, 4096]) # unsupported 1024 dropped + self.assertEqual(ec, ["p256", "p384"]) # unknown brainpool dropped; curves normalized lowercase + # cloud doesnt support ecdsa yet. may be can be enabled in the future # def test_parse_cloud_zone2(self): # conn = CloudConnection(token="") diff --git a/vcert/connection_cloud.py b/vcert/connection_cloud.py index fdc2169..0691ea1 100644 --- a/vcert/connection_cloud.py +++ b/vcert/connection_cloud.py @@ -24,7 +24,7 @@ from .common import (ZoneConfig, CertificateRequest, CommonConnection, Policy, RevocationRequest, get_ip_address, log_errors, MIME_JSON, MIME_TEXT, MIME_ANY, CertField, KeyType, DEFAULT_TIMEOUT, CSR_ORIGIN_SERVICE, CHAIN_OPTION_FIRST, CHAIN_OPTION_LAST) -from .errors import (VenafiConnectionError, ServerUnexptedBehavior, ClientBadData, CertificateRequestError, +from .errors import (VenafiConnectionError, ServerUnexptedBehavior, ClientBadData, BadData, CertificateRequestError, CertificateRenewError, CertificateRevokeError, VenafiError, RetrieveCertificateTimeoutError) from .http_status import HTTPStatus from .logger import get_child @@ -326,11 +326,20 @@ def _parse_policy_response_to_object(d): for kt in d.get('keyTypes', []): key_type = kt['keyType'].lower() if key_type == KeyType.RSA: - for s in kt['keyLengths']: - policy.key_types.append(KeyType(key_type, s)) + for s in kt.get('keyLengths', []): + try: + policy.key_types.append(KeyType(key_type, s)) + except (BadData, KeyError): + # A policy may advertise key sizes the client cannot represent (e.g. RSA 1024). + # Skip them instead of failing the whole policy parse - this keeps read_zone_conf / + # get_policy / enrollment working against such zones (Go tolerates unknown sizes). + log.warning(f"Ignoring unsupported RSA key length advertised by policy: {s}") elif key_type == KeyType.ECDSA: - for s in kt["keyCurves"]: - policy.key_types.append(KeyType(key_type, s)) + for s in kt.get("keyCurves", []): + try: + policy.key_types.append(KeyType(key_type, s)) + except (BadData, KeyError): + log.warning(f"Ignoring unsupported EC curve advertised by policy: {s}") else: log.error(f"Unknown key type: {kt['keyType']}") raise ServerUnexptedBehavior @@ -359,8 +368,12 @@ def _parse_recommended_settings_to_object(d): k_type = key['type'] kl = key['length'] if 'length' in key else None kc = key['curve'] if 'curve' in key else None - kt = KeyType(k_type, kl or kc) - settings.keyType = kt + try: + settings.keyType = KeyType(k_type, kl or kc) + except (BadData, KeyError): + # Same resilience as the keyTypes loop: a recommended key the client cannot + # represent must not abort the whole policy parse. + log.warning(f"Ignoring unsupported recommended key in policy: {k_type} {kl or kc}") return settings