From f09b90460ee993a93cf4eadfcc0e2bfcb7924055 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Tue, 18 Aug 2026 07:40:52 -0400 Subject: [PATCH 1/7] feat: gate Splunk Cloud writes behind two explicit opt-ins Adds create/update/delete for index, role, and hec-token via Splunk Cloud ACS. Every real write, on any backend, now requires SPLUNK_ENABLE_WRITES=true (never a CLI flag, so a saved command line cannot enable one); a Cloud write additionally requires SPLUNK_CLOUD_WRITE=true, checked even for a --dry-run preview since it also proves the object is one of the three ACS-writable resources. --dry-run itself needs neither variable and sends nothing. Adds a Splunk Cloud Write Canary GitHub Actions workflow: workflow_dispatch only, a typed confirm=WRITE input, a commit-subject check, and a protected splunk-cloud-write environment with required reviewers gating the destructive half. It runs the Cloud read canary first, so one approved dispatch is the full read-then-write certification. --- .env.example | 27 +- .github/scripts/detect-cloud-stack.sh | 2 +- .github/workflows/ci.yml | 4 + .github/workflows/cloud-write.yml | 159 ++++++++++++ AGENTS.md | 22 +- CHANGELOG.md | 16 ++ README.md | 40 ++- docs/architecture.md | 22 +- src/vct_splunk/api/acs/__init__.py | 7 +- src/vct_splunk/api/acs/client.py | 81 +++++- src/vct_splunk/api/acs/operations.py | 65 ++++- src/vct_splunk/commands/command_factory.py | 44 ++++ src/vct_splunk/commands/dispatch.py | 23 ++ src/vct_splunk/commands/write.py | 136 ++++++++-- src/vct_splunk/utils/backends.py | 24 +- tests/TESTING.md | 64 ++++- tests/conftest.py | 14 + tests/integration/cloud/write/conftest.py | 79 ++++++ .../cloud/write/test_write_catalog.py | 79 ++++++ .../integration/enterprise/write/conftest.py | 56 +--- tests/integration/test_acs_public_spec.py | 12 +- tests/unit/test_cloud_write_refusal.py | 71 ------ tests/unit/test_write_gating.py | 241 ++++++++++++++++++ tests/write_test_cli.py | 65 +++++ 24 files changed, 1134 insertions(+), 219 deletions(-) create mode 100644 .github/workflows/cloud-write.yml create mode 100644 tests/integration/cloud/write/conftest.py create mode 100644 tests/integration/cloud/write/test_write_catalog.py delete mode 100644 tests/unit/test_cloud_write_refusal.py create mode 100644 tests/unit/test_write_gating.py create mode 100644 tests/write_test_cli.py 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) From 7068e7d1074e018fb9e650fc724144f129bc8abd Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:25:46 -0400 Subject: [PATCH 2/7] ci: gate Splunk Cloud canaries on complete credential sets Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- .github/scripts/detect-cloud-stack.sh | 43 ------------------ .github/workflows/cloud-read.yml | 29 +++++++----- .github/workflows/cloud-write.yml | 63 +++++++++++++-------------- 3 files changed, 48 insertions(+), 87 deletions(-) delete mode 100644 .github/scripts/detect-cloud-stack.sh diff --git a/.github/scripts/detect-cloud-stack.sh b/.github/scripts/detect-cloud-stack.sh deleted file mode 100644 index fcd32f5..0000000 --- a/.github/scripts/detect-cloud-stack.sh +++ /dev/null @@ -1,43 +0,0 @@ -#!/usr/bin/env bash -# Decide what the Splunk Cloud read canary can certify with the secrets present. -# Writes `ready` and `full` to $GITHUB_OUTPUT. -# -# Two credentials matter, and they certify different things: -# SPLUNK_URL + SPLUNK_ACS_TOKEN the reads Cloud serves through ACS -# SPLUNK_TOKEN the reads it does not, which need a splunkd -# credential to reach the dispatch layer at all -# -# Without the second, most of the catalogue stops at the credential check and -# the run looks greener than it is. That case still runs -- the ACS reads are -# worth certifying -- but it says plainly what it did not cover. -# -# Requires: GITHUB_OUTPUT (Actions). Reads SPLUNK_URL, SPLUNK_ACS_TOKEN, SPLUNK_TOKEN. -set -euo pipefail - -: "${GITHUB_OUTPUT:?GITHUB_OUTPUT must be set by GitHub Actions}" - -note() { - echo "$1" - echo "$1" >>"${GITHUB_STEP_SUMMARY:-/dev/null}" -} - -if [ -z "${SPLUNK_URL:-}" ] || [ -z "${SPLUNK_ACS_TOKEN:-}" ]; then - echo "ready=false" >>"$GITHUB_OUTPUT" - 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_write_gating.py." - exit 0 -fi - -echo "ready=true" >>"$GITHUB_OUTPUT" - -if [ -n "${SPLUNK_TOKEN:-}" ]; then - echo "full=true" >>"$GITHUB_OUTPUT" - note "Certifying every catalogued read against the configured Splunk Cloud stack." -else - echo "full=false" >>"$GITHUB_OUTPUT" - note "PARTIAL RUN: no SPLUNK_TOKEN, so only the ACS-served reads are certified." - note "Every other read stops at the splunkd credential check instead of reaching" - note "the dispatch layer. Add SPLUNK_TOKEN to certify the full catalogue." -fi diff --git a/.github/workflows/cloud-read.yml b/.github/workflows/cloud-read.yml index efea0f3..c15f951 100644 --- a/.github/workflows/cloud-read.yml +++ b/.github/workflows/cloud-read.yml @@ -1,7 +1,7 @@ # Certify the Cloud read path against a real Splunk Cloud stack. # # The Cloud contract itself is proved on every pull request, credential-free, by -# tests/unit/test_acs_loopback.py and tests/unit/test_cloud_write_refusal.py. +# tests/unit/test_acs_loopback.py and tests/unit/test_write_gating.py. # What only a live stack can show is that a real one answers the way the ACS # specification says it does. Until a stack is configured, this workflow ends as # an explicit no-op rather than a green run that checked nothing. @@ -33,6 +33,8 @@ jobs: # Untrusted output boundary: hide the Cloud stack name in prompts, JSON # and table output, and error messages and details. VCT_SPLUNK_REDACT_TARGET: "1" + CLOUD_CONFIGURED: ${{ secrets.SPLUNK_URL != '' || secrets.SPLUNK_ACS_TOKEN != '' || secrets.SPLUNK_TOKEN != '' }} + CLOUD_READY: ${{ secrets.SPLUNK_URL != '' && secrets.SPLUNK_ACS_TOKEN != '' && secrets.SPLUNK_TOKEN != '' }} PYTEST_ADDOPTS: >- -q --tb=line -r N -o 'addopts=--strict-markers --import-mode=importlib' steps: @@ -40,45 +42,48 @@ jobs: with: persist-credentials: false - # The secrets context is not available to a job-level `if`, so this is a - # step that reports what can be certified rather than a skipped job. - - name: Check for a configured Cloud stack - id: stack - run: bash .github/scripts/detect-cloud-stack.sh + - name: Require complete Cloud read credentials when configured + if: env.CLOUD_READY != 'true' + run: | + if [ "$CLOUD_CONFIGURED" = "true" ]; then + echo "A partial Cloud read credential set is configured; SPLUNK_URL, SPLUNK_ACS_TOKEN, and SPLUNK_TOKEN are all required." + exit 1 + fi + echo "::notice::No Cloud read credentials are configured; no stack was certified." - - if: steps.stack.outputs.ready == 'true' + - if: env.CLOUD_READY == 'true' uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 with: python-version: "3.14" cache: pip - name: Install project - if: steps.stack.outputs.ready == 'true' + if: env.CLOUD_READY == 'true' 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) - if: steps.stack.outputs.ready == 'true' + if: env.CLOUD_READY == 'true' run: >- .venv/bin/pytest tests/integration/cloud/read/test_catalog.py -m "integration and cloud and read" --junitxml=cloud-read.xml - name: Cloud ACS operations (below the CLI) - if: steps.stack.outputs.ready == 'true' + if: env.CLOUD_READY == 'true' run: >- .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py -m "integration and cloud and read" --junitxml=cloud-acs.xml - name: Publish test summary - if: always() && steps.stack.outputs.ready == 'true' + if: always() && env.CLOUD_READY == 'true' uses: test-summary/action@37b508cfee6d4d080eedd00b5bb240a6a784a6a5 # v2 with: paths: cloud-*.xml - name: Upload test report - if: always() && steps.stack.outputs.ready == 'true' + if: always() && env.CLOUD_READY == 'true' uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: name: cloud-read-${{ github.run_id }} diff --git a/.github/workflows/cloud-write.yml b/.github/workflows/cloud-write.yml index cefdffd..b598c75 100644 --- a/.github/workflows/cloud-write.yml +++ b/.github/workflows/cloud-write.yml @@ -37,6 +37,10 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 2 steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - name: Fail closed unless confirm is WRITE env: CONFIRM: ${{ github.event.inputs.confirm }} @@ -46,6 +50,17 @@ jobs: exit 1 fi + - 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 + read: name: Cloud / ACS reads (pre-write readiness) needs: confirm @@ -59,20 +74,18 @@ jobs: 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 + CLOUD_READY: ${{ secrets.SPLUNK_URL != '' && secrets.SPLUNK_ACS_TOKEN != '' && secrets.SPLUNK_TOKEN != '' }} + PYTEST_ADDOPTS: >- + -q --tb=line -r N -o 'addopts=--strict-markers --import-mode=importlib' 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' + if: env.CLOUD_READY != 'true' run: | - echo "SPLUNK_URL and SPLUNK_ACS_TOKEN are required to certify Cloud reads" + echo "SPLUNK_URL, SPLUNK_ACS_TOKEN, and SPLUNK_TOKEN are required to certify Cloud reads" exit 1 - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 @@ -88,13 +101,13 @@ jobs: - 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" + .venv/bin/pytest tests/integration/cloud/read/test_catalog.py + -m "integration and cloud and read" --junitxml=cloud-read.xml - 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" + .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py + -m "integration and cloud and read" --junitxml=cloud-acs.xml write: name: Cloud / ACS writes @@ -102,31 +115,17 @@ jobs: environment: splunk-cloud-write runs-on: ubuntu-latest timeout-minutes: 15 + env: + CLOUD_READY: ${{ secrets.SPLUNK_URL != '' && secrets.SPLUNK_ACS_WRITE_TOKEN != '' }} + PYTEST_ADDOPTS: >- + -q --tb=line -r N -o 'addopts=--strict-markers --import-mode=importlib' 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' + if: env.CLOUD_READY != 'true' run: | echo "SPLUNK_URL and SPLUNK_ACS_WRITE_TOKEN are required for a WRITE run" exit 1 @@ -155,5 +154,5 @@ jobs: 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" + .venv/bin/pytest tests/integration/cloud/write + -m "integration and cloud and write" --junitxml=cloud-write.xml From a41d347a1ebc5d4fb60200be21377e0018e6d166 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:25:48 -0400 Subject: [PATCH 3/7] fix: distinguish Cloud opt-in refusal and record ACS audit target Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- src/vct_splunk/api/acs/operations.py | 2 +- src/vct_splunk/commands/dispatch.py | 13 +++++-------- src/vct_splunk/commands/write.py | 28 ++++++++++++++++++++-------- src/vct_splunk/config/loader.py | 8 ++++---- 4 files changed, 30 insertions(+), 21 deletions(-) diff --git a/src/vct_splunk/api/acs/operations.py b/src/vct_splunk/api/acs/operations.py index 9da971d..53f918c 100644 --- a/src/vct_splunk/api/acs/operations.py +++ b/src/vct_splunk/api/acs/operations.py @@ -103,7 +103,7 @@ def cloud_write( ``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 + (:func:`vct_splunk.commands.write.refuse_cloud_write`) has already checked both against the same allowlist this reads, so a `KeyError` here would mean that check was bypassed. diff --git a/src/vct_splunk/commands/dispatch.py b/src/vct_splunk/commands/dispatch.py index b1a4623..97e76b2 100644 --- a/src/vct_splunk/commands/dispatch.py +++ b/src/vct_splunk/commands/dispatch.py @@ -57,14 +57,11 @@ def dispatch_list(ctx: Any, resource: str, rest_call: Callable[[Any], Any]) -> A return rest_call(c) -def has_cloud_write(resource: str, verb: str | None = None) -> bool: +def has_cloud_write(resource: str, verb: str) -> 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. + Only create/update/delete for index, role, and hec-token are supported. + Every other Cloud mutation, including enable/disable on those same three + resources, has no ACS route. """ - if resource not in acs.WRITABLE: - return False - return verb is None or verb in _WRITE_VERBS + return resource in acs.WRITABLE and verb in _WRITE_VERBS diff --git a/src/vct_splunk/commands/write.py b/src/vct_splunk/commands/write.py index 3f98312..84e2eba 100644 --- a/src/vct_splunk/commands/write.py +++ b/src/vct_splunk/commands/write.py @@ -17,9 +17,11 @@ from collections.abc import Callable from typing import Any +import click + 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 ..config.loader import load_config, resolve_target from ..output import formatter as out from ..utils import audit from ..utils.backends import cloud_stack_from_url @@ -91,13 +93,15 @@ def do_write( # 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 + 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, safe_target(target)) + audit_target = safe_target(target) if backend == "cloud": stack = cloud_stack_from_url(ctx.base_url) config = acs_config_from_env(stack, write=True) + audit_target = safe_target(f"{config.base_url}/{config.stack}") config.dry_run = ctx.dry_run with AcsClient(config) as c: result = cloud_write(c, resource, verb, name or "", body) @@ -105,7 +109,14 @@ def do_write( with ctx.client() as c: result = run(c) if not (isinstance(result, dict) and result.get("dry_run")): - audit.record({**audit_event, "target": safe_target(target)}) + try: + audit.record({**audit_event, "target": audit_target}) + except OSError: + click.echo( + "Warning: the write succeeded, but its audit record could not be saved; " + "check VCT_SPLUNK_AUDIT and the default audit path.", + err=True, + ) return result @@ -122,10 +133,8 @@ def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None: 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. + skips, since it sends nothing regardless), this one is also required for + a Cloud preview, 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 @@ -137,7 +146,10 @@ def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None: 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") + raise UsageError( + f"Splunk Cloud writes are disabled by default. Set {CLOUD_WRITE_ENV}=true " + f"to allow `{resource} {verb}` (there is no CLI flag for this)." + ) def _require_writes_enabled(resource: str, verb: str) -> None: diff --git a/src/vct_splunk/config/loader.py b/src/vct_splunk/config/loader.py index 12771f1..321eba1 100644 --- a/src/vct_splunk/config/loader.py +++ b/src/vct_splunk/config/loader.py @@ -124,7 +124,7 @@ def load_config(base_url: str | None = None, *, profile: str | None = None) -> S Raises: UsageError: If no URL or no credential source can be resolved. """ - status, prof, verify = _resolve_target(base_url, profile) + status, prof, verify = resolve_target(base_url, profile) env_token = os.environ.get("SPLUNK_TOKEN") env_session_key = os.environ.get("SPLUNK_SESSION_KEY") token = env_token or prof.get("token") @@ -156,7 +156,7 @@ def load_config(base_url: str | None = None, *, profile: str | None = None) -> S def auth_status(base_url: str | None = None, *, profile: str | None = None) -> AuthStatus: """Resolve the active auth scheme without exchanging username/password.""" - status, prof, _verify = _resolve_target(base_url, profile) + status, prof, _verify = resolve_target(base_url, profile) env_token = os.environ.get("SPLUNK_TOKEN") env_session_key = os.environ.get("SPLUNK_SESSION_KEY") if env_token or prof.get("token"): @@ -174,10 +174,10 @@ def auth_status(base_url: str | None = None, *, profile: str | None = None) -> A return AuthStatus(status.base_url, scheme) -def _resolve_target( +def resolve_target( base_url: str | None, profile: str | None ) -> tuple[AuthStatus, dict[str, str], bool | str]: - """Resolve shared URL, profile, and TLS inputs without authenticating.""" + """Resolve the shared URL, profile, and TLS inputs without authenticating.""" prof = load_profile(profile) url = base_url or os.environ.get("SPLUNK_URL") or prof.get("url") if not url: From 6aeca49c2e3fd631da4fdf970f497c408dfff5fc Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:25:51 -0400 Subject: [PATCH 4/7] test: cover Cloud gate combinations and ACS field mapping Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- tests/conftest.py | 8 +- tests/integration/test_acs_public_spec.py | 10 ++ tests/unit/test_acs.py | 2 +- tests/unit/test_cli_matrix.py | 11 +- tests/unit/test_write.py | 52 ++++++++ tests/unit/test_write_gating.py | 145 ++++++++++++++++++++-- tests/write_test_cli.py | 6 +- 7 files changed, 214 insertions(+), 20 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 168b1f4..639fa3e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -20,12 +20,16 @@ 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 + ``test_write_gating.py``'s ``enterprise_env_no_opt_in`` fixture for the pattern of clearing an ambient opt-in before asserting on its absence). """ + previous = os.environ.get("SPLUNK_ENABLE_WRITES") os.environ["SPLUNK_ENABLE_WRITES"] = "true" yield - os.environ.pop("SPLUNK_ENABLE_WRITES", None) + if previous is None: + os.environ.pop("SPLUNK_ENABLE_WRITES", None) + else: + os.environ["SPLUNK_ENABLE_WRITES"] = previous def cli_runner() -> CliRunner: diff --git a/tests/integration/test_acs_public_spec.py b/tests/integration/test_acs_public_spec.py index ee63638..8c0044d 100644 --- a/tests/integration/test_acs_public_spec.py +++ b/tests/integration/test_acs_public_spec.py @@ -8,6 +8,7 @@ import pytest from vct_splunk.api.acs.operations import LIST_ENVELOPES, WRITE_PATHS +from vct_splunk.commands.command_factory import _ACS_FIELD_KEYS pytestmark = [ pytest.mark.integration, @@ -41,3 +42,12 @@ def test_implemented_acs_write_contract_matches_public_spec(): 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" + + index_create = public["paths"]["/{stack}/adminconfig/v2/indexes"]["post"] + schema = index_create["requestBody"]["content"]["application/json"]["schema"] + if "$ref" in schema: + component = schema["$ref"].rsplit("/", 1)[-1] + schema = public["components"]["schemas"][component] + assert set(_ACS_FIELD_KEYS["index"].values()) <= set(schema["properties"]), ( + "ACS index field mappings are missing from the public create schema" + ) diff --git a/tests/unit/test_acs.py b/tests/unit/test_acs.py index 3b7dda5..2b8f25e 100644 --- a/tests/unit/test_acs.py +++ b/tests/unit/test_acs.py @@ -284,7 +284,7 @@ def handler(req: httpx.Request) -> httpx.Response: # --- Unsupported operations stop cleanly ------------------------------------- # -# Cloud write refusal is proved for every mutation in test_cloud_write_refusal.py; +# Cloud write refusal is proved for every mutation in test_write_gating.py; # spot checks here would only restate a subset of it. diff --git a/tests/unit/test_cli_matrix.py b/tests/unit/test_cli_matrix.py index 9ef4bc6..a07fe98 100644 --- a/tests/unit/test_cli_matrix.py +++ b/tests/unit/test_cli_matrix.py @@ -11,6 +11,7 @@ from cli_catalog import CATALOG, help_invocations, iter_leaves from vct_splunk.cli import cli +from vct_splunk.commands.dispatch import has_cloud_write def _handler(req: httpx.Request) -> httpx.Response: @@ -118,5 +119,11 @@ def unexpected_client(*args, **kwargs): monkeypatch.setattr("vct_splunk.commands.context.Ctx.acs_client", unexpected_client) for args in case.argvs: result = CliRunner().invoke(cli, [*case.path, *args, "--output", "json"]) - assert result.exit_code == 4, f"{' '.join(case.path)}: {result.output}" - assert "unsupported_backend" in result.output + payload = json.loads(result.output) + if has_cloud_write(case.path[0], case.path[-1]): + assert result.exit_code == 2, f"{' '.join(case.path)}: {result.output}" + assert payload["error"]["code"] == "usage_error" + assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] + else: + assert result.exit_code == 4, f"{' '.join(case.path)}: {result.output}" + assert payload["error"]["code"] == "unsupported_backend" diff --git a/tests/unit/test_write.py b/tests/unit/test_write.py index 6ead1ee..831122f 100644 --- a/tests/unit/test_write.py +++ b/tests/unit/test_write.py @@ -6,6 +6,9 @@ from __future__ import annotations +import json +from contextlib import nullcontext + import httpx import pytest @@ -22,6 +25,7 @@ def __init__(self, *, dry_run: bool = False, yes: bool = False) -> None: self.base_url = "https://splunk.test:8089" self.dry_run = dry_run self.yes = yes + self.backend = "enterprise" def client(self) -> SplunkClient: cfg = SplunkConfig(base_url=self.base_url, token="T", dry_run=self.dry_run) @@ -60,6 +64,54 @@ def test_do_write_records_audit_on_real_write(monkeypatch, tmp_path): assert "https://splunk.test:8089" in contents # target is recorded +def test_cloud_write_audit_records_the_acs_target(monkeypatch, tmp_path): + monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") + monkeypatch.setenv("SPLUNK_ACS_TOKEN", "T") + monkeypatch.setenv("SPLUNK_CLOUD_WRITE", "true") + for name in ("SPLUNK_ACS_STACK", "SPLUNK_ACS_BASE_URL", "SPLUNK_ACS_WRITE_TOKEN"): + monkeypatch.delenv(name, raising=False) + audit_file = tmp_path / "audit.log" + monkeypatch.setenv("VCT_SPLUNK_AUDIT", str(audit_file)) + ctx = _Ctx(yes=True) + ctx.base_url = "https://acme.splunkcloud.com" + ctx.backend = "cloud" + monkeypatch.setattr("vct_splunk.commands.write.AcsClient", lambda config: nullcontext(object())) + monkeypatch.setattr("vct_splunk.commands.write.cloud_write", lambda *args: {"name": "web"}) + + result = do_write( + ctx, + action="create index 'web'", + audit_event={"action": "index.create", "name": "web"}, + run=lambda client: {}, + name="web", + ) + + assert result == {"name": "web"} + event = json.loads(audit_file.read_text()) + assert event["target"] == "https://admin.splunk.com/acme" + + +def test_audit_failure_warns_after_success(monkeypatch, capsys): + monkeypatch.setenv("SPLUNK_TOKEN", "T") + monkeypatch.setattr( + "vct_splunk.commands.write.audit.record", + lambda event: (_ for _ in ()).throw(OSError("permission denied")), + ) + + result = do_write( + _Ctx(yes=True), + action="do thing", + audit_event={"action": "index.create"}, + run=lambda client: {"name": "web"}, + ) + + assert result == {"name": "web"} + captured = capsys.readouterr() + assert captured.out == "" + assert "write succeeded" in captured.err + assert "audit record could not be saved" in captured.err + + def test_do_write_redacts_credentials_from_confirmation_and_audit(monkeypatch, tmp_path): monkeypatch.setenv("SPLUNK_TOKEN", "T") audit_file = tmp_path / "audit.log" diff --git a/tests/unit/test_write_gating.py b/tests/unit/test_write_gating.py index f0f0358..e64b055 100644 --- a/tests/unit/test_write_gating.py +++ b/tests/unit/test_write_gating.py @@ -31,7 +31,7 @@ from click.testing import CliRunner from cli_catalog import CATALOG, Case -from vct_splunk.api.acs import operations as acs +from conftest import cli_runner 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 @@ -50,8 +50,13 @@ #: 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} +#: CLI resource name -> literal ACS collection path. Keep independent of the +#: implementation so an accidental route change cannot update both sides. +_ACS_PATH = { + "index": "indexes", + "role": "roles", + "hec-token": "inputs/http-event-collectors", +} def _argv(case: Case, *extra: str) -> list[str]: @@ -67,6 +72,11 @@ def refuse(*args: object, **kwargs: object) -> object: monkeypatch.setattr(httpx.HTTPTransport, "handle_request", refuse) +def test_cloud_write_catalog_has_all_nine_mutations() -> None: + """The Cloud success matrices must not silently disappear when routes regress.""" + assert len(CLOUD_WRITABLE_CASES) == 9 + + # --- Enterprise: SPLUNK_ENABLE_WRITES gates every real write ------------------ @@ -91,17 +101,33 @@ 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")) + result = cli_runner().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 result.stdout == "" + payload = json.loads(result.stderr) assert set(payload) == {"error"} assert payload["error"]["code"] == UsageError.code assert "SPLUNK_ENABLE_WRITES" in payload["error"]["message"] +@pytest.mark.parametrize("value", ("", "TRUE", "1", "yes", "false")) +def test_enable_writes_requires_literal_true( + value: str, enterprise_env_no_opt_in: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Enterprise writes reject every opt-in spelling except exact `true`.""" + monkeypatch.setenv("SPLUNK_ENABLE_WRITES", value) + result = cli_runner().invoke(cli, ["index", "create", "example", "--yes", "--output", "json"]) + + assert result.exit_code == UsageError.exit_code + assert result.stdout == "" + payload = json.loads(result.stderr) + 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) @@ -139,15 +165,62 @@ def cloud_env_no_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: @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")) + result = cli_runner().invoke(cli, _argv(case, "--yes")) - assert result.exit_code == UnsupportedBackendError.exit_code, ( + expected = ( + UnsupportedBackendError if not has_cloud_write(case.path[0], case.path[-1]) else UsageError + ) + + assert result.exit_code == expected.exit_code, ( f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" ) - payload = json.loads(result.output) + assert result.stdout == "" + payload = json.loads(result.stderr) assert set(payload) == {"error"} - assert payload["error"]["code"] == UnsupportedBackendError.code - assert "Splunk Cloud" in payload["error"]["message"] + assert payload["error"]["code"] == expected.code + if expected is UsageError: + assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] + + +@pytest.mark.parametrize("value", ("", "TRUE", "1", "yes", "false")) +def test_cloud_write_requires_literal_true( + value: str, cloud_env_no_opt_in: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Cloud writes reject every opt-in spelling except exact `true`.""" + monkeypatch.setenv("SPLUNK_CLOUD_WRITE", value) + result = cli_runner().invoke(cli, ["index", "create", "example", "--yes", "--output", "json"]) + + assert result.exit_code == UsageError.exit_code + assert result.stdout == "" + payload = json.loads(result.stderr) + assert payload["error"]["code"] == UsageError.code + assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] + + +def test_cloud_opt_in_does_not_replace_enable_writes_gate( + cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """The Cloud opt-in still requires the shared real-write gate.""" + monkeypatch.delenv("SPLUNK_ENABLE_WRITES", raising=False) + _refuse_network(monkeypatch) + result = cli_runner().invoke(cli, ["index", "create", "example", "--yes", "--output", "json"]) + + assert result.exit_code == UsageError.exit_code + assert result.stdout == "" + payload = json.loads(result.stderr) + assert payload["error"]["code"] == UsageError.code + assert "SPLUNK_ENABLE_WRITES" in payload["error"]["message"] + + +def test_cloud_gate_precedes_noninteractive_confirmation( + cloud_env_no_opt_in: None, +) -> None: + """A missing Cloud opt-in is reported before the no-`--yes` confirmation gate.""" + result = cli_runner().invoke(cli, ["index", "create", "example", "--output", "json"]) + + assert result.exit_code == UsageError.exit_code + payload = json.loads(result.stderr) + assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] @pytest.fixture @@ -158,6 +231,7 @@ def cloud_write_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") monkeypatch.setenv("SPLUNK_ACS_TOKEN", "T") monkeypatch.setenv("SPLUNK_CLOUD_WRITE", "true") + monkeypatch.setenv("SPLUNK_ENABLE_WRITES", "true") def _patch_acs_write(monkeypatch: pytest.MonkeyPatch, handler) -> None: @@ -175,22 +249,67 @@ def test_non_acs_writes_still_refused_when_opted_in( ) -> None: """The opt-in unlocks only the nine ACS routes -- everything else still refuses.""" _refuse_network(monkeypatch) - result = CliRunner().invoke(cli, _argv(case, "--yes")) + result = cli_runner().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 result.stdout == "" + payload = json.loads(result.stderr) assert set(payload) == {"error"} assert payload["error"]["code"] == UnsupportedBackendError.code +def test_cloud_index_write_rejects_fields_without_an_acs_mapping( + cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """An unmapped typed option fails rather than disappearing from an ACS body.""" + _refuse_network(monkeypatch) + result = cli_runner().invoke( + cli, + ["index", "create", "example", "--frozen-secs", "10", "--yes", "--output", "json"], + ) + + assert result.exit_code == UsageError.exit_code + assert result.stdout == "" + payload = json.loads(result.stderr) + assert "--frozen-secs" in payload["error"]["message"] + + +def test_saved_search_trigger_actions_requires_enable_writes( + enterprise_env_no_opt_in: None, +) -> None: + """Alert actions use the real-write gate even though plain dispatch does not.""" + result = cli_runner().invoke( + cli, + [ + "saved-search", + "run", + "nightly", + "--app", + "my_app", + "--trigger-actions", + "--output", + "json", + ], + ) + + assert result.exit_code == UsageError.exit_code + assert result.stdout == "" + payload = json.loads(result.stderr) + assert "SPLUNK_ENABLE_WRITES" in payload["error"]["message"] + + @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) + + def refuse(req: httpx.Request) -> httpx.Response: + raise AssertionError(f"dry-run sent {req.method} {req.url}") + + _patch_acs_write(monkeypatch, refuse) result = CliRunner().invoke(cli, _argv(case, "--dry-run")) assert result.exit_code == 0, result.output diff --git a/tests/write_test_cli.py b/tests/write_test_cli.py index 599621b..2cea971 100644 --- a/tests/write_test_cli.py +++ b/tests/write_test_cli.py @@ -33,7 +33,7 @@ class WriteTestCli: 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: + def run(self, *argv: str, exit_codes: tuple[int, ...] = (0,)) -> dict[str, Any] | list[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}" @@ -45,7 +45,9 @@ def run(self, *argv: str, exit_codes: tuple[int, ...] = (0,)) -> Any: return payload["error"] def write(self, *argv: str) -> dict[str, Any]: - return self.run(*argv, "--yes") + result = self.run(*argv, "--yes") + assert isinstance(result, dict) + return result def cleanup(self, label: str, *argv: str) -> None: self.cleanups.append( From 52834ef73dfb4f0deb8ebb00aa5dca32316574b3 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:25:54 -0400 Subject: [PATCH 5/7] docs: describe Cloud write opt-ins and canary dispatch Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- .env.example | 5 +++-- AGENTS.md | 8 +++++--- CHANGELOG.md | 7 ++++--- README.md | 27 ++++++++++++++------------- docs/architecture.md | 10 +++++----- tests/TESTING.md | 13 +++++++++++-- 6 files changed, 42 insertions(+), 28 deletions(-) diff --git a/.env.example b/.env.example index c7d8547..4c9b2a7 100644 --- a/.env.example +++ b/.env.example @@ -33,7 +33,8 @@ SPLUNK_TOKEN= # 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. +# nothing and does not need this opt-in; Cloud previews still need +# SPLUNK_CLOUD_WRITE=true and an ACS token. # SPLUNK_ENABLE_WRITES=true # Initial password for `splunk user create` — a secret, so it is read from the @@ -80,7 +81,7 @@ SPLUNK_TOKEN= # 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. +# ACS-writable resources. Previews also need an ACS token to build the request. # SPLUNK_CLOUD_WRITE=true # Optional: a separate ACS token scoped to writes only. Falls back to diff --git a/AGENTS.md b/AGENTS.md index d702e55..2bc46cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,7 +32,7 @@ shell in `commands/` (the "functional core, imperative shell" pattern). See the Authorization header per request, `RetryTransport` retries 429/503) and the envelope-aware `SplunkClient`; `endpoint_factory.py` is the generic CRUD engine (`EndpointConfig`/`Field`/`Endpoints`); `endpoints/` holds one module - per hand-written operation (`server`, `search`, `jobs`, `saved_searches`, + per hand-written operation (`server`, `search`, `jobs`, `saved_searches`, `config`, `kvstore`, `hec`, `apps`, `cluster`, `license`, `deploy`, `lookups`, `datamodel`, `health`, `raw`); `acs/` is the Splunk Cloud ACS management-plane client. @@ -46,8 +46,8 @@ shell in `commands/` (the "functional core, imperative shell" pattern). See - `src/vct_splunk/output/` — `formatter.py`: JSON/table rendering and the error envelope. - `src/vct_splunk/commands/` — Click adapters, one module per hand-written - command group (`server`, `api`, `auth`, `search`, `health`, `inspect`, plus - `saved_search`'s `run`), plus shared plumbing: `context` (the `command` + command group (`server`, `api`, `auth`, `search`, `health`, `inspect`, `config`, + plus `saved_search`'s `run`), plus shared plumbing: `context` (the `command` decorator and `Ctx`), `write` (the single gated write path), `dispatch` (routes a few reads to Cloud ACS), and `registry` + `command_factory` (resource configs declared as data, turned into generated CRUD groups — @@ -99,6 +99,8 @@ Two cross-cutting ideas to know about: - 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`. +- `--dry-run` needs no `SPLUNK_ENABLE_WRITES` opt-in. A Cloud preview still + needs `SPLUNK_CLOUD_WRITE=true` and an ACS token to build its request. - `search run` is bounded by default (time window, row cap, timeout) so an agent cannot trigger an unbounded export by accident. diff --git a/CHANGELOG.md b/CHANGELOG.md index ff2911e..5b15a4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,9 +22,10 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). `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. + ACS-writable resources). Cloud previews also need an ACS token to build the + request; `SPLUNK_ENABLE_WRITES` is not needed for a preview. 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` diff --git a/README.md b/README.md index fb567d0..3037f36 100644 --- a/README.md +++ b/README.md @@ -134,19 +134,19 @@ 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 -- always works +splunk index create payments --dry-run # show the exact request, send nothing 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) ``` -`--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`. +An Enterprise `--dry-run` sends nothing, so it needs no write opt-in. 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 @@ -193,11 +193,12 @@ 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). +line can never enable a write. `--yes` works the same as it does against +Enterprise. A Cloud `--dry-run` preview still needs `SPLUNK_CLOUD_WRITE=true` +(it also proves the object is one of the three ACS-writable resources) and an +ACS token to build the request. 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 5e6284b..fc3039e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -16,10 +16,10 @@ vct_splunk/ client.py # httpx transport stack (AuthTransport, RetryTransport), # SplunkClient, create_client, get/set_client singletons endpoint_factory.py # Generic CRUD Endpoints engine, EndpointConfig/Field - endpoints/ # Hand-written endpoint modules (14 files) + endpoints/ # Hand-written endpoint modules (15 files) 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 + datamodel.py, health.py, raw.py, config.py acs/ # Splunk Cloud ACS management-plane client (reads + # gated index/role/hec-token writes) client.py, operations.py @@ -32,9 +32,9 @@ 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 reads (and gated - # writes) to ACS on Cloud - server.py, api.py, auth.py, search.py, saved_search.py, health.py, + dispatch.py # Routes index/role/hec-token reads to ACS on Cloud; + # checks which writes have an ACS route + server.py, api.py, auth.py, search.py, saved_search.py, config.py, health.py, kvstore.py, hec.py, apps.py, cluster.py, shcluster.py, license.py, deploy.py, lookup.py, datamodel.py, inspect.py diff --git a/tests/TESTING.md b/tests/TESTING.md index 9834f7a..d90b35f 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -299,8 +299,17 @@ 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. +required reviewers around the destructive half. Dispatch from a ref whose HEAD +commit subject starts with `tests: splunk cloud write`; the confirmation job +checks this before the environment approval. If needed, a no-change marker +commit on a validation branch can satisfy that requirement: + +```bash +git commit --allow-empty -m "tests: splunk cloud write" +``` + +The workflow 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: From 0821cfdaa0f3d247ed343f2dbe1a7bcd62684698 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:39:26 -0400 Subject: [PATCH 6/7] ci: reuse the Cloud read canary in the write canary Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- .github/workflows/ci.yml | 4 --- .github/workflows/cloud-read.yml | 19 +++++++++++-- .github/workflows/cloud-write.yml | 47 ++++--------------------------- 3 files changed, 23 insertions(+), 47 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fb10f14..20e7388 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -159,10 +159,6 @@ 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-read.yml b/.github/workflows/cloud-read.yml index c15f951..ff7a8cd 100644 --- a/.github/workflows/cloud-read.yml +++ b/.github/workflows/cloud-read.yml @@ -4,13 +4,26 @@ # tests/unit/test_acs_loopback.py and tests/unit/test_write_gating.py. # What only a live stack can show is that a real one answers the way the ACS # specification says it does. Until a stack is configured, this workflow ends as -# an explicit no-op rather than a green run that checked nothing. +# an explicit no-op rather than a green run that checked nothing. The write +# canary calls it with `require_credentials: true`, where a no-op fails instead. name: Splunk Cloud Read Canary on: schedule: - cron: "41 6 * * 3" workflow_dispatch: + workflow_call: + inputs: + require_credentials: + description: Fail instead of no-op when no Cloud read credentials are configured + type: boolean + default: false + secrets: + SPLUNK_URL: + SPLUNK_ACS_TOKEN: + SPLUNK_ACS_STACK: + SPLUNK_ACS_BASE_URL: + SPLUNK_TOKEN: permissions: contents: read @@ -44,8 +57,10 @@ jobs: - name: Require complete Cloud read credentials when configured if: env.CLOUD_READY != 'true' + env: + CLOUD_REQUIRED: ${{ inputs.require_credentials == true }} run: | - if [ "$CLOUD_CONFIGURED" = "true" ]; then + if [ "$CLOUD_CONFIGURED" = "true" ] || [ "$CLOUD_REQUIRED" = "true" ]; then echo "A partial Cloud read credential set is configured; SPLUNK_URL, SPLUNK_ACS_TOKEN, and SPLUNK_TOKEN are all required." exit 1 fi diff --git a/.github/workflows/cloud-write.yml b/.github/workflows/cloud-write.yml index b598c75..bf4bef4 100644 --- a/.github/workflows/cloud-write.yml +++ b/.github/workflows/cloud-write.yml @@ -1,6 +1,6 @@ # 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 +# Splunk Cloud stack: a `read` job first (the Splunk Cloud Read Canary, called +# as a reusable workflow), 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 @@ -64,50 +64,15 @@ jobs: read: name: Cloud / ACS reads (pre-write readiness) needs: confirm - runs-on: ubuntu-latest - timeout-minutes: 10 - env: - SPLUNK_ACS_LIVE_TEST: "true" + uses: ./.github/workflows/cloud-read.yml + with: + require_credentials: true + secrets: 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" - CLOUD_READY: ${{ secrets.SPLUNK_URL != '' && secrets.SPLUNK_ACS_TOKEN != '' && secrets.SPLUNK_TOKEN != '' }} - PYTEST_ADDOPTS: >- - -q --tb=line -r N -o 'addopts=--strict-markers --import-mode=importlib' - steps: - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 - with: - persist-credentials: false - - - name: Require read secrets when dispatched - if: env.CLOUD_READY != 'true' - run: | - echo "SPLUNK_URL, SPLUNK_ACS_TOKEN, and SPLUNK_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: >- - .venv/bin/pytest tests/integration/cloud/read/test_catalog.py - -m "integration and cloud and read" --junitxml=cloud-read.xml - - - name: Cloud ACS operations (below the CLI) - run: >- - .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py - -m "integration and cloud and read" --junitxml=cloud-acs.xml write: name: Cloud / ACS writes From f309faf2cd4aef45785d67bf6182feb24482b496 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:39:27 -0400 Subject: [PATCH 7/7] refactor: share ACS body mapping with the REST field serializer Assisted-by: Claude:claude-opus-5-5 Claude-Session: https://claude.ai/code/session_01MkCDG3PVVwxfoLyGSpg9is --- .env.example | 6 +- src/vct_splunk/api/acs/operations.py | 16 ++++ src/vct_splunk/commands/command_factory.py | 40 +++------ src/vct_splunk/commands/write.py | 69 ++------------- .../cloud/write/test_write_catalog.py | 8 +- tests/integration/test_acs_public_spec.py | 5 +- tests/unit/test_cli_matrix.py | 29 ------- tests/unit/test_write_gating.py | 86 +++++++++++-------- 8 files changed, 93 insertions(+), 166 deletions(-) diff --git a/.env.example b/.env.example index 4c9b2a7..aa6c3a9 100644 --- a/.env.example +++ b/.env.example @@ -31,10 +31,8 @@ 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 does not need this opt-in; Cloud previews still need -# SPLUNK_CLOUD_WRITE=true and an ACS token. +# Required for any real write on every backend. --dry-run sends no request and +# does not need this opt-in. This is never a CLI flag. # SPLUNK_ENABLE_WRITES=true # Initial password for `splunk user create` — a secret, so it is read from the diff --git a/src/vct_splunk/api/acs/operations.py b/src/vct_splunk/api/acs/operations.py index 53f918c..2488ead 100644 --- a/src/vct_splunk/api/acs/operations.py +++ b/src/vct_splunk/api/acs/operations.py @@ -12,6 +12,7 @@ from __future__ import annotations +from collections.abc import Collection from typing import Any from ...utils.errors import APIError @@ -45,6 +46,11 @@ "hec-token": (HEC_TOKENS, "hec"), } +#: Splunk REST form keys mapped to the equivalent ACS JSON keys. +ACS_FIELD_KEYS: dict[str, dict[str, str]] = { + "index": {"maxTotalDataSizeMB": "maxDataSizeMB"}, +} + #: (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. @@ -59,6 +65,16 @@ ) +def map_acs_body( + resource: str, body: dict[str, Any], *, preserve_keys: Collection[str] = () +) -> dict[str, Any]: + """Translate mapped REST body keys to ACS keys, preserving explicit keys.""" + mapping = ACS_FIELD_KEYS.get(resource, {}) + return { + key if key in preserve_keys else mapping.get(key, key): value for key, value in body.items() + } + + def list_cloud_indexes(client: AcsClient) -> list[dict[str, Any]]: """List indexes on the Cloud stack (ACS).""" return _list(client, INDEXES, LIST_ENVELOPES[INDEXES]) diff --git a/src/vct_splunk/commands/command_factory.py b/src/vct_splunk/commands/command_factory.py index 6e7e4ed..0841822 100644 --- a/src/vct_splunk/commands/command_factory.py +++ b/src/vct_splunk/commands/command_factory.py @@ -17,6 +17,7 @@ import click +from ..api.acs.operations import ACS_FIELD_KEYS, map_acs_body from ..api.endpoint_factory import EndpointConfig, Endpoints, Field from ..output import formatter as out from ..utils.errors import UsageError @@ -28,15 +29,6 @@ _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.""" @@ -56,34 +48,22 @@ 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, {}) +def _acs_body( + spec: EndpointConfig, res: Endpoints, fields: dict[str, Any], sets: dict[str, str] +) -> dict[str, Any]: + """Build an ACS body, refusing explicitly requested fields without a mapping.""" + 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: + if by_opt[opt].key not in mapping: 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 + return map_acs_body(spec.name, res._body(fields, sets), preserve_keys=sets) def _gate_args( @@ -158,7 +138,7 @@ def _create(ctx, name, **opts) -> None: 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, + body=_acs_body(spec, res, fields, sets) if ctx.backend == "cloud" else None, ) out.emit(result, ctx.output_mode, ctx.meta()) @@ -181,7 +161,7 @@ def _update(ctx, name, **opts) -> None: 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, + body=_acs_body(spec, res, fields, sets) if ctx.backend == "cloud" else None, ) out.emit(result, ctx.output_mode, ctx.meta()) diff --git a/src/vct_splunk/commands/write.py b/src/vct_splunk/commands/write.py index 84e2eba..687473b 100644 --- a/src/vct_splunk/commands/write.py +++ b/src/vct_splunk/commands/write.py @@ -29,15 +29,10 @@ from ..utils.redact import 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. +#: Opt-in required for every real mutation. 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`. +#: Additional opt-in for supported Cloud writes and previews. CLOUD_WRITE_ENV = "SPLUNK_CLOUD_WRITE" @@ -51,38 +46,7 @@ def do_write( name: str | None = None, body: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Run one gated mutation: confirm it, execute it, and audit it. - - 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 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. ``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`. - """ + """Confirm, dispatch, and audit a mutation after its backend gates.""" resource, _, verb = str(audit_event.get("action", "")).partition(".") refuse_cloud_write(ctx, resource, verb) if not ctx.dry_run: @@ -121,25 +85,9 @@ def do_write( def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None: - """Stop a Cloud mutation with no ACS route; gate the rest behind opt-in. + """Reject unsupported Cloud mutations or a missing Cloud opt-in. - 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 also required for - a Cloud preview, 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. + The opt-in also applies to previews because it proves the resource has an ACS route. """ if getattr(ctx, "backend", "enterprise") != "cloud": return @@ -153,12 +101,7 @@ def refuse_cloud_write(ctx: Any, resource: str, verb: str) -> None: 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. - """ + """Require the shared opt-in before a real write.""" if os.environ.get(ENABLE_WRITES_ENV) == "true": return raise UsageError( diff --git a/tests/integration/cloud/write/test_write_catalog.py b/tests/integration/cloud/write/test_write_catalog.py index 4bf57a7..e2baf05 100644 --- a/tests/integration/cloud/write/test_write_catalog.py +++ b/tests/integration/cloud/write/test_write_catalog.py @@ -28,7 +28,6 @@ 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 @@ -43,11 +42,16 @@ 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 + interval = 1.0 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) + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(interval, remaining)) + interval = min(interval * 2, 10.0) pytest.fail(f"{resource} {name!r} is still listed {_POLL_TIMEOUT:.0f}s after delete") diff --git a/tests/integration/test_acs_public_spec.py b/tests/integration/test_acs_public_spec.py index 8c0044d..db9392c 100644 --- a/tests/integration/test_acs_public_spec.py +++ b/tests/integration/test_acs_public_spec.py @@ -7,8 +7,7 @@ import httpx import pytest -from vct_splunk.api.acs.operations import LIST_ENVELOPES, WRITE_PATHS -from vct_splunk.commands.command_factory import _ACS_FIELD_KEYS +from vct_splunk.api.acs.operations import ACS_FIELD_KEYS, LIST_ENVELOPES, WRITE_PATHS pytestmark = [ pytest.mark.integration, @@ -48,6 +47,6 @@ def test_implemented_acs_write_contract_matches_public_spec(): if "$ref" in schema: component = schema["$ref"].rsplit("/", 1)[-1] schema = public["components"]["schemas"][component] - assert set(_ACS_FIELD_KEYS["index"].values()) <= set(schema["properties"]), ( + assert set(ACS_FIELD_KEYS["index"].values()) <= set(schema["properties"]), ( "ACS index field mappings are missing from the public create schema" ) diff --git a/tests/unit/test_cli_matrix.py b/tests/unit/test_cli_matrix.py index a07fe98..8e2e753 100644 --- a/tests/unit/test_cli_matrix.py +++ b/tests/unit/test_cli_matrix.py @@ -11,7 +11,6 @@ from cli_catalog import CATALOG, help_invocations, iter_leaves from vct_splunk.cli import cli -from vct_splunk.commands.dispatch import has_cloud_write def _handler(req: httpx.Request) -> httpx.Response: @@ -99,31 +98,3 @@ def record(req: httpx.Request) -> httpx.Response: offline = {("auth", "login"), ("auth", "status"), ("inspect",)} if case.kind == "read" and case.path not in offline: assert requests, f"{' '.join(case.path)} did not exercise its transport" - - -@pytest.mark.parametrize( - "case", - [case for case in CATALOG if case.kind == "write"], - ids=lambda case: " ".join(case.path), -) -def test_every_write_refuses_cloud_before_client_creation(case, cli_env, monkeypatch, tmp_path): - monkeypatch.setenv("SPLUNK_URL", "https://acme.splunkcloud.com") - monkeypatch.setenv("SPLUNK_APP", "my_app") - monkeypatch.setenv("SPLUNK_PASSWORD", "secret") - monkeypatch.setenv("VCT_SPLUNK_AUDIT", str(tmp_path / "audit.log")) - - def unexpected_client(*args, **kwargs): - raise AssertionError("Cloud write must not construct an ACS or splunkd client") - - monkeypatch.setattr("vct_splunk.commands.context.Ctx.client", unexpected_client) - monkeypatch.setattr("vct_splunk.commands.context.Ctx.acs_client", unexpected_client) - for args in case.argvs: - result = CliRunner().invoke(cli, [*case.path, *args, "--output", "json"]) - payload = json.loads(result.output) - if has_cloud_write(case.path[0], case.path[-1]): - assert result.exit_code == 2, f"{' '.join(case.path)}: {result.output}" - assert payload["error"]["code"] == "usage_error" - assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] - else: - assert result.exit_code == 4, f"{' '.join(case.path)}: {result.output}" - assert payload["error"]["code"] == "unsupported_backend" diff --git a/tests/unit/test_write_gating.py b/tests/unit/test_write_gating.py index e64b055..dbae2c3 100644 --- a/tests/unit/test_write_gating.py +++ b/tests/unit/test_write_gating.py @@ -1,30 +1,9 @@ -"""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. -""" +"""Prove write gates refuse before a client or network request can be made.""" from __future__ import annotations import json +from typing import Any import httpx import pytest @@ -163,23 +142,30 @@ def cloud_env_no_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: @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 = cli_runner().invoke(cli, _argv(case, "--yes")) +def test_every_write_is_refused_on_cloud_by_default( + case: Case, cloud_env_no_opt_in: None, monkeypatch: pytest.MonkeyPatch +) -> None: + def unexpected_client(*args: object, **kwargs: object) -> object: + raise AssertionError("Cloud write must not construct an ACS or splunkd client") + monkeypatch.setattr("vct_splunk.commands.context.Ctx.client", unexpected_client) + monkeypatch.setattr("vct_splunk.commands.context.Ctx.acs_client", unexpected_client) expected = ( UnsupportedBackendError if not has_cloud_write(case.path[0], case.path[-1]) else UsageError ) + real_write = tuple(arg for arg in case.argvs[0] if arg != "--dry-run") + ("--yes",) + for args in (*case.argvs, real_write): + result = cli_runner().invoke(cli, [*case.path, *args, "--output", "json"]) - assert result.exit_code == expected.exit_code, ( - f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" - ) - assert result.stdout == "" - payload = json.loads(result.stderr) - assert set(payload) == {"error"} - assert payload["error"]["code"] == expected.code - if expected is UsageError: - assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] + assert result.exit_code == expected.exit_code, ( + f"{' '.join(case.path)} exited {result.exit_code}: {result.stderr}" + ) + assert result.stdout == "" + payload = json.loads(result.stderr) + assert set(payload) == {"error"} + assert payload["error"]["code"] == expected.code + if expected is UsageError: + assert "SPLUNK_CLOUD_WRITE" in payload["error"]["message"] @pytest.mark.parametrize("value", ("", "TRUE", "1", "yes", "false")) @@ -340,6 +326,36 @@ def handler(req: httpx.Request) -> httpx.Response: assert seen["path"].startswith(f"/acme/adminconfig/v2/{_ACS_PATH[resource]}") +def test_cloud_index_body_reuses_rest_serialization_and_maps_acs_keys( + cloud_write_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + seen: dict[str, Any] = {} + + def handler(req: httpx.Request) -> httpx.Response: + seen.update(json.loads(req.content or b"{}")) + return httpx.Response(202, json={"name": "example"}) + + _patch_acs_write(monkeypatch, handler) + result = cli_runner().invoke( + cli, + [ + "index", + "create", + "example", + "--max-gb", + "2", + "--set", + "searchableDays=30", + "--yes", + "--output", + "json", + ], + ) + + assert result.exit_code == 0, result.output + assert seen == {"maxDataSizeMB": 2048, "searchableDays": "30", "name": "example"} + + def test_hec_token_create_redacts_token_in_output( cloud_write_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: