diff --git a/.env.example b/.env.example index 0ee8079..82e2f3b 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,11 @@ SPLUNK_TOKEN= # $XDG_STATE_HOME/vct-splunk/audit.log, else ~/.local/state/vct-splunk/audit.log). # VCT_SPLUNK_AUDIT=/path/to/audit.log +# Required for ANY real (non-dry-run) write, on every backend. Never a CLI +# flag -- a saved command line must not be able to enable one. --dry-run sends +# nothing and needs no opt-in either way. +# SPLUNK_ENABLE_WRITES=true + # Initial password for `splunk user create` — a secret, so it is read from the # environment (or an interactive prompt), never accepted as a CLI flag. # SPLUNK_USER_PASSWORD= @@ -57,10 +62,10 @@ SPLUNK_TOKEN= # VCT_SPLUNK_CONFIG= # --- Splunk Cloud (ACS) ------------------------------------------------------ -# Read-only this release. The backend is deduced from SPLUNK_URL: on a -# *.splunkcloud.com host, supported reads route via the ACS API automatically -# (there is no flag or variable to pick a backend). `splunk inspect` reports -# what the deduced backend supports. +# The backend is deduced from SPLUNK_URL: on a *.splunkcloud.com host, +# supported reads route via the ACS API automatically (there is no flag or +# variable to pick a backend). `splunk inspect` reports what the deduced +# backend supports. # ACS authentication token (Bearer). The stack name is derived from SPLUNK_URL; # set SPLUNK_ACS_STACK only to override it. @@ -71,6 +76,17 @@ SPLUNK_TOKEN= # FedRAMP stacks use https://admin.splunkcloudgc.com. # SPLUNK_ACS_BASE_URL= +# Cloud writes need SPLUNK_ENABLE_WRITES=true (above) AND this, and only unlock +# create/update/delete for index, role, and hec-token -- enable/disable and +# every other resource stay refused regardless. Never a CLI flag. Checked even +# for a --dry-run preview, since it also proves the object is one of the three +# ACS-writable resources. +# SPLUNK_CLOUD_WRITE=true + +# Optional: a separate ACS token scoped to writes only. Falls back to +# SPLUNK_ACS_TOKEN when unset; reads always use SPLUNK_ACS_TOKEN, never this one. +# SPLUNK_ACS_WRITE_TOKEN= + # Hide the Cloud stack name at untrusted output boundaries (e.g. CI logs) in # the target shown by prompts, JSON metadata, and error text. The audit log is # unaffected -- it always records the real host. @@ -82,7 +98,8 @@ SPLUNK_TOKEN= # SPLUNK_INTEGRATION_TEST=true # SPLUNK_WRITE_TEST=true -# Enables the read-only Cloud ACS canary. +# Enables the Cloud ACS integration suites (read and write). Cloud write tests +# also require SPLUNK_CLOUD_WRITE=true, same as the CLI itself. # SPLUNK_ACS_LIVE_TEST=true # Server-side staging directory used only by disposable Enterprise write tests. diff --git a/.github/scripts/detect-cloud-stack.sh b/.github/scripts/detect-cloud-stack.sh index 636ca32..fcd32f5 100644 --- a/.github/scripts/detect-cloud-stack.sh +++ b/.github/scripts/detect-cloud-stack.sh @@ -26,7 +26,7 @@ if [ -z "${SPLUNK_URL:-}" ] || [ -z "${SPLUNK_ACS_TOKEN:-}" ]; then echo "full=false" >>"$GITHUB_OUTPUT" note "No Splunk Cloud stack configured, so there is nothing to certify." note "The Cloud read and write contracts still run on every pull request, in" - note "tests/unit/test_acs_loopback.py and tests/unit/test_cloud_write_refusal.py." + note "tests/unit/test_acs_loopback.py and tests/unit/test_write_gating.py." exit 0 fi diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0374dc7..8070060 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -159,6 +159,10 @@ jobs: env: SPLUNK_INTEGRATION_TEST: "true" SPLUNK_WRITE_TEST: "true" + # The test suite itself also force-enables this for every pytest session + # (tests/conftest.py); set explicitly here too so the job documents its + # own intent without a reader having to know that. + SPLUNK_ENABLE_WRITES: "true" SPLUNK_URL: https://localhost:8089 SPLUNK_USERNAME: admin SPLUNK_PASSWORD: Ch4ng3d-CI-Pass! diff --git a/.github/workflows/cloud-write.yml b/.github/workflows/cloud-write.yml new file mode 100644 index 0000000..cefdffd --- /dev/null +++ b/.github/workflows/cloud-write.yml @@ -0,0 +1,159 @@ +# Certify the Cloud write path end-to-end against a real, non-production +# Splunk Cloud stack: a `read` job first (same secrets and steps as the +# scheduled Splunk Cloud Read Canary), then a destructive `write` job gated +# behind a protected `splunk-cloud-write` GitHub Environment with required +# reviewers. The approval pause sits in front of the destructive half only -- +# a credential or connectivity problem surfaces in `read` before anyone is +# asked to approve anything. One dispatch is the full read-then-write +# certification. +# +# Never runs on pull_request or a schedule -- workflow_dispatch only, and even +# then the confirm job fails closed unless `confirm` is typed exactly `WRITE` +# and HEAD's commit subject starts with `tests: splunk cloud write`. The Cloud +# write contract itself (opt-in gating, the ACS-writable allowlist) is proved +# credential-free on every pull request in tests/unit/test_write_gating.py; +# what only a live stack can show is that a real one answers the way that +# contract expects. +name: Splunk Cloud Write Canary + +on: + workflow_dispatch: + inputs: + confirm: + description: Type WRITE to run destructive ACS tests on the configured stack + required: true + type: string + +permissions: + contents: read + +concurrency: + group: splunk-cloud-write + cancel-in-progress: false + +jobs: + confirm: + name: Confirm WRITE + runs-on: ubuntu-latest + timeout-minutes: 2 + steps: + - name: Fail closed unless confirm is WRITE + env: + CONFIRM: ${{ github.event.inputs.confirm }} + run: | + if [ "$CONFIRM" != "WRITE" ]; then + echo "confirm must be exactly WRITE" + exit 1 + fi + + read: + name: Cloud / ACS reads (pre-write readiness) + needs: confirm + runs-on: ubuntu-latest + timeout-minutes: 10 + env: + SPLUNK_ACS_LIVE_TEST: "true" + SPLUNK_URL: ${{ secrets.SPLUNK_URL }} + SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_TOKEN }} + SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }} + SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }} + SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }} + VCT_SPLUNK_REDACT_TARGET: "1" + VCT_SPLUNK_AUDIT: ${{ runner.temp }}/vct-splunk-audit.log + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + + - name: Check for a configured Cloud stack + id: stack + run: bash .github/scripts/detect-cloud-stack.sh + + - name: Require read secrets when dispatched + if: steps.stack.outputs.ready != 'true' + run: | + echo "SPLUNK_URL and SPLUNK_ACS_TOKEN are required to certify Cloud reads" + exit 1 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: "3.14" + cache: pip + + - name: Install project + run: | + python -m venv .venv + .venv/bin/python -m pip install --require-hashes -r requirements-ci.txt + .venv/bin/python -m pip install -e . --no-deps + + - name: Cloud reads (every catalogued read command) + run: >- + bash .github/scripts/run-cloud-suite.sh cloud-read + tests/integration/cloud/read/test_catalog.py "integration and cloud and read" + + - name: Cloud ACS operations (below the CLI) + run: >- + bash .github/scripts/run-cloud-suite.sh cloud-acs + tests/integration/cloud/read/test_acs_operations.py "integration and cloud and read" + + write: + name: Cloud / ACS writes + needs: read + environment: splunk-cloud-write + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + + - name: Require the write-canary commit prefix + run: | + subject=$(git log -1 --format=%s) + case "$subject" in + "tests: splunk cloud write"*) ;; + *) + echo "HEAD subject must start with: tests: splunk cloud write" + exit 1 + ;; + esac + + - name: Check for a configured write-capable Cloud stack + id: stack + env: + SPLUNK_URL: ${{ secrets.SPLUNK_URL }} + SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }} + run: bash .github/scripts/detect-cloud-stack.sh + + - name: Require write secrets when dispatched + if: steps.stack.outputs.ready != 'true' + run: | + echo "SPLUNK_URL and SPLUNK_ACS_WRITE_TOKEN are required for a WRITE run" + exit 1 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: "3.14" + cache: pip + + - name: Install project + run: | + python -m venv .venv + .venv/bin/python -m pip install --require-hashes -r requirements-ci.txt + .venv/bin/python -m pip install -e . --no-deps + + - name: Cloud writes (index, role, hec-token) with undo + env: + SPLUNK_ACS_LIVE_TEST: "true" + SPLUNK_ENABLE_WRITES: "true" + SPLUNK_CLOUD_WRITE: "true" + SPLUNK_URL: ${{ secrets.SPLUNK_URL }} + SPLUNK_ACS_WRITE_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }} + SPLUNK_ACS_TOKEN: ${{ secrets.SPLUNK_ACS_WRITE_TOKEN }} + SPLUNK_ACS_STACK: ${{ secrets.SPLUNK_ACS_STACK }} + SPLUNK_ACS_BASE_URL: ${{ secrets.SPLUNK_ACS_BASE_URL }} + VCT_SPLUNK_REDACT_TARGET: "1" + VCT_SPLUNK_AUDIT: ${{ runner.temp }}/vct-splunk-audit.log + run: >- + bash .github/scripts/run-cloud-suite.sh cloud-write + tests/integration/cloud/write "integration and cloud and write" diff --git a/AGENTS.md b/AGENTS.md index fad0470..d702e55 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -69,9 +69,13 @@ Two cross-cutting ideas to know about: require an explicit app and never silently default to `search`. - **Transparent backend.** When `SPLUNK_URL` points at `*.splunkcloud.com`, a few reads (`index list`, `role list`, `hec-token list`) route through the - Cloud ACS API and writes are refused; everything else talks to splunkd REST. - The backend is deduced from the URL — there is no flag to pick it. `splunk - inspect` reports what the deduced backend supports, offline. + Cloud ACS API; everything else talks to splunkd REST. Cloud writes need both + `SPLUNK_ENABLE_WRITES=true` (every write, every backend) and + `SPLUNK_CLOUD_WRITE=true`, and only unlock `create`/`update`/`delete` for + `index`, `role`, and `hec-token` (never a CLI flag; enable/disable and every + other resource stay refused regardless). The backend is deduced from the + URL — there is no flag to pick it. `splunk inspect` reports what the deduced + backend supports, offline. ## Conventions @@ -85,11 +89,13 @@ Two cross-cutting ideas to know about: ## Safety -- Writes are gated. Every mutation — index lifecycle, saved-search CRUD and - `search cancel`, and all factory-generated create/update/delete/enable/ - disable — funnels through one shared path (`commands/write.py`): `--dry-run` - previews the exact request and sends nothing; otherwise it confirms on a TTY - or requires `--yes` when non-interactive (it never hangs on a hidden prompt). +- Writes are gated and disabled by default. Every mutation — index lifecycle, + saved-search CRUD and `search cancel`, and all factory-generated + create/update/delete/enable/disable — funnels through one shared path + (`commands/write.py`): `--dry-run` previews the exact request and sends + nothing, needing no opt-in; a real write needs `SPLUNK_ENABLE_WRITES=true` + first (there is no CLI flag), then confirms on a TTY or requires `--yes` + when non-interactive (it never hangs on a hidden prompt). - Each applied write is appended to a local audit log: `$VCT_SPLUNK_AUDIT` if set, else `$XDG_STATE_HOME/vct-splunk/audit.log`, else `~/.local/state/vct-splunk/audit.log`. diff --git a/CHANGELOG.md b/CHANGELOG.md index 4dcc984..8e9a93a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,11 +16,27 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). boundaries: `VCT_SPLUNK_REDACT_TARGET=1` hides it in prompts, JSON metadata, and transport error text. The audit log is unaffected and always records the real host. The `Splunk Cloud Read Canary` workflow sets this. +- Opt-in Splunk Cloud writes for `index`, `role`, and `hec-token` + create/update/delete via ACS. Two env-only gates, never a CLI flag: + `SPLUNK_ENABLE_WRITES=true` for any real write on any backend, and + additionally `SPLUNK_CLOUD_WRITE=true` on a Cloud target (checked even for a + `--dry-run` preview, since it also proves the object is one of the three + ACS-writable resources). `--dry-run` and `--yes` behave the same as on + Enterprise. A separate `SPLUNK_ACS_WRITE_TOKEN` can scope the write + credential apart from the read token. +- A `Splunk Cloud Write Canary` GitHub Actions workflow: `workflow_dispatch` + only, gated by a typed `confirm=WRITE` input, a `tests: splunk cloud write` + HEAD commit-subject requirement, and a protected `splunk-cloud-write` + environment with required reviewers around the destructive half. It runs the + Cloud read canary first, so one approved dispatch certifies both. ### Changed - `splunk inspect` no longer echoes the Cloud stack name in its report body. It reports `stack_configured: bool` instead. +- **Breaking:** every real (non-`--dry-run`) write, on every backend, now + requires `SPLUNK_ENABLE_WRITES=true`. There is no CLI flag, so a saved + command line cannot enable one. `--dry-run` is unaffected. - Lower the supported Python floor to 3.9, so the CLI runs under the interpreter bundled with Splunk Enterprise 9.x. Shipped code needed no change: the package diff --git a/README.md b/README.md index 4ec105d..fb567d0 100644 --- a/README.md +++ b/README.md @@ -129,18 +129,24 @@ so an object is never created somewhere you did not intend. ### Changes are guarded -Every command that changes the server behaves the same way: +Writes are disabled by default, on every backend. Set `SPLUNK_ENABLE_WRITES=true` +to allow any real (non-preview) mutation -- there is no CLI flag, so a saved +command line can never turn this on by itself: ```bash -splunk index create payments --dry-run # show the exact request, send nothing -splunk index create payments # ask for confirmation, then do it -splunk index create payments --yes # skip the question (required in scripts) +splunk index create payments --dry-run # show the exact request, send nothing -- always works +export SPLUNK_ENABLE_WRITES=true +splunk index create payments # ask for confirmation, then do it +splunk index create payments --yes # skip the question (required in scripts) ``` -In a script with no person watching, a change without `--yes` stops immediately -rather than waiting forever for an answer. Every applied change is appended to an -audit log — the first of `$VCT_SPLUNK_AUDIT`, -`$XDG_STATE_HOME/vct-splunk/audit.log`, or `~/.local/state/vct-splunk/audit.log`. +`--dry-run` sends nothing, so it needs no opt-in and always works. Once writes +are enabled, every command that changes the server behaves the same way: ask +for confirmation on a TTY, or require `--yes` in a script with no person +watching -- a change without `--yes` there stops immediately rather than +waiting forever for an answer. Every applied change is appended to an audit +log — the first of `$VCT_SPLUNK_AUDIT`, `$XDG_STATE_HOME/vct-splunk/audit.log`, +or `~/.local/state/vct-splunk/audit.log`. ## Output and exit codes @@ -178,10 +184,20 @@ export SPLUNK_ACS_TOKEN="" export SPLUNK_ACS_BASE_URL="https://admin.splunkcloudgc.com" # only for FedRAMP ``` -Cloud support is **read-only** today, and covers `index list`, `role list`, and -`hec-token list`. Anything else stops with a clear "not supported here" error -instead of guessing. Run `splunk inspect` to see which backend your address -resolves to and what it can do; it answers offline, without contacting anything. +Cloud reads cover `index list`, `role list`, and `hec-token list`. Anything else +stops with a clear "not supported here" error instead of guessing. Run +`splunk inspect` to see which backend your address resolves to and what it can +do; it answers offline, without contacting anything. + +Cloud **writes** need both `SPLUNK_ENABLE_WRITES=true` (every write, every +backend) and `SPLUNK_CLOUD_WRITE=true`, and only unlock `create`/`update`/ +`delete` for `index`, `role`, and `hec-token` -- enable/disable and every other +resource stay refused regardless. Neither is a CLI flag, so a saved command +line can never enable a write. `--dry-run` and `--yes` work the same as they do +against Enterprise, except a Cloud preview still needs `SPLUNK_CLOUD_WRITE=true` +(it also proves the object is one of the three ACS-writable resources). An ACS +write token can be scoped separately from the read token via +`SPLUNK_ACS_WRITE_TOKEN` (falls back to `SPLUNK_ACS_TOKEN` when unset). ## Security diff --git a/docs/architecture.md b/docs/architecture.md index 4d0cf9f..c5eec9d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -20,7 +20,8 @@ vct_splunk/ server.py, search.py, jobs.py, saved_searches.py, kvstore.py, hec.py, apps.py, cluster.py, license.py, deploy.py, lookups.py, datamodel.py, health.py, raw.py - acs/ # Splunk Cloud ACS management-plane client (read-only) + acs/ # Splunk Cloud ACS management-plane client (reads + + # gated index/role/hec-token writes) client.py, operations.py auth/ @@ -31,7 +32,8 @@ vct_splunk/ registry.py # Declarative list of factory-generated resources context.py # Shared `command` decorator + Ctx (builds clients) write.py # The single gated write path (confirm + audit) - dispatch.py # Routes index/role/hec-token list to ACS on Cloud + dispatch.py # Routes index/role/hec-token reads (and gated + # writes) to ACS on Cloud server.py, api.py, auth.py, search.py, saved_search.py, health.py, kvstore.py, hec.py, apps.py, cluster.py, shcluster.py, license.py, deploy.py, lookup.py, datamodel.py, inspect.py @@ -148,12 +150,16 @@ CLI flags (--base-url, --profile, --app, ...) ### Write safety -Every mutation funnels through `commands/write.py:do_write()`: `--dry-run` -previews the exact request and sends nothing; otherwise it confirms on a TTY or -requires `--yes` when non-interactive, then appends a record to the local audit -log. Reads redact secret-named fields by default; only commands whose purpose -is to mint a credential reveal one. Splunk Cloud targets refuse writes and -route supported reads through ACS. +Every mutation funnels through `commands/write.py:do_write()`. A real write -- +`--dry-run` sends nothing and needs neither gate below -- requires +`SPLUNK_ENABLE_WRITES=true` first, on every backend; there is no CLI flag, so a +saved command line cannot enable one. It then confirms on a TTY or requires +`--yes` when non-interactive, and appends a record to the local audit log. +Reads redact secret-named fields by default; only commands whose purpose is to +mint a credential reveal one. Splunk Cloud targets route supported reads +through ACS; writes there need `SPLUNK_CLOUD_WRITE=true` as well (checked even +for a `--dry-run` preview, since it also proves the object is one of the three +ACS-writable resources -- index, role, hec-token create/update/delete only). ### Error handling diff --git a/src/vct_splunk/api/acs/__init__.py b/src/vct_splunk/api/acs/__init__.py index 3fe49bf..2f3faa4 100644 --- a/src/vct_splunk/api/acs/__init__.py +++ b/src/vct_splunk/api/acs/__init__.py @@ -1 +1,6 @@ -"""Read-only Splunk Cloud ACS (adminconfig/v2) client and operations.""" +"""Splunk Cloud ACS (adminconfig/v2) client and operations. + +Reads are unrestricted; writes exist for index/role/hec-token create, update, +and delete but are opt-in and gated -- see +:func:`vct_splunk.commands.write.refuse_cloud_write`. +""" diff --git a/src/vct_splunk/api/acs/client.py b/src/vct_splunk/api/acs/client.py index 08cf519..4a1395e 100644 --- a/src/vct_splunk/api/acs/client.py +++ b/src/vct_splunk/api/acs/client.py @@ -1,10 +1,14 @@ -"""A thin, read-only client for the Splunk Cloud ACS adminconfig/v2 API. +"""A thin client for the Splunk Cloud ACS adminconfig/v2 API. ACS is a different surface from splunkd: a different base URL (``https://admin.splunk.com//adminconfig/v2``), a stack auth token, and plain JSON responses (not the form-encoded ``entry[].content`` shape). So it gets its own small client rather than reusing :class:`~vct_splunk.api.client.SplunkClient`. -Writes are intentionally absent this release. + +Reads (:meth:`get`) are unrestricted. Mutations (:meth:`write`) exist too, but +this client sends whatever it is told -- the opt-in gate and the allowlist of +which (resource, verb) pairs are ever reachable live one layer up, in +:mod:`vct_splunk.commands.write` and :mod:`vct_splunk.commands.dispatch`. """ from __future__ import annotations @@ -18,7 +22,7 @@ import httpx from ...utils.errors import APIError, AuthError, NotFoundError, TransportError, UsageError -from ...utils.redact import safe_target +from ...utils.redact import public_target, redact_exception_text ACS_BASE_URL = "https://admin.splunk.com" _STACK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*$") @@ -31,25 +35,35 @@ class AcsConfig: token: str base_url: str = ACS_BASE_URL timeout: float = 30.0 + dry_run: bool = False -def acs_config_from_env(stack: str | None = None) -> AcsConfig: - """Build an ACS config: the stack (derived from SPLUNK_URL) + ``SPLUNK_ACS_TOKEN``. +def acs_config_from_env(stack: str | None = None, *, write: bool = False) -> AcsConfig: + """Build an ACS config: the stack (derived from SPLUNK_URL) + an ACS token. The Cloud stack is normally derived from the ``*.splunkcloud.com`` host in ``SPLUNK_URL`` and passed in as ``stack``; ``SPLUNK_ACS_STACK`` is a rare - explicit override. ``SPLUNK_ACS_TOKEN`` (a Bearer token, separate from the - Enterprise auth token) is always required for ACS operations. + explicit override. + + Args: + stack: The Cloud stack name, or None to require ``SPLUNK_ACS_STACK``. + write: When True, prefer ``SPLUNK_ACS_WRITE_TOKEN`` (falling back to + ``SPLUNK_ACS_TOKEN`` when unset) -- lets an operator scope a + write-capable token separately from the read token. Reads always + use ``SPLUNK_ACS_TOKEN`` only, never the write token. """ stack = os.environ.get("SPLUNK_ACS_STACK") or stack token = os.environ.get("SPLUNK_ACS_TOKEN") + if write: + token = os.environ.get("SPLUNK_ACS_WRITE_TOKEN") or token if not stack: raise UsageError( "Could not determine the Splunk Cloud stack. Set SPLUNK_URL to your " "https://.splunkcloud.com host (or set SPLUNK_ACS_STACK)." ) if not token: - raise UsageError("No ACS token. Set SPLUNK_ACS_TOKEN for Splunk Cloud operations.") + env_name = "SPLUNK_ACS_WRITE_TOKEN or SPLUNK_ACS_TOKEN" if write else "SPLUNK_ACS_TOKEN" + raise UsageError(f"No ACS token. Set {env_name} for Splunk Cloud operations.") if not _STACK_RE.fullmatch(stack): raise UsageError( "Invalid ACS stack name. Use only letters, numbers, and hyphens, " @@ -60,7 +74,15 @@ def acs_config_from_env(stack: str | None = None) -> AcsConfig: class AcsClient: - """Read-only GET access to one Splunk Cloud stack's ACS adminconfig/v2 API.""" + """Access to one Splunk Cloud stack's ACS adminconfig/v2 API. + + Reads (:meth:`get`) are unrestricted. Mutations (:meth:`write`) are dry-run + gated the same way :meth:`vct_splunk.api.client.SplunkClient.write` is -- + ``config.dry_run`` sends nothing and returns a preview instead. This client + does not itself decide *which* resources may be written or whether the + caller opted in; that gate is one layer up, in + :func:`vct_splunk.commands.write.refuse_cloud_write`. + """ def __init__(self, config: AcsConfig, *, transport: httpx.BaseTransport | None = None) -> None: if not _STACK_RE.fullmatch(config.stack): @@ -81,13 +103,46 @@ def __exit__(self, exc_type: object, exc: object, tb: object) -> None: def get(self, path: str, params: dict[str, Any] | None = None) -> Any: """GET an ACS read endpoint and return the parsed JSON.""" + return self.request("GET", path, params=params) + + def write(self, method: str, path: str, json_body: Any | None = None) -> Any: + """Mutating ACS request. When dry_run is set, sends nothing and returns a preview. + + Mirrors :meth:`vct_splunk.api.client.SplunkClient.write`: the caller (an + ACS operation function) is trusted to have already decided this mutation + is allowed to run; this method only decides whether to send it. + """ + if self.config.dry_run: + base = f"{self.config.base_url}/{self.config.stack}/adminconfig/v2" + return { + "dry_run": True, + "request": {"method": method, "path": "/" + path.lstrip("/"), "body": json_body}, + "target": public_target(base), + } + return self.request(method, path, json_body=json_body) + + def request( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + json_body: Any | None = None, + ) -> Any: + """Send one ACS request and return the parsed JSON response. + + Shared by every read and (non-dry-run) write: retries 429/5xx honoring + ``Retry-After``, and maps status codes to the same typed errors GET has + always raised. + """ url = "/" + path.lstrip("/") for attempt in range(_MAX_RETRIES + 1): try: - resp = self._http.get(url, params=params) + resp = self._http.request(method, url, params=params, json=json_body) except httpx.HTTPError as exc: raise TransportError( - f"Could not reach ACS at {safe_target(self.config.base_url)}: {exc}" + f"Could not reach ACS at {public_target(self.config.base_url)}: " + f"{redact_exception_text(str(exc))}" ) from exc if (resp.status_code == 429 or 500 <= resp.status_code < 600) and ( attempt < _MAX_RETRIES @@ -99,13 +154,13 @@ def get(self, path: str, params: dict[str, Any] | None = None) -> Any: if resp.status_code == 404: raise NotFoundError(f"ACS endpoint not found: {url}") if resp.status_code >= 400: - raise APIError(f"ACS returned {resp.status_code} for GET {url}") + raise APIError(f"ACS returned {resp.status_code} for {method} {url}") if not resp.content: return {} try: return resp.json() except ValueError as exc: - raise APIError(f"ACS returned malformed JSON for GET {url}") from exc + raise APIError(f"ACS returned malformed JSON for {method} {url}") from exc raise TransportError("ACS retries exhausted") # pragma: no cover diff --git a/src/vct_splunk/api/acs/operations.py b/src/vct_splunk/api/acs/operations.py index c263617..9da971d 100644 --- a/src/vct_splunk/api/acs/operations.py +++ b/src/vct_splunk/api/acs/operations.py @@ -1,10 +1,21 @@ -"""Read-only ACS operations.""" +"""ACS operations: unrestricted reads plus the three writable resources. + +Only index, role, and HTTP Event Collector token support create/update/delete +on Splunk Cloud (see :data:`WRITABLE`, checked against Splunk's public OpenAPI +by ``tests/integration/test_acs_public_spec.py`` via :data:`WRITE_PATHS`). +Every write goes through :meth:`~vct_splunk.api.acs.client.AcsClient.write`, +so ``--dry-run`` sends nothing here exactly as it does for the Enterprise REST +path; the opt-in gate and the allowlist of which (resource, verb) pairs even +reach :func:`cloud_write` live one layer up, in +:mod:`vct_splunk.commands.write` and :mod:`vct_splunk.commands.dispatch`. +""" from __future__ import annotations from typing import Any from ...utils.errors import APIError +from ...utils.path import path_segment from ...utils.redact import redact_secrets from .client import AcsClient @@ -23,6 +34,30 @@ #: Every ACS path the CLI reads. READ_PATHS = tuple(LIST_ENVELOPES) +#: CLI resource name -> (ACS collection path, Splunk's own OpenAPI +#: path-parameter name for one item). The parameter name differs per resource +#: (`{index}`, `{roleName}`, `{hec}`), so :data:`WRITE_PATHS` below carries it +#: rather than a generic placeholder -- the spec drift check looks paths up +#: verbatim in Splunk's published contract. +WRITABLE: dict[str, tuple[str, str]] = { + "index": (INDEXES, "index"), + "role": (ROLES, "roleName"), + "hec-token": (HEC_TOKENS, "hec"), +} + +#: (path template, HTTP method) pairs ACS exposes for create/update/delete on +#: the three writable resources, derived from :data:`WRITABLE` so this and +#: :func:`cloud_write` can never drift apart. +WRITE_PATHS: tuple[tuple[str, str], ...] = tuple( + (path, method) + for base, param in WRITABLE.values() + for path, method in ( + (base, "post"), + (f"{base}/{{{param}}}", "patch"), + (f"{base}/{{{param}}}", "delete"), + ) +) + def list_cloud_indexes(client: AcsClient) -> list[dict[str, Any]]: """List indexes on the Cloud stack (ACS).""" @@ -59,3 +94,31 @@ def _list(client: AcsClient, path: str, envelope: str) -> list[dict[str, Any]]: if len(page) < 100: return output offset += len(page) + + +def cloud_write( + client: AcsClient, resource: str, verb: str, name: str, body: dict[str, Any] | None = None +) -> Any: + """Create, update, or delete one object of a Cloud-writable resource via ACS. + + ``resource`` must be a key of :data:`WRITABLE` and ``verb`` one of + create/update/delete -- the caller + (:func:`vct_splunk.commands.dispatch.dispatch_write`) has already checked + both against the same allowlist this reads, so a `KeyError` here would + mean that check was bypassed. + + Unlike the Enterprise ``hec-token create`` (whose response is allowed to + reveal the minted token because it is the only way to learn it), every ACS + write response is redacted here -- Splunk Cloud CI output must never carry + a live credential, so there is no reveal-once escape hatch on this path. + """ + collection, _ = WRITABLE[resource] + if verb == "create": + result = client.write("POST", collection, {**(body or {}), "name": name}) + else: + path = f"{collection}/{path_segment(name, label='name')}" + method = "PATCH" if verb == "update" else "DELETE" + result = client.write(method, path, body or {}) + if isinstance(result, dict) and result.get("dry_run"): + return result + return redact_secrets(result) diff --git a/src/vct_splunk/commands/command_factory.py b/src/vct_splunk/commands/command_factory.py index f5d596e..6e7e4ed 100644 --- a/src/vct_splunk/commands/command_factory.py +++ b/src/vct_splunk/commands/command_factory.py @@ -28,6 +28,15 @@ _VERB_ALIASES = {"add": "create", "edit": "update", "remove": "delete"} +#: For a Cloud-writable spec, the one typed `Field` (if any) that maps onto an +#: ACS JSON key. Every other typed field has no ACS equivalent -- `--set` with +#: ACS's own field name (e.g. `searchableDays`, not a Splunk REST form field) +#: is the escape hatch for those, same as it is for anything --field-options +#: does not cover on Enterprise. +_ACS_FIELD_KEYS: dict[str, dict[str, str]] = { + "index": {"max_gb": "maxDataSizeMB"}, +} + def _help_for(spec: EndpointConfig, verb: str) -> str: """One-line help for a generated command, in the hand-written commands' style.""" @@ -47,6 +56,36 @@ def _help_for(spec: EndpointConfig, verb: str) -> str: return texts[verb] +def _acs_body(spec: EndpointConfig, fields: dict[str, Any], sets: dict[str, str]) -> dict[str, Any]: + """Build the ACS JSON body for a Cloud-routed create/update. + + ``--set KEY=VALUE`` pairs pass straight through as given -- ``KEY`` must be + the ACS field's own JSON name, which is not always the same as the Splunk + REST form field the same option sends on Enterprise. A typed field option + (e.g. ``--max-gb``) is honored only when this spec declares an + ACS-equivalent key in :data:`_ACS_FIELD_KEYS`; anything else raises rather + than silently dropping or mis-mapping a value the caller explicitly asked + to send. + """ + mapping = _ACS_FIELD_KEYS.get(spec.name, {}) + by_opt = {f.opt: f for f in spec.fields} + body: dict[str, Any] = {} + for opt, value in fields.items(): + if value is None or value == (): + continue + key = mapping.get(opt) + if key is None: + dashed = opt.replace("_", "-") + raise UsageError( + f"--{dashed} has no Splunk Cloud (ACS) equivalent. " + f"Use --set with the ACS field's own JSON name instead." + ) + f = by_opt[opt] + body[key] = int(float(value) * f.scale) if f.scale else value + body.update(sets) + return body + + def _gate_args( spec: EndpointConfig, verb: str, name: str, owner, app ) -> tuple[str, dict[str, Any]]: @@ -118,6 +157,8 @@ def _create(ctx, name, **opts) -> None: action=action, audit_event=event, run=lambda c: res.create(c, name, fields=fields, sets=sets, owner=owner, app=app), + name=name, + body=_acs_body(spec, fields, sets) if ctx.backend == "cloud" else None, ) out.emit(result, ctx.output_mode, ctx.meta()) @@ -139,6 +180,8 @@ def _update(ctx, name, **opts) -> None: action=action, audit_event=event, run=lambda c: res.update(c, name, fields=fields, sets=sets, owner=owner, app=app), + name=name, + body=_acs_body(spec, fields, sets) if ctx.backend == "cloud" else None, ) out.emit(result, ctx.output_mode, ctx.meta()) @@ -156,6 +199,7 @@ def _delete(ctx, name) -> None: action=action, audit_event=event, run=lambda c: res.delete(c, name, owner=owner, app=app), + name=name, ) out.emit(result, ctx.output_mode, ctx.meta()) diff --git a/src/vct_splunk/commands/dispatch.py b/src/vct_splunk/commands/dispatch.py index 2f82198..b1a4623 100644 --- a/src/vct_splunk/commands/dispatch.py +++ b/src/vct_splunk/commands/dispatch.py @@ -6,6 +6,14 @@ with a clean :class:`UnsupportedBackendError` rather than falling through to an unofficial endpoint. This is the one place that knows both clients exist; the Click-free core stays unaware of backends. + +:func:`has_cloud_write` is the write-side counterpart, checked from +:func:`vct_splunk.commands.write.refuse_cloud_write` before a mutation ever +opens a client. The ACS write call itself +(:func:`vct_splunk.api.acs.operations.cloud_write`) is invoked directly from +:func:`vct_splunk.commands.write.do_write`, which already knows which client it +opened for the deduced backend -- there is no write-side counterpart to +``dispatch_list`` here, only the capability check. """ from __future__ import annotations @@ -24,6 +32,8 @@ "hec-token": acs.list_hec_tokens, } +_WRITE_VERBS = ("create", "update", "delete") + def has_cloud_list(resource: str) -> bool: """True if ``resource``'s list is served by ACS on the Cloud backend.""" @@ -45,3 +55,16 @@ def dispatch_list(ctx: Any, resource: str, rest_call: Callable[[Any], Any]) -> A return op(c) with ctx.client() as c: return rest_call(c) + + +def has_cloud_write(resource: str, verb: str | None = None) -> bool: + """True if ``(resource, verb)`` is a Cloud-writable mutation via ACS. + + With ``verb`` omitted, true if ``resource`` has any Cloud write route at + all -- create/update/delete for index, role, and hec-token only. Every + other Cloud mutation, including enable/disable on those same three + resources, has none. + """ + if resource not in acs.WRITABLE: + return False + return verb is None or verb in _WRITE_VERBS diff --git a/src/vct_splunk/commands/write.py b/src/vct_splunk/commands/write.py index 44f9a33..c7dde97 100644 --- a/src/vct_splunk/commands/write.py +++ b/src/vct_splunk/commands/write.py @@ -1,26 +1,42 @@ """The one shared write path for every mutation. Shell layer. Every gated write -- hand-written or factory-generated -- runs through -:func:`do_write`, so target resolution, the confirmation gate, and audit logging -live in exactly one place. +:func:`do_write`, so target resolution, both write-enable gates, backend +routing, the confirmation gate, and audit logging all live in exactly one +place. This function is also the seam for the larger write-safety framework (#12): deferred pieces such as plan tokens, optimistic concurrency, and a freeze-writes kill switch would all belong *inside* it. Add each only when a real second -writer needs it; today the call sites need exactly the gate + audit below. +writer needs it; today the call sites need exactly the gates + audit below. """ from __future__ import annotations +import os from collections.abc import Callable from typing import Any -from ..api.client import SplunkClient -from ..config.loader import load_config +from ..api.acs.client import AcsClient, acs_config_from_env +from ..api.acs.operations import cloud_write +from ..config.loader import _resolve_target, load_config from ..output import formatter as out from ..utils import audit -from ..utils.errors import UnsupportedBackendError -from ..utils.redact import safe_target +from ..utils.backends import cloud_stack_from_url +from ..utils.errors import UnsupportedBackendError, UsageError +from ..utils.redact import public_target, safe_target +from .dispatch import has_cloud_write + +#: Opt-in for ANY real mutation, on any backend. Never a CLI flag -- a saved +#: command line must not be able to enable a write. ``--dry-run`` needs neither +#: this nor :data:`CLOUD_WRITE_ENV`: a preview sends nothing regardless, so +#: :func:`do_write` skips both checks for it. +ENABLE_WRITES_ENV = "SPLUNK_ENABLE_WRITES" + +#: Additional opt-in required, on top of :data:`ENABLE_WRITES_ENV`, for a real +#: write against a Splunk Cloud target. Never a CLI flag either. See +#: :func:`refuse_cloud_write`. +CLOUD_WRITE_ENV = "SPLUNK_CLOUD_WRITE" def do_write( @@ -28,48 +44,112 @@ def do_write( *, action: str, audit_event: dict[str, Any], - run: Callable[[SplunkClient], dict[str, Any]], + run: Callable[[Any], dict[str, Any]], target: str | None = None, + name: str | None = None, + body: dict[str, Any] | None = None, ) -> dict[str, Any]: """Run one gated mutation: confirm it, execute it, and audit it. - Resolves the target up front (so a missing ``SPLUNK_URL`` / ``SPLUNK_TOKEN`` + Refuses outright unless :data:`ENABLE_WRITES_ENV` is set (every backend, + real writes only -- ``--dry-run`` sends nothing regardless, so it is exempt) + and, on a Cloud target, unless the object is one of the three ACS-writable + resources and :data:`CLOUD_WRITE_ENV` is also set. That second check is + enforced even for a preview -- see :func:`refuse_cloud_write`. Then resolves + the target up front (so a missing ``SPLUNK_URL`` / credential fails before we prompt), gates the write via :func:`output.confirm_write` - (dry-run / ``--yes`` / non-interactive fail-fast), runs ``run(client)``, and - appends an audit record for any real (non-dry-run) write. + (dry-run / ``--yes`` / non-interactive fail-fast), runs the operation on the + backend-appropriate client, and appends an audit record for any real + (non-dry-run) write. Args: ctx: The command :class:`~vct_splunk.commands.context.Ctx`. action: A human phrase for the prompt, e.g. ``"create index 'web'"``. - audit_event: Fields to record on a real write; ``target`` is added here. - run: Callable given an open client, returning the operation result. + audit_event: Fields to record on a real write; ``target`` is added + here. ``audit_event["action"]`` is ``"."``, e.g. + ``"index.create"``. + run: Callable given an open Enterprise ``SplunkClient``, returning the + operation result. Not called for a Cloud write -- see ``name`` and + ``body``. target: The Splunk URL; resolved from the environment when omitted. + name: The object name, needed only for a Cloud write. The REST path + already has its name closed over inside ``run``. + body: The ACS JSON body for a Cloud create/update. Built by the caller + -- it needs the resource's own field mapping, which this module + does not have. Ignored for delete and on Enterprise. Returns: The operation result, ready for :func:`output.emit`. """ - # audit_event["action"] is ".", e.g. "index.create". resource, _, verb = str(audit_event.get("action", "")).partition(".") refuse_cloud_write(ctx, resource, verb) - target = safe_target( - target or load_config(ctx.base_url, profile=getattr(ctx, "profile", None)).base_url - ) - out.confirm_write(ctx, action, target) - with ctx.client() as c: - result = run(c) + if not ctx.dry_run: + _require_writes_enabled(resource, verb) + backend = getattr(ctx, "backend", "enterprise") + if target is None: + if backend == "cloud": + # A Cloud write authenticates to ACS, not splunkd, so resolving the + # target must not go through `load_config` -- that function demands + # a splunkd credential (SPLUNK_TOKEN et al.) this path never needs. + target = _resolve_target(ctx.base_url, getattr(ctx, "profile", None))[0].base_url + else: + target = load_config(ctx.base_url, profile=getattr(ctx, "profile", None)).base_url + out.confirm_write(ctx, action, public_target(target)) + if backend == "cloud": + stack = cloud_stack_from_url(ctx.base_url) + config = acs_config_from_env(stack, write=True) + config.dry_run = ctx.dry_run + with AcsClient(config) as c: + result = cloud_write(c, resource, verb, name or "", body) + else: + with ctx.client() as c: + result = run(c) if not (isinstance(result, dict) and result.get("dry_run")): - audit.record({**audit_event, "target": target}) + audit.record({**audit_event, "target": safe_target(target)}) return result def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None: - """Stop a mutation aimed at a Splunk Cloud stack, before anything else runs. + """Stop a Cloud mutation with no ACS route; gate the rest behind opt-in. - Writes are Enterprise-only this release. :func:`do_write` calls this, which - is enough for a command that reaches the gate directly. A command that first - resolves a namespace calls it earlier as well, so a Cloud target is told that - writes are unsupported rather than being asked for an ``--app`` that would - not have helped. + Two independent checks, in order, both enforced unconditionally -- + dry-run or not: + + 1. **Capability**. Only create/update/delete for index, role, and + hec-token (see :func:`vct_splunk.commands.dispatch.has_cloud_write`) + exist on Cloud at all; every other Cloud mutation -- including + enable/disable on those same three resources -- has no route to opt + into, so this always raises. + 2. **Opt-in** (:data:`CLOUD_WRITE_ENV`). Unlike the top-level + :data:`ENABLE_WRITES_ENV` gate in :func:`do_write` (which a preview + skips, since it sends nothing regardless), this one is a capability + fact as much as a safety switch: even a Cloud *preview* needs to prove + the object is one of the three ACS-writable resources, so ``--dry-run`` + does not exempt it. + + :func:`do_write` calls this, which is enough for a command that reaches the + gate directly. A command that first resolves a namespace calls it earlier + as well, so a Cloud target is told writes are unsupported rather than + asked for an ``--app`` that would not have helped. """ - if getattr(ctx, "backend", "enterprise") == "cloud": + if getattr(ctx, "backend", "enterprise") != "cloud": + return + if not has_cloud_write(resource, verb): + raise UnsupportedBackendError(resource or "this resource", verb or "write", "cloud") + if os.environ.get(CLOUD_WRITE_ENV) != "true": raise UnsupportedBackendError(resource or "this resource", verb or "write", "cloud") + + +def _require_writes_enabled(resource: str, verb: str) -> None: + """Refuse a real (non-dry-run) mutation unless :data:`ENABLE_WRITES_ENV` is set. + + Every backend, always -- there is no CLI flag, so a saved command line can + never enable a write. Only :func:`do_write` calls this, and only when + ``ctx.dry_run`` is False; a preview sends nothing and needs no opt-in. + """ + if os.environ.get(ENABLE_WRITES_ENV) == "true": + return + raise UsageError( + f"Writes are disabled by default. Set {ENABLE_WRITES_ENV}=true to allow " + f"`{resource} {verb}` (there is no CLI flag for this)." + ) diff --git a/src/vct_splunk/utils/backends.py b/src/vct_splunk/utils/backends.py index 5365456..b17701e 100644 --- a/src/vct_splunk/utils/backends.py +++ b/src/vct_splunk/utils/backends.py @@ -19,9 +19,12 @@ _CLOUD_HOST_MARKER = "splunkcloud" #: What each backend supports, for the `splunk inspect` report. Values are True -#: (full support) or a short string naming the limit. Cloud is read-only this -#: release. This is informational only -- routing is decided per command, and an -#: unavailable operation stops with a typed error, never a silent fallthrough. +#: (full support) or a short string naming the limit. Cloud writes are opt-in +#: (SPLUNK_ENABLE_WRITES=true and SPLUNK_CLOUD_WRITE=true) and narrow: only +#: index/role/hec-token create, update, and delete go through ACS -- everything +#: else on Cloud is read-only. This is informational only -- routing is decided +#: per command, and an unavailable operation stops with a typed error, never a +#: silent fallthrough. CAPABILITIES: dict[str, dict[str, Any]] = { "enterprise": { "search": True, @@ -35,11 +38,14 @@ "health": True, }, "cloud": { - "indexes": "read-only (ACS)", - "hec_tokens": "read-only (ACS)", - "roles": "read-only (ACS)", + "indexes": "read via ACS; create/update/delete gated behind opt-in (see 'writes')", + "hec_tokens": "read via ACS; create/update/delete gated behind opt-in (see 'writes')", + "roles": "read via ACS; create/update/delete gated behind opt-in (see 'writes')", "search": "via the search head REST (your SPLUNK_URL), where the stack permits", - "writes": "not supported this release (read-only)", + "writes": ( + "opt-in: SPLUNK_ENABLE_WRITES=true and SPLUNK_CLOUD_WRITE=true, " + "index/role/hec-token create/update/delete only" + ), }, } @@ -90,6 +96,8 @@ def inspect_report(url: str | None = None) -> dict[str, Any]: # live Cloud target's identity is shown, and only with consent to leak it. report["stack_configured"] = cloud_stack_from_url(url) is not None report["note"] = ( - "Cloud/ACS coverage is read-only and not yet certified against a live stack." + "Cloud/ACS reads are certified; writes are opt-in " + "(SPLUNK_ENABLE_WRITES=true, SPLUNK_CLOUD_WRITE=true, index/role/hec-token " + "create/update/delete only) and not yet certified against a live stack." ) return report diff --git a/tests/TESTING.md b/tests/TESTING.md index 451fa64..9834f7a 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -1,6 +1,6 @@ # Running the tests -Six groups. Only the first needs nothing at all — start there. +Seven groups. Only the first needs nothing at all — start there. | Group | What it checks | What you must provide | Directory | | --- | --- | --- | --- | @@ -8,6 +8,7 @@ Six groups. Only the first needs nothing at all — start there. | Enterprise reads | Every read command against a real server | A reachable Splunk | `tests/integration/enterprise/read/` | | Enterprise writes | Every change, then undoes it | A **disposable** Splunk | `tests/integration/enterprise/write/` | | Cloud reads | Every read command against a real Cloud stack | A Cloud stack and an ACS token | `tests/integration/cloud/read/` | +| Cloud writes | Index/role/HEC create-update-delete, then undo | A **non-production** Cloud stack + write ACS token | `tests/integration/cloud/write/` | | ACS contract | Whether Splunk changed its public Cloud API | Nothing | `tests/integration/` | | Fuzz | That a credentialed URL never survives redaction | Linux on x86_64 | `tests/fuzz/` | @@ -33,8 +34,8 @@ Two files in this group cover the Cloud path in full, so you can check it without an account: ```bash -.venv/bin/python -m pytest tests/unit/test_acs_loopback.py # every Cloud read -.venv/bin/python -m pytest tests/unit/test_cloud_write_refusal.py # every Cloud write +.venv/bin/python -m pytest tests/unit/test_acs_loopback.py # every Cloud read +.venv/bin/python -m pytest tests/unit/test_write_gating.py # every write gate, both backends ``` `test_acs_loopback.py` starts a small HTTP server on a loopback port, points @@ -42,9 +43,13 @@ the tool's Cloud address at it, and runs each read command the whole way through. Nothing is stubbed out, so it checks the address the tool builds, the token it sends, and that a returned secret never reaches your screen. -`test_cloud_write_refusal.py` runs every command that changes something against -a Cloud address, in the form that would really do it, and fails if any of them -so much as opens a connection. +`test_write_gating.py` proves both write-enable gates: every mutation on +either backend refuses by default (`SPLUNK_ENABLE_WRITES` unset), a Cloud +mutation additionally refuses without `SPLUNK_CLOUD_WRITE` -- even for the +three ACS-writable resources, even as a `--dry-run` preview -- and, opted in, +a mocked ACS create/update/delete actually reaches the client. Every refusal +case installs a transport that fails the test if any request leaves the +process. Group 4 below is what these cannot be: proof that a real stack answers the way Splunk's specification says it does. @@ -107,6 +112,11 @@ export SPLUNK_TEST_SERVER_FIXTURE_DIR=/opt/splunk/var/run/splunk/lookup_tmp .venv/bin/python -m pytest tests/integration/enterprise/write -v ``` +`SPLUNK_ENABLE_WRITES` (the top-level write-enable gate every real mutation +needs, on any backend) does not need to be set here -- the whole test session +force-enables it (`tests/conftest.py`), the same as every other unit and +integration suite. + Clean up when you are finished: `docker rm -f splunk-test`. ## Group 4: human-operated Splunk Cloud validation @@ -145,7 +155,7 @@ current public ACS OpenAPI contract: ```bash .venv/bin/python -m pytest \ tests/unit/test_acs_loopback.py \ - tests/unit/test_cloud_write_refusal.py \ + tests/unit/test_write_gating.py \ -q --tb=short SPLUNK_ACS_SPEC_TEST=true .venv/bin/python -m pytest \ @@ -280,6 +290,36 @@ when every required row passes and the cleanup is complete. | Cleanup | Variables unset and short-lived tokens revoked | | | **Overall approval** | **PASS** | | +## Group 4b: Cloud writes + +> **Warning.** This group creates, changes, and deletes real objects on a +> Splunk Cloud stack. Point it only at a **non-production** stack. + +The primary path is the `Splunk Cloud Write Canary` GitHub Actions workflow, +not a human-operated runbook: `workflow_dispatch` only, gated by a typed +`confirm=WRITE` input, a `tests: splunk cloud write` HEAD commit-subject +requirement, and a protected `splunk-cloud-write` GitHub Environment with +required reviewers around the destructive half. It runs the Cloud read canary +first, then the write suite below, so one approved dispatch certifies both. + +To run the suite locally instead, on top of the read-only setup in the runbook +above: + +```bash +export SPLUNK_ENABLE_WRITES=true +export SPLUNK_CLOUD_WRITE=true +export SPLUNK_ACS_WRITE_TOKEN="$( + .venv/bin/python -c 'import getpass; print(getpass.getpass("Write-scoped ACS token: "))' +)" + +.venv/bin/python -m pytest tests/integration/cloud/write -v +``` + +Each test creates one uniquely named object, updates it, deletes it, and polls +until it is gone (ACS index and HEC-token deletes complete asynchronously). A +cleanup failure fails the test loudly rather than leaking a `vct_ci_*` object +silently. + ## Group 5: ACS public contract No credentials. It downloads Splunk's public Cloud API description and reports @@ -326,7 +366,9 @@ global state, fails on a cleanup leak, and restarts Splunk last. Every pull request runs group 1 — including the two Cloud contract files above — plus lint and type checks. Pull requests that touch code also run groups 2, 3, and 6 against a throwaway container and a Linux runner. Groups 4 and 5 run -weekly; group 4 reports -that there is nothing to certify until a Cloud stack is configured, rather than -passing without checking anything. A single check named **Merge Gate** -summarizes the pull-request jobs. +weekly, and group 4 reports that there is nothing to certify until a Cloud +stack is configured, rather than passing without checking anything. Group 4b +(Cloud writes) never runs on a schedule or a pull request — it runs only on an +approved, human-gated `workflow_dispatch` of the `Splunk Cloud Write Canary` +workflow. A single check named **Merge Gate** summarizes the pull-request +jobs. diff --git a/tests/conftest.py b/tests/conftest.py index 7e2c750..168b1f4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,7 @@ from __future__ import annotations import inspect +import os from collections.abc import Callable from typing import Any @@ -14,6 +15,19 @@ _TEST_URL = "https://splunk.test:8089" +@pytest.fixture(scope="session", autouse=True) +def _writes_enabled_for_tests(): + """Every test may attempt a real write. ``SPLUNK_ENABLE_WRITES`` gates + production usage, not the suite that proves the gate itself works -- a + test proving the *default* refusal deletes this var for itself (see + ``test_cloud_write_refusal.py``'s fixture for the existing pattern of + clearing an ambient opt-in before asserting on its absence). + """ + os.environ["SPLUNK_ENABLE_WRITES"] = "true" + yield + os.environ.pop("SPLUNK_ENABLE_WRITES", None) + + def cli_runner() -> CliRunner: """Return a ``CliRunner`` that captures stderr separately on every Click. diff --git a/tests/integration/cloud/write/conftest.py b/tests/integration/cloud/write/conftest.py new file mode 100644 index 0000000..a32fd9d --- /dev/null +++ b/tests/integration/cloud/write/conftest.py @@ -0,0 +1,79 @@ +"""Write-specific safety gate for destructive Splunk Cloud tests. + +Nested under ``tests/integration/cloud/``, so ``_require_cloud_target`` (that +package's ``conftest.py``) already gates ``SPLUNK_ACS_LIVE_TEST``, a Cloud +``SPLUNK_URL``, and ``SPLUNK_ACS_TOKEN`` before any test here runs. This file +adds the write-specific opt-in on top -- ``SPLUNK_ENABLE_WRITES`` is already +``"true"`` for the whole test session (``tests/conftest.py``), mirroring +production usage where a real write needs both variables set. The CLI harness +itself (``WriteTestCli``) is shared with the Enterprise suite; see +``tests/write_test_cli.py``. +""" + +from __future__ import annotations + +import json +import os +import uuid +from collections.abc import Iterator + +import pytest +from click.testing import CliRunner + +from vct_splunk.cli import cli +from write_test_cli import WriteTestCli + +#: Groups every name this test run creates, so a stale object from another run +#: (or one still mid-async-delete) is never mistaken for one of this run's own. +RUN_ID = os.environ.get("GITHUB_RUN_ID") or uuid.uuid4().hex[:8] + + +def unique_name(prefix: str) -> str: + """A ``vct_ci___`` name unique to this test run.""" + return f"vct_ci_{RUN_ID}_{prefix}_{uuid.uuid4().hex[:8]}" + + +@pytest.fixture(autouse=True) +def _require_cloud_write_opt_in() -> None: + if os.environ.get("SPLUNK_CLOUD_WRITE") == "true": + return + message = "set SPLUNK_CLOUD_WRITE=true to run destructive Splunk Cloud tests" + if os.environ.get("CI") == "true": + pytest.fail(message, pytrace=False) + pytest.skip(message) + + +@pytest.fixture(scope="session", autouse=True) +def _preflight_leftover_report() -> None: + """List existing `vct_ci_*` objects before this run starts, without failing. + + ACS index and HEC-token deletes complete asynchronously, so an object from + a run that finished (and whose own cleanup succeeded) moments ago can still + be listed here. Asserting on that would be a false failure, so this is a + report only. The enforceable guarantee is narrower and per-test: each test + below polls its own delete to completion before returning, and + `WriteTestCli.finish()` fails loudly if any registered reverse-cleanup + command itself errors. + """ + runner = CliRunner() + for resource in ("index", "role", "hec-token"): + result = runner.invoke(cli, [resource, "list", "--output", "json"]) + if result.exit_code != 0: + continue + payload = json.loads(result.stdout) + leftover = [ + item.get("name") + for item in payload.get("data", []) + if isinstance(item, dict) and str(item.get("name", "")).startswith("vct_ci_") + ] + if leftover: + print(f"[cloud write preflight] leftover {resource} objects: {leftover}") + + +@pytest.fixture +def cloud_cli() -> Iterator[WriteTestCli]: + harness = WriteTestCli() + try: + yield harness + finally: + harness.finish() diff --git a/tests/integration/cloud/write/test_write_catalog.py b/tests/integration/cloud/write/test_write_catalog.py new file mode 100644 index 0000000..4bf57a7 --- /dev/null +++ b/tests/integration/cloud/write/test_write_catalog.py @@ -0,0 +1,79 @@ +"""Apply every Cloud-writable mutation against a real ACS-backed stack, then undo it. + +The write-side mirror of `tests/integration/enterprise/write/test_write_catalog.py`, +narrowed to the three resources ACS actually lets this CLI mutate: `index`, +`role`, and `hec-token`. No restart, no app install, no file inputs -- ACS has +no equivalent surface for any of those. All three follow the same shape +(create, change one setting, delete, poll until gone), so they run as one +parametrized test rather than three near-identical copies. + +Each case creates one uniquely named object, updates it, deletes it, and polls +the resource's list until the name is gone (ACS index and HEC-token deletes +complete asynchronously, so a `202` response does not mean the object is gone +yet). `cloud_cli.finish()` still fails loudly if the registered reverse +cleanup itself errors -- the poll only proves the delete this test issued +actually completed. +""" + +from __future__ import annotations + +import time + +import pytest + +from write_test_cli import WriteTestCli + +from .conftest import unique_name + +pytestmark = [pytest.mark.integration, pytest.mark.cloud, pytest.mark.write] + +_POLL_TIMEOUT = 120.0 +_POLL_INTERVAL = 3.0 + +#: resource -> (create args beyond the name, update args). Each case creates +#: with the given extra args, updates with the given args, deletes, and polls +#: until the name is gone from ` list`. +_CASES: tuple[tuple[str, tuple[str, ...], tuple[str, ...]], ...] = ( + ("index", (), ("--max-gb", "2")), + ("role", ("--set", "defaultApp=search"), ("--set", "srchFilter=search index=main")), + ("hec-token", ("--set", "defaultIndex=main"), ("--set", "defaultSourcetype=vct_ci")), +) + + +def _poll_until_gone(cloud_cli: WriteTestCli, resource: str, name: str) -> None: + """Poll `resource list` until *name* no longer appears, or fail after the timeout.""" + deadline = time.monotonic() + _POLL_TIMEOUT + while time.monotonic() < deadline: + items = cloud_cli.run(resource, "list") + if not any(isinstance(item, dict) and item.get("name") == name for item in items): + return + time.sleep(_POLL_INTERVAL) + pytest.fail(f"{resource} {name!r} is still listed {_POLL_TIMEOUT:.0f}s after delete") + + +@pytest.mark.parametrize("resource, create_args, update_args", _CASES, ids=lambda v: v) +def test_create_update_delete( + cloud_cli: WriteTestCli, + resource: str, + create_args: tuple[str, ...], + update_args: tuple[str, ...], +) -> None: + """Create one object, change a setting, delete it, and confirm ACS finished deleting it. + + The hec-token case also proves the ACS create response never shows the + minted token -- unlike the Enterprise `hec-token create`, which is allowed + to reveal it once. + """ + name = unique_name(resource.replace("-", "_")) + label = f"delete {resource} {name}" + cloud_cli.cleanup(label, resource, "delete", name) + + created = cloud_cli.write(resource, "create", name, *create_args) + if resource == "hec-token": + assert created.get("token") in (None, "") + + cloud_cli.write(resource, "update", name, *update_args) + cloud_cli.write(resource, "delete", name) + + cloud_cli.drop_cleanup(label) # already deleted above; nothing left to undo + _poll_until_gone(cloud_cli, resource, name) diff --git a/tests/integration/enterprise/write/conftest.py b/tests/integration/enterprise/write/conftest.py index 56444a8..5d54419 100644 --- a/tests/integration/enterprise/write/conftest.py +++ b/tests/integration/enterprise/write/conftest.py @@ -1,17 +1,13 @@ -"""Shared safety gate and CLI harness for destructive Enterprise tests.""" +"""Shared safety gate for destructive Enterprise tests. Harness lives in write_test_cli.""" from __future__ import annotations -import json import os -from collections.abc import Callable, Iterator -from dataclasses import dataclass, field -from typing import Any +from collections.abc import Iterator import pytest -from click.testing import CliRunner, Result -from vct_splunk.cli import cli +from write_test_cli import WriteTestCli @pytest.fixture(autouse=True) @@ -24,51 +20,9 @@ def _require_write_opt_in() -> None: pytest.skip(message) -@dataclass -class EnterpriseCli: - """Invoke the public CLI and fail explicitly if reverse cleanup leaks state.""" - - runner: CliRunner = field(default_factory=CliRunner) - cleanups: list[tuple[str, Callable[[], Result]]] = field(default_factory=list) - - def run( - self, - *argv: str, - exit_codes: tuple[int, ...] = (0,), - ) -> dict[str, Any]: - result = self.runner.invoke(cli, [*argv, "--output", "json"]) - assert result.exit_code in exit_codes, ( - f"{' '.join(argv)} exited {result.exit_code}\n{result.output}" - ) - payload = json.loads(result.stdout) - if result.exit_code == 0: - assert set(payload) == {"data", "meta"} - return payload["data"] - return payload["error"] - - def write(self, *argv: str) -> dict[str, Any]: - return self.run(*argv, "--yes") - - def cleanup(self, label: str, *argv: str) -> None: - self.cleanups.append( - ( - label, - lambda: self.runner.invoke(cli, [*argv, "--yes", "--output", "json"]), - ) - ) - - def finish(self) -> None: - failures: list[str] = [] - for label, cleanup in reversed(self.cleanups): - result = cleanup() - if result.exit_code != 0: - failures.append(f"{label}: exit {result.exit_code}: {result.output}") - assert not failures, "cleanup failures:\n" + "\n".join(failures) - - @pytest.fixture -def enterprise_cli() -> Iterator[EnterpriseCli]: - harness = EnterpriseCli() +def enterprise_cli() -> Iterator[WriteTestCli]: + harness = WriteTestCli() try: yield harness finally: diff --git a/tests/integration/test_acs_public_spec.py b/tests/integration/test_acs_public_spec.py index 5a64e88..ee63638 100644 --- a/tests/integration/test_acs_public_spec.py +++ b/tests/integration/test_acs_public_spec.py @@ -7,7 +7,7 @@ import httpx import pytest -from vct_splunk.api.acs.operations import LIST_ENVELOPES +from vct_splunk.api.acs.operations import LIST_ENVELOPES, WRITE_PATHS pytestmark = [ pytest.mark.integration, @@ -31,3 +31,13 @@ def test_implemented_acs_contract_matches_public_spec(): ) schema = operation["responses"]["200"]["content"]["application/json"]["schema"] assert schema["properties"][envelope]["type"] == "array" + + +def test_implemented_acs_write_contract_matches_public_spec(): + if os.environ.get("SPLUNK_ACS_SPEC_TEST") != "true": + pytest.skip("set SPLUNK_ACS_SPEC_TEST=true to check the public ACS OpenAPI") + + public = httpx.get(_SOURCE, timeout=30).raise_for_status().json() + for path, method in WRITE_PATHS: + operations = public["paths"][f"/{{stack}}/adminconfig/v2/{path}"] + assert method in operations, f"{method.upper()} {path} is missing from the public spec" diff --git a/tests/unit/test_cloud_write_refusal.py b/tests/unit/test_cloud_write_refusal.py deleted file mode 100644 index f5b4435..0000000 --- a/tests/unit/test_cloud_write_refusal.py +++ /dev/null @@ -1,71 +0,0 @@ -"""Every mutation must refuse a Splunk Cloud target before touching the network. - -Cloud support is read-only. The property that makes that safe is not "writes -usually fail" — it is that a write against a Cloud stack stops at the gate, with -a typed error, having sent nothing. A refusal that happened only because a -credential was missing, or one that fired after a request went out, would both -look like success in a weaker test. - -So this suite runs every write leaf in the catalog with `--yes`, the form that -would otherwise execute, and installs a transport that fails the test if any -request leaves the process. - -Its read counterpart is `test_acs_loopback.py`. Both run credential-free on -every change; neither needs a Cloud stack. -""" - -from __future__ import annotations - -import json - -import httpx -import pytest -from click.testing import CliRunner - -from cli_catalog import CATALOG, Case -from vct_splunk.cli import cli -from vct_splunk.utils.errors import UnsupportedBackendError - -WRITE_CASES = tuple(case for case in CATALOG if case.kind == "write") - - -@pytest.fixture(autouse=True) -def cloud_target_with_no_network(monkeypatch: pytest.MonkeyPatch) -> None: - """Point the CLI at a Cloud stack and make any real request an error. - - Patching the transport rather than the client covers both clients at once, - including a future one, and cannot be satisfied by a command that simply - fails earlier for an unrelated reason. - """ - # Clear the ambient settings a developer may have exported, so the suite - # tests the tool rather than the machine it runs on. - for name in ("SPLUNK_APP", "SPLUNK_OWNER", "SPLUNK_PROFILE", "SPLUNK_ACS_BASE_URL"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") - monkeypatch.setenv("SPLUNK_ACS_TOKEN", "unused") - monkeypatch.setenv("SPLUNK_TOKEN", "unused") - - def refuse(*args: object, **kwargs: object) -> object: - raise AssertionError("a write against a Cloud target reached the network") - - monkeypatch.setattr(httpx.HTTPTransport, "handle_request", refuse) - - -def _argv(case: Case) -> list[str]: - """Build the executing form of a write: no preview, no prompt.""" - args = [arg for arg in case.argvs[0] if arg != "--dry-run"] - return [*case.path, *args, "--yes", "--output", "json"] - - -@pytest.mark.parametrize("case", WRITE_CASES, ids=lambda case: " ".join(case.path)) -def test_every_write_is_refused_on_cloud(case: Case) -> None: - """The gate stops the write, names the backend, and sends nothing.""" - result = CliRunner().invoke(cli, _argv(case)) - - assert result.exit_code == UnsupportedBackendError.exit_code, ( - f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" - ) - payload = json.loads(result.output) - assert set(payload) == {"error"} - assert payload["error"]["code"] == UnsupportedBackendError.code - assert "Splunk Cloud" in payload["error"]["message"] diff --git a/tests/unit/test_write_gating.py b/tests/unit/test_write_gating.py new file mode 100644 index 0000000..f0f0358 --- /dev/null +++ b/tests/unit/test_write_gating.py @@ -0,0 +1,241 @@ +"""Writes are refused by default, on every backend, until explicitly opted in. + +Two independent env-only gates (never a CLI flag, so a saved command line can +never turn one on by itself): + +- `SPLUNK_ENABLE_WRITES=true` -- required for any real (non-dry-run) mutation, + on every backend. `--dry-run` sends nothing and needs no opt-in. +- `SPLUNK_CLOUD_WRITE=true` -- required *additionally* on a Cloud target, and + only unlocks create/update/delete for `index`, `role`, and `hec-token` (the + three ACS mutation routes). This one is enforced even for a `--dry-run` + preview, since it also proves the object is one of those three -- see + `vct_splunk.commands.write.refuse_cloud_write`. + +The property that makes each refusal safe is not "writes usually fail" -- it is +that the write stops at the gate, with a typed error, having sent nothing. A +refusal that happened only because a credential was missing, or one that fired +after a request went out, would both look like success in a weaker test. So +every refusal case here installs a transport that fails the test if any real +request leaves the process. + +`tests/unit/test_acs_loopback.py` covers the Cloud read contract; this and +that both run credential-free on every change, needing no live target. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest +from click.testing import CliRunner + +from cli_catalog import CATALOG, Case +from vct_splunk.api.acs import operations as acs +from vct_splunk.api.acs.client import AcsClient as RealAcsClient +from vct_splunk.cli import cli +from vct_splunk.commands.dispatch import has_cloud_write +from vct_splunk.utils.errors import UnsupportedBackendError, UsageError + +WRITE_CASES = tuple(case for case in CATALOG if case.kind == "write") +CLOUD_WRITABLE_CASES = tuple( + case for case in WRITE_CASES if has_cloud_write(case.path[0], case.path[-1]) +) +CLOUD_UNWRITABLE_CASES = tuple( + case for case in WRITE_CASES if not has_cloud_write(case.path[0], case.path[-1]) +) +#: `saved-search run` (no `--trigger-actions`) dispatches a job -- it changes no +#: configuration, so it deliberately does not go through `do_write` at all (see +#: its docstring). It is cataloged as "write" for other purposes, but does not +#: belong in a suite proving `SPLUNK_ENABLE_WRITES` refuses every gated write. +_GATED_WRITE_CASES = tuple(case for case in WRITE_CASES if case.path != ("saved-search", "run")) + +#: CLI resource name -> the ACS collection path it writes to. +_ACS_PATH = {"index": acs.INDEXES, "role": acs.ROLES, "hec-token": acs.HEC_TOKENS} + + +def _argv(case: Case, *extra: str) -> list[str]: + """Build one invocation of *case*, replacing any `--dry-run` with *extra*.""" + args = [arg for arg in case.argvs[0] if arg != "--dry-run"] + return [*case.path, *args, *extra, "--output", "json"] + + +def _refuse_network(monkeypatch: pytest.MonkeyPatch) -> None: + def refuse(*args: object, **kwargs: object) -> object: + raise AssertionError("a refused write reached the network") + + monkeypatch.setattr(httpx.HTTPTransport, "handle_request", refuse) + + +# --- Enterprise: SPLUNK_ENABLE_WRITES gates every real write ------------------ + + +@pytest.fixture +def enterprise_env_no_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: + """Point the CLI at Enterprise with the top-level gate off; refuse the network. + + A namespaced write needs an explicit app regardless of this gate (that + check runs before `do_write`), so `SPLUNK_APP` is set, not cleared, here -- + this suite is testing the write-enable gate, not namespace resolution. + """ + for name in ("SPLUNK_OWNER", "SPLUNK_PROFILE", "SPLUNK_ENABLE_WRITES"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("SPLUNK_URL", "https://sh.corp:8089") + monkeypatch.setenv("SPLUNK_TOKEN", "T") + monkeypatch.setenv("SPLUNK_APP", "my_app") + _refuse_network(monkeypatch) + + +@pytest.mark.parametrize("case", _GATED_WRITE_CASES, ids=lambda case: " ".join(case.path)) +def test_every_write_is_refused_without_enable_writes( + case: Case, enterprise_env_no_opt_in: None +) -> None: + """Every write leaf refuses before touching the network when the gate is off.""" + result = CliRunner().invoke(cli, _argv(case, "--yes")) + + assert result.exit_code == UsageError.exit_code, ( + f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" + ) + payload = json.loads(result.output) + assert set(payload) == {"error"} + assert payload["error"]["code"] == UsageError.code + assert "SPLUNK_ENABLE_WRITES" in payload["error"]["message"] + + +def test_enterprise_dry_run_needs_no_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: + """A preview sends nothing, so it works with SPLUNK_ENABLE_WRITES unset.""" + monkeypatch.delenv("SPLUNK_ENABLE_WRITES", raising=False) + monkeypatch.setenv("SPLUNK_URL", "https://sh.corp:8089") + monkeypatch.setenv("SPLUNK_TOKEN", "T") + + result = CliRunner().invoke( + cli, ["index", "create", "example", "--dry-run", "--output", "json"] + ) + + assert result.exit_code == 0, result.output + assert json.loads(result.output)["data"]["dry_run"] is True + + +# --- Cloud: SPLUNK_CLOUD_WRITE additionally gates the three ACS routes -------- + + +@pytest.fixture +def cloud_env_no_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: + """Point the CLI at a Cloud stack with neither opt-in set; refuse the network.""" + for name in ( + "SPLUNK_APP", + "SPLUNK_OWNER", + "SPLUNK_PROFILE", + "SPLUNK_ACS_BASE_URL", + "SPLUNK_CLOUD_WRITE", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") + monkeypatch.setenv("SPLUNK_ACS_TOKEN", "unused") + monkeypatch.setenv("SPLUNK_TOKEN", "unused") + _refuse_network(monkeypatch) + + +@pytest.mark.parametrize("case", WRITE_CASES, ids=lambda case: " ".join(case.path)) +def test_every_write_is_refused_on_cloud_by_default(case: Case, cloud_env_no_opt_in: None) -> None: + """The gate stops the write, names the backend, and sends nothing.""" + result = CliRunner().invoke(cli, _argv(case, "--yes")) + + assert result.exit_code == UnsupportedBackendError.exit_code, ( + f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" + ) + payload = json.loads(result.output) + assert set(payload) == {"error"} + assert payload["error"]["code"] == UnsupportedBackendError.code + assert "Splunk Cloud" in payload["error"]["message"] + + +@pytest.fixture +def cloud_write_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Point the CLI at a Cloud stack with the write opt-in set.""" + for name in ("SPLUNK_APP", "SPLUNK_OWNER", "SPLUNK_PROFILE", "SPLUNK_ACS_BASE_URL"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") + monkeypatch.setenv("SPLUNK_ACS_TOKEN", "T") + monkeypatch.setenv("SPLUNK_CLOUD_WRITE", "true") + + +def _patch_acs_write(monkeypatch: pytest.MonkeyPatch, handler) -> None: + """Back the ACS write client `do_write` opens with an `httpx.MockTransport`.""" + + def _make(config): + return RealAcsClient(config, transport=httpx.MockTransport(handler)) + + monkeypatch.setattr("vct_splunk.commands.write.AcsClient", _make) + + +@pytest.mark.parametrize("case", CLOUD_UNWRITABLE_CASES, ids=lambda case: " ".join(case.path)) +def test_non_acs_writes_still_refused_when_opted_in( + case: Case, cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """The opt-in unlocks only the nine ACS routes -- everything else still refuses.""" + _refuse_network(monkeypatch) + result = CliRunner().invoke(cli, _argv(case, "--yes")) + + assert result.exit_code == UnsupportedBackendError.exit_code, ( + f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" + ) + payload = json.loads(result.output) + assert set(payload) == {"error"} + assert payload["error"]["code"] == UnsupportedBackendError.code + + +@pytest.mark.parametrize("case", CLOUD_WRITABLE_CASES, ids=lambda case: " ".join(case.path)) +def test_acs_write_dry_run_sends_nothing( + case: Case, cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """--dry-run previews an ACS write and sends no request, even when opted in.""" + _refuse_network(monkeypatch) + result = CliRunner().invoke(cli, _argv(case, "--dry-run")) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["data"]["dry_run"] is True + assert payload["data"]["request"]["method"] in {"POST", "PATCH", "DELETE"} + + +@pytest.mark.parametrize("case", CLOUD_WRITABLE_CASES, ids=lambda case: " ".join(case.path)) +def test_acs_write_succeeds_when_opted_in_and_confirmed( + case: Case, cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """With the opt-in and --yes, a mocked ACS create/update/delete succeeds.""" + seen: dict[str, str] = {} + + def handler(req: httpx.Request) -> httpx.Response: + seen["method"] = req.method + seen["path"] = req.url.path + return httpx.Response(202, json={"name": "example"}) + + _patch_acs_write(monkeypatch, handler) + result = CliRunner().invoke(cli, _argv(case, "--yes")) + + assert result.exit_code == 0, result.output + resource, verb = case.path + expected_method = {"create": "POST", "update": "PATCH", "delete": "DELETE"}[verb] + assert seen["method"] == expected_method + assert seen["path"].startswith(f"/acme/adminconfig/v2/{_ACS_PATH[resource]}") + + +def test_hec_token_create_redacts_token_in_output( + cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """ACS mints the token on create; Cloud CI output must never show it.""" + + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response( + 202, json={"http-event-collector": {"name": "example", "token": "SECRET-DO-NOT-LEAK"}} + ) + + _patch_acs_write(monkeypatch, handler) + result = CliRunner().invoke( + cli, ["hec-token", "create", "example", "--yes", "--output", "json"] + ) + + assert result.exit_code == 0, result.output + assert "SECRET-DO-NOT-LEAK" not in result.output + assert "" in result.output diff --git a/tests/write_test_cli.py b/tests/write_test_cli.py new file mode 100644 index 0000000..599621b --- /dev/null +++ b/tests/write_test_cli.py @@ -0,0 +1,65 @@ +"""Shared CLI harness for destructive integration suites (Enterprise + Cloud). + +Both suites need the same three operations: invoke the CLI and assert on its +JSON envelope, invoke it as a real write (``--yes``), and register a +reverse-cleanup command that runs (in reverse order) when the test finishes, +failing loudly if any of those cleanups itself errors. Only the *opt-in* gate +differs per suite (``SPLUNK_WRITE_TEST`` vs ``SPLUNK_CLOUD_WRITE``), so each +keeps its own autouse fixture in its own ``conftest.py``; this module holds +only what is identical between them. + +A flat top-level module rather than a nested ``tests/integration/conftest.py``: +this test suite has no ``__init__.py`` packages (see ``pythonpath = ["tests"]`` +in ``pyproject.toml``), so a shared helper is imported the same way +``cli_catalog.py`` already is -- ``from write_test_cli import WriteTestCli``. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +from click.testing import CliRunner, Result + +from vct_splunk.cli import cli + + +@dataclass +class WriteTestCli: + """Invoke the public CLI and fail explicitly if reverse cleanup leaks state.""" + + runner: CliRunner = field(default_factory=CliRunner) + cleanups: list[tuple[str, Callable[[], Result]]] = field(default_factory=list) + + def run(self, *argv: str, exit_codes: tuple[int, ...] = (0,)) -> Any: + result = self.runner.invoke(cli, [*argv, "--output", "json"]) + assert result.exit_code in exit_codes, ( + f"{' '.join(argv)} exited {result.exit_code}\n{result.output}" + ) + payload = json.loads(result.stdout) + if result.exit_code == 0: + assert set(payload) == {"data", "meta"} + return payload["data"] + return payload["error"] + + def write(self, *argv: str) -> dict[str, Any]: + return self.run(*argv, "--yes") + + def cleanup(self, label: str, *argv: str) -> None: + self.cleanups.append( + (label, lambda: self.runner.invoke(cli, [*argv, "--yes", "--output", "json"])) + ) + + def drop_cleanup(self, label: str) -> None: + """Remove a previously registered cleanup, e.g. once a test's own delete succeeds.""" + self.cleanups = [c for c in self.cleanups if c[0] != label] + + def finish(self) -> None: + failures: list[str] = [] + for label, cleanup in reversed(self.cleanups): + result = cleanup() + if result.exit_code != 0: + failures.append(f"{label}: exit {result.exit_code}: {result.output}") + assert not failures, "cleanup failures:\n" + "\n".join(failures)