Vc 60045 vcert python support for workspaces

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,47 @@ 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 id>",
+    client_secret="<client secret>",
+    tsg_id="<tsg id>",
+    workspace="1234567890",
+)
+```
+
+Equivalently, on an existing connection:
+
+```python
+conn.set_workspace("1234567890")
+```
+
+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=<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.
+
+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`.

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 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
 * Fixed EC curve casing validation issue in service-generated CSR enrollment for Cloud/NGTS that caused `ClientBadData` errors with elliptic curve keys
 

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()

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,197 @@ 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_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")
+
+    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")
+
+    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,