From 82a629aaeefacde6cd2a4515f5394752fe139613 Mon Sep 17 00:00:00 2001 From: JacobPEvans <20714140+JacobPEvans-personal@users.noreply.github.com> Date: Sun, 16 Aug 2026 21:16:49 -0400 Subject: [PATCH] feat: add read-only configuration commands --- CHANGELOG.md | 7 ++ README.md | 2 + src/vct_splunk/cli.py | 2 + src/vct_splunk/commands/config.py | 42 ++++++++ src/vct_splunk/core/config.py | 52 +++++++++ tests/cli_catalog.py | 7 ++ tests/unit/test_cli_matrix.py | 4 +- tests/unit/test_config.py | 168 ++++++++++++++++++++++++++++++ 8 files changed, 282 insertions(+), 2 deletions(-) create mode 100644 src/vct_splunk/commands/config.py create mode 100644 src/vct_splunk/core/config.py create mode 100644 tests/unit/test_config.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a27c16..1a814e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Added + +- Read configuration files through `splunk config list [FILE]` and `splunk + 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. + ### Changed - Lower the supported Python floor to 3.9, so the CLI runs under the interpreter diff --git a/README.md b/README.md index d3c2c2b..4ec105d 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,8 @@ splunk search run --query 'index=_internal | stats count by sourcetype' --earlie splunk search list # search jobs, running and finished splunk health check # server health; exit code 5 if anything is warn or fail splunk saved-search list --app my_app # saved searches in an app +splunk config list props.conf # stanzas in a configuration file +splunk config get props host::web-01 # properties from one stanza splunk api get /services/data/indexes # raw read-only escape hatch for any endpoint ``` diff --git a/src/vct_splunk/cli.py b/src/vct_splunk/cli.py index 21234c3..97bb445 100644 --- a/src/vct_splunk/cli.py +++ b/src/vct_splunk/cli.py @@ -9,6 +9,7 @@ from .commands.apps import app_install from .commands.auth import auth from .commands.cluster import cluster +from .commands.config import config from .commands.datamodel import datamodel_accelerate from .commands.deploy import deploy_client, deploy_server from .commands.factory import build_group @@ -41,6 +42,7 @@ def cli() -> None: auth, kvstore, cluster, + config, shcluster, license, deploy_server, diff --git a/src/vct_splunk/commands/config.py b/src/vct_splunk/commands/config.py new file mode 100644 index 0000000..f24e5dd --- /dev/null +++ b/src/vct_splunk/commands/config.py @@ -0,0 +1,42 @@ +"""`splunk config` read-only commands. Shell layer (imports Click).""" + +from __future__ import annotations + +import click + +from ..core import config as core +from ..core.namespace import resolve_ns +from . import output as out +from .context import command + + +@click.group() +def config() -> None: + """Read visible Splunk configuration files and stanzas.""" + + +@config.command("list") +@click.argument("file", required=False) +@command +def list_(ctx, file: str | None) -> None: + """List visible configuration files, or every stanza in FILE.""" + owner, app = resolve_ns(ctx.owner, ctx.app, for_write=False) + with ctx.client() as c: + data = ( + core.list_stanzas(c, file, owner=owner, app=app) + if file is not None + else core.list_files(c, owner=owner, app=app) + ) + out.emit(data, ctx.output_mode, ctx.meta()) + + +@config.command("get") +@click.argument("file") +@click.argument("stanza") +@command +def get(ctx, file: str, stanza: str) -> None: + """Show one stanza's properties from FILE.""" + owner, app = resolve_ns(ctx.owner, ctx.app, for_write=False) + with ctx.client() as c: + data = core.get_stanza(c, file, stanza, owner=owner, app=app) + out.emit(data, ctx.output_mode, ctx.meta()) diff --git a/src/vct_splunk/core/config.py b/src/vct_splunk/core/config.py new file mode 100644 index 0000000..67f1d6a --- /dev/null +++ b/src/vct_splunk/core/config.py @@ -0,0 +1,52 @@ +"""Read Splunk configuration files and stanzas. Click-free core.""" + +from __future__ import annotations + +from typing import Any + +from .client import SplunkClient +from .namespace import ns_path +from .path import path_segment +from .redact import REDACTED, is_secret_key, redact_secrets + + +def list_files(client: SplunkClient, *, owner: str, app: str) -> list[dict[str, Any]]: + """Return configuration files visible in one Splunk namespace.""" + path = ns_path("properties", owner=owner, app=app) + return [_name(entry) for entry in client.get_collection(path)] + + +def list_stanzas(client: SplunkClient, file: str, *, owner: str, app: str) -> list[dict[str, Any]]: + """Return every stanza in a visible configuration file.""" + path = _base(file, owner=owner, app=app) + return [_name(entry) for entry in client.get_collection(path)] + + +def get_stanza( + client: SplunkClient, file: str, stanza: str, *, owner: str, app: str +) -> dict[str, Any]: + """Return the properties in one visible configuration stanza.""" + encoded_stanza = path_segment(stanza, label="configuration stanza") + path = f"{_base(file, owner=owner, app=app)}/{encoded_stanza}" + properties: dict[str, Any] = {} + for entry in client.get_collection(path): + name = str(entry.get("name")) + content = entry.get("content") + properties[name] = REDACTED if is_secret_key(name) else redact_secrets(content) + return {"name": stanza, "properties": properties} + + +def _base(file: str, *, owner: str, app: str) -> str: + """Build a namespace-qualified configuration-file endpoint.""" + return f"{ns_path('properties', owner=owner, app=app)}/{_normal_file(file)}" + + +def _normal_file(file: str) -> str: + """Validate and encode a config-file name, accepting one optional .conf suffix.""" + name = file[:-5] if file.endswith(".conf") else file + return path_segment(name, label="configuration file") + + +def _name(entry: dict[str, Any]) -> dict[str, Any]: + """Return only the authoritative REST entry name for a file or stanza listing.""" + return {"name": entry.get("name")} diff --git a/tests/cli_catalog.py b/tests/cli_catalog.py index 8161600..25cd26c 100644 --- a/tests/cli_catalog.py +++ b/tests/cli_catalog.py @@ -58,6 +58,13 @@ class Case: Case(("auth", "login"), "read", (("--username", "admin"),)), Case(("auth", "status"), "read", ((),)), Case(("cluster", "status"), "read", ((),)), + Case( + ("config", "get"), + "read", + (("server", "general"),), + live_argv=("server", "general"), + ), + Case(("config", "list"), "read", ((), ("server.conf",)), live_argv=()), Case(("datamodel", "accelerate"), "write", (("model", "--app", "my_app", "--dry-run"),)), Case(("deploy-client", "list"), "read", ((),)), Case(("deploy-server", "reload"), "write", (("--dry-run",),)), diff --git a/tests/unit/test_cli_matrix.py b/tests/unit/test_cli_matrix.py index 8cf9ccc..9ef4bc6 100644 --- a/tests/unit/test_cli_matrix.py +++ b/tests/unit/test_cli_matrix.py @@ -58,8 +58,8 @@ def test_catalog_is_exactly_complete_and_unique(): assert duplicates == [], f"duplicate catalog commands: {duplicates}" assert sorted(" ".join(path) for path in live - catalogued) == [], "missing catalog commands" assert sorted(" ".join(path) for path in catalogued - live) == [], "stale catalog commands" - assert len(CATALOG) == 154 - assert sum(case.kind == "read" for case in CATALOG) == 61 + assert len(CATALOG) == 156 + assert sum(case.kind == "read" for case in CATALOG) == 63 assert sum(case.kind == "write" for case in CATALOG) == 93 assert all(1 <= len(case.argvs) <= 2 for case in CATALOG) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..0efb940 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,168 @@ +"""Configuration endpoint and CLI coverage.""" + +from __future__ import annotations + +import json + +import httpx +import pytest +from click.testing import CliRunner + +from vct_splunk.cli import cli +from vct_splunk.core import config +from vct_splunk.core.errors import UsageError + + +def _entry(name: str, content: object | None = None) -> dict[str, object]: + return {"name": name, "content": {} if content is None else content} + + +def test_list_files_paginates_without_returning_unknown_content(client_for) -> None: + seen: list[dict[str, str]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(dict(request.url.params)) + if request.url.params["offset"] == "0": + entries = [_entry("props", {"token": "never-print"})] + [ + _entry(f"other-{index}") for index in range(199) + ] + else: + entries = [_entry("transforms")] + return httpx.Response(200, json={"entry": entries, "paging": {"total": 201}}) + + files = config.list_files(client_for(handler), owner="-", app="-") + + assert files[0] == {"name": "props"} + assert files[-1] == {"name": "transforms"} + assert len(files) == 201 + assert seen == [ + {"count": "200", "offset": "0", "output_mode": "json"}, + {"count": "200", "offset": "200", "output_mode": "json"}, + ] + + +def test_stanza_paths_normalize_file_and_encode_dynamic_segments(client_for) -> None: + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.url.raw_path.decode().partition("?")[0]) + return httpx.Response( + 200, + json={"entry": [_entry("TRANSFORMS-routing", "main")], "paging": {"total": 1}}, + ) + + stanza = config.get_stanza( + client_for(handler), "props file.conf", "host::web one", owner="nobody", app="search" + ) + + assert stanza == {"name": "host::web one", "properties": {"TRANSFORMS-routing": "main"}} + assert seen == ["/servicesNS/nobody/search/properties/props%20file/host%3A%3Aweb%20one"] + + +def test_config_get_paginates_and_redacts_scalar_secret_keys(client_for) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.params["offset"] == "0": + entries = [_entry("password", "hidden")] + [ + _entry(f"other-{index}", index) for index in range(199) + ] + else: + entries = [_entry("nested", {"token": "also-hidden"})] + return httpx.Response(200, json={"entry": entries, "paging": {"total": 201}}) + + result = config.get_stanza(client_for(handler), "props", "actual", owner="-", app="-") + + assert result["name"] == "actual" + assert result["properties"]["password"] == "" + assert result["properties"]["nested"] == {"token": ""} + assert len(result["properties"]) == 201 + + +def test_config_lists_use_the_authoritative_entry_name(client_for) -> None: + body = {"entry": [_entry("actual", {"name": "forged", "nested": {"token": "hidden"}})]} + result = config.list_stanzas( + client_for(lambda request: httpx.Response(200, json=body)), + "props", + owner="-", + app="-", + ) + + assert result == [{"name": "actual"}] + + +@pytest.mark.parametrize("file", ["", ".conf", "../props", "props/other", "%252fetc"]) +def test_config_file_rejects_unsafe_paths_before_request(client_for, file: str) -> None: + requests: list[httpx.Request] = [] + client = client_for(lambda request: requests.append(request) or httpx.Response(200, json={})) + + with pytest.raises(UsageError): + config.list_stanzas(client, file, owner="-", app="-") + + assert requests == [] + + +@pytest.mark.parametrize("stanza", ["", "../default", "a/b", "%252fetc"]) +def test_config_stanza_rejects_unsafe_paths_before_request(client_for, stanza: str) -> None: + requests: list[httpx.Request] = [] + client = client_for(lambda request: requests.append(request) or httpx.Response(200, json={})) + + with pytest.raises(UsageError): + config.get_stanza(client, "props", stanza, owner="-", app="-") + + assert requests == [] + + +def test_get_stanza_empty_response_is_an_empty_stanza(client_for) -> None: + result = config.get_stanza( + client_for(lambda request: httpx.Response(200, json={"entry": []})), + "props", + "default", + owner="-", + app="-", + ) + + assert result == {"name": "default", "properties": {}} + + +def test_config_cli_uses_read_namespace_envelope_and_redaction(cli_env, patch_client) -> None: + seen: dict[str, str] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["path"] = request.url.path + return httpx.Response( + 200, + json={"entry": [_entry("password", "hidden")], "paging": {"total": 1}}, + ) + + patch_client(handler) + result = CliRunner().invoke( + cli, + [ + "config", + "get", + "props.conf", + "default", + "--owner", + "alice", + "--app", + "search", + "--output", + "json", + ], + ) + + assert result.exit_code == 0, result.output + assert seen["path"] == "/servicesNS/alice/search/properties/props/default" + payload = json.loads(result.output) + assert set(payload) == {"data", "meta"} + assert payload["data"] == {"name": "default", "properties": {"password": ""}} + assert "hidden" not in result.output + + +def test_config_cli_not_found_is_a_typed_error(cli_env, patch_client) -> None: + patch_client(lambda request: httpx.Response(404, json={"messages": []})) + + result = CliRunner().invoke(cli, ["config", "get", "props", "default", "--output", "json"]) + + assert result.exit_code == 4 + payload = json.loads(result.output) + assert payload["error"]["code"] == "not_found"