Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
11 changes: 7 additions & 4 deletions .github/workflows/cloud-read.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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'
Expand Down
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/<owner>/<app>/ path building + policy
path.py # Path-segment validation/encoding (traversal-safe)
validation.py # KEY=VALUE parsing
Expand Down
3 changes: 2 additions & 1 deletion src/vct_splunk/commands/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
5 changes: 2 additions & 3 deletions src/vct_splunk/config/loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
11 changes: 7 additions & 4 deletions src/vct_splunk/output/formatter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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:
Expand Down
3 changes: 2 additions & 1 deletion src/vct_splunk/utils/backends.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."
)
Expand Down
24 changes: 24 additions & 0 deletions src/vct_splunk/utils/redact.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = "<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)(?<![A-Za-z0-9-])[A-Za-z0-9-]+(?=\.splunkcloud\.com(?:[^A-Za-z0-9.-]|$))"
)
_ACS_STACK_PATH_RE = re.compile(
rf"(?i)(//[^/\s?#]+/|/)(?!{re.escape(REDACTED)}(?:/|$))([^/?#\s]+)(?=/adminconfig(?:/|$))"
)


def is_secret_key(key: object) -> bool:
Expand Down Expand Up @@ -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)
9 changes: 9 additions & 0 deletions src/vct_splunk/utils/validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import os
from collections.abc import Iterable

from .errors import UsageError
Expand All @@ -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"}
4 changes: 2 additions & 2 deletions tests/TESTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 `<redacted>`; an actual token value must never
appear.

Expand Down Expand Up @@ -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 `<redacted>` | |
Expand Down
2 changes: 1 addition & 1 deletion tests/integration/cloud/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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://<stack>.splunkcloud.com",
pytrace=False,
)
Expand Down
7 changes: 2 additions & 5 deletions tests/integration/cloud/read/test_catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand All @@ -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)

Expand Down
6 changes: 5 additions & 1 deletion tests/unit/test_acs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
13 changes: 13 additions & 0 deletions tests/unit/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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://<redacted>.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")
Expand Down
17 changes: 16 additions & 1 deletion tests/unit/test_output.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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://<redacted>.splunkcloud.com:8089"
assert payload["error"]["details"]["request_url"] == "https://<redacted>.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")
Expand Down
Loading
Loading