Skip to content

Commit 7feecf4

Browse files
authored
t18697: aidevops secret dependents: show what uses a credential before rename, rotation or revocation
* feat: non-secret credential registry with explicit provider defaults (GH#34717) * refactor: extract set option parsing to keep cmd_set under complexity threshold (GH#34717) * refactor: split credential registry into fields/schema/names/store modules under qlty file-complexity (GH#34717) * feat: resolve credential aliases and explicit provider defaults at injection (GH#34718) * feat: add read-only credential dependents search (GH#34719) * fix: use absolute bash for inventory subprocess (CodeFactor B603/B607, GH#34717) * fix: resolve git/crontab to absolute paths in dependents scan (B603/B607) * fix: annotate subprocess import (Bandit B404, GH#34717) * fix: annotate subprocess imports (Bandit B404, GH#34719) * fix(credentials): narrow plist fallback exceptions in dependents scan (GH#34719)
1 parent 202ef29 commit 7feecf4

6 files changed

Lines changed: 381 additions & 2 deletions

File tree

‎.agents/scripts/credential_inventory.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,18 @@ def cmd_validate(_args: argparse.Namespace) -> int:
150150
return 0
151151

152152

153+
def cmd_dependents(args: argparse.Namespace) -> int:
154+
"""Local, read-only search for what uses a credential (GH#34719)."""
155+
import credential_inventory_dependents as dependents
156+
157+
result = dependents.find_dependents(names.related_names(store.load(), args.name))
158+
if args.json:
159+
emit_json(result)
160+
else:
161+
print(dependents.render(result))
162+
return 0
163+
164+
153165
# --------------------------------------------------------------------------
154166
# parser
155167
# --------------------------------------------------------------------------
@@ -211,6 +223,10 @@ def add_simple_parsers(sub: argparse._SubParsersAction) -> None:
211223
resolve_name = sub.add_parser("resolve-name", help="source credential for an alias or default variable")
212224
resolve_name.add_argument("name")
213225
resolve_name.set_defaults(func=cmd_resolve_name)
226+
dependents = sub.add_parser("dependents", help="find what uses a credential before rename/rotation")
227+
dependents.add_argument("name")
228+
dependents.add_argument("--json", action="store_true")
229+
dependents.set_defaults(func=cmd_dependents)
214230

215231

216232
def build_parser() -> argparse.ArgumentParser:
Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
#!/usr/bin/env python3
2+
# SPDX-License-Identifier: MIT
3+
# SPDX-FileCopyrightText: 2025-2026 Marcus Quinn
4+
"""What uses a credential? Local, read-only dependents search (GH#34719).
5+
6+
Searches registered repos and their linked worktrees, custom agents, local
7+
aidevops configs, scheduler definitions and the user crontab for a secret
8+
name and its aliases. Reads no secret values, makes no network calls, and
9+
reports `complete: false` whenever coverage is unproven (dynamic name
10+
construction, timeouts, truncation or unreadable sources), so a result
11+
never claims "unused" without evidence.
12+
"""
13+
14+
from __future__ import annotations
15+
16+
import json
17+
import subprocess # nosec B404 -- exception classes only; commands run via credential_inventory_scan.run_tool
18+
from concurrent.futures import ThreadPoolExecutor
19+
from pathlib import Path
20+
from typing import Any
21+
22+
import credential_inventory_scan as scan
23+
24+
GREP_TIMEOUT = 30
25+
MAX_WORKERS = 8
26+
CONFIG_SUFFIXES = {".json", ".conf", ".toml", ".yaml", ".yml"}
27+
REGISTRY_FILE = "credential-inventory.json"
28+
29+
30+
def registered_repos(home: Path) -> list[tuple[str, Path]]:
31+
"""(label, path) for each existing repo in repos.json."""
32+
try:
33+
data = json.loads((home / ".config" / "aidevops" / "repos.json").read_text(encoding="utf-8"))
34+
except (OSError, ValueError):
35+
return []
36+
entries = data.get("initialized_repos", []) if isinstance(data, dict) else []
37+
repos = []
38+
for entry in entries if isinstance(entries, list) else []:
39+
path = Path(str((entry or {}).get("path", ""))).expanduser()
40+
if str(path) != "." and path.is_dir():
41+
repos.append((str(entry.get("slug") or path.name), path))
42+
return repos
43+
44+
45+
def worktrees(repo: Path) -> list[Path]:
46+
"""Existing worktree paths (main checkout included) for a repo."""
47+
try:
48+
result = scan.run_tool("git", ["-C", str(repo), "worktree", "list", "--porcelain"], GREP_TIMEOUT)
49+
except (OSError, subprocess.SubprocessError):
50+
return [repo]
51+
paths = [Path(line[9:]) for line in result.stdout.splitlines() if line.startswith("worktree ")]
52+
return [path for path in paths if path.is_dir()] or [repo]
53+
54+
55+
def local_files(home: Path) -> dict[str, list[Path]]:
56+
"""Config and scheduler files to scan, grouped by source."""
57+
config_dir = home / ".config" / "aidevops"
58+
configs = sorted(
59+
path for path in config_dir.glob("*")
60+
if path.is_file() and path.suffix in CONFIG_SUFFIXES and path.name != REGISTRY_FILE
61+
) if config_dir.is_dir() else []
62+
schedulers = sorted((home / "Library" / "LaunchAgents").glob("*.plist"))
63+
systemd = home / ".config" / "systemd" / "user"
64+
schedulers += sorted(path for path in systemd.glob("*") if path.suffix in {".service", ".timer"})
65+
return {"config": configs, "scheduler": schedulers}
66+
67+
68+
def crontab_hits(needles: scan.Needles) -> tuple[list[dict], str | None]:
69+
"""Scan `crontab -l`; no crontab is an empty result, not a gap."""
70+
try:
71+
result = scan.run_tool("crontab", ["-l"], 10)
72+
except FileNotFoundError:
73+
return [], None
74+
except (OSError, subprocess.SubprocessError):
75+
return [], "crontab unreadable"
76+
if result.returncode != 0:
77+
return [], None
78+
return [{"path": "crontab", **hit} for hit in scan.scan_lines(result.stdout.splitlines(), needles)], None
79+
80+
81+
class Report:
82+
"""Accumulates hits, skipped sources and coverage counts."""
83+
84+
def __init__(self, names: list[str]) -> None:
85+
self.names = names
86+
self.hits: list[dict] = []
87+
self.skipped: list[dict] = []
88+
self.scanned: dict[str, int] = {}
89+
90+
def add(self, source: str, location: str, hits: list[dict], reason: str | None = None) -> None:
91+
self.hits += [{"source": source, "location": location, **hit} for hit in hits]
92+
if reason:
93+
self.skipped.append({"source": source, "location": location, "reason": reason})
94+
self.scanned[source] = self.scanned.get(source, 0) + 1
95+
96+
def result(self) -> dict[str, Any]:
97+
literal = [hit for hit in self.hits if hit["kind"] == "literal"]
98+
dynamic = [hit for hit in self.hits if hit["kind"] == "possible-dynamic"]
99+
return {
100+
"name": self.names[0],
101+
"aliases": self.names[1:],
102+
"literal": literal,
103+
"possible_dynamic": dynamic,
104+
"skipped": self.skipped,
105+
"scanned": self.scanned,
106+
"complete": not dynamic and not self.skipped,
107+
}
108+
109+
110+
def scan_repos(report: Report, home: Path, needles: scan.Needles) -> None:
111+
"""git grep every worktree of every registered repo, in parallel."""
112+
targets = [(label, tree) for label, repo in registered_repos(home) for tree in worktrees(repo)]
113+
targets = list({str(tree): (label, tree) for label, tree in targets}.values())
114+
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
115+
results = pool.map(lambda target: scan.grep_worktree(target[1], needles, GREP_TIMEOUT), targets)
116+
for (label, tree), (hits, reason) in zip(targets, results):
117+
report.add("repo", str(tree), [{"repo": label, **hit} for hit in hits], reason)
118+
119+
120+
def scan_local(report: Report, home: Path, needles: scan.Needles) -> None:
121+
"""Custom agents, aidevops configs, schedulers and crontab."""
122+
custom, capped = scan.tree_files(home / ".aidevops" / "agents" / "custom")
123+
groups = {"custom-agents": custom, **local_files(home)}
124+
for source, paths in groups.items():
125+
hits, skipped = scan.scan_files(paths, needles)
126+
report.add(source, source, hits, "file limit reached" if source == "custom-agents" and capped else None)
127+
report.scanned[source] = len(paths)
128+
for path in skipped:
129+
report.skipped.append({"source": source, "location": path, "reason": "unreadable or too large"})
130+
hits, reason = crontab_hits(needles)
131+
report.add("crontab", "crontab", hits, reason)
132+
133+
134+
def find_dependents(names: list[str], home: Path | None = None) -> dict[str, Any]:
135+
"""Search all local sources for `names` (canonical first, then aliases)."""
136+
home = home or Path.home()
137+
needles = scan.build_needles(names)
138+
report = Report(names)
139+
scan_repos(report, home, needles)
140+
scan_local(report, home, needles)
141+
return report.result()
142+
143+
144+
def render(result: dict[str, Any]) -> str:
145+
"""Human-readable report; never states 'unused' when incomplete."""
146+
aliases = f" (aliases: {', '.join(result['aliases'])})" if result["aliases"] else ""
147+
lines = [f"Dependents of {result['name']}{aliases}"]
148+
for label, key in (("literal", "literal"), ("possible-dynamic", "possible_dynamic")):
149+
for hit in result[key]:
150+
lines.append(f" {label:16} {hit['location']} {hit['path']}:{hit['line']} [{hit['matched']}]")
151+
for skip in result["skipped"]:
152+
lines.append(f" skipped {skip['location']} ({skip['reason']})")
153+
scanned = ", ".join(f"{source} {count}" for source, count in sorted(result["scanned"].items()))
154+
lines.append(f"Scanned: {scanned}")
155+
if not result["complete"]:
156+
lines.append("INCOMPLETE: dynamic name construction or skipped sources; do not treat this name as unused.")
157+
elif not result["literal"]:
158+
lines.append("No references found in the scanned sources.")
159+
return "\n".join(lines)

‎.agents/scripts/credential_inventory_names.py‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,14 @@ def alias_owner(data: dict[str, Any], alias: str) -> str | None:
7272
return None
7373

7474

75+
def related_names(data: dict[str, Any], name: str) -> list[str]:
76+
"""Canonical name first, then its declared aliases (works from an alias too)."""
77+
check_name("secret name", name)
78+
canonical = alias_owner(data, name) or name
79+
record = (data.get("credentials") or {}).get(canonical) or {}
80+
return [canonical, *sorted(set(record.get("aliases") or []) - {canonical})]
81+
82+
7583
def add_alias(data: dict[str, Any], name: str, alias: str) -> None:
7684
"""Declare `alias` as another name for recorded credential `name`."""
7785
check_name("secret name", name)
Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
#!/usr/bin/env python3
2+
# SPDX-License-Identifier: MIT
3+
# SPDX-FileCopyrightText: 2025-2026 Marcus Quinn
4+
"""Bounded reference scanning for credential names (GH#34719).
5+
6+
Classifies lines as a literal use of a secret name (token-exact) or a
7+
possible dynamic construction (`PREFIX_${X}_TAIL`, concatenation, f-strings,
8+
`%s`/`.format`). Matched line text is used only for classification and is
9+
never returned, so values that happen to sit in scanned files never leak.
10+
"""
11+
12+
from __future__ import annotations
13+
14+
import os
15+
import plistlib
16+
import re
17+
import shutil
18+
import subprocess # nosec B404 -- absolute git/crontab, list argv, never a shell
19+
from dataclasses import dataclass
20+
from pathlib import Path
21+
from typing import Iterable
22+
from xml.parsers.expat import ExpatError
23+
24+
MAX_HITS = 500
25+
MAX_FILE_BYTES = 1_000_000
26+
MAX_TREE_FILES = 5000
27+
MAX_GAP = 80
28+
TOKEN_RE = re.compile(r"[A-Za-z0-9_]")
29+
DYNAMIC_RE = re.compile(r"\$\{|\$\(|\$[A-Za-z_]|[\"']\s*\+|\+\s*[\"']|%s|\{[A-Za-z_][\w.]*\}|\.format\(|\.join\(")
30+
SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv"}
31+
32+
33+
@dataclass(frozen=True)
34+
class Needles:
35+
"""Exact names plus (prefix, tail) stems for dynamic-construction checks."""
36+
37+
names: tuple[str, ...]
38+
stems: tuple[tuple[str, str], ...]
39+
40+
@property
41+
def patterns(self) -> list[str]:
42+
"""Fixed strings for a first-pass grep; every name contains its prefix."""
43+
return sorted({prefix for prefix, _ in self.stems} | set(self.names))
44+
45+
46+
def build_needles(names: Iterable[str]) -> Needles:
47+
"""Build needles from a secret name and its aliases."""
48+
unique = sorted(set(names))
49+
stems = {(parts[0], parts[-1]) for parts in (name.split("_") for name in unique) if len(parts) >= 2}
50+
return Needles(tuple(unique), tuple(sorted(stems)))
51+
52+
53+
def has_token(line: str, name: str) -> bool:
54+
"""True when `name` appears with no identifier characters either side."""
55+
start = line.find(name)
56+
while start >= 0:
57+
end = start + len(name)
58+
before = line[start - 1] if start else ""
59+
after = line[end] if end < len(line) else ""
60+
if not TOKEN_RE.match(before) and not TOKEN_RE.match(after):
61+
return True
62+
start = line.find(name, start + 1)
63+
return False
64+
65+
66+
def looks_dynamic(line: str, prefix: str, tail: str) -> bool:
67+
"""True when prefix..tail on one expression is joined by a dynamic part."""
68+
start = line.find(prefix)
69+
while start >= 0:
70+
gap_start = start + len(prefix)
71+
end = line.find(tail, gap_start)
72+
gap = line[gap_start:end] if end >= 0 else ""
73+
if 0 < len(gap) <= MAX_GAP and prefix not in gap and not set(gap) & {";", "="} and DYNAMIC_RE.search(gap):
74+
return True
75+
start = line.find(prefix, start + 1)
76+
return False
77+
78+
79+
def classify(line: str, needles: Needles) -> tuple[str, str] | None:
80+
"""Return ("literal", NAME), ("possible-dynamic", PATTERN) or None."""
81+
for name in needles.names:
82+
if has_token(line, name):
83+
return "literal", name
84+
for prefix, tail in needles.stems:
85+
if looks_dynamic(line, prefix, tail):
86+
return "possible-dynamic", f"{prefix}_*_{tail}"
87+
return None
88+
89+
90+
def scan_lines(lines: Iterable[str], needles: Needles) -> list[dict]:
91+
"""Classify numbered lines; returns path-less hit records."""
92+
hits = []
93+
for number, line in enumerate(lines, 1):
94+
hit = classify(line, needles)
95+
if hit:
96+
hits.append({"line": number, "kind": hit[0], "matched": hit[1]})
97+
return hits
98+
99+
100+
def run_tool(tool: str, args: list[str], timeout: int) -> subprocess.CompletedProcess:
101+
"""Run a fixed system tool by absolute path; FileNotFoundError if absent."""
102+
executable = shutil.which(tool)
103+
if not executable:
104+
raise FileNotFoundError(tool)
105+
# Absolute executable, list argv (no shell); args are paths and fixed-string needles.
106+
return subprocess.run( # nosec B603
107+
[executable, *args], capture_output=True, text=True, errors="replace", timeout=timeout, check=False,
108+
)
109+
110+
111+
def grep_worktree(path: Path, needles: Needles, timeout: int) -> tuple[list[dict], str | None]:
112+
"""git grep a worktree (tracked + untracked, honouring .gitignore)."""
113+
args = ["-C", str(path), "grep", "-I", "-n", "--null", "--untracked", "-F"]
114+
for pattern in needles.patterns:
115+
args += ["-e", pattern]
116+
try:
117+
result = run_tool("git", args, timeout)
118+
except subprocess.TimeoutExpired:
119+
return [], "timeout"
120+
except OSError:
121+
return [], "git unavailable"
122+
if result.returncode not in (0, 1):
123+
return [], "git grep failed"
124+
hits: list[dict] = []
125+
for raw in result.stdout.splitlines():
126+
parts = raw.split("\0", 2)
127+
hit = classify(parts[2], needles) if len(parts) == 3 and parts[1].isdigit() else None
128+
if hit:
129+
hits.append({"path": parts[0], "line": int(parts[1]), "kind": hit[0], "matched": hit[1]})
130+
if len(hits) >= MAX_HITS:
131+
return hits, "truncated"
132+
return hits, None
133+
134+
135+
def plist_as_xml(data: bytes) -> str | None:
136+
"""XML text for a parseable plist, or None so callers fall back to raw text."""
137+
try:
138+
return plistlib.dumps(plistlib.loads(data), fmt=plistlib.FMT_XML).decode("utf-8", "replace")
139+
except (plistlib.InvalidFileException, ExpatError, ValueError, TypeError, OverflowError):
140+
return None
141+
142+
143+
def read_text(path: Path) -> str | None:
144+
"""Readable text (XML for binary plists); "" for binary, None if unscanned."""
145+
try:
146+
if path.stat().st_size > MAX_FILE_BYTES:
147+
return None
148+
data = path.read_bytes()
149+
except OSError:
150+
return None
151+
if path.suffix == ".plist":
152+
xml_text = plist_as_xml(data)
153+
if xml_text is not None:
154+
return xml_text
155+
if b"\0" in data[:4096]:
156+
return "" # binary, skipped like `git grep -I`
157+
return data.decode("utf-8", "replace")
158+
159+
160+
def scan_files(paths: Iterable[Path], needles: Needles) -> tuple[list[dict], list[str]]:
161+
"""Scan explicit files; returns hits and paths skipped as unreadable/large."""
162+
hits: list[dict] = []
163+
skipped: list[str] = []
164+
for path in paths:
165+
text = read_text(path)
166+
if text is None:
167+
skipped.append(str(path))
168+
continue
169+
hits += [{"path": str(path), **hit} for hit in scan_lines(text.splitlines(), needles)]
170+
return hits, skipped
171+
172+
173+
def tree_files(root: Path) -> tuple[list[Path], bool]:
174+
"""Files under root (skipping VCS/dependency dirs); True if capped."""
175+
found: list[Path] = []
176+
for dirpath, dirnames, filenames in os.walk(root):
177+
dirnames[:] = [name for name in dirnames if name not in SKIP_DIRS]
178+
found += [Path(dirpath) / name for name in filenames]
179+
if len(found) >= MAX_TREE_FILES:
180+
return found[:MAX_TREE_FILES], True
181+
return found, False

0 commit comments

Comments
 (0)