feat(discovery): resolve assess and report assets

Discovery/adr_discovery/judge/__init__.py

@@ -0,0 +1,66 @@
+"""M6 -- turns an inventory into something a policy can act on.
+
+Its currency is operator trust, and it spends that on every finding it
+raises. Ambiguity resolves toward the safe verdict: a shape the parser
+cannot read confidently produces no finding and a recorded ambiguity,
+rather than a guess dressed as a verdict.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+
+from ..contracts.records import Asset, Declaration, Finding, Risk
+from .findings import findings_for
+from .risk import credential_reach, pinning, transport_of, unattended
+from .sanction import EMPTY, Policy, state_for
+
+__all__ = ["judge", "Policy", "EMPTY"]
+
+
+def judge(assets: tuple[Asset, ...], declarations: dict[str, Declaration],
+          policy: Policy = EMPTY, ledger=None) -> tuple[tuple[Asset, ...], tuple[Finding, ...]]:
+    """Attach risk facts and sanction state, then raise what is worth raising."""
+    judged: list[Asset] = []
+    findings: list[Finding] = []
+
+    for asset in assets:
+        declaration = declarations.get(asset.asset_id)
+        risk = _risk_for(asset, declaration, ledger)
+        with_risk = replace(asset, risk=risk, sanction=state_for(asset.catalog_id, policy))
+        judged.append(with_risk)
+        findings.extend(findings_for(with_risk, policy))
+
+    return tuple(judged), tuple(findings)
+
+
+def _risk_for(asset: Asset, declaration: Declaration | None, ledger) -> Risk:
+    if declaration is None:
+        # Nothing new to say. Keep whatever earlier stages established
+        # rather than overwriting it with an empty verdict.
+        return asset.risk
+
+    pinned, factors = pinning(declaration.command, declaration.args)
+    # A shape we could not read confidently: record the ambiguity and raise
+    # nothing, which is the safe verdict.
+    if pinned is None and declaration.command and not declaration.url and ledger is not None:
+        ledger.probe("judge", "degraded", f"unreadable launch shape: {declaration.name}")
+
+    env_declared = bool(declaration.env_names) or "env" in declaration.raw
+    env_names, kinds = credential_reach(declaration.env_names, env_declared)
+    transport, transport_factors = transport_of(declaration.raw, declaration.url)
+
+    destinations = ()
+    if declaration.url:
+        host = declaration.url.split("//", 1)[-1].split("/", 1)[0].split(":", 1)[0]
+        destinations = (host,)
+
+    return Risk(
+        pinned=pinned,
+        factors=tuple(factors) + transport_factors,
+        credential_kinds=kinds,
+        env_names=env_names,
+        transport=transport,
+        destinations=destinations,
+        unattended=asset.risk.unattended or unattended(declaration.args),
+    )

Discovery/adr_discovery/judge/findings.py

@@ -0,0 +1,97 @@
+"""The findings worth surfacing.
+
+Findings are the narrow end on purpose. Every one an operator dismisses
+costs the ones that follow it, so the gate below is deliberately strict and
+the strictness is recovered by correlation elsewhere: a server a config
+declares is recognized even when its name says nothing.
+
+Two lessons already paid for, both caught on real machines and neither by
+the suite: a loose runtime heuristic put a high-severity finding on
+`npx eslint .`, and a command that merely mentioned a path containing "mcp"
+became an undeclared server. Hence the first rule below.
+"""
+
+from __future__ import annotations
+
+from ..contracts.evidence import Channel, Evidence
+from ..contracts.records import Asset, Finding, Kind
+from .sanction import Policy, is_third_party
+
+
+#: A finding may only be raised about an asset whose identity was
+#: established by evidence -- never about one inferred from a command line.
+#: This is the precision gate, and it is what `npx eslint .` fails.
+def _may_be_judged(asset: Asset) -> bool:
+    if asset.kind is Kind.MCP_SERVER:
+        # Declared in a config, or correlated with one. A bare process whose
+        # argv merely resembles a server does not reach this branch.
+        return bool(asset.catalog_id) or "declared" in asset.flags or asset.confidence.label != "none"
+    return bool(asset.catalog_id)
+
+
+def findings_for(asset: Asset, policy: Policy) -> tuple[Finding, ...]:
+    if not _may_be_judged(asset):
+        return ()
+
+    out: list[Finding] = []
+
+    def ev(proof: str) -> tuple[Evidence, ...]:
+        return (Evidence("judge", Channel.CONFIG, asset.install_path or "", proof, 0.9),)
+
+    if asset.kind is Kind.MCP_SERVER and asset.risk.pinned is False:
+        out.append(
+            Finding(
+                rule="unpinned_mcp_server",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} resolves its package at launch time",
+                evidence=ev("no version in the resolved operand"),
+            )
+        )
+
+    if asset.kind is Kind.MCP_SERVER and "undeclared" in asset.flags:
+        out.append(
+            Finding(
+                rule="undeclared_mcp_server",
+                severity="high",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} is running but no configuration declares it",
+                evidence=ev("running server correlated with no declaration"),
+            )
+        )
+
+    if "plaintext_transport" in asset.risk.factors:
+        out.append(
+            Finding(
+                rule="plaintext_transport",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} is reached over http://",
+                evidence=ev("declared url uses a plaintext scheme"),
+            )
+        )
+
+    if asset.risk.unattended:
+        out.append(
+            Finding(
+                rule="unattended_execution",
+                severity="high",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} launches with permission checks bypassed",
+                evidence=ev("bypass flag present in the agent's own launch"),
+            )
+        )
+
+    third_party = [d for d in asset.risk.destinations if is_third_party(d, policy)]
+    if third_party and policy.tenant_domains:
+        out.append(
+            Finding(
+                rule="third_party_destination",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} reaches {', '.join(sorted(third_party))}",
+                evidence=ev("destination outside the tenant domain list"),
+            )
+        )
+
+    return tuple(out)

Discovery/adr_discovery/judge/risk.py

@@ -0,0 +1,129 @@
+"""Risk facts per asset.
+
+Pinning is the highest-yield verdict in the module and the easiest to get
+wrong, because it has to read the *specification* rather than look for an
+`@`, and parse options before choosing an operand. A volume mount is not an
+image; a registry URL is not a package.
+"""
+
+from __future__ import annotations
+
+import re
+from types import MappingProxyType
+
+from ..redact import rules as redact
+
+#: runner -> flags that consume the next argument. Anything not listed and
+#: starting with '-' is a switch; the first bare word after the options is
+#: the operand. This table is the whole difference between reading a
+#: specification and pattern-matching one.
+VALUE_FLAGS = MappingProxyType({
+    "npx": frozenset({"--package", "-p", "--call", "-c", "--node-options"}),
+    "bunx": frozenset({"--package"}),
+    "uvx": frozenset({"--from", "--with", "--python", "-p", "--index-url"}),
+    "pipx": frozenset({"--spec", "--python", "--index-url", "--pip-args"}),
+    "pip": frozenset({"--index-url", "-i", "--extra-index-url", "--find-links", "-f",
+                      "--target", "-t", "--requirement", "-r"}),
+    "docker": frozenset({"-v", "--volume", "-e", "--env", "-p", "--publish", "--name",
+                         "--network", "--mount", "-w", "--workdir", "--user", "-u",
+                         "--entrypoint", "--label", "-l"}),
+    "podman": frozenset({"-v", "--volume", "-e", "--env", "-p", "--publish", "--name"}),
+})
+
+#: Subcommands that must be skipped before the operand is read.
+SUBCOMMANDS = MappingProxyType({
+    "docker": frozenset({"run", "create", "start", "exec"}),
+    "podman": frozenset({"run", "create"}),
+    "pip": frozenset({"install"}),
+    "pipx": frozenset({"run", "install"}),
+    "uv": frozenset({"tool", "run", "pip"}),
+})
+
+#: Runners that resolve a package at launch time -- the shapes where a
+#: missing version means "whatever upstream published this morning".
+EPHEMERAL_RUNNERS = frozenset({"npx", "bunx", "uvx", "pipx", "pip"})
+CONTAINER_RUNNERS = frozenset({"docker", "podman"})
+
+_NPM_PINNED = re.compile(r"^(@[^/]+/)?[^@/]+@[^@]+$")
+_DIGEST = re.compile(r"@sha256:[0-9a-f]{8,}$")
+_TAGGED = re.compile(r":[^/:]+$")
+
+
+def operand_of(command: str | None, args: tuple[str, ...]) -> tuple[str | None, str]:
+    """The thing being run, having parsed the options first."""
+    runner = (command or "").rsplit("/", 1)[-1]
+    value_flags = VALUE_FLAGS.get(runner, frozenset())
+    subcommands = SUBCOMMANDS.get(runner, frozenset())
+
+    i = 0
+    seen_subcommand = not subcommands
+    while i < len(args):
+        arg = args[i]
+        if arg.startswith("-"):
+            flag = arg.split("=", 1)[0]
+            if flag in value_flags and "=" not in arg:
+                i += 2
+                continue
+            i += 1
+            continue
+        if not seen_subcommand and arg in subcommands:
+            seen_subcommand = True
+            i += 1
+            continue
+        return arg, runner
+    return None, runner
+
+
+def pinning(command: str | None, args: tuple[str, ...]) -> tuple[bool | None, tuple[str, ...]]:
+    """(pinned, factors). `None` means the shape carries no pinning question."""
+    operand, runner = operand_of(command, args)
+    if operand is None:
+        return None, ()
+
+    if runner in CONTAINER_RUNNERS:
+        if _DIGEST.search(operand):
+            return True, ()
+        if _TAGGED.search(operand) and not operand.endswith(":latest"):
+            return False, ("unpinned_supply_chain",)
+        return False, ("unpinned_supply_chain",)
+
+    if runner in EPHEMERAL_RUNNERS:
+        if _NPM_PINNED.match(operand) or "==" in operand or operand.count("@") >= 1 and not operand.startswith("@"):
+            return True, ()
+        return False, ("unpinned_supply_chain",)
+
+    return None, ()
+
+
+def credential_reach(env_names: tuple[str, ...], env_declared: bool) -> tuple[tuple[str, ...], tuple[str, ...]]:
+    """Which secrets could be touched, by name and kind, never by value.
+
+    The absence of an env block means the process inherits the parent
+    environment -- which is *all* of them, not none. Reading absence as
+    "no credentials" is the safe-looking answer and the wrong one.
+    """
+    if not env_declared:
+        return ("<inherits parent environment>",), ("inherited",)
+    return env_names, redact.credential_kinds(env_names)
+
+
+def unattended(argv: tuple[str, ...]) -> bool:
+    """The flag comes from the agent's own launch, never from the scheduler
+    that started it: `-p` with `--dangerously-skip-permissions` is the
+    finding whether cron, a timer or a person typed it."""
+    flags = set(argv)
+    headless = bool(flags & {"-p", "--print", "--headless", "--non-interactive", "--yes", "-y"})
+    bypass = bool(
+        flags & {"--dangerously-skip-permissions", "--yolo", "--auto-approve", "--no-confirm"}
+    )
+    return headless and bypass or bypass
+
+
+def transport_of(declaration_raw: dict, url: str | None) -> tuple[str | None, tuple[str, ...]]:
+    transport = str(declaration_raw.get("transport") or ("http" if url else "stdio"))
+    factors: list[str] = []
+    if url and url.startswith("http://"):
+        factors.append("plaintext_transport")
+    if transport == "sse":
+        factors.append("deprecated_transport")
+    return transport, tuple(factors)

Discovery/adr_discovery/judge/sanction.py

@@ -0,0 +1,57 @@
+"""Sanction state, from tenant policy.
+
+Tenant-specific values come from configuration, never from code. The test
+for this is to run one world against two policies and require two verdicts
+-- a value reachable from code alone cannot pass it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class Policy:
+    approved: frozenset[str] = frozenset()
+    forbidden: frozenset[str] = frozenset()
+    #: Domains the tenant owns. A destination outside this set is third
+    #: party; there is no default, because a default would be somebody's
+    #: wrong answer shipped as a constant.
+    tenant_domains: frozenset[str] = frozenset()
+
+    @staticmethod
+    def from_dict(raw: dict) -> "Policy":
+        if not isinstance(raw, dict):
+            raise ValueError("policy root must be an object")
+
+        def string_set(field: str) -> frozenset[str]:
+            value = raw.get(field) or ()
+            if isinstance(value, str) or not isinstance(value, (list, tuple, set, frozenset)):
+                raise ValueError(f"policy {field!r} must be an array of strings")
+            if not all(isinstance(item, str) and item for item in value):
+                raise ValueError(f"policy {field!r} must contain non-empty strings")
+            return frozenset(value)
+
+        return Policy(
+            approved=string_set("approved"),
+            forbidden=string_set("forbidden"),
+            tenant_domains=frozenset(domain.lower().rstrip(".") for domain in string_set("tenant_domains")),
+        )
+
+
+EMPTY = Policy()
+
+
+def state_for(catalog_id: str | None, policy: Policy) -> str:
+    if catalog_id is None:
+        return "unknown"
+    if catalog_id in policy.forbidden:
+        return "forbidden"
+    if catalog_id in policy.approved:
+        return "approved"
+    return "unsanctioned" if policy.approved else "unknown"
+
+
+def is_third_party(host: str, policy: Policy) -> bool:
+    h = host.lower().rstrip(".")
+    return not any(h == d or h.endswith("." + d) for d in policy.tenant_domains)

Discovery/adr_discovery/reporter/__init__.py

@@ -0,0 +1,8 @@
+"""M7 -- the snapshot, and the difference between this one and the last."""
+
+from __future__ import annotations
+
+from .delta import Delta, DifferentEndpoints, diff
+from .snapshot import from_dict, stats, to_dict, to_json
+
+__all__ = ["diff", "Delta", "DifferentEndpoints", "from_dict", "to_dict", "to_json", "stats"]