From 1ab81a2b40c1409542024694657732323329d4fd Mon Sep 17 00:00:00 2001 From: "erkhem.unenbat" Date: Thu, 24 Sep 2026 12:16:36 -0700 Subject: [PATCH 1/2] first iteration --- README-NGTS.md | 48 ++++++++- docs/version_history.md | 4 + tests/test_env.py | 2 + tests/test_local_methods.py | 197 +++++++++++++++++++++++++++++++++++- tests/test_ngts.py | 23 ++++- vcert/__init__.py | 18 +++- vcert/connection_ngts.py | 99 ++++++++++++++++-- 7 files changed, 373 insertions(+), 18 deletions(-) diff --git a/README-NGTS.md b/README-NGTS.md index 8df63b8..ce16348 100644 --- a/README-NGTS.md +++ b/README-NGTS.md @@ -15,9 +15,9 @@ enrollment. This guide covers using it against **Palo Alto Networks Next-Gen Tru (NGTS)**, also known as Strata Cloud Manager. > ๐Ÿ“Œ **NOTE:** Unlike the [Go VCert](https://github.com/Venafi/vcert) project, vcert-python is -> **SDK-only** โ€” there is no CLI, playbook, or certificate provisioning. NGTS support in this -> SDK is **certificate-lifecycle only**: `get_policy`/`set_policy`, SSH, and `get_version` -> raise `NotImplementedError`. +> **SDK-only** โ€” there is no CLI, playbook, or certificate provisioning. NGTS support covers the +> certificate lifecycle plus `get_policy`/`set_policy`; SSH and `get_version` raise +> `NotImplementedError`. ## Quick Links @@ -26,6 +26,7 @@ enrollment. This guide covers using it against **Palo Alto Networks Next-Gen Tru - [Connection Parameters](#connection-parameters) - [API URL Default and Token URL](#api-url-default-and-token-url) - [Zone Format](#zone-format) +- [Workspaces](#workspaces) - [Examples](#examples) - [Connect with service-account credentials](#connect-with-service-account-credentials) - [Connect with a pre-issued access token](#connect-with-a-pre-issued-access-token) @@ -92,6 +93,7 @@ conn = venafi_connection( | `access_token` | noยน | A pre-issued OAuth access token. When supplied, `client_id`/`client_secret` become optional (but are still used to refresh the token if present). | | `token_url` | no | OAuth token endpoint. Defaults to the Palo Alto production endpoint (see below); override it for non-production environments. | | `url` | no | NGTS API base URL. Defaults to the Palo Alto production endpoint (see below). | +| `workspace` | no | Numeric ID of the NGTS workspace to operate in โ€” a workspace is identified by its **ID, not its name**. When omitted, no workspace is sent and NGTS applies its own default. See [Workspaces](#workspaces). | | `http_request_kwargs` | no | Passed through to `requests` (e.g. a trust bundle via `verify`). | ยน Provide **either** `access_token`, **or** `client_id` + `client_secret`. @@ -142,6 +144,46 @@ separator. zone = "PublicTrust" # the Issuing Template API alias ``` +## Workspaces + +NGTS calls can optionally be scoped to a **workspace**. A workspace is **orthogonal to the zone** โ€” +the zone stays a bare Issuing Template alias โ€” and is identified by its **numeric ID (1โ€“10 digits), +not its name**. + +```python +conn = venafi_connection( + platform=VenafiPlatform.NGTS, + client_id="", + client_secret="", + tsg_id="", + workspace="1234567890", +) +``` + +Equivalently, on an existing connection: + +```python +conn.set_workspace("1234567890") +``` + +Set the workspace **before the first request**. It scopes the access token the SDK mints as well as +the resource calls, so setting it once a token already exists leaves that token unscoped until it is +renewed. + +When a workspace is set, VCert appends `?workspace_id=` to API, GraphQL, and token requests. +Any query string a request already carries is preserved. When no workspace is set, requests are +unchanged โ€” omit the parameter and NGTS applies its own default. + +A non-numeric or over-long workspace raises `ClientBadData` locally, and passing `workspace` to a +non-NGTS platform raises `VenafiError`, since no other platform supports workspaces. + +> โš ๏ธ **`set_policy` must run without a workspace.** Issuing templates (Request Policies) belong to the +> tenant, not to a workspace. Within a workspace they are read-only: `get_policy` works, but creating +> or updating one requires the Superuser role at the tenant level. Calling `set_policy` on a +> connection with a workspace set raises `ClientBadData` before any request is sent (NGTS would +> otherwise reject it with `403`, code `1002`). Use a separate connection with no `workspace` for +> policy management. + ## Examples For the examples below, assume the Issuing Template has an API alias of `PublicTrust`. diff --git a/docs/version_history.md b/docs/version_history.md index 2a128b9..519ae09 100644 --- a/docs/version_history.md +++ b/docs/version_history.md @@ -2,6 +2,10 @@ ## Version History +#### 0.23.0 +* Added NGTS workspace support: `NGTSConnection(workspace=...)`, `venafi_connection(workspace=...)` and `set_workspace()` scope every API, GraphQL, and token request to a workspace via the `workspace_id` query parameter. A workspace is identified by its numeric ID, is independent of the zone, and is optional โ€” omitting it leaves requests unchanged +* NGTS `set_policy` raises `ClientBadData` on a workspace-scoped connection: issuing templates belong to the tenant and are read-only within a workspace, so policy management must use a connection without a workspace + #### 0.22.1 * Fixed EC curve casing validation issue in service-generated CSR enrollment for Cloud/NGTS that caused `ClientBadData` errors with elliptic curve keys diff --git a/tests/test_env.py b/tests/test_env.py index 0d1a1b6..a14b2c7 100644 --- a/tests/test_env.py +++ b/tests/test_env.py @@ -50,6 +50,8 @@ NGTS_TSG_ID = environ.get('NGTS_TSG_ID') NGTS_SCOPE = environ.get('NGTS_SCOPE') NGTS_ZONE = environ.get('NGTS_ZONE') +# Optional: omitting sets the workspace to the tenant default +NGTS_WORKSPACE = environ.get('NGTS_WORKSPACE') if RANDOM_DOMAIN and not isinstance(RANDOM_DOMAIN, str): RANDOM_DOMAIN = RANDOM_DOMAIN.decode() diff --git a/tests/test_local_methods.py b/tests/test_local_methods.py index 5bc9600..8e164f4 100644 --- a/tests/test_local_methods.py +++ b/tests/test_local_methods.py @@ -24,10 +24,11 @@ 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) + NGTSConnection, RevocationRequest, logger, CSR_ORIGIN_SERVICE, VenafiPlatform, + venafi_connection) from vcert.connection_cloud import (URLS, CSR_ATTR_CN, CSR_ATTR_SANS_BY_TYPE, CSR_ATTR_SANS_IP_ADDR, CSR_ATTR_KEY_TYPE_PARAMS, CSR_ATTR_KEY_CURVE) -from vcert.connection_ngts import (_parse_ngts_zone, DEFAULT_API_URL, DEFAULT_TOKEN_URL, +from vcert.connection_ngts import (_parse_ngts_zone, _with_workspace_id, DEFAULT_API_URL, DEFAULT_TOKEN_URL, TRUSTED_TOKEN_HOST_SUFFIX) from vcert.errors import (ClientBadData, ServerUnexptedBehavior, VenafiError, VenafiConnectionError, CertificateRevokeError) @@ -675,6 +676,198 @@ def test_ngts_get_sends_bearer_header(self): _, kwargs = get.call_args self.assertEqual(kwargs['headers']['Authorization'], 'Bearer pre.issued.token') + # -- NGTS workspaces (offline) ------------------------------------------------------------ + # + # A workspace is orthogonal to the zone: it never appears in the zone string, the URL path or + # a request body, only as a `workspace_id` query parameter. + + def test_with_workspace_id_no_workspace_leaves_url_untouched(self): + # The regression that matters most: existing callers pass no workspace, so their URLs must + # be byte-for-byte what they were before workspaces existed. + for url in ("https://api.example.com/ngts/outagedetection/v1/certificaterequests", + "https://api.example.com/ngts/x?chainOrder=EE_FIRST&format=PEM"): + self.assertEqual(_with_workspace_id(url, ""), url) + self.assertEqual(_with_workspace_id(url, None), url) + + def test_with_workspace_id_appends_to_url_without_query(self): + self.assertEqual(_with_workspace_id("https://api.example.com/ngts/v1/x", "1234567890"), + "https://api.example.com/ngts/v1/x?workspace_id=1234567890") + + def test_with_workspace_id_preserves_existing_query(self): + # retrieve_cert hand-builds ?chainOrder=...&format=PEM; those must survive. + self.assertEqual( + _with_workspace_id("https://api.example.com/ngts/v1/x?chainOrder=EE_FIRST&format=PEM", "7"), + "https://api.example.com/ngts/v1/x?chainOrder=EE_FIRST&format=PEM&workspace_id=7") + + def test_with_workspace_id_is_idempotent(self): + once = _with_workspace_id("https://api.example.com/ngts/v1/x?a=b", "7") + self.assertEqual(_with_workspace_id(once, "7"), once) + + def test_with_workspace_id_overwrites_existing_workspace(self): + self.assertEqual(_with_workspace_id("https://api.example.com/ngts/v1/x?workspace_id=1", "2"), + "https://api.example.com/ngts/v1/x?workspace_id=2") + + def test_ngts_workspace_accepts_numeric_ids(self): + # A uint32 rendered as a string: 1 to 10 digits. + for workspace in ("7", "1234567890"): + self.assertEqual(self._ngts_conn(workspace=workspace)._workspace, workspace) + + def test_ngts_workspace_defaults_to_none(self): + self.assertIsNone(self._ngts_conn()._workspace) + + def test_ngts_workspace_rejects_non_numeric_ids(self): + # A workspace is identified by its numeric ID, never its name. + for bad in ("12345678901", "my-workspace", "123abc", "-1", " 7"): + with self.assertRaises(ClientBadData): + self._ngts_conn(workspace=bad) + + def test_ngts_set_workspace_can_clear(self): + conn = self._ngts_conn(workspace="7") + conn.set_workspace(None) + self.assertIsNone(conn._workspace) + self.assertNotIn("workspace_id", conn._resolve_url("v1/certificateissuingtemplates")) + + def test_ngts_get_sends_workspace(self): + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None, workspace='1234567890') + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.headers = {'content-type': 'application/json'} + fake_resp.json.return_value = {} + with mock.patch('vcert.connection_ngts.requests.get', return_value=fake_resp) as get: + conn._get("v1/certificateissuingtemplates") + args, _ = get.call_args + self.assertEqual(args[0], f"{DEFAULT_API_URL}/v1/certificateissuingtemplates" + f"?workspace_id=1234567890") + + def test_ngts_get_sends_workspace_alongside_existing_query(self): + # The retrieve_cert path: Cloud appends its own query before _get sees the URL. + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None, workspace='7') + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.headers = {'content-type': 'application/json'} + fake_resp.json.return_value = {} + with mock.patch('vcert.connection_ngts.requests.get', return_value=fake_resp) as get: + conn._get("outagedetection/v1/certificates/abc/contents?chainOrder=EE_FIRST&format=PEM") + args, _ = get.call_args + self.assertEqual(args[0], f"{DEFAULT_API_URL}/outagedetection/v1/certificates/abc/contents" + f"?chainOrder=EE_FIRST&format=PEM&workspace_id=7") + + def test_ngts_post_and_put_send_workspace(self): + # _post also covers GraphQL (CloudConnection._graphql posts through it), so REST and + # GraphQL share this single choke point. + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None, workspace='7') + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.headers = {'content-type': 'application/json'} + fake_resp.json.return_value = {} + with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: + conn._post(URLS.CERTIFICATE_REQUESTS, data={}) + self.assertEqual(post.call_args[0][0], + f"{DEFAULT_API_URL}/{URLS.CERTIFICATE_REQUESTS}?workspace_id=7") + with mock.patch('vcert.connection_ngts.requests.put', return_value=fake_resp) as put: + conn._put(URLS.ISSUING_TEMPLATES_UPDATE.format("cit-id"), data={}) + self.assertEqual(put.call_args[0][0], + f"{DEFAULT_API_URL}/v1/certificateissuingtemplates/cit-id?workspace_id=7") + + def test_ngts_requests_omit_workspace_when_unset(self): + # Complements the helper-level check: no workspace means no query parameter is added at + # the verb level either. + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None) + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.headers = {'content-type': 'application/json'} + fake_resp.json.return_value = {} + with mock.patch('vcert.connection_ngts.requests.get', return_value=fake_resp) as get: + conn._get("v1/certificateissuingtemplates") + self.assertEqual(get.call_args[0][0], f"{DEFAULT_API_URL}/v1/certificateissuingtemplates") + + def test_ngts_access_token_request_sends_workspace(self): + # The workspace scopes the minted token itself, so it rides on the token URL too. + conn = self._ngts_conn(workspace='1234567890') + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.json.return_value = {'access_token': 'a.b.c', 'token_type': 'Bearer', 'expires_in': 900} + with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: + conn._get_access_token() + self.assertEqual(post.call_args[0][0], + "https://auth.example.com/oauth2/token?workspace_id=1234567890") + + def test_ngts_access_token_request_preserves_token_url_query(self): + conn = self._ngts_conn(token_url="https://auth.example.com/oauth2/token?foo=bar", workspace='7') + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.json.return_value = {'access_token': 'a.b.c', 'token_type': 'Bearer', 'expires_in': 900} + with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: + conn._get_access_token() + self.assertEqual(post.call_args[0][0], + "https://auth.example.com/oauth2/token?foo=bar&workspace_id=7") + + def test_ngts_access_token_request_omits_workspace_when_unset(self): + conn = self._ngts_conn() + fake_resp = mock.MagicMock() + fake_resp.status_code = 200 + fake_resp.json.return_value = {'access_token': 'a.b.c', 'token_type': 'Bearer', 'expires_in': 900} + with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: + conn._get_access_token() + self.assertEqual(post.call_args[0][0], "https://auth.example.com/oauth2/token") + + def test_ngts_set_policy_rejected_within_workspace(self): + # Issuing templates are tenant-owned and read-only inside a workspace, so set_policy must + # fail locally with a clear error rather than sending a request NGTS will reject (403/1002). + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None, workspace='7') + ps = PolicySpecification() + ps.policy = Policy(domains=["venafi.example"]) + with mock.patch('vcert.connection_ngts.requests.get') as get, \ + mock.patch('vcert.connection_ngts.requests.post') as post, \ + mock.patch('vcert.connection_ngts.requests.put') as put: + with self.assertRaises(ClientBadData) as cm: + conn.set_policy("MyTemplate", ps) + self.assertIn("workspace", str(cm.exception)) + # Nothing may reach the network. + get.assert_not_called() + post.assert_not_called() + put.assert_not_called() + + def test_ngts_get_policy_allowed_within_workspace(self): + # The read side stays available in a workspace (templates are read-only there, not hidden). + conn = self._ngts_conn(access_token='pre.issued.token', token_url=None, workspace='7') + with mock.patch.object(conn, '_get_cit_or_fail', return_value={}) as get_cit, \ + mock.patch.object(conn, '_parse_policy_response_to_object'), \ + mock.patch.object(conn, '_get_ca_info', return_value=mock.MagicMock()), \ + mock.patch('vcert.connection_ngts.build_policy_spec', return_value="spec"): + self.assertEqual(conn.get_policy("MyTemplate"), "spec") + get_cit.assert_called_once_with("MyTemplate") + + def test_venafi_connection_passes_workspace_to_ngts(self): + # Both NGTS paths: explicit platform, and auto-detection via client_id + client_secret. + conn = venafi_connection(platform=VenafiPlatform.NGTS, client_id="cid", client_secret="csecret", + tsg_id="1000000001", workspace="7") + self.assertEqual(conn._workspace, "7") + conn = venafi_connection(client_id="cid", client_secret="csecret", tsg_id="1000000001", workspace="7") + self.assertEqual(conn._workspace, "7") + + def test_venafi_connection_rejects_workspace_for_non_ngts(self): + # A workspace on a non-NGTS connector is an error, not a no-op. + with self.assertRaises(VenafiError): + venafi_connection(platform=VenafiPlatform.VAAS, api_key="key", workspace="7") + with self.assertRaises(VenafiError): + venafi_connection(platform=VenafiPlatform.TPP, url="https://tpp.example.com", + access_token="tok", workspace="7") + with self.assertRaises(VenafiError): + venafi_connection(fake=True, workspace="7") + + def test_venafi_connection_rejects_workspace_when_client_creds_do_not_select_ngts(self): + # client_id + client_secret only imply NGTS when nothing overrides them: an explicit + # non-NGTS platform, or fake=True (checked first during auto-detection), wins. The + # workspace would be silently dropped in those cases, so it must raise instead. + creds = dict(client_id="cid", client_secret="csecret", tsg_id="1000000001") + with self.assertRaises(VenafiError): + venafi_connection(platform=VenafiPlatform.VAAS, api_key="key", workspace="7", **creds) + with self.assertRaises(VenafiError): + venafi_connection(platform=VenafiPlatform.FAKE, workspace="7", **creds) + with self.assertRaises(VenafiError): + venafi_connection(fake=True, workspace="7", **creds) + # -- NGTS policy management (offline) ----------------------------------------------------- # # NGTS reuses Cloud's CIT/CA/policy-spec helpers (validate_policy_spec, build_cit_request, diff --git a/tests/test_ngts.py b/tests/test_ngts.py index 1b56bc0..62c6bed 100644 --- a/tests/test_ngts.py +++ b/tests/test_ngts.py @@ -25,11 +25,11 @@ from cryptography.hazmat.primitives import hashes from test_env import (NGTS_URL, NGTS_TOKEN_URL, NGTS_CLIENT_ID, NGTS_CLIENT_SECRET, NGTS_TSG_ID, NGTS_SCOPE, - NGTS_ZONE) + NGTS_ZONE, NGTS_WORKSPACE) from test_utils import random_word, enroll, renew, renew_by_thumbprint from vcert import NGTSConnection, KeyType, logger, CertificateRevokeError from vcert.common import RetireRequest, RevocationRequest -from vcert.policy.policy_spec import Policy, PolicySpecification +from vcert.policy.policy_spec import KeyPair, Policy, PolicySpecification log = logger.get_child("test-ngts") @@ -43,8 +43,11 @@ def setUp(self): # Built in setUp (not __init__) so collecting this module without NGTS_* creds does not # try to construct a connection - the class is skipped before setUp runs. self.ngts_zone = NGTS_ZONE + # NGTS_WORKSPACE is optional: unset, every call goes to the tenant's default workspace, so + # the suite exercises both paths depending on the environment it runs in. self.ngts_conn = NGTSConnection(client_id=NGTS_CLIENT_ID, client_secret=NGTS_CLIENT_SECRET, - token_url=NGTS_TOKEN_URL, scope=NGTS_SCOPE, tsg_id=NGTS_TSG_ID, url=NGTS_URL) + token_url=NGTS_TOKEN_URL, scope=NGTS_SCOPE, tsg_id=NGTS_TSG_ID, url=NGTS_URL, + workspace=NGTS_WORKSPACE) def test_ngts_auth(self): token = self.ngts_conn.auth() @@ -112,6 +115,11 @@ def test_ngts_get_policy(self): self.assertEqual(ps.users, []) def test_ngts_set_get_policy_roundtrip(self): + # Issuing templates (Request Policies) belong to the tenant and are read-only inside a + # workspace: creating or editing one requires tenant-level Superuser, so NGTS rejects + # set_policy with 403/1002 whenever a workspace is set. + if NGTS_WORKSPACE: + self.skipTest("set_policy requires tenant level; issuing templates are read-only in a workspace") # set_policy mutates a CIT by alias, so use a THROWAWAY alias - never self.ngts_zone, # which the other tests depend on. The CA is read from the existing zone so it is # guaranteed valid for this tenant. @@ -123,13 +131,20 @@ def test_ngts_set_get_policy_roundtrip(self): domains=["venafi.example"], max_valid_days=90, cert_auth=ca, + # serviceGenerated must be explicit. Omitting key_pair makes build_cit_request send + # csrUploadAllowed AND keyGeneratedByVenafiAllowed both true (pm_cloud.py:636), which + # NGTS rejects with a malformed 405 that its gateway then surfaces as a 502. + key_pair=KeyPair(service_generated=False), ) throwaway_zone = f"vcert-python-pmtest-{random_word(8)}" self.ngts_conn.set_policy(throwaway_zone, ps) result = self.ngts_conn.get_policy(throwaway_zone) self.assertEqual(result.policy.certificate_authority, ca) - self.assertEqual(result.policy.max_valid_days, 90) + # max_valid_days is deliberately not asserted: set_policy writes validityPeriod nested + # inside the product object (pm_cloud.py:491) while the CIT parser only looks for it at the + # top level (connection_cloud.py:316), so it always reads back None for Cloud and NGTS + # alike. Pre-existing, unrelated to workspaces, tracked separately. self.assertIn("venafi.example", result.policy.domains) self.assertEqual(result.users, []) diff --git a/vcert/__init__.py b/vcert/__init__.py index 8d28d98..31ccf13 100644 --- a/vcert/__init__.py +++ b/vcert/__init__.py @@ -56,7 +56,7 @@ def Connection(url=None, token=None, user=None, password=None, fake=False, http_ def venafi_connection(url=None, api_key=None, user=None, password=None, access_token=None, refresh_token=None, fake=False, http_request_kwargs=None, platform=None, client_id=None, client_secret=None, - token_url=None, scope=None, tsg_id=None): + token_url=None, scope=None, tsg_id=None, workspace=None): """ Return connection based on credentials list. CyberArk Platform (CyberArk Certificate Manager, Self-Hosted) requires URL and access_token (or user and password for getting a new access_token) @@ -77,8 +77,20 @@ def venafi_connection(url=None, api_key=None, user=None, password=None, access_t :param str token_url: NGTS OAuth2 token endpoint (optional; defaults to the Palo Alto production endpoint, override for non-production environments) :param str scope: NGTS OAuth2 scope (``tsg_id:``); derived from tsg_id when omitted :param str tsg_id: NGTS tenant service group id + :param str workspace: NGTS workspace numeric id (optional); scopes every call to that workspace :rtype CommonConnection: """ + # A workspace is NGTS-only: specifying one for + # another platform is an error rather than a silently dropped argument. targets_ngts must + # mirror the dispatch below exactly: an explicit platform wins, and without one `fake` is + # checked before client_id + client_secret auto-detect NGTS. + if platform: + targets_ngts = platform == VenafiPlatform.NGTS + else: + targets_ngts = not fake and bool(client_id and client_secret) + if workspace and not targets_ngts: + raise VenafiError("a workspace was specified but this connector does not support workspaces") + if platform: if platform == VenafiPlatform.FAKE: return FakeConnection() @@ -90,7 +102,7 @@ def venafi_connection(url=None, api_key=None, user=None, password=None, access_t elif platform == VenafiPlatform.NGTS: return NGTSConnection(client_id=client_id, client_secret=client_secret, token_url=token_url, scope=scope, tsg_id=tsg_id, access_token=access_token, url=url, - http_request_kwargs=http_request_kwargs) + http_request_kwargs=http_request_kwargs, workspace=workspace) else: raise VenafiError(f"Invalid Platform: {platform}. Cannot instantiate a Connector.") else: @@ -102,7 +114,7 @@ def venafi_connection(url=None, api_key=None, user=None, password=None, access_t if client_id and client_secret: return NGTSConnection(client_id=client_id, client_secret=client_secret, token_url=token_url, scope=scope, tsg_id=tsg_id, access_token=access_token, url=url, - http_request_kwargs=http_request_kwargs) + http_request_kwargs=http_request_kwargs, workspace=workspace) if url and (access_token or refresh_token or (user and password)): return TPPTokenConnection(url=url, user=user, password=password, access_token=access_token, refresh_token=refresh_token, http_request_kwargs=http_request_kwargs) diff --git a/vcert/connection_ngts.py b/vcert/connection_ngts.py index fa482e8..d535503 100644 --- a/vcert/connection_ngts.py +++ b/vcert/connection_ngts.py @@ -15,7 +15,7 @@ # import re from datetime import datetime, timedelta -from urllib.parse import urlsplit +from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit import requests @@ -61,6 +61,15 @@ # garbage. SCOPE_PATTERN = re.compile(r"tsg_id:[0-9]{10}") +# NGTS workspaces are identified by a numeric ID (a uint32 rendered as a string, so 1-10 digits), +# never by name. Validated in the connector so that passing a workspace *name* fails locally with a +# clear error rather than surfacing as an opaque NGTS API error. +WORKSPACE_ID_PATTERN = re.compile(r"[0-9]{1,10}") + +# Query parameter carrying the workspace on every NGTS request (REST, GraphQL, and the OAuth token +# exchange). The workspace is never part of the zone, the URL path, or a request body. +WORKSPACE_QUERY_PARAM = "workspace_id" + log = get_child("connection-ngts") @@ -98,6 +107,29 @@ def _warn_if_untrusted_token_host(url): "there - verify this endpoint is trusted", host, TRUSTED_TOKEN_HOST_SUFFIX) +def _with_workspace_id(url, workspace_id): + """ + Append ``workspace_id=`` to ``url``, preserving any query string it already carries. + + Merging rather than overwriting matters: ``retrieve_cert`` hand-builds + ``?chainOrder=...&format=PEM`` (``connection_cloud.py``), and those parameters must survive. + The operation is idempotent, replaces a pre-existing ``workspace_id``, and returns the URL + untouched when no workspace is set. + + :param str url: + :param str workspace_id: + :rtype: str + """ + if not workspace_id: + return url + parts = urlsplit(url) + # doseq keeps repeated parameters intact; parse_qs with keep_blank_values preserves an + # existing "?foo=" rather than silently dropping it. + query = parse_qs(parts.query, keep_blank_values=True) + query[WORKSPACE_QUERY_PARAM] = [workspace_id] + return urlunsplit(parts._replace(query=urlencode(query, doseq=True))) + + def _parse_ngts_zone(zone): """ NGTS zones are a Certificate Issuing Template alias only - the entire zone string is the @@ -125,6 +157,10 @@ class NGTSConnection(CloudConnection): - Zones are a Certificate Issuing Template alias only (no ``Application\\CIT`` split), and request payloads omit ``applicationId``. + Calls can optionally be scoped to an NGTS *workspace* (see :meth:`set_workspace`). A workspace + is orthogonal to the zone: the zone stays a bare CIT alias, and the workspace rides along as a + ``workspace_id`` query parameter on every request. + ``url`` is optional: when omitted it defaults to the published Palo Alto production API endpoint (:data:`DEFAULT_API_URL`), matching Go's ``normalizeURL`` fallback. ``token_url`` is likewise optional and defaults to the production OAuth2 token endpoint @@ -140,7 +176,7 @@ class NGTSConnection(CloudConnection): """ def __init__(self, client_id, client_secret, token_url=None, scope=None, tsg_id=None, access_token=None, url=None, - http_request_kwargs=None): + http_request_kwargs=None, workspace=None): # url defaults to the published Palo Alto production endpoint (Go defaults the base URL # too); it must be defaulted before the super().__init__ call, which normalizes whatever # base URL it receives. token_url likewise defaults to the production OAuth2 endpoint; @@ -177,8 +213,13 @@ def __init__(self, client_id, client_secret, token_url=None, scope=None, tsg_id= self._tsg_id = tsg_id self._access_token = access_token self._token_expires = None + self._workspace = None + if workspace: + self.set_workspace(workspace) def __str__(self): + if self._workspace: + return f"[NGTS] {self._base_url} (workspace {self._workspace})" return f"[NGTS] {self._base_url}" def _normalize_and_verify_base_url(self): @@ -195,6 +236,41 @@ def _normalize_and_verify_base_url(self): raise ClientBadData self._base_url = u + # -- Workspaces ------------------------------------------------------------------------------ + + def set_workspace(self, workspace): + """ + Scope every subsequent call to an NGTS workspace. A workspace is independent of the zone - + the zone remains a bare Certificate Issuing Template alias. + + Set this before the first request: the workspace also scopes the access token obtained by + :meth:`_get_access_token`, so setting it once a token already exists leaves that token + unscoped until it is renewed. + + :param str workspace: the workspace's numeric ID (1-10 digits), not its name + """ + if workspace and not WORKSPACE_ID_PATTERN.fullmatch(workspace): + raise ClientBadData(f'invalid workspace "{workspace}". A workspace is identified by its ' + f"numeric ID, not its name") + self._workspace = workspace or None + + def _resolve_url(self, url): + """ + Turn a relative resource path into the absolute URL to call, carrying the workspace when + one is set. + + The workspace is applied here - after the base URL is joined and after callers have + substituted their ``{}`` placeholders and appended any hand-built query (e.g. + ``retrieve_cert``'s ``?chainOrder=...&format=PEM``) - so that query is merged with rather + than clobbered. This is the single choke point for both REST and GraphQL: Cloud's + ``_graphql`` posts through ``_post`` too, so there is no separate GraphQL URL builder to keep + in sync. + + :param str url: + :rtype: str + """ + return _with_workspace_id(self._base_url + url, self._workspace) + # -- Authentication -------------------------------------------------------------------------- def _get_access_token(self): @@ -213,7 +289,10 @@ def _get_access_token(self): 'grant_type': 'client_credentials', 'scope': self._scope, } - r = requests.post(self._token_url, data=data, auth=(self._client_id, self._client_secret), + # The workspace scopes the token itself, so it goes on the token URL as well as on resource + # calls: the resulting access token is already workspace-scoped when issued. + token_url = _with_workspace_id(self._token_url, self._workspace) + r = requests.post(token_url, data=data, auth=(self._client_id, self._client_secret), headers=headers, **self._http_request_kwargs) # nosec B113 if r.status_code != HTTPStatus.OK: log.error(f"Failed to obtain access token. Server status: {r.status_code}") @@ -281,7 +360,7 @@ def _auth_headers(self, accept): def _get(self, url, params=None): self._ensure_token() headers = self._auth_headers(MIME_ANY) - r = requests.get(self._base_url + url, params=params, headers=headers, + r = requests.get(self._resolve_url(url), params=params, headers=headers, **self._http_request_kwargs) # nosec B113 return self.process_server_response(r) @@ -289,7 +368,7 @@ def _post(self, url, data=None): self._ensure_token() headers = self._auth_headers(MIME_JSON) if isinstance(data, dict): - r = requests.post(self._base_url + url, json=data, headers=headers, + r = requests.post(self._resolve_url(url), json=data, headers=headers, **self._http_request_kwargs) # nosec B113 else: log.error(f"Unexpected client data type: {type(data)} for {url}") @@ -300,7 +379,7 @@ def _put(self, url, data=None): self._ensure_token() headers = self._auth_headers(MIME_JSON) if isinstance(data, dict): - r = requests.put(self._base_url + url, json=data, headers=headers, + r = requests.put(self._resolve_url(url), json=data, headers=headers, **self._http_request_kwargs) # nosec B113 else: log.error(f"Unexpected client data type: {type(data)} for {url}") @@ -537,9 +616,17 @@ def set_policy(self, zone, policy_spec): CIT is created/updated directly on the global issuing-template endpoint and ``policy_spec.users`` is ignored (parity with Go NGTS). + Not supported on a workspace-scoped connection: issuing templates (Request Policies) belong + to the tenant and are read-only within a workspace, so NGTS rejects the write with 403/1002. + This raises locally instead, before any request is sent, with an actionable message. + :param str zone: the CIT alias (NGTS zones are a CIT alias only - no Application\\CIT split) :param PolicySpecification policy_spec: """ + if self._workspace: + raise ClientBadData(f"set_policy is not supported within a workspace (workspace {self._workspace}): " + f"issuing templates belong to the tenant and are read-only in a workspace. " + f"Use a connection without a workspace for policy management") validate_policy_spec(policy_spec) cit_alias = _parse_ngts_zone(zone) From 79da6f272b31b995bcca18c9f0c35286d314bd44 Mon Sep 17 00:00:00 2001 From: "erkhem.unenbat" Date: Thu, 24 Sep 2026 13:53:06 -0700 Subject: [PATCH 2/2] workspace support added --- README-NGTS.md | 9 +++++---- docs/version_history.md | 2 +- tests/test_local_methods.py | 11 +++++------ vcert/connection_ngts.py | 17 ++++++++--------- 4 files changed, 19 insertions(+), 20 deletions(-) diff --git a/README-NGTS.md b/README-NGTS.md index ce16348..87a3498 100644 --- a/README-NGTS.md +++ b/README-NGTS.md @@ -166,11 +166,12 @@ Equivalently, on an existing connection: conn.set_workspace("1234567890") ``` -Set the workspace **before the first request**. It scopes the access token the SDK mints as well as -the resource calls, so setting it once a token already exists leaves that token unscoped until it is -renewed. +The workspace applies to every request made after it is set. Access tokens are not +workspace-scoped: the tenant is selected by `scope`/`tsg_id`, and the workspace is applied per +request, so an existing token keeps working when the workspace changes. -When a workspace is set, VCert appends `?workspace_id=` to API, GraphQL, and token requests. +When a workspace is set, VCert appends `?workspace_id=` to API and GraphQL requests (not to the +OAuth token request, which ignores it). Any query string a request already carries is preserved. When no workspace is set, requests are unchanged โ€” omit the parameter and NGTS applies its own default. diff --git a/docs/version_history.md b/docs/version_history.md index 519ae09..4cae4a4 100644 --- a/docs/version_history.md +++ b/docs/version_history.md @@ -3,7 +3,7 @@ ## Version History #### 0.23.0 -* Added NGTS workspace support: `NGTSConnection(workspace=...)`, `venafi_connection(workspace=...)` and `set_workspace()` scope every API, GraphQL, and token request to a workspace via the `workspace_id` query parameter. A workspace is identified by its numeric ID, is independent of the zone, and is optional โ€” omitting it leaves requests unchanged +* Added NGTS workspace support: `NGTSConnection(workspace=...)`, `venafi_connection(workspace=...)` and `set_workspace()` scope every API and GraphQL request to a workspace via the `workspace_id` query parameter. A workspace is identified by its numeric ID, is independent of the zone, and is optional โ€” omitting it leaves requests unchanged * NGTS `set_policy` raises `ClientBadData` on a workspace-scoped connection: issuing templates belong to the tenant and are read-only within a workspace, so policy management must use a connection without a workspace #### 0.22.1 diff --git a/tests/test_local_methods.py b/tests/test_local_methods.py index 8e164f4..8b7da39 100644 --- a/tests/test_local_methods.py +++ b/tests/test_local_methods.py @@ -781,16 +781,16 @@ def test_ngts_requests_omit_workspace_when_unset(self): conn._get("v1/certificateissuingtemplates") self.assertEqual(get.call_args[0][0], f"{DEFAULT_API_URL}/v1/certificateissuingtemplates") - def test_ngts_access_token_request_sends_workspace(self): - # The workspace scopes the minted token itself, so it rides on the token URL too. + def test_ngts_access_token_request_does_not_send_workspace(self): + # The token endpoint ignores workspace_id (tokens minted with and without it carry identical + # claims), so sending it would only imply a scoping that does not happen. conn = self._ngts_conn(workspace='1234567890') fake_resp = mock.MagicMock() fake_resp.status_code = 200 fake_resp.json.return_value = {'access_token': 'a.b.c', 'token_type': 'Bearer', 'expires_in': 900} with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: conn._get_access_token() - self.assertEqual(post.call_args[0][0], - "https://auth.example.com/oauth2/token?workspace_id=1234567890") + self.assertEqual(post.call_args[0][0], "https://auth.example.com/oauth2/token") def test_ngts_access_token_request_preserves_token_url_query(self): conn = self._ngts_conn(token_url="https://auth.example.com/oauth2/token?foo=bar", workspace='7') @@ -799,8 +799,7 @@ def test_ngts_access_token_request_preserves_token_url_query(self): fake_resp.json.return_value = {'access_token': 'a.b.c', 'token_type': 'Bearer', 'expires_in': 900} with mock.patch('vcert.connection_ngts.requests.post', return_value=fake_resp) as post: conn._get_access_token() - self.assertEqual(post.call_args[0][0], - "https://auth.example.com/oauth2/token?foo=bar&workspace_id=7") + self.assertEqual(post.call_args[0][0], "https://auth.example.com/oauth2/token?foo=bar") def test_ngts_access_token_request_omits_workspace_when_unset(self): conn = self._ngts_conn() diff --git a/vcert/connection_ngts.py b/vcert/connection_ngts.py index d535503..38d513d 100644 --- a/vcert/connection_ngts.py +++ b/vcert/connection_ngts.py @@ -66,8 +66,8 @@ # clear error rather than surfacing as an opaque NGTS API error. WORKSPACE_ID_PATTERN = re.compile(r"[0-9]{1,10}") -# Query parameter carrying the workspace on every NGTS request (REST, GraphQL, and the OAuth token -# exchange). The workspace is never part of the zone, the URL path, or a request body. +# Query parameter carrying the workspace on every NGTS API request (REST and GraphQL). It is not sent +# to the OAuth token endpoint, and is never part of the zone, the URL path, or a request body. WORKSPACE_QUERY_PARAM = "workspace_id" log = get_child("connection-ngts") @@ -243,9 +243,8 @@ def set_workspace(self, workspace): Scope every subsequent call to an NGTS workspace. A workspace is independent of the zone - the zone remains a bare Certificate Issuing Template alias. - Set this before the first request: the workspace also scopes the access token obtained by - :meth:`_get_access_token`, so setting it once a token already exists leaves that token - unscoped until it is renewed. + Takes effect on the next request. Access tokens are not workspace-scoped, so an existing + token keeps working after the workspace changes. :param str workspace: the workspace's numeric ID (1-10 digits), not its name """ @@ -289,10 +288,10 @@ def _get_access_token(self): 'grant_type': 'client_credentials', 'scope': self._scope, } - # The workspace scopes the token itself, so it goes on the token URL as well as on resource - # calls: the resulting access token is already workspace-scoped when issued. - token_url = _with_workspace_id(self._token_url, self._workspace) - r = requests.post(token_url, data=data, auth=(self._client_id, self._client_secret), + # The workspace is deliberately not sent here. The token endpoint ignores workspace_id (tokens + # minted with and without it carry identical claims); the tenant is selected by the scope, and + # the workspace is applied per API request instead. + r = requests.post(self._token_url, data=data, auth=(self._client_id, self._client_secret), headers=headers, **self._http_request_kwargs) # nosec B113 if r.status_code != HTTPStatus.OK: log.error(f"Failed to obtain access token. Server status: {r.status_code}")