diff --git a/.env.example b/.env.example index 47f2c3b..fb299c0 100644 --- a/.env.example +++ b/.env.example @@ -71,6 +71,11 @@ SPLUNK_TOKEN= # FedRAMP stacks use https://admin.splunkcloudgc.com. # SPLUNK_ACS_BASE_URL= +# Hide the Cloud stack name at untrusted output boundaries (e.g. CI logs) in +# prompts, rendered JSON/table output, and error messages and details. The audit +# log is unaffected -- it always records the real host. +# VCT_SPLUNK_REDACT_TARGET=1 + # --- Live test opt-ins ------------------------------------------------------- # Enables live read tests. Enterprise writes also require SPLUNK_WRITE_TEST=true. diff --git a/.github/workflows/cloud-read.yml b/.github/workflows/cloud-read.yml index a34636d..efea0f3 100644 --- a/.github/workflows/cloud-read.yml +++ b/.github/workflows/cloud-read.yml @@ -30,6 +30,11 @@ jobs: # the credential check instead of reaching the dispatch layer, so the run # covers far less than it appears to. The guard below says so out loud. SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }} + # Untrusted output boundary: hide the Cloud stack name in prompts, JSON + # and table output, and error messages and details. + VCT_SPLUNK_REDACT_TARGET: "1" + PYTEST_ADDOPTS: >- + -q --tb=line -r N -o 'addopts=--strict-markers --import-mode=importlib' steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 with: @@ -58,15 +63,13 @@ jobs: if: steps.stack.outputs.ready == 'true' run: >- .venv/bin/pytest tests/integration/cloud/read/test_catalog.py - -m "integration and cloud and read" - -vv --color=yes --tb=short --junitxml=cloud-read.xml + -m "integration and cloud and read" --junitxml=cloud-read.xml - name: Cloud ACS operations (below the CLI) if: steps.stack.outputs.ready == 'true' run: >- .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py - -m "integration and cloud and read" - -vv --color=yes --tb=short --junitxml=cloud-acs.xml + -m "integration and cloud and read" --junitxml=cloud-acs.xml - name: Publish test summary if: always() && steps.stack.outputs.ready == 'true' diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a814e5..136ce27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,9 +12,17 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). config get FILE STANZA`. The commands use the normal read namespace, accept a file name with or without `.conf`, paginate collection results, and redact secret-valued properties. +- Opt-in redaction of Splunk Cloud stack names at untrusted CLI output + boundaries: `VCT_SPLUNK_REDACT_TARGET=1` hides them in prompts, rendered + JSON/table output, and error-envelope messages and details. The audit log is + unaffected and always records the real host. The `Splunk Cloud Read Canary` + workflow sets this. ### Changed +- `splunk inspect` no longer echoes the Cloud stack name in its report body. It + reports `stack_configured: bool` instead. + - 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 already uses only 3.9-compatible syntax and APIs. Declarations move diff --git a/docs/architecture.md b/docs/architecture.md index 37c962f..fecdd20 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -45,7 +45,7 @@ vct_splunk/ utils/ errors.py # Typed SplunkError hierarchy with exit codes - redact.py # Secret redaction by field name; safe_target for URLs + redact.py # Secret redaction by field name; safe_target; Cloud stack-name redaction namespace.py # /servicesNS/// path building + policy path.py # Path-segment validation/encoding (traversal-safe) validation.py # KEY=VALUE parsing diff --git a/src/vct_splunk/commands/auth.py b/src/vct_splunk/commands/auth.py index 01b01fe..1d8f9f6 100644 --- a/src/vct_splunk/commands/auth.py +++ b/src/vct_splunk/commands/auth.py @@ -18,13 +18,14 @@ from ..output import formatter as out from ..utils.errors import UsageError from ..utils.redact import safe_target +from ..utils.validation import env_flag from .context import command def _verify() -> bool | str: """TLS verification from the environment, matching the main client.""" ca = os.environ.get("SPLUNK_CA_BUNDLE") - on = os.environ.get("SPLUNK_VERIFY", "true").strip().lower() not in {"0", "false", "no"} + on = env_flag("SPLUNK_VERIFY", default=True) return ca or on diff --git a/src/vct_splunk/config/loader.py b/src/vct_splunk/config/loader.py index 446474b..12771f1 100644 --- a/src/vct_splunk/config/loader.py +++ b/src/vct_splunk/config/loader.py @@ -24,6 +24,7 @@ from pathlib import Path from ..utils.errors import UsageError +from ..utils.validation import env_flag from .types import AuthStatus, SplunkConfig #: The profile keys a section may define. Anything else is ignored. @@ -182,7 +183,5 @@ def _resolve_target( if not url: raise UsageError("No Splunk URL. Set SPLUNK_URL or pass --base-url.") ca = os.environ.get("SPLUNK_CA_BUNDLE") - verify = ca or ( - os.environ.get("SPLUNK_VERIFY", "true").strip().lower() not in {"0", "false", "no"} - ) + verify = ca or (env_flag("SPLUNK_VERIFY", default=True)) return AuthStatus(url.rstrip("/"), "none"), prof, verify diff --git a/src/vct_splunk/output/formatter.py b/src/vct_splunk/output/formatter.py index 98f6907..069d60d 100644 --- a/src/vct_splunk/output/formatter.py +++ b/src/vct_splunk/output/formatter.py @@ -13,6 +13,7 @@ import click from ..utils.errors import SplunkError, UsageError +from ..utils.redact import redact_public_text def resolve_mode(output: str | None, table: bool) -> str: @@ -27,15 +28,16 @@ def fail(exc: SplunkError) -> None: payload: dict[str, Any] = {"error": {"code": exc.code, "message": exc.message}} if exc.details is not None: payload["error"]["details"] = exc.details - click.echo(json.dumps(payload, default=str), err=True) + click.echo(redact_public_text(json.dumps(payload, default=str)), err=True) raise SystemExit(exc.exit_code) def emit(data: Any, mode: str, meta: dict[str, Any] | None = None) -> None: if mode == "json": - click.echo(json.dumps({"data": data, "meta": meta or {}}, indent=2, default=str)) + rendered = json.dumps({"data": data, "meta": meta or {}}, indent=2, default=str) else: - click.echo(_table(data)) + rendered = _table(data) + click.echo(redact_public_text(rendered)) def _table(data: Any) -> str: @@ -58,7 +60,8 @@ def confirm_write(ctx: Any, action: str, target: str | None) -> None: return if not (sys.stdin.isatty() and sys.stderr.isatty()): raise UsageError(f"Refusing to {action} on {target} without --yes (non-interactive).") - click.confirm(f"About to {action} on {target}. Continue?", abort=True, err=True) + prompt = redact_public_text(f"About to {action} on {target}. Continue?") + click.confirm(prompt, abort=True, err=True) def resolve_query(query: str | None, file_: Any, stdin_token: str | None) -> str: diff --git a/src/vct_splunk/utils/backends.py b/src/vct_splunk/utils/backends.py index d896791..ab9ce22 100644 --- a/src/vct_splunk/utils/backends.py +++ b/src/vct_splunk/utils/backends.py @@ -85,7 +85,8 @@ def inspect_report(url: str | None = None) -> dict[str, Any]: backend = deduce_backend(url) report: dict[str, Any] = {"backend": backend, "capabilities": CAPABILITIES[backend]} if backend == "cloud": - report["stack"] = cloud_stack_from_url(url) + # Identify whether a stack is configured without printing its name. + 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." ) diff --git a/src/vct_splunk/utils/redact.py b/src/vct_splunk/utils/redact.py index 631445f..74ac548 100644 --- a/src/vct_splunk/utils/redact.py +++ b/src/vct_splunk/utils/redact.py @@ -14,14 +14,26 @@ from __future__ import annotations +import re from typing import Any from urllib.parse import urlsplit, urlunsplit +from .validation import env_flag + #: What a hidden value is replaced with. The key stays, so a caller can still #: see that the field exists. REDACTED = "" +#: Set to hide a Splunk Cloud stack name at untrusted output boundaries. +REDACT_TARGET_ENV = "VCT_SPLUNK_REDACT_TARGET" + _SECRET_MARKERS = ("pass4symmkey", "password", "passwd", "secret", "token") +_CLOUD_STACK_HOST_RE = re.compile( + r"(?i)(? bool: @@ -98,3 +110,15 @@ def safe_target(target: str) -> str: pass path = parsed.path if "@" not in parsed.path else f"/{REDACTED}" return urlunsplit((parsed.scheme, host, path, "", "")) + + +def redact_public_text(text: str) -> str: + """Hide Cloud stack labels in rendered output when :data:`REDACT_TARGET_ENV` is on. + + The audit trail keeps the real host; only untrusted output boundaries such as + CI logs opt in. + """ + if not env_flag(REDACT_TARGET_ENV, default=False): + return text + text = _CLOUD_STACK_HOST_RE.sub(REDACTED, text) + return _ACS_STACK_PATH_RE.sub(rf"\g<1>{REDACTED}", text) diff --git a/src/vct_splunk/utils/validation.py b/src/vct_splunk/utils/validation.py index 470f16d..bc2ca27 100644 --- a/src/vct_splunk/utils/validation.py +++ b/src/vct_splunk/utils/validation.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from collections.abc import Iterable from .errors import UsageError @@ -20,3 +21,11 @@ def parse_key_value_pairs(pairs: Iterable[str]) -> dict[str, str]: raise UsageError(f"Duplicate key {key!r}.") parsed[key] = value return parsed + + +def env_flag(name: str, *, default: bool) -> bool: + """Read a boolean env var: unset gives ``default``; ``0``/``false``/``no`` are false.""" + value = os.environ.get(name) + if value is None: + return default + return value.strip().lower() not in {"0", "false", "no"} diff --git a/tests/TESTING.md b/tests/TESTING.md index 451fa64..b5101a3 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -200,7 +200,7 @@ output: ``` Every command must exit `0` and print a top-level `data` and `meta` object. -`inspect` must report the `cloud` backend and the intended stack. A secret field +`inspect` must report the `cloud` backend and `"stack_configured": true`. A secret field in HEC output must contain ``; an actual token value must never appear. @@ -271,7 +271,7 @@ when every required row passes and the cleanup is complete. | Revision and clean checkout | Intended revision; no unexpected changes | | | Credential-free Cloud routes and write refusals | Pytest passed | | | Public ACS OpenAPI contract | Pytest passed | | -| Backend inspection | `cloud` and intended stack | | +| Backend inspection | `cloud` and `stack_configured: true` | | | Index list | Exit 0; `data` + `meta` | | | Role list | Exit 0; `data` + `meta` | | | HEC token list | Exit 0; secrets absent or `` | | diff --git a/tests/integration/cloud/conftest.py b/tests/integration/cloud/conftest.py index ee51101..7422bcd 100644 --- a/tests/integration/cloud/conftest.py +++ b/tests/integration/cloud/conftest.py @@ -28,7 +28,7 @@ def _require_cloud_target() -> None: # would quietly test Enterprise instead of what this suite claims to cover. if deduce_backend(url) != "cloud": pytest.fail( - f"SPLUNK_URL={url!r} is not a Splunk Cloud host; " + "SPLUNK_URL is not a Splunk Cloud host; " "expected something like https://.splunkcloud.com", pytrace=False, ) diff --git a/tests/integration/cloud/read/test_catalog.py b/tests/integration/cloud/read/test_catalog.py index 1189eee..74cf339 100644 --- a/tests/integration/cloud/read/test_catalog.py +++ b/tests/integration/cloud/read/test_catalog.py @@ -55,9 +55,7 @@ def test_cloud_backed_read_succeeds(case: Case) -> None: """A read Cloud serves must succeed and return the typed data envelope.""" result = _invoke(case) - assert result.exit_code == 0, ( - f"{' '.join(case.path)} exited {result.exit_code}: {result.output}" - ) + assert result.exit_code == 0, f"{' '.join(case.path)} exited {result.exit_code}" payload = json.loads(result.output) assert set(payload) == {"data", "meta"} assert payload["meta"]["target"] @@ -69,8 +67,7 @@ def test_every_cloud_read_leaf_honors_the_contract(case: Case) -> None: result = _invoke(case) assert result.exit_code in CONTRACT_EXITS, ( - f"{' '.join(case.path)} exited {result.exit_code}, " - f"outside the documented contract: {result.output}" + f"{' '.join(case.path)} exited {result.exit_code}, outside the documented contract" ) payload = json.loads(result.output) diff --git a/tests/unit/test_acs.py b/tests/unit/test_acs.py index 61a1a47..3b7dda5 100644 --- a/tests/unit/test_acs.py +++ b/tests/unit/test_acs.py @@ -296,7 +296,11 @@ def test_inspect_reports_deduced_cloud(monkeypatch): result = CliRunner().invoke(cli, ["inspect", "--output", "json"]) assert result.exit_code == 0 assert '"backend": "cloud"' in result.output - assert '"stack": "acme"' in result.output + # The report body says whether a stack is configured, never its name -- + # unlike `meta.target`, whose opt-in redaction is covered separately in + # test_redact_public_text.py. + assert '"stack_configured": true' in result.output + assert '"stack"' not in result.output assert "not yet certified" in result.output diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py index 5a63367..d565aa1 100644 --- a/tests/unit/test_auth.py +++ b/tests/unit/test_auth.py @@ -200,6 +200,19 @@ def test_auth_status_reports_none_when_unset(monkeypatch): assert '"auth_scheme": "none"' in result.output +def test_auth_status_redacts_cloud_target_in_data_and_meta(monkeypatch): + _clear_auth_env(monkeypatch) + monkeypatch.setenv("SPLUNK_URL", "https://acme-test.splunkcloud.com:8089") + monkeypatch.setenv("VCT_SPLUNK_REDACT_TARGET", "1") + result = CliRunner().invoke(cli, ["auth", "status", "--output", "json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + expected = "https://.splunkcloud.com:8089" + assert payload["data"]["target"] == payload["meta"]["target"] == expected + assert "acme-test" not in result.output + + def test_auth_status_reports_username_password_without_logging_in(monkeypatch): _clear_auth_env(monkeypatch) monkeypatch.setenv("SPLUNK_URL", "https://splunk.test:8089") diff --git a/tests/unit/test_output.py b/tests/unit/test_output.py index 5fc8dc5..15d2a82 100644 --- a/tests/unit/test_output.py +++ b/tests/unit/test_output.py @@ -5,7 +5,7 @@ import pytest from vct_splunk.output import formatter as out -from vct_splunk.utils.errors import UsageError +from vct_splunk.utils.errors import TransportError, UsageError def test_resolve_mode_explicit(): @@ -33,6 +33,21 @@ def test_emit_json_envelope(capsys): assert parsed["meta"]["target"] == "splunk.test" +def test_fail_redacts_cloud_target_in_message_and_details(monkeypatch, capsys): + target = "https://acme-test.splunkcloud.com:8089" + monkeypatch.setenv("VCT_SPLUNK_REDACT_TARGET", "yes") + + with pytest.raises(SystemExit) as exc_info: + out.fail(TransportError(f"Could not reach {target}", details={"request_url": target})) + + assert exc_info.value.code == 1 + rendered = capsys.readouterr().err + payload = json.loads(rendered) + assert "acme-test" not in rendered + assert payload["error"]["message"] == "Could not reach https://.splunkcloud.com:8089" + assert payload["error"]["details"]["request_url"] == "https://.splunkcloud.com:8089" + + def test_emit_table(capsys): # A bare dict renders the same as a one-row list. out.emit({"name": "main", "size": 10}, "table") diff --git a/tests/unit/test_redact_public_text.py b/tests/unit/test_redact_public_text.py new file mode 100644 index 0000000..c129d35 --- /dev/null +++ b/tests/unit/test_redact_public_text.py @@ -0,0 +1,80 @@ +"""`redact_public_text` hides a Cloud stack name only when explicitly asked. + +`safe_target` always strips credentials -- that's unconditional. A Cloud stack +name is not a credential, but it does identify a specific customer's instance, +so hiding it is opt-in via `VCT_SPLUNK_REDACT_TARGET`, for untrusted output +boundaries such as CI logs. The output formatter applies it to everything it +renders. +""" + +from __future__ import annotations + +import pytest + +from vct_splunk.utils.redact import REDACTED, redact_public_text, safe_target + + +@pytest.fixture +def redact_enabled(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("VCT_SPLUNK_REDACT_TARGET", "1") + + +def test_without_the_env_var_the_stack_name_is_left_readable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("VCT_SPLUNK_REDACT_TARGET", raising=False) + assert redact_public_text("https://acme.splunkcloud.com") == "https://acme.splunkcloud.com" + + +@pytest.mark.parametrize("value", ["1", "TRUE", " yes "]) +def test_truthy_env_values_hide_a_cloud_host_label( + monkeypatch: pytest.MonkeyPatch, value: str +) -> None: + monkeypatch.setenv("VCT_SPLUNK_REDACT_TARGET", value) + assert ( + redact_public_text("https://acme.splunkcloud.com") == f"https://{REDACTED}.splunkcloud.com" + ) + + +def test_a_port_survives_the_rewrite(redact_enabled) -> None: + assert ( + redact_public_text("https://acme.splunkcloud.com:8089") + == f"https://{REDACTED}.splunkcloud.com:8089" + ) + + +def test_an_enterprise_target_is_unaffected(redact_enabled) -> None: + assert redact_public_text("https://sh.corp:8089") == "https://sh.corp:8089" + + +def test_url_credentials_are_still_stripped_under_redaction(redact_enabled) -> None: + target = redact_public_text(safe_target("https://admin:secret@acme.splunkcloud.com")) + assert target == f"https://{REDACTED}.splunkcloud.com" + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ( + "ACS request failed: https://admin.splunk.com/acme/adminconfig/v2/indexes", + "https://admin.splunk.com//adminconfig/v2/indexes", + ), + ( + "ACS request failed: https://admin.splunkcloudgc.com/acme/adminconfig/v2/indexes", + "https://admin.splunkcloudgc.com//adminconfig/v2/indexes", + ), + ( + "ACS returned 404 for GET /acme/adminconfig/v2/indexes", + "GET //adminconfig/v2/indexes", + ), + ], +) +def test_an_acs_stack_path_segment_is_hidden(redact_enabled, message: str, expected: str) -> None: + text = redact_public_text(message) + assert expected in text + assert "acme" not in text + + +def test_redaction_is_idempotent(redact_enabled) -> None: + once = redact_public_text("https://acme.splunkcloud.com") + assert redact_public_text(once) == once == f"https://{REDACTED}.splunkcloud.com"