feat: add read-only configuration commands

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

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
 ```
 

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,

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())

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")}

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",),)),

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)
 

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"] == "<redacted>"
+    assert result["properties"]["nested"] == {"token": "<redacted>"}
+    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": "<redacted>"}}
+    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"