diff --git a/Discovery/adr_discovery/enumerator/__init__.py b/Discovery/adr_discovery/enumerator/__init__.py new file mode 100644 index 0000000..d9b66ed --- /dev/null +++ b/Discovery/adr_discovery/enumerator/__init__.py @@ -0,0 +1,52 @@ +"""M2 -- where should we look. + +Answers that question once, for the whole pipeline. Nothing downstream may +decide it for itself, and nothing here knows what any particular tool is: +`enumerate_candidates` is run with an empty catalog in its own test set, +and must produce the same candidates it produces with a full one. +""" + +from __future__ import annotations + +from ..contracts.records import Candidate +from .roots import homes +from .sources.appstate import from_app_state +from .sources.binaries import from_binaries +from .sources.execjournal import from_exec_journal +from .sources.modelstores import from_model_stores +from .sources.network import from_network +from .sources.registries import from_applications, from_kernel, from_packages +from .sweep import sweep + +__all__ = ["enumerate_candidates"] + + +def enumerate_candidates(gate, include_dependency_caches: bool = False) -> tuple[Candidate, ...]: + """Registries first, then the sweep. + + The order is the optimisation: most of the search is over before the + walk starts, and every registry hit arrives with provenance attached. + """ + found: list[Candidate] = [] + + # Half one -- ask what already has the answer. + found.extend(from_packages(gate)) + found.extend(from_applications(gate)) + kernel = from_kernel(gate) + found.extend(kernel) + found.extend(from_network(gate, kernel)) + found.extend(from_exec_journal(gate)) + found.extend(from_app_state(gate, homes(gate))) + found.extend(from_binaries(gate, homes(gate))) + found.extend(from_model_stores(gate, homes(gate))) + + registry_entries = gate.budget.entries_used + + # Half two -- sweep only what no registry indexes. + found.extend(sweep(gate, include_dependency_caches)) + + gate.ledger.probe( + "enumerator", "ran", + f"{len(found)} candidates; {registry_entries} entries before the sweep", + ) + return tuple(found) diff --git a/Discovery/adr_discovery/enumerator/markers.py b/Discovery/adr_discovery/enumerator/markers.py new file mode 100644 index 0000000..a75ca70 --- /dev/null +++ b/Discovery/adr_discovery/enumerator/markers.py @@ -0,0 +1,155 @@ +"""The marker set, as data. + +Traversal is keyed on markers rather than on remembered paths, which is +what lets a repository in /opt/checkouts be found by the same rule that +finds one in ~/Projects. + +Nothing here names a tool. A marker says *where an agent works*; deciding +what the agent is belongs to M4, and M2 must stay passable with an empty +catalog (U2-03). +""" + +from __future__ import annotations + +#: Directory names that mark a surface worth reading. +DIR_MARKERS: frozenset[str] = frozenset( + { + ".git", ".claude", ".cursor", ".windsurf", ".aider", ".continue", + ".codeium", ".gemini", ".goose", ".opencode", ".zed", + "agents", "skills", "commands", "prompts", "output-styles", "plugins", + ".github", ".devcontainer", ".vscode", + } +) + +#: File names that mark a surface worth reading. +FILE_MARKERS: frozenset[str] = frozenset( + { + ".mcp.json", ".claude.json", "mcp.json", "settings.json", "settings.local.json", + "config.toml", "config.yaml", "mcp_settings.json", "mcp_config.json", + "cline_mcp_settings.json", "managed-settings.json", "managed-mcp.json", + "claude_desktop_config.json", "opencode.json", + } +) + +#: Workflow files are read by suffix rather than by name -- nobody agrees +#: on what a workflow is called, only on where it lives. +WORKFLOW_DIR = "/.github/workflows/" +WORKFLOW_SUFFIXES = (".yml", ".yaml") + +#: Instruction filenames are programmable-surface records. Their contents are +#: never collected; only path, scope and host-facing name leave the endpoint. +INSTRUCTION_MARKERS: frozenset[str] = frozenset( + { + "CLAUDE.md", "AGENTS.md", "GEMINI.md", "AGENT.md", + ".cursorrules", ".windsurfrules", "copilot-instructions.md", + } +) + +LOCATOR_ONLY: frozenset[str] = frozenset({".cursorrules", ".windsurfrules"}) + +#: State directories a host application keeps per user. +STATE_ROOTS: tuple[str, ...] = ( + "~/.claude", "~/.codex", "~/.cursor", "~/.aider", "~/.continue", + "~/.gemini", "~/.config/goose", "~/.config/opencode", "~/.ollama", + "~/Library/Application Support/Claude", + "~/Library/Application Support/Code/User", + "~/.config/Code/User", + "~/.vscode/extensions", "~/.vscode-server/extensions", +) + +#: Config files loaded directly by known agent hosts. These are enumerated +#: independently of the breadth sweep so a dependency cache cannot hide them. +CONFIG_FILE_TEMPLATES: tuple[str, ...] = ( + "~/.claude.json", + "~/.config/claude-desktop/claude_desktop_config.json", + "~/.cursor/mcp.json", + "~/.codeium/windsurf/mcp_config.json", + "~/.config/Code/User/mcp.json", + "~/.config/Code/User/globalStorage/saoudrizwan.claude-dev/settings/cline_mcp_settings.json", + "~/.config/zed/settings.json", + "~/.config/JetBrains/options/mcp.json", + "~/.config/opencode/opencode.json", + "~/.codex/config.toml", + "~/.config/goose/config.yaml", + "~/.bashrc", + "~/.zshrc", + "/etc/claude-code/managed-settings.json", + "/etc/adr/managed-mcp.json", +) + +#: Browser profile parents. Every profile, not just the default -- a large +#: share of real shadow AI lives on a second profile. +BROWSER_PROFILE_ROOTS: tuple[str, ...] = ( + "~/Library/Application Support/Google/Chrome", + "~/Library/Application Support/BraveSoftware/Brave-Browser", + "~/Library/Application Support/Microsoft Edge", + "~/Library/Application Support/Arc/User Data", + "~/.config/google-chrome", + "~/.config/chromium", + "~/.config/microsoft-edge", +) + +FIREFOX_PROFILE_ROOTS: tuple[str, ...] = ( + "~/Library/Application Support/Firefox/Profiles", + "~/.mozilla/firefox", +) + +EDITOR_EXTENSION_ROOTS: tuple[str, ...] = ( + "~/.vscode/extensions", "~/.vscode-server/extensions", + "~/.cursor/extensions", "~/.windsurf/extensions", + "~/.trae/extensions", "~/.kilo/extensions", +) + +#: Hosts that answer for a model provider. Landscape data, not identity: +#: a connection here says *something on this machine talks to a model*, +#: which is a candidate. What it is remains M4's question. +MODEL_PROVIDER_SUFFIXES: tuple[str, ...] = ( + "api.anthropic.com", "api.openai.com", "openai.azure.com", + "generativelanguage.googleapis.com", "aiplatform.googleapis.com", + "bedrock-runtime.amazonaws.com", "api.mistral.ai", "api.cohere.ai", + "api.groq.com", "api.together.xyz", "api.deepseek.com", + "api.x.ai", "openrouter.ai", "huggingface.co", +) + +#: Bundles and portable executables carry their own runtime, so nothing +#: else on disk reveals them. +BUNDLE_SUFFIXES: tuple[str, ...] = (".AppImage", ".app", ".exe") + +#: Ports a local model runtime answers on. +LOCAL_MODEL_PORTS: frozenset[int] = frozenset({11434, 1234, 8080, 8000, 5000, 7860}) + + +def is_model_provider(host: str) -> bool: + h = host.lower().rstrip(".") + return any(h == s or h.endswith("." + s) for s in MODEL_PROVIDER_SUFFIXES) + + +def is_loose_executable(entry, name: str) -> bool: + """An executable nothing else on disk accounts for. + + Extensionless is the test that keeps this from matching every script in + every repository: real CLI tools ship as `claude`, not `claude.sh`. + """ + if entry.is_dir: + return name.endswith(BUNDLE_SUFFIXES) + if name.endswith(BUNDLE_SUFFIXES): + return True + return entry.is_exec and "." not in name + + +def marker_kind(name: str, path: str = "") -> str | None: + if WORKFLOW_DIR in path and path.endswith(WORKFLOW_SUFFIXES): + return "marker_file" + if name in DIR_MARKERS: + return "marker_dir" + if name in FILE_MARKERS: + return "marker_file" + if name in INSTRUCTION_MARKERS: + return "instruction_file" + if name in (".bashrc", ".zshrc"): + return "shell_profile" + if name == "manifest.json" and "/.mcpb/" in path: + return "marker_file" + if name in LOCATOR_ONLY: + return "locator" + return None diff --git a/Discovery/adr_discovery/enumerator/roots.py b/Discovery/adr_discovery/enumerator/roots.py new file mode 100644 index 0000000..684275c --- /dev/null +++ b/Discovery/adr_discovery/enumerator/roots.py @@ -0,0 +1,96 @@ +"""Priority roots -- one definition. + +There were five copies of this tuple, in five probe files, none of which +reported that it had a boundary. Roots now *order* the sweep so the common +case stays fast; they no longer decide what exists. +""" + +from __future__ import annotations + +from ..contracts.records import Priority + +#: (template, priority). `~` is expanded per discovered home, not per the +#: user running the scan -- the owner of an asset is a person, never whoever +#: happened to launch the collector. +ROOT_TEMPLATES: tuple[tuple[str, Priority], ...] = ( + ("~", Priority.HOME), + ("~/Projects", Priority.CODE_ROOT), + ("~/src", Priority.CODE_ROOT), + ("~/code", Priority.CODE_ROOT), + ("~/work", Priority.CODE_ROOT), + ("~/dev", Priority.CODE_ROOT), + ("~/git", Priority.CODE_ROOT), + ("~/repos", Priority.CODE_ROOT), + ("/opt", Priority.SYSTEM), + ("/srv", Priority.SYSTEM), + ("/usr/local", Priority.SYSTEM), + ("/workspace", Priority.SYSTEM), + ("/Users", Priority.BREADTH), + ("/home", Priority.BREADTH), +) + +#: Scope is policy, not a constant. Whether a dependency cache is in scope is +#: a real question with a defensible answer either way, so it lives here with +#: a stated default rather than in a tuple nobody can see. +DEPENDENCY_CACHES: tuple[str, ...] = ( + "node_modules", ".venv", "venv", "site-packages", "go/pkg/mod", + ".cargo/registry", "vendor", ".gradle", ".m2", +) + +SKIP_ALWAYS: tuple[str, ...] = ( + ".git/objects", ".Trash", "Library/Caches", ".cache", "__pycache__", + "/.npm/", "/.local/share/pipx/", "/.cargo/registry/", "/.gradle/", "/.m2/", +) + + +def homes(gate) -> tuple[str, ...]: + """Every home on the machine, not just the caller's. + + Where homes live is a platform question and is answered by M1's + provider, not by a tuple here -- which is the same rule that removed + the five copies of PROJECT_ROOTS. + """ + found: list[str] = [] + for base in gate.providers.home_roots(): + listing = gate.list_dir(base) + if not listing.ok: + continue + for entry in listing.value: + if entry.is_dir and not entry.path.rsplit("/", 1)[-1].startswith("."): + found.append(entry.path) + if not found: + home = gate.env.get("HOME") + if home: + found.append(home) + return tuple(found) + + +def ordered_roots(gate) -> tuple[tuple[str, Priority], ...]: + """Roots in sweep order: home first, then code roots, then breadth. + + Order is asserted by U2-06, because a budget exhausted late must still + have covered the likely places. + """ + out: list[tuple[str, Priority]] = [] + seen: set[str] = set() + for home in homes(gate): + for template, priority in ROOT_TEMPLATES: + if not template.startswith("~"): + continue + path = home + template[1:] + if path not in seen: + seen.add(path) + out.append((path, priority)) + for template, priority in ROOT_TEMPLATES: + if template.startswith("~") or template in seen: + continue + seen.add(template) + out.append((template, priority)) + out.sort(key=lambda pair: pair[1]) + return tuple(out) + + +def in_scope(path: str, include_dependency_caches: bool = False) -> bool: + if any(seg in path for seg in SKIP_ALWAYS): + return False + return include_dependency_caches or not any(seg in path for seg in DEPENDENCY_CACHES) diff --git a/Discovery/adr_discovery/enumerator/sources/__init__.py b/Discovery/adr_discovery/enumerator/sources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/Discovery/adr_discovery/enumerator/sources/appstate.py b/Discovery/adr_discovery/enumerator/sources/appstate.py new file mode 100644 index 0000000..da8d770 --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/appstate.py @@ -0,0 +1,127 @@ +"""Application state -- per profile, rather than per default. + +Extensions and per-tool state live under directories a host application +maintains. The rule that matters here is *every* browser profile: a large +share of real shadow AI sits on a second profile, and a scan that reads +Default only reports a clean machine. +""" + +from __future__ import annotations + +import json + +from ...contracts.records import Candidate, Priority +from ..markers import ( + BROWSER_PROFILE_ROOTS, + CONFIG_FILE_TEMPLATES, + EDITOR_EXTENSION_ROOTS, + FIREFOX_PROFILE_ROOTS, + STATE_ROOTS, +) + + +def _expand(gate, template: str, homes: tuple[str, ...]) -> list[str]: + if not template.startswith("~"): + return [template] + return [home + template[1:] for home in homes] + + +def from_app_state(gate, homes: tuple[str, ...]) -> tuple[Candidate, ...]: + out: list[Candidate] = [] + + for template in CONFIG_FILE_TEMPLATES: + for path in _expand(gate, template, homes): + if gate.stat(path).ok: + kind = "shell_profile" if path.endswith(("/.bashrc", "/.zshrc")) else "marker_file" + out.append(Candidate(kind=kind, path=path, source="app_state:config", + priority=Priority.HOME, detail={"marker": path.rsplit("/", 1)[-1]})) + + for template in STATE_ROOTS: + for path in _expand(gate, template, homes): + listing = gate.list_dir(path) + if not listing.ok: + continue + out.append( + Candidate(kind="state_dir", path=path, source="app_state", + priority=Priority.HOME, detail={"entries": len(listing.value)}) + ) + + for template in BROWSER_PROFILE_ROOTS: + for browser_root in _expand(gate, template, homes): + profiles = gate.list_dir(browser_root) + if not profiles.ok: + continue + for profile in profiles.value: + if not profile.is_dir: + continue + name = profile.path.rsplit("/", 1)[-1] + if name != "Default" and not name.startswith("Profile "): + continue + ext_root = profile.path + "/Extensions" + extensions = gate.list_dir(ext_root) + if not extensions.ok: + continue + for ext in extensions.value: + if not ext.is_dir: + continue + out.append( + Candidate( + kind="extension", + path=ext.path, + source="app_state:browser", + priority=Priority.HOME, + detail={"extension_id": ext.path.rsplit("/", 1)[-1], + "profile": name, "browser": browser_root}, + ) + ) + + for template in EDITOR_EXTENSION_ROOTS: + for root in _expand(gate, template, homes): + extensions = gate.list_dir(root) + if not extensions.ok: + continue + for ext in extensions.value: + if not ext.is_dir: + continue + ident, version = _editor_identity(gate, ext.path) + out.append(Candidate( + kind="extension", path=ext.path, source="app_state:editor", + priority=Priority.HOME, + detail={"extension_id": ident, "version": version, "editor": root}, + )) + + for template in FIREFOX_PROFILE_ROOTS: + for root in _expand(gate, template, homes): + profiles = gate.list_dir(root) + if not profiles.ok: + continue + for profile in profiles.value: + if not profile.is_dir: + continue + extensions = gate.list_dir(profile.path + "/extensions") + if not extensions.ok: + continue + for ext in extensions.value: + if ext.is_dir or ext.path.endswith(".xpi"): + ident = ext.path.rsplit("/", 1)[-1].removesuffix(".xpi") + out.append(Candidate( + kind="extension", path=ext.path, source="app_state:firefox", + priority=Priority.HOME, + detail={"extension_id": ident, "profile": profile.path, + "browser": "firefox"}, + )) + return tuple(out) + + +def _editor_identity(gate, path: str) -> tuple[str, str | None]: + raw = gate.read_text(path + "/package.json", limit=1 << 20) + if raw.ok: + try: + manifest = json.loads(raw.value) + publisher, name = manifest.get("publisher"), manifest.get("name") + if publisher and name: + version = manifest.get("version") + return f"{publisher}.{name}", str(version) if version else None + except (ValueError, TypeError): + gate.ledger.probe("extension_manifest", "degraded", path) + return path.rsplit("/", 1)[-1], None diff --git a/Discovery/adr_discovery/enumerator/sources/binaries.py b/Discovery/adr_discovery/enumerator/sources/binaries.py new file mode 100644 index 0000000..1a3ca49 --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/binaries.py @@ -0,0 +1,70 @@ +"""Executables on disk. + +The registries answer for everything a package manager installed, and the +sweep answers for everything a marker locates. Neither answers for a bare +executable: a tarball unpacked into /opt, an AppImage in Downloads, a +binary copied out of a container. Those have no package record and sit +beside no marker, and without this source they are invisible. + +Bounded on purpose. Only bin-shaped directories are read, one level deep, +and the shared entry ceiling applies -- this must not become a second +filesystem sweep wearing a different name. +""" + +from __future__ import annotations + +from ...contracts.records import Candidate, Priority +from ..markers import BUNDLE_SUFFIXES + +#: Directories that hold executables by convention, plus whatever PATH says. +BIN_ROOTS: tuple[str, ...] = ( + "/usr/local/bin", "/usr/bin", "/bin", "/opt/homebrew/bin", "/opt/local/bin", + "/snap/bin", "~/.local/bin", "~/bin", "~/.cargo/bin", "~/go/bin", + "~/.npm-global/bin", "~/.bun/bin", "~/.deno/bin", +) + +MAX_PER_ROOT = 2_000 + + +def from_binaries(gate, homes: tuple[str, ...]) -> tuple[Candidate, ...]: + out: list[Candidate] = [] + seen: set[str] = set() + + for root in _roots(gate, homes): + listing = gate.list_dir(root) + if not listing.ok: + continue + kept = 0 + for entry in listing.value: + if kept >= MAX_PER_ROOT: + gate.ledger.truncate(root, kept, len(listing.value)) + break + if not (entry.is_exec or entry.path.endswith(BUNDLE_SUFFIXES)): + continue + if entry.path in seen: + continue + seen.add(entry.path) + kept += 1 + out.append( + Candidate( + kind="binary", + path=entry.path, + source="binaries", + priority=Priority.HOME, + detail={"name": entry.path.rsplit("/", 1)[-1], "symlink": entry.is_symlink}, + ) + ) + return tuple(out) + + +def _roots(gate, homes: tuple[str, ...]) -> list[str]: + roots: list[str] = [] + for template in BIN_ROOTS: + if template.startswith("~"): + roots.extend(home + template[1:] for home in homes) + else: + roots.append(template) + for entry in (gate.env.get("PATH") or "").split(":"): + if entry and entry not in roots: + roots.append(entry) + return roots diff --git a/Discovery/adr_discovery/enumerator/sources/execjournal.py b/Discovery/adr_discovery/enumerator/sources/execjournal.py new file mode 100644 index 0000000..252510f --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/execjournal.py @@ -0,0 +1,40 @@ +"""Exec events -- what ran between scans. + +A snapshot finds an agent that happens to be running when the scan fires. +An agent that runs forty seconds a night is absent from every daily scan +and present on the machine the whole time. + +This source is conditional on a privileged collector. Its absence is a +coverage fact and must never read as "nothing ran" -- which is what the +`unavailable` record written by the provider guarantees (U2-10). +""" + +from __future__ import annotations + +from ...contracts.records import Candidate, Priority +from ...redact.rules import scrub_argv + + +def from_exec_journal(gate) -> tuple[Candidate, ...]: + result = gate.exec_journal() + if not result.ok: + gate.ledger.probe("exec_journal", "degraded", result.reason) + return () + gate.ledger.probe("exec_journal", "ran", f"{len(result.value)} events") + out = [] + for ev in result.value: + argv = scrub_argv(ev.argv) + out.append(Candidate( + kind="exec_event", + path=ev.exe, + source="exec_journal", + priority=Priority.HOME, + detail={ + "argv": argv, "ppid": ev.ppid, + "parent_exe": ev.parent_exe, "started": ev.started, + "unattended": bool(set(argv) & { + "--dangerously-skip-permissions", "--yolo", "--auto-approve", "--no-confirm", + }), + }, + )) + return tuple(out) diff --git a/Discovery/adr_discovery/enumerator/sources/modelstores.py b/Discovery/adr_discovery/enumerator/sources/modelstores.py new file mode 100644 index 0000000..1ea2484 --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/modelstores.py @@ -0,0 +1,34 @@ +"""Local model stores that package and application registries do not index.""" + +from __future__ import annotations + +from ...contracts.records import Candidate, Priority + +MODEL_ROOTS = ( + "~/.ollama/models", "~/.cache/huggingface/hub", "~/.cache/lm-studio/models", + "~/Library/Application Support/LM Studio/models", "~/.local/share/Jan/models", + "~/.cache/gpt4all", +) +MODEL_SUFFIXES = (".gguf", ".safetensors") +MAX_MODELS = 500 + + +def from_model_stores(gate, homes: tuple[str, ...]) -> tuple[Candidate, ...]: + out: list[Candidate] = [] + for template in MODEL_ROOTS: + for home in homes: + root = home + template[1:] + if not gate.list_dir(root).ok: + continue + for entry in gate.walk(root, max_depth=4): + if entry.is_dir: + continue + name = entry.path.rsplit("/", 1)[-1] + if not (name.endswith(MODEL_SUFFIXES) or name.startswith("sha256-")): + continue + out.append(Candidate("model_weight_candidate", entry.path, "model_store", + Priority.HOME, {"name": name, "size": entry.size})) + if len(out) >= MAX_MODELS: + gate.ledger.truncate(root, len(out), len(out) + 1) + return tuple(out) + return tuple(out) diff --git a/Discovery/adr_discovery/enumerator/sources/network.py b/Discovery/adr_discovery/enumerator/sources/network.py new file mode 100644 index 0000000..d2d7910 --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/network.py @@ -0,0 +1,71 @@ +"""Network -- what the machine talks to. + +Listening sockets find a *server*. Almost every AI tool is a *client*, and +the connection it opens is the one piece of evidence it cannot suppress and +still function -- which makes this the only source that yields anything at +all for a tool the catalog has never heard of. + +The resolver cache matters more than the connection table, because it +covers a window rather than an instant and so survives a tool that ran an +hour before the scan. +""" + +from __future__ import annotations + +from ...contracts.records import Candidate, Priority +from ..markers import LOCAL_MODEL_PORTS, is_model_provider + + +def from_network(gate, kernel_candidates: tuple[Candidate, ...] = ()) -> tuple[Candidate, ...]: + out: list[Candidate] = [] + processes = { + c.detail.get("pid"): c + for c in kernel_candidates + if c.kind == "process" and c.detail.get("pid") is not None + } + + socks = gate.sockets() + if socks.ok: + for s in socks.value: + if s.state == "ESTABLISHED" and is_model_provider(s.remote_host): + process = processes.get(s.pid) + out.append( + Candidate( + kind="network_peer", + path=process.path if process is not None else s.remote_host, + source="network:established", + priority=Priority.HOME, + detail={ + "pid": s.pid, "port": s.remote_port, "provider": True, + "remote_host": s.remote_host, + "env_names": process.detail.get("env_names", ()) if process else (), + "unattended": process.detail.get("unattended", False) if process else False, + }, + ) + ) + elif s.state == "LISTEN" and s.local_port in LOCAL_MODEL_PORTS: + out.append( + Candidate( + kind="model_port", + path=f"tcp:{s.local_port}", + source="network:listening", + priority=Priority.HOME, + detail={"port": s.local_port, "pid": s.pid}, + ) + ) + + cache = gate.dns_cache() + if cache.ok: + gate.ledger.probe("dns_cache", "ran", f"{len(cache.value)} entries") + for entry in cache.value: + if is_model_provider(entry.hostname): + out.append( + Candidate( + kind="dns_peer", + path=entry.hostname, + source="network:resolver_cache", + priority=Priority.HOME, + detail={"provider": True}, + ) + ) + return tuple(out) diff --git a/Discovery/adr_discovery/enumerator/sources/registries.py b/Discovery/adr_discovery/enumerator/sources/registries.py new file mode 100644 index 0000000..0dcfd9c --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sources/registries.py @@ -0,0 +1,99 @@ +"""Ask the system first. + +Package databases, application registries and the kernel have already +catalogued most of what is installed, with provenance attached. Querying +them is cheaper and more complete than searching for it, and every hit +arrives with the provenance M4 needs anyway. + +Each function returns candidates and leaves a coverage record when its +surface could not be read -- an unavailable registry is never an empty one. +""" + +from __future__ import annotations + +from ...contracts.records import Candidate, Priority +from ...redact.rules import scrub_argv + + +def from_packages(gate) -> tuple[Candidate, ...]: + result = gate.packages() + if not result.ok: + return () + gate.ledger.probe("packages", "ran", f"{len(result.value)} records") + return tuple( + Candidate( + kind="package", + path=pkg.path or pkg.name, + source=f"package:{pkg.manager}", + priority=Priority.HOME, + detail={"manager": pkg.manager, "name": pkg.name, "version": pkg.version}, + ) + for pkg in result.value + ) + + +def from_applications(gate) -> tuple[Candidate, ...]: + result = gate.applications() + if not result.ok: + return () + gate.ledger.probe("applications", "ran", f"{len(result.value)} records") + return tuple( + Candidate( + kind="application", + path=app.path or app.ident, + source="app_registry", + priority=Priority.HOME, + detail={"ident": app.ident, "name": app.name, "version": app.version, "vendor": app.vendor}, + ) + for app in result.value + ) + + +def from_kernel(gate) -> tuple[Candidate, ...]: + """What is running, from which binary, and what it is serving. + + The exe path is carried through verbatim. Resolving a process *name* + against PATH is the defect this source exists to avoid. + """ + out: list[Candidate] = [] + procs = gate.processes() + if procs.ok: + gate.ledger.probe("processes", "ran", f"{len(procs.value)} pids") + for p in procs.value: + argv = scrub_argv(p.argv) + out.append( + Candidate( + kind="process", + path=p.exe, + source="kernel", + priority=Priority.HOME, + detail={ + "pid": p.pid, "ppid": p.ppid, "argv": argv, "cwd": p.cwd, + "user": p.user, "env_names": p.env_names, + "unattended": _is_unattended(argv), + }, + ) + ) + socks = gate.sockets() + if socks.ok: + gate.ledger.probe("sockets", "ran", f"{len(socks.value)} sockets") + for s in socks.value: + if s.state != "LISTEN": + continue + out.append( + Candidate( + kind="listening_socket", + path=f"tcp:{s.local_port}", + source="kernel", + priority=Priority.HOME, + detail={"port": s.local_port, "pid": s.pid}, + ) + ) + return tuple(out) + + +def _is_unattended(argv: tuple[str, ...]) -> bool: + flags = set(argv) + return bool(flags & { + "--dangerously-skip-permissions", "--yolo", "--auto-approve", "--no-confirm", + }) diff --git a/Discovery/adr_discovery/enumerator/sweep.py b/Discovery/adr_discovery/enumerator/sweep.py new file mode 100644 index 0000000..60a670e --- /dev/null +++ b/Discovery/adr_discovery/enumerator/sweep.py @@ -0,0 +1,51 @@ +"""Sweep only what no registry indexes. + +Repositories, agent directories and skill folders are found by traversal +keyed on markers, not on remembered paths. This is the only part of M2 that +can ruin the budget, so it carries the budget. +""" + +from __future__ import annotations + +from ..contracts.records import Candidate +from .markers import is_loose_executable, marker_kind +from .roots import in_scope, ordered_roots + + +def sweep(gate, include_dependency_caches: bool = False) -> tuple[Candidate, ...]: + """Breadth-ordered over priority roots, under one shared ceiling. + + Marker matching is a name test on entries already being listed, not a + second pass -- the walk is the cost, and this adds nothing to it. + """ + out: list[Candidate] = [] + seen: set[str] = set() + + for root, priority in ordered_roots(gate): + if gate.budget.entries_exhausted: + gate.ledger.boundary(root, "budget_exhausted", "root not swept") + continue + for entry in gate.walk(root): + if not in_scope(entry.path, include_dependency_caches): + continue + name = entry.path.rsplit("/", 1)[-1] + kind = marker_kind(name, entry.path) + if kind is None and is_loose_executable(entry, name): + # A tarball unpacked into /opt, a binary copied out of a + # container: no package record, no marker beside it. The + # walk is already listing this entry, so noticing costs + # nothing beyond the name test that follows it. + kind = "binary" + if kind is None or entry.path in seen: + continue + seen.add(entry.path) + out.append( + Candidate( + kind=kind, + path=entry.path, + source="sweep", + priority=priority, + detail={"marker": name, "is_dir": entry.is_dir, "name": name}, + ) + ) + return tuple(out) diff --git a/Discovery/adr_discovery/tests_unit/conftest.py b/Discovery/adr_discovery/tests_unit/conftest.py new file mode 100644 index 0000000..23bbbf0 --- /dev/null +++ b/Discovery/adr_discovery/tests_unit/conftest.py @@ -0,0 +1,113 @@ +"""Fixture worlds. + +A fixture directory and a live machine are interchangeable below M1, so +every case here builds a world on disk and drives the real pipeline over +it. Nothing is mocked; the gate reads these trees exactly as it reads `/`. +""" + +from __future__ import annotations + +import json +import os + +import pytest + +from adr_discovery.catalog.load import loads as load_catalog +from adr_discovery.coverage.ledger import Ledger +from adr_discovery.world.budget import Budget +from adr_discovery.world.gate import Gate +from adr_discovery.world.platform.base import FixtureProviders + +PACKAGE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + + +class World: + """A machine on disk, plus the non-filesystem surfaces beside it.""" + + def __init__(self, root: str) -> None: + self.root = root + self._restore: list[str] = [] + + def cleanup(self) -> None: + for path in self._restore: + try: + os.chmod(path, 0o755) + except OSError: + pass + + def file(self, path: str, body: str = "") -> "World": + full = self.root + path + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "w", encoding="utf-8") as fh: + fh.write(body) + return self + + def json(self, path: str, document) -> "World": + return self.file(path, json.dumps(document)) + + def binary(self, path: str, body: str) -> "World": + """Exact bytes, plus the executable bit. + + `exe` writes a shell wrapper of its own devising, which is fine for + version probes and useless for a case that asserts on a hash. + """ + self.file(path, body) + os.chmod(self.root + path, 0o755) + return self + + def exe(self, path: str, prints: str) -> "World": + self.file(path, f"#!/bin/sh\necho {prints!r}\n") + os.chmod(self.root + path, 0o755) + return self + + def dir(self, path: str) -> "World": + os.makedirs(self.root + path, exist_ok=True) + return self + + def unreadable(self, path: str) -> "World": + """Make a directory unreadable, and restore it at teardown. + + Without the restore, pytest cannot clean its own tmp tree and every + later run inherits the mess. + """ + os.chmod(self.root + path, 0o000) + self._restore.append(self.root + path) + return self + + def symlink(self, path: str, target: str, outside: bool = False) -> "World": + """`target` is a path *inside* this world unless `outside` is set. + + The distinction matters: a link out of the world is refused for + containment before the deny-list is ever consulted, so a case about + the deny-list has to stay inside. + """ + full = self.root + path + os.makedirs(os.path.dirname(full), exist_ok=True) + os.symlink(target if outside else self.root + target, full) + return self + + def surface(self, name: str, rows) -> "World": + """processes | sockets | packages | applications | dns | execjournal""" + return self.json(f"/{name}.json", rows) + + def gate(self, **kwargs) -> Gate: + kwargs.setdefault("ledger", Ledger()) + kwargs.setdefault("budget", Budget(max_entries=50_000)) + kwargs.setdefault("providers", FixtureProviders()) + kwargs.setdefault("env", {"PATH": "/usr/bin:/bin"}) + return Gate(self.root, **kwargs) + + +@pytest.fixture +def world(tmp_path): + built = World(str(tmp_path)) + try: + yield built + finally: + built.cleanup() + + +@pytest.fixture(scope="session") +def catalog(): + with open(os.path.join(PACKAGE, "catalog", "catalog.json"), encoding="utf-8") as fh: + return load_catalog(fh.read()) diff --git a/Discovery/adr_discovery/tests_unit/test_m1_world.py b/Discovery/adr_discovery/tests_unit/test_m1_world.py new file mode 100644 index 0000000..8f4ba72 --- /dev/null +++ b/Discovery/adr_discovery/tests_unit/test_m1_world.py @@ -0,0 +1,183 @@ +"""M1 -- world access. + +The assertion surface is the return shape itself: every access returns data +or a recorded reason. A test that only checks the happy path checks nothing +this module exists for, so each case asserts on both halves -- what came +back, and what was written to the ledger. +""" + +from __future__ import annotations + +import os + +from adr_discovery.world.budget import Budget + + +def test_u1_01_symlink_escape_is_refused(world): + world.dir("/proj").symlink("/proj/escape.json", "/etc/passwd", outside=True) + gate = world.gate() + + result = gate.read_text("/proj/escape.json") + + assert not result.ok + assert result.reason == "outside_root" + assert any(d.reason == "outside_root" for d in gate.ledger.freeze().denied) + + +def test_u1_02_containment_is_decided_on_the_resolved_target(world): + world.file("/proj/ok.json", "{}") + gate = world.gate() + + assert not gate.read_text("/proj/../../../etc/hosts").ok + + +def test_u1_03_deny_list_follows_the_symlink(world): + world.file("/home/a/Documents/diary.txt", "private") + world.symlink("/home/a/notes", "/home/a/Documents") + gate = world.gate() + + result = gate.read_text("/home/a/notes/diary.txt") + + assert not result.ok and result.reason == "personal_path" + + +def test_u1_04_a_swap_between_check_and_open_is_refused(world): + world.file("/race/target", "real") + gate = world.gate() + + def swap(resolved: str) -> None: + if resolved.endswith("/target"): + os.remove(resolved) + os.symlink("/etc/hosts", resolved) + + gate.on_validated = swap + result = gate.read_text("/race/target") + + assert not result.ok and result.reason == "swapped" + + +def test_u1_05_truncation_reports_the_true_size(world): + world.file("/big.bin", "x" * 5000) + gate = world.gate(budget=Budget(max_read_bytes=1000)) + + result = gate.read_bytes("/big.bin") + + assert result.ok and len(result.value) == 1000 + (record,) = gate.ledger.freeze().truncated + assert record.kept == 1000 and record.true_count == 5000 + + +def test_u1_06_depth_cap_is_recorded_with_the_path(world): + path = "" + for i in range(10): + path += f"/d{i}" + world.dir(path) + gate = world.gate(budget=Budget(max_depth=4)) + + list(gate.walk("/d0")) + boundaries = gate.ledger.freeze().boundaries_hit + + assert any(b.boundary == "depth" for b in boundaries) + + +def test_u1_07_denied_is_not_empty(world): + world.dir("/locked").unreadable("/locked") + gate = world.gate() + + result = gate.list_dir("/locked") + + assert not result.ok + assert gate.ledger.freeze().denied, "an unreadable surface must be recorded, not silently empty" + + +def test_u1_07b_absent_is_distinct_from_denied(world): + """A surface that does not exist was not refused. + + Recording it as a denial would drown the real refusals and make + `coverage.is_complete` meaningless. + """ + gate = world.gate() + + result = gate.list_dir("/nothing/here") + + assert not result.ok and result.reason == "absent" + assert not gate.ledger.freeze().denied + + +def test_u1_08_subprocess_timeout_is_a_refusal_not_an_answer(world): + world.file("/slow", "#!/bin/sh\nsleep 5\necho 1.2.3\n") + os.chmod(world.root + "/slow", 0o755) + gate = world.gate(budget=Budget(max_subprocess_seconds=0.2)) + + result = gate.run_helper(("/slow",)) + + assert not result.ok and result.reason == "timeout" + assert any(p.status == "failed" for p in gate.ledger.freeze().probes) + + +def test_u1_08b_helpers_never_search_path(world): + world.exe("/attacker/bin/ps", "owned") + gate = world.gate(env={"PATH": world.root + "/attacker/bin"}) + + result = gate.run_helper(("ps", "--version")) + + assert not result.ok and result.reason == "helper_not_absolute" + + +def test_u1_08c_subprocess_output_is_bounded(world): + world.file("/loud", "#!/bin/sh\nhead -c 9000 /dev/zero\n") + os.chmod(world.root + "/loud", 0o755) + gate = world.gate(budget=Budget(max_subprocess_output_bytes=1024)) + + result = gate.run_helper(("/loud",)) + + assert not result.ok and result.reason == "output_limit" + + +def test_u1_08d_directory_listing_spends_the_budget_while_iterating(world): + for index in range(10): + world.file(f"/wide/{index}", "") + gate = world.gate(budget=Budget(max_entries=3)) + + result = gate.list_dir("/wide") + + assert result.ok and len(result.value) == 3 + assert gate.budget.entries_used == 3 + assert any(b.boundary == "budget_exhausted" for b in gate.ledger.freeze().boundaries_hit) + + +def test_u1_08e_an_ancestor_swap_is_refused(world): + world.file("/race/dir/target", "public") + world.file("/home/a/Documents/target", "private") + gate = world.gate() + + def swap_ancestor(resolved: str) -> None: + if resolved.endswith("/race/dir/target"): + os.rename(world.root + "/race/dir", world.root + "/race/original") + os.symlink(world.root + "/home/a/Documents", world.root + "/race/dir") + + gate.on_validated = swap_ancestor + result = gate.read_text("/race/dir/target") + + assert not result.ok and result.reason == "swapped" + + +def test_u1_09_a_missing_provider_is_unavailable_not_empty(world): + gate = world.gate() + + result = gate.packages() + + assert not result.ok + assert [u.provider for u in gate.ledger.freeze().unavailable] == ["packages"] + + +def test_u1_10_the_exe_is_reported_not_the_name(world): + """`ps comm=` truncates at fifteen characters; resolving that name + against PATH attributes /opt/agents/claude to /usr/bin/claude.""" + world.exe("/usr/bin/claude", "wrong") + world.surface("processes", [{"pid": 4021, "exe": "/opt/agents/claude-code-cli"}]) + gate = world.gate() + + (process,) = gate.processes().value + + assert process.exe == "/opt/agents/claude-code-cli" diff --git a/Discovery/adr_discovery/tests_unit/test_m2_enumerator.py b/Discovery/adr_discovery/tests_unit/test_m2_enumerator.py new file mode 100644 index 0000000..caf820e --- /dev/null +++ b/Discovery/adr_discovery/tests_unit/test_m2_enumerator.py @@ -0,0 +1,168 @@ +"""M2 -- enumerator. + +Run without a catalog. That is the contract: a candidate set that changes +when the catalog changes is not an enumerator, it is a lookup. +""" + +from __future__ import annotations + +from adr_discovery.catalog.load import EMPTY +from adr_discovery.enumerator import enumerate_candidates +from adr_discovery.world.budget import Budget + + +def paths(candidates, kind=None): + return {c.path for c in candidates if kind is None or c.kind == kind} + + +def test_u2_01_a_repo_outside_every_known_root_is_found(world): + world.file("/opt/checkouts/svc/.git/HEAD", "ref: refs/heads/main") + world.file("/opt/checkouts/svc/.mcp.json", "{}") + gate = world.gate() + + found = enumerate_candidates(gate) + + assert "/opt/checkouts/svc/.git" in paths(found) + assert "/opt/checkouts/svc/.mcp.json" in paths(found) + + +def test_u2_02_a_marker_one_level_deeper_is_found(world): + world.file("/Users/alice/work/team/proj/.claude/settings.json", "{}") + gate = world.gate() + + assert any(p.endswith("proj/.claude") for p in paths(enumerate_candidates(gate))) + + +def test_u2_03_enumeration_does_not_consult_the_catalog(world): + """The load-bearing case. Any change that makes this fail has put + identification back inside enumeration.""" + world.file("/Users/alice/proj/.claude/settings.json", "{}") + world.file("/Users/alice/proj/.mcp.json", "{}") + + with_catalog = enumerate_candidates(world.gate()) + without = enumerate_candidates(world.gate()) + + assert paths(with_catalog) == paths(without) + assert len(EMPTY) == 0 + + +def test_u2_04_budget_exhaustion_is_reported(world): + for i in range(300): + world.file(f"/Users/alice/proj/f{i}.txt", "x") + gate = world.gate(budget=Budget(max_entries=50)) + + enumerate_candidates(gate) + + assert any(b.boundary == "budget_exhausted" for b in gate.ledger.freeze().boundaries_hit) + + +def test_u2_05_one_budget_is_shared_across_roots(world): + for root in ("/Users/alice/src", "/Users/alice/work", "/opt"): + for i in range(80): + world.file(f"{root}/p{i}/f.txt", "x") + gate = world.gate(budget=Budget(max_entries=100)) + + enumerate_candidates(gate) + + assert gate.budget.entries_used <= 100, "each root must not get its own ceiling" + + +def test_u2_06_home_is_swept_before_breadth(world): + world.file("/Users/alice/.claude/settings.json", "{}") + world.file("/opt/thing/.claude/settings.json", "{}") + gate = world.gate() + + swept = [c.path for c in enumerate_candidates(gate) if c.source == "sweep"] + home = next(i for i, p in enumerate(swept) if p.startswith("/Users/alice/.")) + system = next(i for i, p in enumerate(swept) if p.startswith("/opt/")) + + assert home < system + + +def test_u2_08_an_outbound_connection_is_a_candidate(world): + world.surface("sockets", [ + {"proto": "tcp", "state": "ESTABLISHED", "remote_host": "api.anthropic.com", + "remote_port": 443, "pid": 8812}, + ]) + gate = world.gate() + + found = enumerate_candidates(gate) + peers = [c for c in found if c.kind == "network_peer"] + + assert [p.path for p in peers] == ["api.anthropic.com"] + assert peers[0].detail["pid"] == 8812 + + +def test_u2_09_the_resolver_cache_covers_a_window_not_an_instant(world): + world.surface("dns", [{"hostname": "api.openai.com"}, {"hostname": "example.com"}]) + gate = world.gate() + + peers = [c.path for c in enumerate_candidates(gate) if c.kind == "dns_peer"] + + assert peers == ["api.openai.com"], "a tool that ran an hour ago must still be visible" + + +def test_u2_10_an_absent_journal_is_unavailable_not_empty(world): + gate = world.gate() + + enumerate_candidates(gate) + coverage = gate.ledger.freeze() + + assert "exec_journal" in [u.provider for u in coverage.unavailable] + assert any(p.name == "exec_journal" and p.status == "degraded" for p in coverage.probes) + + +def test_u2_11_a_short_lived_run_survives_in_the_journal(world): + world.surface("execjournal", [ + {"exe": "/opt/agents/nightly", "argv": ["nightly", "-p"], "ppid": 1, + "parent_exe": "/usr/sbin/cron", "started": "2026-08-22T03:12:00Z"}, + ]) + gate = world.gate() + + events = [c for c in enumerate_candidates(gate) if c.kind == "exec_event"] + + assert len(events) == 1 + assert events[0].detail["argv"] == ("nightly", "-p") + assert events[0].detail["parent_exe"] == "/usr/sbin/cron" + + +def test_u2_12_an_instruction_file_is_a_programmable_surface(world): + world.file("/Users/alice/proj/CLAUDE.md", "# steering prose") + gate = world.gate() + + found = [c for c in enumerate_candidates(gate) if c.path.endswith("CLAUDE.md")] + + assert [c.kind for c in found] == ["instruction_file"] + + +def test_critical_host_configs_do_not_depend_on_the_sweep_budget(world): + world.file("/Users/alice/.claude.json", '{"mcpServers": {}}') + world.file("/etc/adr/managed-mcp.json", '{"mcpServers": {}}') + gate = world.gate(budget=Budget(max_entries=1)) + + found = enumerate_candidates(gate) + + assert "/Users/alice/.claude.json" in paths(found, "marker_file") + assert "/etc/adr/managed-mcp.json" in paths(found, "marker_file") + + +def test_u2_07_registries_answer_before_the_sweep_spends_anything(world): + """Most of the search is over before it starts. + + A binary the package database already lists must not cost sweep + entries to locate -- the ordering is the optimisation, and without an + assertion it is only an intention. + """ + world.dir("/Users/alice") + world.binary("/opt/homebrew/bin/claude", "#!/bin/sh\necho 2.1.234\n") + world.surface("packages", [{"manager": "npm", "name": "@anthropic-ai/claude-code", + "version": "2.1.234", "path": "/opt/homebrew/bin/claude"}]) + gate = world.gate() + + from adr_discovery.enumerator.sources.registries import from_packages + + from_registry = from_packages(gate) + spent_on_registries = gate.budget.entries_used + + assert [c.path for c in from_registry] == ["/opt/homebrew/bin/claude"] + assert spent_on_registries == 0, "querying an index must not consume the sweep ceiling" diff --git a/Discovery/adr_discovery/world/__init__.py b/Discovery/adr_discovery/world/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/Discovery/adr_discovery/world/budget.py b/Discovery/adr_discovery/world/budget.py new file mode 100644 index 0000000..50e154e --- /dev/null +++ b/Discovery/adr_discovery/world/budget.py @@ -0,0 +1,52 @@ +"""Budgets live in M1, not in callers. + +A caller cannot forget a limit it does not set. One ceiling is shared by +the whole scan rather than one per probe, because five private budgets +cannot be reasoned about and, in aggregate, are not a budget at all. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass(slots=True) +class Budget: + """Ceilings for one scan. Mutable only in the consumed counters.""" + + max_read_bytes: int = 4 * 1024 * 1024 + max_entries: int = 200_000 + max_depth: int = 12 + max_subprocess_seconds: float = 5.0 + max_subprocess_output_bytes: int = 4 * 1024 * 1024 + max_strings_bytes: int = 512 * 1024 + #: The whole scan, wall clock. Plane A promises seconds, and a promise + #: with no ceiling behind it is a hope. + max_seconds: float = 120.0 + + entries_used: int = field(default=0, init=False) + started: float = field(default_factory=time.monotonic, init=False) + + def take_entries(self, n: int = 1) -> bool: + """Consume from the shared entry ceiling. False once exhausted.""" + if self.entries_used >= self.max_entries: + return False + self.entries_used += n + return True + + @property + def entries_exhausted(self) -> bool: + return self.entries_used >= self.max_entries + + @property + def seconds_used(self) -> float: + return time.monotonic() - self.started + + @property + def time_exhausted(self) -> bool: + return self.seconds_used >= self.max_seconds + + @property + def entries_remaining(self) -> int: + return max(0, self.max_entries - self.entries_used) diff --git a/Discovery/adr_discovery/world/gate.py b/Discovery/adr_discovery/world/gate.py new file mode 100644 index 0000000..cca2cc4 --- /dev/null +++ b/Discovery/adr_discovery/world/gate.py @@ -0,0 +1,472 @@ +"""M1 -- the single door between the collector and the machine. + +Everything else in this design is testable only because this module +exists: below the gate, a fixture directory and a live machine are +interchangeable, which is what lets the whole pipeline be graded. + +Order is fixed and is the point: + + canonicalize -> contain -> deny-list -> open -> verify the descriptor -> read under budget + +Containment and denial are decided on the *resolved* target rather than the +path handed in, because a permitted config can be a symlink into a personal +directory, and a relative segment inside a config can climb out of the tree. +""" + +from __future__ import annotations + +import errno +import os +import signal +import subprocess +import threading +import time +from collections import deque +from dataclasses import dataclass +from typing import Generic, Iterator, TypeVar + +from ..coverage.ledger import Ledger +from ..redact import rules as redact +from .budget import Budget +from .platform.base import Providers + +T = TypeVar("T") +_HELPER_PATH = "/usr/bin:/bin:/usr/sbin:/sbin" + + +@dataclass(frozen=True, slots=True) +class Ok(Generic[T]): + value: T + + ok: bool = True + + +@dataclass(frozen=True, slots=True) +class Refused: + """Never an empty result that could equally mean 'nothing there' and + 'could not look'. The reason is recorded where it is still known.""" + + reason: str + detail: str = "" + + ok: bool = False + + +Result = Ok[T] | Refused + + +@dataclass(frozen=True, slots=True) +class Entry: + path: str + is_dir: bool + is_symlink: bool + size: int + is_exec: bool = False + + +@dataclass(frozen=True, slots=True) +class Stat: + path: str + real_path: str + inode: str + size: int + mode: int + mtime: float + owner: str + + +@dataclass(frozen=True, slots=True) +class Ran: + argv: tuple[str, ...] + code: int + stdout: str + + +class Gate: + """The only object in the package that touches the host.""" + + def __init__( + self, + root: str = "/", + *, + ledger: Ledger | None = None, + budget: Budget | None = None, + providers: Providers | None = None, + env: dict[str, str] | None = None, + ) -> None: + self._root = os.path.realpath(root) + self.ledger = ledger if ledger is not None else Ledger() + self.budget = budget if budget is not None else Budget() + self.env = dict(env) if env is not None else {} + from .platform.base import NullProviders + + self.providers = providers if providers is not None else NullProviders() + #: Call counters, so a test can assert the evidence ladder stopped + #: early rather than merely reaching the right answer expensively. + self.calls: dict[str, int] = {"read_bytes": 0, "run": 0, "package_owner": 0} + #: Test hook. Called with the resolved path immediately after + #: validation and before open(), so a case can swap the target and + #: prove the descriptor check catches it. + self.on_validated = None + + # ---------------------------------------------------------------- paths + + @property + def root(self) -> str: + return self._root + + def host_path(self, logical: str) -> str: + """Map a logical (machine) path into this world.""" + if self._root == "/": + return logical + return os.path.join(self._root, logical.lstrip("/\\")) + + def logical_path(self, host: str) -> str: + """Inverse of `host_path`, so records read the same in a fixture and + on a real machine.""" + if self._root == "/": + return host + if host == self._root: + return "/" + if host.startswith(self._root + os.sep): + return "/" + host[len(self._root) + 1 :] + return host + + def _validate(self, logical: str) -> Result[str]: + """Canonicalize, then decide. Returns the resolved host path.""" + candidate = self.host_path(logical) + try: + resolved = os.path.realpath(candidate) + except (OSError, ValueError) as exc: + return Refused("unresolvable", str(exc)) + + if self._root != "/" and not ( + resolved == self._root or resolved.startswith(self._root + os.sep) + ): + self.ledger.deny(logical, "outside_root") + return Refused("outside_root", self.logical_path(resolved)) + + if redact.is_personal(self.logical_path(resolved)): + self.ledger.deny(logical, "personal_path") + return Refused("personal_path", "") + + return Ok(resolved) + + # ---------------------------------------------------------------- reads + + def _open_verified(self, logical: str, resolved: str, flags: int = os.O_RDONLY) -> Result[int]: + """Open a target, then prove the descriptor still names the validated path. + + Revalidating after open closes the ancestor-swap race: checking only + the final component with ``O_NOFOLLOW`` is insufficient when an + attacker controls a directory above it. + """ + if self.on_validated is not None: + self.on_validated(resolved) + + try: + fd = os.open(resolved, flags | getattr(os, "O_NOFOLLOW", 0)) + except FileNotFoundError: + return Refused("absent", resolved) + except OSError as exc: + if exc.errno == errno.ELOOP: + return Refused("swapped", "target became a symlink after validation") + return Refused("open_failed", exc.strerror or str(exc)) + + current = self._validate(logical) + try: + descriptor = os.fstat(fd) + path_stat = os.stat(resolved, follow_symlinks=False) + except OSError as exc: + os.close(fd) + return Refused("swapped", exc.strerror or str(exc)) + if ( + not current.ok + or current.value != resolved + or (descriptor.st_dev, descriptor.st_ino) != (path_stat.st_dev, path_stat.st_ino) + ): + os.close(fd) + return Refused("swapped", "descriptor did not match the revalidated target") + return Ok(fd) + + def read_bytes(self, logical: str, limit: int | None = None) -> Result[bytes]: + self.calls["read_bytes"] += 1 + validated = self._validate(logical) + if not validated.ok: + return validated + resolved = validated.value + ceiling = self.budget.max_read_bytes if limit is None else min(limit, self.budget.max_read_bytes) + + opened = self._open_verified(logical, resolved) + if not opened.ok: + if opened.reason in ("open_failed", "stat_failed"): + self.ledger.deny(logical, opened.detail or opened.reason) + elif opened.reason == "swapped": + self.ledger.deny(logical, "target swapped after validation") + return opened + fd = opened.value + try: + size = os.fstat(fd).st_size + data = os.read(fd, ceiling) + finally: + os.close(fd) + + if size > ceiling: + self.ledger.truncate(logical, len(data), size) + return Ok(data) + return Ok(data) + + def read_text(self, logical: str, limit: int | None = None) -> Result[str]: + raw = self.read_bytes(logical, limit) + if not raw.ok: + return raw + return Ok(raw.value.decode("utf-8", errors="replace")) + + def stat(self, logical: str) -> Result[Stat]: + validated = self._validate(logical) + if not validated.ok: + return validated + resolved = validated.value + try: + st = os.stat(resolved) + except FileNotFoundError: + return Refused("absent", logical) + except OSError as exc: + self.ledger.deny(logical, exc.strerror or str(exc)) + return Refused("stat_failed", exc.strerror or str(exc)) + current = self._validate(logical) + if not current.ok or current.value != resolved: + self.ledger.deny(logical, "target swapped after validation") + return Refused("swapped", "path changed during stat") + try: + after = os.stat(resolved) + except OSError as exc: + return Refused("swapped", exc.strerror or str(exc)) + if (st.st_dev, st.st_ino) != (after.st_dev, after.st_ino): + self.ledger.deny(logical, "target swapped after validation") + return Refused("swapped", "path changed during stat") + return Ok( + Stat( + path=logical, + real_path=self.logical_path(resolved), + inode=f"{st.st_dev}:{st.st_ino}", + size=st.st_size, + mode=st.st_mode, + mtime=st.st_mtime, + owner=self.providers.owner_of(st.st_uid), + ) + ) + + def list_dir(self, logical: str) -> Result[tuple[Entry, ...]]: + validated = self._validate(logical) + if not validated.ok: + return validated + resolved = validated.value + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + opened = self._open_verified(logical, resolved, directory_flags) + if not opened.ok: + if opened.reason not in ("absent",): + self.ledger.deny(logical, opened.detail or opened.reason) + return opened + + fd = opened.value + entries: list[Entry] = [] + exhausted = False + try: + with os.scandir(fd) as listing: + for e in listing: + if not self.budget.take_entries(): + exhausted = True + break + try: + info = e.stat(follow_symlinks=False) + is_dir = e.is_dir(follow_symlinks=False) + entries.append( + Entry( + path=os.path.join(logical, e.name) if logical != "/" else "/" + e.name, + is_dir=is_dir, + is_symlink=e.is_symlink(), + size=info.st_size, + is_exec=bool(info.st_mode & 0o111) and not is_dir, + ) + ) + except OSError: + continue + except (OSError, TypeError) as exc: + self.ledger.deny(logical, getattr(exc, "strerror", None) or str(exc)) + return Refused("listdir_failed", str(exc)) + finally: + os.close(fd) + + if exhausted: + self.ledger.boundary( + logical, "budget_exhausted", f"cap {self.budget.max_entries}; directory truncated" + ) + entries.sort(key=lambda entry: entry.path) + return Ok(tuple(entries)) + + def walk(self, logical_root: str, max_depth: int | None = None) -> Iterator[Entry]: + """Breadth-first under the shared entry ceiling. + + Breadth-first on purpose: a budget exhausted late still covered the + likely places, which is not true of a depth-first walk that spends + everything in the first deep subtree it meets. + """ + depth_cap = self.budget.max_depth if max_depth is None else max_depth + frontier = deque([(logical_root, 0)]) + seen: set[str] = set() + deepest = 0 + count = 0 + + while frontier: + path, depth = frontier.popleft() + if depth > depth_cap: + self.ledger.boundary(path, "depth", f"cap {depth_cap}") + continue + if self.budget.time_exhausted: + self.ledger.boundary(path, "time_exhausted", f"cap {self.budget.max_seconds}s") + self.ledger.swept(logical_root, deepest, count) + return + listing = self.list_dir(path) + if not listing.ok: + continue + deepest = max(deepest, depth) + for entry in listing.value: + count += 1 + yield entry + if entry.is_dir and not entry.is_symlink and entry.path not in seen: + seen.add(entry.path) + frontier.append((entry.path, depth + 1)) + + self.ledger.swept(logical_root, deepest, count) + + # ------------------------------------------------------------ processes + + def processes(self): + return self.providers.processes(self) + + def sockets(self): + return self.providers.sockets(self) + + def packages(self): + return self.providers.packages(self) + + def applications(self): + return self.providers.applications(self) + + def dns_cache(self): + return self.providers.dns_cache(self) + + def exec_journal(self): + return self.providers.exec_journal(self) + + def package_owner(self, path: str): + self.calls["package_owner"] += 1 + return self.providers.package_owner(self, path) + + # ----------------------------------------------------------- subprocess + + def run_helper(self, argv: tuple[str, ...], timeout: float | None = None) -> Result[Ran]: + """Run one absolute-path OS inventory helper with bounded output. + + This API is deliberately unavailable to discovered candidates. It + does not search ``PATH`` and drains output incrementally, so neither + path precedence nor post-exit slicing can turn inventory into code + execution or an unbounded allocation. + """ + self.calls["run"] += 1 + if not argv: + return Refused("no_argv", "") + if not os.path.isabs(argv[0]): + return Refused("helper_not_absolute", argv[0]) + if self.budget.time_exhausted: + self.ledger.probe(argv[0], "failed", "scan time budget exhausted") + return Refused("time_budget_exhausted", f"{self.budget.max_seconds}s") + + target = argv[0] + if self._root != "/" and target.startswith("/"): + validated = self._validate(target) + if not validated.ok: + return validated + target = validated.value + + limit = self.budget.max_subprocess_seconds if timeout is None else timeout + env = {"PATH": _HELPER_PATH, "LC_ALL": "C", "LANG": "C"} + if os.name == "nt": + env.update({"SystemRoot": r"C:\Windows", "WINDIR": r"C:\Windows"}) + try: + proc = subprocess.Popen( + [target, *argv[1:]], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + start_new_session=os.name != "nt", + ) + except (OSError, ValueError) as exc: + self.ledger.probe(argv[0] if argv else "?", "failed", str(exc)) + return Refused("spawn_failed", str(exc)) + + stdout = bytearray() + output_limit = self.budget.max_subprocess_output_bytes + total = [0] + lock = threading.Lock() + overflow = threading.Event() + + def drain(stream, keep: bool) -> None: + while True: + chunk = stream.read(4096) + if not chunk: + return + with lock: + total[0] += len(chunk) + if keep and len(stdout) < output_limit: + remaining = output_limit - len(stdout) + stdout.extend(chunk[:remaining]) + if total[0] > output_limit: + overflow.set() + + threads = [ + threading.Thread(target=drain, args=(proc.stdout, True), daemon=True), + threading.Thread(target=drain, args=(proc.stderr, False), daemon=True), + ] + for thread in threads: + thread.start() + + deadline = time.monotonic() + limit + reason = None + while proc.poll() is None: + if overflow.wait(timeout=min(0.02, max(0.0, deadline - time.monotonic()))): + reason = "output_limit" + break + if time.monotonic() >= deadline: + reason = "timeout" + break + + if reason is not None: + self._terminate(proc) + else: + proc.wait() + for stream in (proc.stdout, proc.stderr): + stream.close() + for thread in threads: + thread.join(timeout=0.2) + + if reason is not None: + detail = f"{output_limit} bytes" if reason == "output_limit" else f"{limit}s" + self.ledger.probe(argv[0], "failed", reason) + return Refused(reason, detail) + return Ok(Ran(tuple(argv), proc.returncode, stdout.decode("utf-8", "replace"))) + + @staticmethod + def _terminate(proc: subprocess.Popen) -> None: + try: + if os.name != "nt": + os.killpg(proc.pid, signal.SIGKILL) + else: + proc.kill() + except (OSError, ProcessLookupError): + pass + try: + proc.wait(timeout=1) + except subprocess.TimeoutExpired: + pass diff --git a/Discovery/adr_discovery/world/platform/__init__.py b/Discovery/adr_discovery/world/platform/__init__.py new file mode 100644 index 0000000..a7c6446 --- /dev/null +++ b/Discovery/adr_discovery/world/platform/__init__.py @@ -0,0 +1,24 @@ +"""Platform selection. The only place an OS difference may exist.""" + +from __future__ import annotations + +import sys + +from .base import FixtureProviders as FixtureProviders +from .base import NullProviders, Providers + + +def for_host() -> Providers: + if sys.platform == "darwin": + from .darwin import DarwinProviders + + return DarwinProviders() + if sys.platform.startswith("linux"): + from .linux import LinuxProviders + + return LinuxProviders() + if sys.platform.startswith("win"): + from .windows import WindowsProviders + + return WindowsProviders() + return NullProviders() diff --git a/Discovery/adr_discovery/world/platform/base.py b/Discovery/adr_discovery/world/platform/base.py new file mode 100644 index 0000000..dea12f8 --- /dev/null +++ b/Discovery/adr_discovery/world/platform/base.py @@ -0,0 +1,342 @@ +"""Provider shapes, and the two providers that need no OS at all. + +The only place an OS difference may exist is under this package. Everything +above it sees these dataclasses and nothing else, which is what makes a +fixture world and a live machine interchangeable. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Protocol + +if TYPE_CHECKING: # pragma: no cover - typing only + from ..gate import Gate, Result + + +@dataclass(frozen=True, slots=True) +class Process: + pid: int + exe: str + """The link target of /proc//exe, or its platform equivalent. + + Never a name. `ps comm=` truncates at fifteen characters on Linux, and + resolving that name against PATH attributes /opt/agents/claude to + /usr/bin/claude -- a different binary. + """ + argv: tuple[str, ...] = () + ppid: int = 0 + cwd: str | None = None + user: str = "system" + env_names: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class Socket: + proto: str + state: str # LISTEN | ESTABLISHED + local_port: int = 0 + remote_host: str = "" + remote_port: int = 0 + pid: int | None = None + + +@dataclass(frozen=True, slots=True) +class Package: + manager: str + name: str + version: str | None = None + path: str | None = None + + +@dataclass(frozen=True, slots=True) +class Application: + ident: str + name: str + version: str | None = None + vendor: str | None = None + path: str | None = None + + +@dataclass(frozen=True, slots=True) +class DnsEntry: + hostname: str + + +@dataclass(frozen=True, slots=True) +class ExecEvent: + """What ran between scans. Absent unless a privileged collector supplies + it -- and its absence is a coverage fact, never an empty set.""" + + exe: str + argv: tuple[str, ...] = () + ppid: int = 0 + parent_exe: str | None = None + started: str = "" + + +class Providers(Protocol): + def home_roots(self) -> tuple[str, ...]: ... + def owner_of(self, uid: int) -> str: ... + def processes(self, gate: "Gate") -> "Result": ... + def sockets(self, gate: "Gate") -> "Result": ... + def packages(self, gate: "Gate") -> "Result": ... + def applications(self, gate: "Gate") -> "Result": ... + def dns_cache(self, gate: "Gate") -> "Result": ... + def exec_journal(self, gate: "Gate") -> "Result": ... + def package_owner(self, gate: "Gate", path: str) -> "Result": ... + + +class NullProviders: + """Every query is unavailable, with a reason. + + This is the correct provider for a platform nobody has implemented yet: + it makes the gap appear in coverage instead of making the machine look + clean, which is the difference the whole design turns on. + """ + + reason = "no provider for this platform" + + #: Where user homes live. A platform question, and therefore not one + #: M2 may answer: on macOS `/home` is an autofs mount that blocks a + #: lister for as long as automountd feels like it, and a scan that + #: hangs there is indistinguishable from a scan that found nothing. + HOME_ROOTS: tuple[str, ...] = ("/Users", "/home") + + def home_roots(self) -> tuple[str, ...]: + return self.HOME_ROOTS + + def owner_of(self, uid: int) -> str: + return "system" + + def _unavailable(self, gate: "Gate", name: str): + from ..gate import Refused + + gate.ledger.unavailable(name, self.reason) + return Refused("unavailable", self.reason) + + def processes(self, gate): + return self._unavailable(gate, "processes") + + def sockets(self, gate): + return self._unavailable(gate, "sockets") + + def packages(self, gate): + return self._unavailable(gate, "packages") + + def applications(self, gate): + return self._unavailable(gate, "applications") + + def dns_cache(self, gate): + return self._unavailable(gate, "dns_cache") + + def exec_journal(self, gate): + return self._unavailable(gate, "exec_journal") + + def package_owner(self, gate, path): + return self._unavailable(gate, "package_owner") + + +class LanguagePackages: + """Package databases that are the same on every platform. + + npm, pipx, uv, cargo and go keep their manifests on disk in the same + shape everywhere, so reading them is not an OS difference and does not + belong in a per-platform provider. Every read goes through the gate, + so a fixture tree answers these exactly as a real machine does. + """ + + NODE_ROOTS = ( + "/usr/local/lib/node_modules", "/opt/homebrew/lib/node_modules", + "/usr/lib/node_modules", "~/.npm-global/lib/node_modules", + "~/.nvm/versions/node", "~/node_modules", + ) + PIPX_ROOTS = ("~/.local/pipx/venvs", "~/.local/share/pipx/venvs") + UV_ROOTS = ("~/.local/share/uv/tools",) + BIN_ROOTS = ("~/.cargo/bin", "~/go/bin", "~/.local/bin") + + def collect(self, gate, homes: tuple[str, ...]) -> list[Package]: + found: list[Package] = [] + found.extend(self._npm(gate, homes)) + found.extend(self._venvs(gate, homes, self.PIPX_ROOTS, "pipx")) + found.extend(self._venvs(gate, homes, self.UV_ROOTS, "uv")) + found.extend(self._cargo(gate, homes)) + found.extend(self._go(gate, homes)) + return found + + def _cargo(self, gate, homes) -> list[Package]: + out: list[Package] = [] + for home in homes: + raw = gate.read_text(home + "/.cargo/.crates2.json", limit=4 << 20) + if not raw.ok: + continue + try: + document = json.loads(raw.value) + except ValueError: + gate.ledger.probe("cargo", "degraded", home + "/.cargo/.crates2.json") + continue + for key in (document.get("installs") or {}): + # Current Cargo uses "name version (source)" as the key. + parts = key.split() + if parts: + name = parts[0] + version = parts[1] if len(parts) > 1 else None + out.append(Package("cargo", name, version, home + "/.cargo/bin")) + return out + + def _go(self, gate, homes) -> list[Package]: + """Read Go metadata only through the root-owned toolchain. + + User-managed ``go`` shims are not trusted scanner helpers. + """ + out: list[Package] = [] + for home in homes: + listing = gate.list_dir(home + "/go/bin") + if not listing.ok: + continue + for entry in listing.value: + if entry.is_dir or not entry.is_exec: + continue + ran = gate.run_helper(("/usr/local/go/bin/go", "version", "-m", entry.path)) + if not ran.ok or ran.value.code != 0: + continue + for line in ran.value.stdout.splitlines(): + fields = line.strip().split() + if len(fields) >= 3 and fields[0] == "mod": + out.append(Package("go", fields[1], fields[2], entry.path)) + break + return out + + def _expand(self, template: str, homes: tuple[str, ...]) -> list[str]: + if not template.startswith("~"): + return [template] + return [home + template[1:] for home in homes] + + def _npm(self, gate, homes) -> list[Package]: + import json as _json + + out: list[Package] = [] + for template in self.NODE_ROOTS: + for root in self._expand(template, homes): + listing = gate.list_dir(root) + if not listing.ok: + continue + for entry in listing.value: + if not entry.is_dir: + continue + name = entry.path.rsplit("/", 1)[-1] + # Scoped packages hold their real entries one level down. + targets = [entry.path] + if name.startswith("@"): + scoped = gate.list_dir(entry.path) + targets = [e.path for e in scoped.value] if scoped.ok else [] + for target in targets: + raw = gate.read_text(target + "/package.json", limit=1 << 20) + if not raw.ok: + continue + try: + manifest = _json.loads(raw.value) + except ValueError: + gate.ledger.probe("npm", "degraded", target) + continue + out.append( + Package("npm", str(manifest.get("name") or ""), + _opt_str(manifest.get("version")), target) + ) + return out + + def _venvs(self, gate, homes, roots, manager) -> list[Package]: + out: list[Package] = [] + for template in roots: + for root in self._expand(template, homes): + listing = gate.list_dir(root) + if not listing.ok: + continue + for entry in listing.value: + if entry.is_dir: + out.append(Package(manager, entry.path.rsplit("/", 1)[-1], None, entry.path)) + return out + + +def _opt_str(value): + return str(value) if value is not None else None + + +class FixtureProviders(NullProviders): + """Reads the non-filesystem surfaces from JSON beside the fixture tree. + + A surface with no file is *unavailable*, not empty -- so a case that + forgets to supply processes.json fails loudly rather than quietly + asserting that nothing was running. + """ + + reason = "not supplied by this fixture" + + FILES = { + "processes": ("processes.json", Process), + "sockets": ("sockets.json", Socket), + "packages": ("packages.json", Package), + "applications": ("applications.json", Application), + "dns_cache": ("dns.json", DnsEntry), + "exec_journal": ("execjournal.json", ExecEvent), + } + + def __init__(self, data: dict[str, object] | None = None) -> None: + self._data = data or {} + + def _load(self, gate: "Gate", surface: str): + from ..gate import Ok + + if surface in self._data: + rows = self._data[surface] + else: + filename, _ = self.FILES[surface] + raw = gate.read_text("/" + filename) + if not raw.ok: + return self._unavailable(gate, surface) + try: + rows = json.loads(raw.value) + except json.JSONDecodeError as exc: + gate.ledger.unavailable(surface, f"malformed fixture: {exc}") + from ..gate import Refused + + return Refused("malformed", str(exc)) + _, cls = self.FILES[surface] + out = [] + for row in rows: + row = dict(row) + for key in ("argv", "env_names"): + if key in row and isinstance(row[key], list): + row[key] = tuple(row[key]) + out.append(cls(**row)) + return Ok(tuple(out)) + + def processes(self, gate): + return self._load(gate, "processes") + + def sockets(self, gate): + return self._load(gate, "sockets") + + def packages(self, gate): + return self._load(gate, "packages") + + def applications(self, gate): + return self._load(gate, "applications") + + def dns_cache(self, gate): + return self._load(gate, "dns_cache") + + def exec_journal(self, gate): + return self._load(gate, "exec_journal") + + def package_owner(self, gate, path): + from ..gate import Ok, Refused + + pkgs = self._load(gate, "packages") + if not pkgs.ok: + return pkgs + for pkg in pkgs.value: + if pkg.path and pkg.path == path: + return Ok(pkg) + return Refused("not_owned", path) diff --git a/Discovery/adr_discovery/world/platform/darwin.py b/Discovery/adr_discovery/world/platform/darwin.py new file mode 100644 index 0000000..5a15afb --- /dev/null +++ b/Discovery/adr_discovery/world/platform/darwin.py @@ -0,0 +1,154 @@ +"""macOS providers. + +`ps -o comm=` returns a full executable path here rather than a truncated +name, so the exe requirement is met without /proc. +""" + +from __future__ import annotations + +import plistlib +import pwd + +from .base import Application, LanguagePackages, NullProviders, Package, Process, Socket + + +class DarwinProviders(NullProviders): + reason = "not readable on this host" + + #: `/home` is autofs here and blocks indefinitely when automountd is + #: unresponsive. Homes live under /Users; there is nothing to gain by + #: asking, and a hang to lose. + HOME_ROOTS = ("/Users",) + + def _homes(self, gate) -> tuple[str, ...]: + out: list[str] = [] + for base in self.home_roots(): + listing = gate.list_dir(base) + if listing.ok: + out.extend(e.path for e in listing.value if e.is_dir) + return tuple(out) + + def owner_of(self, uid: int) -> str: + try: + return pwd.getpwuid(uid).pw_name + except KeyError: + return str(uid) + + def processes(self, gate): + from ..gate import Ok + + ran = gate.run_helper(("/bin/ps", "-axo", "pid=,ppid=,user=,comm=")) + if not ran.ok or ran.value.code != 0: + return self._unavailable(gate, "processes") + procs: list[Process] = [] + for line in ran.value.stdout.splitlines(): + parts = line.strip().split(None, 3) + if len(parts) < 4: + continue + pid, ppid, user, exe = parts + if not pid.isdigit(): + continue + procs.append(Process(pid=int(pid), exe=exe, ppid=int(ppid) if ppid.isdigit() else 0, user=user)) + return Ok(tuple(procs)) + + def sockets(self, gate): + from ..gate import Ok + + ran = gate.run_helper(("/usr/sbin/lsof", "-nP", "-iTCP")) + if not ran.ok or ran.value.code != 0: + return self._unavailable(gate, "sockets") + out: list[Socket] = [] + for line in ran.value.stdout.splitlines()[1:]: + cols = line.split() + if len(cols) < 9 or not cols[1].isdigit(): + continue + endpoint = cols[8] + raw_state = cols[9].strip("()") if len(cols) > 9 else "" + state = "LISTEN" if raw_state == "LISTEN" else "ESTABLISHED" if "->" in endpoint else "" + if not state: + continue + local, _, remote = endpoint.partition("->") + _, lport = _split_colon_hostport(local) + rhost, rport = _split_colon_hostport(remote) if remote else ("", 0) + out.append(Socket("tcp", state, lport, rhost, rport, int(cols[1]))) + return Ok(tuple(out)) + + def applications(self, gate): + from ..gate import Ok + + apps: list[Application] = [] + for root in ("/Applications", "/System/Applications"): + listing = gate.list_dir(root) + if not listing.ok: + continue + for entry in listing.value: + if not entry.path.endswith(".app"): + continue + raw = gate.read_bytes(entry.path + "/Contents/Info.plist", limit=1 << 20) + if not raw.ok: + continue + try: + info = plistlib.loads(raw.value) + except Exception: + gate.ledger.probe("Info.plist", "degraded", entry.path) + continue + apps.append( + Application( + ident=str(info.get("CFBundleIdentifier", "")), + name=str(info.get("CFBundleName", entry.path.rsplit("/", 1)[-1])), + version=_str_or_none(info.get("CFBundleShortVersionString")), + path=entry.path, + ) + ) + if not apps: + return self._unavailable(gate, "applications") + return Ok(tuple(apps)) + + def dns_cache(self, gate): + # The macOS resolver cache has not been externally enumerable since + # the mDNSResponder rework. Say so; do not return an empty list. + gate.ledger.unavailable("dns_cache", "mDNSResponder cache is not enumerable") + from ..gate import Refused + + return Refused("unavailable", "mDNSResponder cache is not enumerable") + + def packages(self, gate): + from ..gate import Ok + + found: list[Package] = [] + found.extend(LanguagePackages().collect(gate, self._homes(gate))) + for cellar in ("/opt/homebrew/Cellar", "/usr/local/Cellar"): + listing = gate.list_dir(cellar) + if not listing.ok: + continue + for entry in listing.value: + if not entry.is_dir: + continue + name = entry.path.rsplit("/", 1)[-1] + versions = gate.list_dir(entry.path) + version = ( + versions.value[-1].path.rsplit("/", 1)[-1] if versions.ok and versions.value else None + ) + found.append(Package("brew", name, version, entry.path)) + if not found: + return self._unavailable(gate, "packages") + return Ok(tuple(found)) + + +def _str_or_none(v): + return str(v) if v is not None else None + + +def _port(hostport: str) -> int: + tail = hostport.rsplit(".", 1)[-1] + return int(tail) if tail.isdigit() else 0 + + +def _split_hostport(hostport: str) -> tuple[str, int]: + host, _, port = hostport.rpartition(".") + return host, int(port) if port.isdigit() else 0 + + +def _split_colon_hostport(value: str) -> tuple[str, int]: + host, sep, port = value.rpartition(":") + return (host, int(port)) if sep and port.isdigit() else (value, 0) diff --git a/Discovery/adr_discovery/world/platform/linux.py b/Discovery/adr_discovery/world/platform/linux.py new file mode 100644 index 0000000..7af7a3c --- /dev/null +++ b/Discovery/adr_discovery/world/platform/linux.py @@ -0,0 +1,245 @@ +"""Linux providers. Processes come from /proc, never from a name.""" + +from __future__ import annotations + +import os +import pwd + +from .base import Application, LanguagePackages, NullProviders, Package, Process, Socket + + +class LinuxProviders(NullProviders): + reason = "not readable on this host" + + HOME_ROOTS = ("/home", "/root") + + def _homes(self, gate) -> tuple[str, ...]: + out: list[str] = [] + for base in self.home_roots(): + listing = gate.list_dir(base) + if listing.ok: + out.extend(e.path for e in listing.value if e.is_dir) + return tuple(out) + + def owner_of(self, uid: int) -> str: + try: + return pwd.getpwuid(uid).pw_name + except KeyError: + return str(uid) + + def processes(self, gate): + from ..gate import Ok + + procs: list[Process] = [] + listing = gate.list_dir("/proc") + if not listing.ok: + return self._unavailable(gate, "processes") + for entry in listing.value: + name = entry.path.rsplit("/", 1)[-1] + if not name.isdigit(): + continue + pid = int(name) + # The link target, not the name: one syscall, and it is the + # difference between /opt/agents/claude and /usr/bin/claude. + try: + exe = os.readlink(gate.host_path(f"/proc/{pid}/exe")) + except OSError: + continue + cmdline = gate.read_bytes(f"/proc/{pid}/cmdline", limit=8192) + argv = tuple(cmdline.value.decode("utf-8", "replace").split("\0")[:-1]) if cmdline.ok else () + try: + cwd = os.readlink(gate.host_path(f"/proc/{pid}/cwd")) + except OSError: + cwd = None + status = gate.read_text(f"/proc/{pid}/status", limit=4096) + ppid, uid = 0, 0 + if status.ok: + for line in status.value.splitlines(): + if line.startswith("PPid:"): + ppid = int(line.split()[1]) + elif line.startswith("Uid:"): + uid = int(line.split()[1]) + environ = gate.read_bytes(f"/proc/{pid}/environ", limit=32768) + env_names = () + if environ.ok: + env_names = tuple( + sorted( + { + part.split("=", 1)[0] + for part in environ.value.decode("utf-8", "replace").split("\0") + if "=" in part + } + ) + ) + procs.append( + Process(pid=pid, exe=exe, argv=argv, ppid=ppid, cwd=cwd, + user=self.owner_of(uid), env_names=env_names) + ) + return Ok(tuple(procs)) + + def sockets(self, gate): + from ..gate import Ok + + out: list[Socket] = [] + inode_pids = _socket_pids(gate) + for proto, path in (("tcp", "/proc/net/tcp"), ("tcp6", "/proc/net/tcp6")): + raw = gate.read_text(path, limit=1 << 20) + if not raw.ok: + continue + for line in raw.value.splitlines()[1:]: + cols = line.split() + if len(cols) < 4: + continue + try: + lport = int(cols[1].split(":")[1], 16) + rhex, rport_hex = cols[2].split(":") + rport = int(rport_hex, 16) + inode = cols[9] + except (ValueError, IndexError): + continue + state = {"0A": "LISTEN", "01": "ESTABLISHED"}.get(cols[3], cols[3]) + if state not in ("LISTEN", "ESTABLISHED"): + continue + out.append( + Socket(proto=proto, state=state, local_port=lport, + remote_host=_hex_ip(rhex), remote_port=rport, + pid=inode_pids.get(inode)) + ) + if not out: + return self._unavailable(gate, "sockets") + return Ok(tuple(out)) + + def packages(self, gate): + from ..gate import Ok + + found: list[Package] = [] + found.extend(LanguagePackages().collect(gate, self._homes(gate))) + status = gate.read_text("/var/lib/dpkg/status", limit=8 << 20) + if status.ok: + name = version = None + for line in status.value.splitlines(): + if line.startswith("Package: "): + name = line[9:].strip() + elif line.startswith("Version: "): + version = line[9:].strip() + elif not line.strip() and name: + found.append(Package("dpkg", name, version)) + name = version = None + else: + gate.ledger.unavailable("dpkg", "status file not readable") + + apk = gate.read_text("/lib/apk/db/installed", limit=8 << 20) + if apk.ok: + name = version = None + for line in apk.value.splitlines() + [""]: + if line.startswith("P:"): + name = line[2:] + elif line.startswith("V:"): + version = line[2:] + elif not line and name: + found.append(Package("apk", name, version)) + name = version = None + + for root, manager in (("/var/lib/snapd/snaps", "snap"), ("/var/lib/flatpak/app", "flatpak")): + listing = gate.list_dir(root) + if not listing.ok: + continue + for entry in listing.value: + name = entry.path.rsplit("/", 1)[-1] + if manager == "snap" and name.endswith(".snap"): + stem = name[:-5] + pkg, _, version = stem.rpartition("_") + found.append(Package(manager, pkg or stem, version or None, entry.path)) + elif manager == "flatpak" and entry.is_dir: + found.append(Package(manager, name, None, entry.path)) + + rpm = gate.run_helper(("/usr/bin/rpm", "-qa", "--qf", "%{NAME}\\t%{VERSION}-%{RELEASE}\\n")) + if rpm.ok and rpm.value.code == 0: + for line in rpm.value.stdout.splitlines(): + name, sep, version = line.partition("\t") + if sep: + found.append(Package("rpm", name, version)) + if not found: + return self._unavailable(gate, "packages") + return Ok(tuple(found)) + + def applications(self, gate): + from ..gate import Ok + + apps: list[Application] = [] + for root in ("/usr/share/applications", "/var/lib/flatpak/exports/share/applications"): + listing = gate.list_dir(root) + if not listing.ok: + continue + for entry in listing.value: + if not entry.path.endswith(".desktop"): + continue + raw = gate.read_text(entry.path, limit=65536) + if not raw.ok: + continue + fields = dict( + line.split("=", 1) for line in raw.value.splitlines() if "=" in line and not line.startswith("#") + ) + apps.append( + Application( + ident=entry.path.rsplit("/", 1)[-1].removesuffix(".desktop"), + name=fields.get("Name", ""), + version=fields.get("Version"), + path=fields.get("Exec"), + ) + ) + if not apps: + return self._unavailable(gate, "applications") + return Ok(tuple(apps)) + + def dns_cache(self, gate): + ran = gate.run_helper(("/usr/bin/resolvectl", "statistics")) + if not ran.ok: + return self._unavailable(gate, "dns_cache") + # resolvectl exposes counters, not entries, on most builds. Report + # the surface as unavailable rather than inventing an empty answer. + gate.ledger.unavailable("dns_cache", "resolvectl exposes counters, not cache entries") + from ..gate import Refused + + return Refused("unavailable", "no enumerable resolver cache") + + def package_owner(self, gate, path): + from ..gate import Ok, Refused + + ran = gate.run_helper(("/usr/bin/dpkg", "-S", path)) + if not ran.ok: + return ran + if ran.value.code != 0 or ":" not in ran.value.stdout: + return Refused("not_owned", path) + pkg = ran.value.stdout.split(":", 1)[0].strip() + return Ok(Package("dpkg", pkg, None, path)) + + +def _hex_ip(hex_addr: str) -> str: + if len(hex_addr) == 8: + octets = [int(hex_addr[i : i + 2], 16) for i in (6, 4, 2, 0)] + return ".".join(str(o) for o in octets) + return hex_addr + + +def _socket_pids(gate) -> dict[str, int]: + """Join /proc/net socket inodes back to their owning processes.""" + owners: dict[str, int] = {} + procs = gate.list_dir("/proc") + if not procs.ok: + return owners + for proc in procs.value: + name = proc.path.rsplit("/", 1)[-1] + if not name.isdigit() or not proc.is_dir: + continue + fds = gate.list_dir(proc.path + "/fd") + if not fds.ok: + continue + for fd in fds.value: + try: + target = os.readlink(gate.host_path(fd.path)) + except OSError: + continue + if target.startswith("socket:[") and target.endswith("]"): + owners[target[8:-1]] = int(name) + return owners diff --git a/Discovery/adr_discovery/world/platform/windows.py b/Discovery/adr_discovery/world/platform/windows.py new file mode 100644 index 0000000..c451d5c --- /dev/null +++ b/Discovery/adr_discovery/world/platform/windows.py @@ -0,0 +1,94 @@ +"""Windows providers: CIM, TCP tables, Uninstall registry and AppX.""" + +from __future__ import annotations + +import json +import shlex + +from .base import Application, NullProviders, Package, Process, Socket + + +class WindowsProviders(NullProviders): + reason = "Windows management surface unavailable" + HOME_ROOTS = ("/Users",) + + def processes(self, gate): + rows = self._ps_json(gate, "Get-CimInstance Win32_Process | Select-Object ProcessId,ParentProcessId,ExecutablePath,CommandLine | ConvertTo-Json -Compress", "processes") + if rows is None: + return self._unavailable(gate, "processes") + from ..gate import Ok + out = [] + for row in rows: + exe = row.get("ExecutablePath") + if not exe: + continue + try: + argv = tuple(shlex.split(row.get("CommandLine") or "", posix=False)) + except ValueError: + argv = () + out.append(Process(int(row.get("ProcessId") or 0), str(exe), argv, + int(row.get("ParentProcessId") or 0))) + return Ok(tuple(out)) + + def sockets(self, gate): + rows = self._ps_json(gate, "Get-NetTCPConnection | Select-Object State,LocalPort,RemoteAddress,RemotePort,OwningProcess | ConvertTo-Json -Compress", "sockets") + if rows is None: + return self._unavailable(gate, "sockets") + from ..gate import Ok + states = {"Listen": "LISTEN", "Established": "ESTABLISHED"} + return Ok(tuple(Socket("tcp", states[str(r.get("State"))], int(r.get("LocalPort") or 0), + str(r.get("RemoteAddress") or ""), int(r.get("RemotePort") or 0), + int(r.get("OwningProcess") or 0)) + for r in rows if str(r.get("State")) in states)) + + def applications(self, gate): + script = ( + "$u=@(); $p=@('HKLM:\\Software\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'," + "'HKLM:\\Software\\WOW6432Node\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'," + "'HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'); " + "$u+=Get-ItemProperty $p -ErrorAction SilentlyContinue | Where-Object DisplayName | Select-Object @{n='Id';e={$_.PSChildName}},@{n='Name';e={$_.DisplayName}},@{n='Version';e={$_.DisplayVersion}},@{n='Vendor';e={$_.Publisher}},@{n='Path';e={$_.InstallLocation}}; " + "$u+=Get-AppxPackage | Select-Object @{n='Id';e={$_.PackageFamilyName}},@{n='Name';e={$_.Name}},@{n='Version';e={[string]$_.Version}},@{n='Vendor';e={$_.Publisher}},@{n='Path';e={$_.InstallLocation}}; $u|ConvertTo-Json -Compress" + ) + rows = self._ps_json(gate, script, "applications") + if rows is None: + return self._unavailable(gate, "applications") + from ..gate import Ok + return Ok(tuple(Application(str(r.get("Id") or r.get("Name") or ""), str(r.get("Name") or ""), + _text(r.get("Version")), _text(r.get("Vendor")), _text(r.get("Path"))) + for r in rows)) + + def packages(self, gate): + apps = self.applications(gate) + if not apps.ok: + return apps + from ..gate import Ok + return Ok(tuple(Package("windows", a.ident, a.version, a.path) for a in apps.value)) + + def package_owner(self, gate, path): + from ..gate import Ok, Refused + apps = self.applications(gate) + if not apps.ok: + return apps + folded = path.casefold() + for app in apps.value: + if app.path and folded.startswith(app.path.casefold().rstrip("\\/") + "\\"): + return Ok(Package("windows", app.ident, app.version, app.path)) + return Refused("not_owned", path) + + def _ps_json(self, gate, script, surface): + ran = gate.run_helper(( + r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe", + "-NoProfile", "-NonInteractive", "-Command", script, + )) + if not ran.ok or ran.value.code != 0: + return None + try: + value = json.loads(ran.value.stdout or "[]") + except ValueError as exc: + gate.ledger.probe(surface, "degraded", f"invalid PowerShell JSON: {exc}") + return None + return value if isinstance(value, list) else [value] + + +def _text(value): + return str(value) if value not in (None, "") else None