From a3e96d3840839a5b4903e76383711134697dfef9 Mon Sep 17 00:00:00 2001 From: Tomasz Swierszcz Date: Wed, 2 Sep 2026 16:53:28 +0200 Subject: [PATCH] fix(ngts): enable service-generated CSR enrollment on CIT-only zones [VC-59232] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NGTS + csr_origin=service failed with "the zone format is incorrect" on a bare CIT alias (local CSR worked). Root cause: NGTS overrode public get_policy but not private _get_policy, so the inherited service-CSR builder hit Cloud's zone backslash-split and Application-owner resolution — both invalid for NGTS. - Override private _get_policy(zone, subject_cn_to_str) in NGTSConnection: resolve the CIT via the CIT-only path, no owner resolution (Go GetPolicy / GetPolicyWithRegex parity). Public get_policy passes True, service path False. - _retrieve_service_generated_cert: encode key_password to bytes (None -> b"") so SealedBox.encrypt no longer raises TypeError; fixes Cloud/VaaS retrieve too. - Service-CSR SANs: emit each type independently instead of gating on san_dns, which silently dropped IP/email/URI-only SANs (Go getCsrAttributes parity). Verified live end-to-end against an NGTS tenant; offline suite green (60) + bandit clean. Tests added/hardened in tests/test_local_methods.py. --- tests/test_local_methods.py | 169 +++++++++++++++++++++++++++++++++++- vcert/connection_cloud.py | 32 ++++--- vcert/connection_ngts.py | 22 ++++- 3 files changed, 208 insertions(+), 15 deletions(-) diff --git a/tests/test_local_methods.py b/tests/test_local_methods.py index 94c8066..dd50aa0 100644 --- a/tests/test_local_methods.py +++ b/tests/test_local_methods.py @@ -25,7 +25,7 @@ from assets import POLICY_CLOUD1, POLICY_TPP1, EXAMPLE_CSR, EXAMPLE_CHAIN from vcert import (CloudConnection, KeyType, TPPConnection, CertificateRequest, ZoneConfig, CertField, FakeConnection, NGTSConnection, RevocationRequest, logger, CSR_ORIGIN_SERVICE) -from vcert.connection_cloud import URLS +from vcert.connection_cloud import (URLS, CSR_ATTR_CN, CSR_ATTR_SANS_BY_TYPE, CSR_ATTR_SANS_IP_ADDR) from vcert.connection_ngts import (_parse_ngts_zone, DEFAULT_API_URL, DEFAULT_TOKEN_URL, TRUSTED_TOKEN_HOST_SUFFIX) from vcert.errors import (ClientBadData, ServerUnexptedBehavior, VenafiError, VenafiConnectionError, @@ -679,6 +679,173 @@ def test_ngts_get_policy_missing_cit_raises(self): with self.assertRaises(VenafiError): conn.get_policy("does-not-exist") + # -- NGTS service-generated CSR (offline) ------------------------------------------------- + # + # Regression guard for the "Invalid Zone [...]. The zone format is incorrect" bug: on a bare + # CIT-alias zone, csr_origin=service used to fall through to Cloud's _get_policy -> + # _get_template_by_id -> _parse_zone (backslash split) and then resolve_cloud_owners_names -> + # _get_app_details_by_name (Application API NGTS has no layer for). NGTS now overrides the + # PRIVATE _get_policy, so the service path resolves the CIT the CIT-only, owner-free way. + + @staticmethod + def _ngts_cit_dict(name="openssl-360d-max", cn_regexes=None): + return { + 'id': 'cit-123', + 'name': name, + 'certificateAuthority': 'DIGICERT', + 'certificateAuthorityAccountId': 'acct-1', + 'certificateAuthorityProductOptionId': 'po-1', + 'subjectCNRegexes': cn_regexes or ['.*'], + 'sanRegexes': ['.*'], + 'keyReuse': False, + 'validityPeriod': 'P90D', + 'csrUploadAllowed': True, + 'keyGeneratedByVenafiAllowed': True, + 'keyTypes': [{'keyType': 'RSA', 'keyLengths': [2048, 4096]}], + } + + def test_ngts_service_csr_uses_cit_only_zone(self): + # THE regression test: a bare CIT-alias zone must NOT raise "zone format is incorrect", and + # the owner-resolution landmine must stay dead. + conn = self._ngts_conn(access_token='t', token_url=None) + info = CertificateAuthorityInfo('DIGICERT', 'acct-key', 'Product') + captured = {} + + def fake_post(url, data=None): + captured['url'] = url + captured['data'] = data + return HTTPStatus.CREATED, {'certificateRequests': [{'id': 'req-1', 'certificateIds': ['c1']}]} + + req = CertificateRequest(common_name="host.example.com", csr_origin=CSR_ORIGIN_SERVICE) + with mock.patch.object(conn, '_get_cit', return_value=self._ngts_cit_dict()), \ + mock.patch.object(conn, '_get_ca_info', return_value=info), \ + mock.patch.object(conn, '_post', side_effect=fake_post), \ + mock.patch.object(conn, 'resolve_cloud_owners_names') as resolve_owners: + result = conn.request_cert(req, "openssl-360d-max") # bare CIT alias, no backslash + + self.assertTrue(result) + body = captured['data'] + self.assertTrue(body.get('isVaaSGenerated')) + self.assertIn('applicationServerTypeId', body) + self.assertEqual(body.get('certificateIssuingTemplateId'), 'cit-123') + self.assertNotIn('applicationId', body) # NGTS has no Application layer + self.assertNotIn('certificateSigningRequest', body) # service-generated: server makes the CSR + resolve_owners.assert_not_called() # owner-resolution landmine must stay dead + self.assertEqual(req.id, 'req-1') + # Assert the csrAttributes CONTENTS, not just the key's presence: a build that dropped or + # mangled the CN would still leave the key in place. + self.assertEqual(body['csrAttributes'][CSR_ATTR_CN], 'host.example.com') + + def test_ngts_service_csr_includes_ip_only_sans(self): + # Regression guard for the SAN-gating bug: the service-CSR csrAttributes builder used to nest + # the whole subjectAlternativeNamesByType map under "if len(san_dns) > 0", silently dropping + # IP/email/URI-only SANs. Each SAN type must now be emitted independently (Go parity). + conn = self._ngts_conn(access_token='t', token_url=None) + info = CertificateAuthorityInfo('DIGICERT', 'acct-key', 'Product') + captured = {} + + def fake_post(url, data=None): + captured['data'] = data + return HTTPStatus.CREATED, {'certificateRequests': [{'id': 'req-1', 'certificateIds': ['c1']}]} + + req = CertificateRequest(common_name="host.example.com", csr_origin=CSR_ORIGIN_SERVICE) + req.ip_addresses = ['10.0.0.5'] # IP SAN only; no DNS SAN + with mock.patch.object(conn, '_get_cit', return_value=self._ngts_cit_dict()), \ + mock.patch.object(conn, '_get_ca_info', return_value=info), \ + mock.patch.object(conn, '_post', side_effect=fake_post), \ + mock.patch.object(conn, 'resolve_cloud_owners_names'): + conn.request_cert(req, "openssl-360d-max") + + sans = captured['data']['csrAttributes'][CSR_ATTR_SANS_BY_TYPE] + self.assertEqual(sans[CSR_ATTR_SANS_IP_ADDR], ['10.0.0.5']) + + def test_ngts_private_get_policy_is_cit_only_and_ownerless(self): + # The PRIVATE _get_policy (used by the service-CSR path) resolves via the CIT-only path, + # never resolves owners, and honours subject_cn_to_str (the service path passes False). + conn = self._ngts_conn(access_token='t', token_url=None) + info = CertificateAuthorityInfo('DIGICERT', 'acct-key', 'Product') + + def get_policy(subject_cn_to_str): + with mock.patch.object(conn, '_get_cit', + return_value=self._ngts_cit_dict(name='my-template', + cn_regexes=['.*\\.example\\.com'])), \ + mock.patch.object(conn, '_get_ca_info', return_value=info), \ + mock.patch.object(conn, 'resolve_cloud_owners_names') as resolve_owners: + ps = conn._get_policy("my-template", subject_cn_to_str=subject_cn_to_str) + resolve_owners.assert_not_called() # no Application owners on NGTS, either way + return ps + + # subject_cn_to_str=False (the service-CSR path): domains stay the RAW regexes so the + # inherited CN-vs-policy validation in _get_service_generated_csr_attr can re.match them. + ps_false = get_policy(subject_cn_to_str=False) + self.assertIsInstance(ps_false, PolicySpecification) + self.assertEqual(ps_false.policy.domains, ['.*\\.example\\.com']) + + # subject_cn_to_str=True (the public get_policy path): the regex escaping is stripped to a + # human-readable domain string. This is what proves the argument is actually honoured. + ps_true = get_policy(subject_cn_to_str=True) + self.assertEqual(ps_true.policy.domains, ['.*.example.com']) + + def test_ngts_service_csr_missing_cit_raises_clean_error(self): + # An unknown CIT alias must fail with the NGTS "issuing template not found" error, not a + # zone-format error. Two paths need covering: + conn = self._ngts_conn(access_token='t', token_url=None) + + # (1) The NEW _get_policy override's own missing-CIT branch (this is the code the service-CSR + # path reaches via _get_service_generated_csr_attr). Call it directly: request_cert would + # short-circuit at its earlier _get_cit_or_fail and never enter _get_policy. + with mock.patch.object(conn, '_get_cit', return_value=None): + with self.assertRaises(VenafiError) as ctx: + conn._get_policy("no-such-cit", subject_cn_to_str=False) + self.assertIn("issuing template not found", str(ctx.exception)) + self.assertNotIn("zone format", str(ctx.exception)) + + # (2) End-to-end request_cert also surfaces a clean error (from its line-327 guard), never a + # zone-format error. + req = CertificateRequest(common_name="host.example.com", csr_origin=CSR_ORIGIN_SERVICE) + with mock.patch.object(conn, '_get_cit', return_value=None): + with self.assertRaises(VenafiError) as ctx: + conn.request_cert(req, "no-such-cit") + self.assertNotIn("zone format", str(ctx.exception)) + + def test_service_generated_retrieve_encodes_passphrase_to_bytes(self): + # Regression: SealedBox.encrypt requires bytes, but request.key_password is a str. + # Without encoding, service-generated-CSR retrieval raised "TypeError: input message + # must be bytes" (verified live on NGTS). The retrieve path is inherited from Cloud, so + # this also covers VaaS service-generated CSR. + conn = self._ngts_conn(access_token='t', token_url=None) + captured = {} + + class _FakeBox: + def __init__(self, _pk): + pass + + def encrypt(self, msg): + captured['type'] = type(msg).__name__ + captured['msg'] = msg + return b'ciphertext' + + dek = mock.Mock() + dek.public_key = b'pub' + req = CertificateRequest(common_name="x.example.com", csr_origin=CSR_ORIGIN_SERVICE) + req.key_password = "s3cret" + req.cert_guid = "guid-1" + with mock.patch('vcert.connection_cloud.SealedBox', _FakeBox), \ + mock.patch('vcert.connection_cloud.zip_to_pem', return_value=("CERT", "CHAIN", "KEY")), \ + mock.patch.object(conn, '_post', return_value=(HTTPStatus.OK, b'zipdata')): + cert = conn._retrieve_service_generated_cert(req, dek) + self.assertEqual(captured['type'], 'bytes') # not str -> no TypeError + self.assertEqual(captured['msg'], b"s3cret") + self.assertEqual(cert.cert, "CERT") + + # None passphrase must not blow up either (encodes to empty bytes). + req.key_password = None + with mock.patch('vcert.connection_cloud.SealedBox', _FakeBox), \ + mock.patch('vcert.connection_cloud.zip_to_pem', return_value=("CERT", "CHAIN", "KEY")), \ + mock.patch.object(conn, '_post', return_value=(HTTPStatus.OK, b'zipdata')): + conn._retrieve_service_generated_cert(req, dek) + self.assertEqual(captured['msg'], b"") + # -- Cloud / NGTS revoke (offline) -------------------------------------------------------- # # Cloud and NGTS revoke via the GraphQL CA-operations `revokeCertificate` mutation (no REST diff --git a/vcert/connection_cloud.py b/vcert/connection_cloud.py index fdc2169..7523174 100644 --- a/vcert/connection_cloud.py +++ b/vcert/connection_cloud.py @@ -1149,17 +1149,20 @@ def _get_service_generated_csr_attr(self, request, zone): elif ps.defaults and ps.defaults.subject and ps.defaults.subject.country: csr_attr_map[CSR_ATTR_COUNTRY] = ps.defaults.subject.country - if len(request.san_dns) > 0: - sans = dict() - if request.san_dns and len(request.san_dns) > 0: - sans[CSR_ATTR_SANS_DNS] = request.san_dns - if request.ip_addresses and len(request.ip_addresses) > 0: - sans[CSR_ATTR_SANS_IP_ADDR] = request.ip_addresses - if request.email_addresses and len(request.email_addresses) > 0: - sans[CSR_ATTR_SANS_EMAIL_ADDR] = request.email_addresses - if request.uniform_resource_identifiers and len(request.uniform_resource_identifiers) > 0: - sans[CSR_ATTR_SANS_URIS] = request.uniform_resource_identifiers - + # Emit each SAN type independently (parity with Go's getCsrAttributes): gating the whole + # block on san_dns silently dropped IP/email/URI-only service CSRs. Only attach the map when + # at least one SAN type is present. + sans = dict() + if request.san_dns and len(request.san_dns) > 0: + sans[CSR_ATTR_SANS_DNS] = request.san_dns + if request.ip_addresses and len(request.ip_addresses) > 0: + sans[CSR_ATTR_SANS_IP_ADDR] = request.ip_addresses + if request.email_addresses and len(request.email_addresses) > 0: + sans[CSR_ATTR_SANS_EMAIL_ADDR] = request.email_addresses + if request.uniform_resource_identifiers and len(request.uniform_resource_identifiers) > 0: + sans[CSR_ATTR_SANS_URIS] = request.uniform_resource_identifiers + + if sans: csr_attr_map[CSR_ATTR_SANS_BY_TYPE] = sans if request.key_type: @@ -1263,7 +1266,12 @@ def _retrieve_service_generated_cert(self, request, dek_info): :rtype: Certificate """ box = SealedBox(dek_info.public_key) - encrypted_key_pass = box.encrypt(request.key_password) + # SealedBox.encrypt requires bytes; request.key_password is a str (or None). Encode it so + # service-generated-CSR retrieval doesn't raise "TypeError: input message must be bytes". + key_password = request.key_password or "" + if isinstance(key_password, str): + key_password = key_password.encode("utf-8") + encrypted_key_pass = box.encrypt(key_password) body = { 'exportFormat': 'PEM', 'encryptedPrivateKeyPassphrase': base64.b64encode(encrypted_key_pass).decode('utf-8'), diff --git a/vcert/connection_ngts.py b/vcert/connection_ngts.py index cf99efb..10b3b25 100644 --- a/vcert/connection_ngts.py +++ b/vcert/connection_ngts.py @@ -470,6 +470,25 @@ def get_policy(self, zone): :param str zone: the CIT alias (NGTS zones are a CIT alias only - no Application\\CIT split) :rtype: PolicySpecification """ + return self._get_policy(zone, subject_cn_to_str=True) + + def _get_policy(self, zone, subject_cn_to_str): + """ + NGTS override of ``CloudConnection._get_policy``. Besides backing the public ``get_policy`` + it is the method the inherited service-generated-CSR builder calls + (``_get_service_generated_csr_attr`` -> ``self._get_policy(zone, subject_cn_to_str=False)``). + Cloud's version splits the zone on a backslash (``Application\\CIT``) via + ``_get_template_by_id`` -> ``_parse_zone`` and then resolves Application owners + (``resolve_cloud_owners_names`` -> ``_get_app_details_by_name``); both are wrong for NGTS + (CIT-alias-only zone, no Application layer). This resolves the CIT through the NGTS CIT-only + path and builds the spec without owner resolution, matching Go's NGTS ``GetPolicyWithRegex`` + (which passes ``subject_cn_to_str=false`` for the service-CSR path). Without this override, + ``csr_origin=service`` on NGTS raised "Invalid Zone [...]. The zone format is incorrect". + + :param str zone: the CIT alias (NGTS zones are a CIT alias only - no Application\\CIT split) + :param bool subject_cn_to_str: + :rtype: PolicySpecification + """ cit_data = self._get_cit_or_fail(zone) cit = self._parse_policy_response_to_object(cit_data) @@ -478,8 +497,7 @@ def get_policy(self, zone): if not info: raise VenafiError("Certificate Authority info not found") - ps = build_policy_spec(cit, info, subject_cn_to_str=True) - return ps + return build_policy_spec(cit, info, subject_cn_to_str=subject_cn_to_str) def set_policy(self, zone, policy_spec): """