Splunk Enterprise CVE-2026-20253: KEV Arbitrary File Creation via PostgreSQL Sidecar

#!/usr/bin/env python3
"""Audit Splunk Enterprise CVE-2026-20253 exposure and abuse evidence.

The script scans a repository tree and optional telemetry export tree for
incident-specific selectors drawn from the CISA KEV entry, the Splunk advisory,
and NVD. It reports whether the tree contains version strings, file-path hints,
or mitigation evidence tied to the PostgreSQL sidecar service.
"""

import json
import os
import sys
from pathlib import Path
from typing import Iterable

CVE_ID = "CVE-2026-20253"
ADVISORY_ID = "SVD-2026-0603"
CISA_FEED_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
SPLUNK_ADVISORY_URL = "https://advisory.splunk.com/advisories/SVD-2026-0603"
NVD_URL = "https://nvd.nist.gov/vuln/detail/CVE-2026-20253"
URLS = ["https://advisory.splunk.com/advisories/SVD-2026-0603","https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"]

# Collect unique indicators
indicators = set()
for group in [URLS]:
    for val in group:
        if val:
            indicators.add(val)

AFFECTED_VERSION_STRINGS = [
    "10.2 versions below 10.2.4",
    "10 versions below 10.0.7",
    "10.2.0 to 10.2.3",
    "10.0.0 to 10.0.6",
    "10.2.3",
    "10.0.6",
    "10.2.4",
    "10.0.7",
]
MITIGATION_STRINGS = [
    "[postgres]",
    "disabled = true",
    "$SPLUNK_HOME/etc/system/local/server.conf",
    "disable the PostgreSQL sidecar service",
    "Sidecar Configuration Settings",
    "Postgresql Configuration",
]
CONTEXT_STRINGS = [
    CVE_ID,
    ADVISORY_ID,
    "Splunk Enterprise",
    "PostgreSQL sidecar service endpoint",
    "create or truncate arbitrary files",
]
INDICATORS = sorted({
    *AFFECTED_VERSION_STRINGS,
    *MITIGATION_STRINGS,
    *CONTEXT_STRINGS,
    *DOMAINS,
    *PROCESS_PATTERNS,
    CISA_FEED_URL,
    SPLUNK_ADVISORY_URL,
    NVD_URL,
})
EXCLUDED_DIR_NAMES = {".git", "node_modules", "vendor", "dist", "__pycache__", ".venv"}
TEXT_SUFFIXES = {".conf", ".txt", ".log", ".md", ".json", ".yaml", ".yml", ".py", ".ini", ".cfg", ".xml", ".toml"}


def _scan_tree(root: Path) -> list[dict[str, object]]:
    matches: list[dict[str, object]] = []
    if not root.exists():
        return matches

    for path in root.rglob("*"):
        if path.is_dir():
            continue
        if any(part in EXCLUDED_DIR_NAMES for part in path.parts):
            continue
        if path.suffix and path.suffix not in TEXT_SUFFIXES and path.name not in {"server.conf"}:
            # Keep the scan focused on text-like files, but still allow key config names.
            continue
        try:
            content = path.read_text(errors="ignore")
        except Exception:
            continue
        hits = [indicator for indicator in INDICATORS if indicator.lower() in content.lower()]
        if hits:
            matches.append({
                "path": str(path),
                "hits": sorted(set(hits)),
            })
    return matches


def _ensure_out_dir(out_dir: Path) -> None:
    out_dir.mkdir(parents=True, exist_ok=True)


def _write_lines(path: Path, lines: Iterable[str]) -> None:
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> int:
    root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
    log_root_env = os.environ.get("LOG_ROOT", "").strip()
    log_root = Path(log_root_env) if log_root_env else None
    out_dir = Path(os.environ.get("OUT", "hp-splunk-enterprise-cve-2026-20253-kev-scope"))
    _ensure_out_dir(out_dir)

    selectors_file = out_dir / "selectors.txt"
    _write_lines(selectors_file, INDICATORS)

    repo_matches = _scan_tree(root)
    log_matches = _scan_tree(log_root) if log_root else []

    report = {
        "cve_id": CVE_ID,
        "advisory_id": ADVISORY_ID,
        "root": str(root),
        "log_root": str(log_root) if log_root else "",
        "indicator_count": len(INDICATORS),
        "repository_matches": repo_matches,
        "telemetry_matches": log_matches,
        "exposure_signals": {
            "affected_version_seen": any(
                any(version.lower() in hit.lower() for hit in entry["hits"])  # type: ignore[index]
                for entry in repo_matches
                for version in AFFECTED_VERSION_STRINGS
            ),
            "mitigation_seen": any(
                any(mitig.lower() in hit.lower() for hit in entry["hits"])  # type: ignore[index]
                for entry in repo_matches
                for mitig in MITIGATION_STRINGS
            ),
        },
    }

    report_path = out_dir / "audit-report.json"
    report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    summary = [
        f"[+] selectors written: {selectors_file}",
        f"[+] repository matches: {len(repo_matches)}",
        f"[+] telemetry matches: {len(log_matches)}",
        f"[+] report written: {report_path}",
    ]
    print("\n".join(summary))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())