feat(prover): add OpenShell Policy Prover (OPP) to Python SDK

Integrate Z3-based formal verification for sandbox policies as
openshell.prover. Detects data exfiltration paths, write bypass
attacks, L7 enforcement gaps, overpermissive methods, and privilege
escalation via binary capability chains.

Closes #699

Signed-off-by: Alexander Watson <zredlined@gmail.com>

python/openshell/prover/__init__.py

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""OpenShell Policy Prover (OPP) — formal verification for AI agent sandbox policies."""

python/openshell/prover/__main__.py

@@ -0,0 +1,6 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from .cli import main
+
+main()

python/openshell/prover/accepted_risks.py

@@ -0,0 +1,133 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Load and match accepted risk annotations against findings."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import yaml
+
+from .queries import Finding
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@dataclass
+class AcceptedRisk:
+    query: str
+    reason: str
+    accepted_by: str = ""
+    binary: str = ""
+    endpoint: str = ""
+
+
+def load_accepted_risks(path: Path) -> list[AcceptedRisk]:
+    """Load accepted risks from YAML."""
+    with open(path) as f:  # noqa: PTH123
+        raw = yaml.safe_load(f)
+
+    return [
+        AcceptedRisk(
+            query=r.get("query", ""),
+            reason=r.get("reason", ""),
+            accepted_by=r.get("accepted_by", ""),
+            binary=r.get("binary", ""),
+            endpoint=r.get("endpoint", ""),
+        )
+        for r in (raw.get("accepted_risks") or [])
+    ]
+
+
+def _path_matches_risk(path, risk: AcceptedRisk) -> bool:
+    """Check if a single finding path matches an accepted risk."""
+    if risk.binary:
+        path_binary = getattr(path, "binary", "") or getattr(path, "parent", "")
+        if path_binary != risk.binary:
+            return False
+
+    if risk.endpoint:
+        endpoint_host = getattr(path, "endpoint_host", "")
+        if endpoint_host != risk.endpoint:
+            return False
+
+    return True
+
+
+def apply_accepted_risks(
+    findings: list[Finding],
+    accepted: list[AcceptedRisk],
+) -> list[Finding]:
+    """Mark findings as accepted where they match accepted risk annotations.
+
+    A finding is accepted if ALL of its paths match at least one accepted risk
+    entry for that query. If only some paths match, the finding stays active
+    with the unmatched paths.
+    """
+    if not accepted:
+        return findings
+
+    result = []
+    for finding in findings:
+        matching_risks = [r for r in accepted if r.query == finding.query]
+        if not matching_risks:
+            result.append(finding)
+            continue
+
+        if not finding.paths:
+            # Pathless finding (e.g., inference relay) — accept if query matches
+            accepted_finding = Finding(
+                query=finding.query,
+                title=finding.title,
+                description=finding.description,
+                risk=finding.risk,
+                paths=finding.paths,
+                remediation=finding.remediation,
+                accepted=True,
+                accepted_reason=matching_risks[0].reason,
+            )
+            result.append(accepted_finding)
+            continue
+
+        unmatched_paths = []
+        matched_reason = ""
+        for path in finding.paths:
+            path_accepted = False
+            for risk in matching_risks:
+                if _path_matches_risk(path, risk):
+                    path_accepted = True
+                    matched_reason = risk.reason
+                    break
+            if not path_accepted:
+                unmatched_paths.append(path)
+
+        if not unmatched_paths:
+            accepted_finding = Finding(
+                query=finding.query,
+                title=finding.title,
+                description=finding.description,
+                risk=finding.risk,
+                paths=finding.paths,
+                remediation=finding.remediation,
+                accepted=True,
+                accepted_reason=matched_reason,
+            )
+            result.append(accepted_finding)
+        elif len(unmatched_paths) < len(finding.paths):
+            result.append(
+                Finding(
+                    query=finding.query,
+                    title=finding.title,
+                    description=finding.description,
+                    risk=finding.risk,
+                    paths=unmatched_paths,
+                    remediation=finding.remediation,
+                )
+            )
+        else:
+            result.append(finding)
+
+    return result

python/openshell/prover/binary_registry.py

@@ -0,0 +1,171 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Load binary capability descriptors from YAML registry."""
+
+from __future__ import annotations
+
+import fnmatch
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING
+
+import yaml
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class ActionType(Enum):
+    READ = "read"
+    WRITE = "write"
+    DESTRUCTIVE = "destructive"
+
+
+@dataclass
+class BinaryAction:
+    name: str
+    type: ActionType
+    description: str = ""
+
+
+@dataclass
+class BinaryProtocol:
+    name: str
+    transport: str = ""
+    description: str = ""
+    bypasses_l7: bool = False
+    actions: list[BinaryAction] = field(default_factory=list)
+
+    @property
+    def can_write(self) -> bool:
+        return any(
+            a.type in (ActionType.WRITE, ActionType.DESTRUCTIVE) for a in self.actions
+        )
+
+    @property
+    def can_destroy(self) -> bool:
+        return any(a.type == ActionType.DESTRUCTIVE for a in self.actions)
+
+
+@dataclass
+class BinaryCapability:
+    path: str
+    description: str = ""
+    protocols: list[BinaryProtocol] = field(default_factory=list)
+    spawns: list[str] = field(default_factory=list)
+    can_exfiltrate: bool = False
+    exfil_mechanism: str = ""
+    can_construct_http: bool = False
+
+    @property
+    def bypasses_l7(self) -> bool:
+        return any(p.bypasses_l7 for p in self.protocols)
+
+    @property
+    def can_write(self) -> bool:
+        return any(p.can_write for p in self.protocols) or self.can_construct_http
+
+    @property
+    def can_destroy(self) -> bool:
+        return any(p.can_destroy for p in self.protocols)
+
+    @property
+    def write_mechanisms(self) -> list[str]:
+        mechanisms = []
+        for p in self.protocols:
+            if p.can_write:
+                for a in p.actions:
+                    if a.type in (ActionType.WRITE, ActionType.DESTRUCTIVE):
+                        mechanisms.append(f"{p.name}: {a.name}")
+        if self.can_construct_http:
+            mechanisms.append("arbitrary HTTP request construction")
+        return mechanisms
+
+
+@dataclass
+class BinaryRegistry:
+    binaries: dict[str, BinaryCapability] = field(default_factory=dict)
+
+    def get(self, path: str) -> BinaryCapability | None:
+        return self.binaries.get(path)
+
+    def get_or_unknown(self, path: str) -> BinaryCapability:
+        """Return known capability or a default 'unknown' descriptor."""
+        if path in self.binaries:
+            return self.binaries[path]
+        # Check glob patterns (e.g., registry has /usr/bin/python* matching /usr/bin/python3.13)
+        for reg_path, cap in self.binaries.items():
+            if "*" in reg_path and fnmatch.fnmatch(path, reg_path):
+                return cap
+        return BinaryCapability(
+            path=path,
+            description="Unknown binary — not in registry",
+            can_exfiltrate=True,  # conservative: assume unknown can exfil
+            can_construct_http=True,  # conservative: assume unknown can make HTTP requests
+        )
+
+    def transitive_spawns(self, path: str, visited: set[str] | None = None) -> set[str]:
+        """Return all binaries transitively spawned by this binary."""
+        if visited is None:
+            visited = set()
+        if path in visited:
+            return set()
+        visited.add(path)
+
+        cap = self.get(path)
+        if not cap:
+            return set()
+
+        result = set(cap.spawns)
+        for child in cap.spawns:
+            result |= self.transitive_spawns(child, visited)
+        return result
+
+
+def load_binary_capability(path: Path) -> BinaryCapability:
+    """Load a single binary capability descriptor from YAML."""
+    with open(path) as f:  # noqa: PTH123
+        raw = yaml.safe_load(f)
+
+    protocols = []
+    for p_raw in raw.get("protocols", []):
+        actions = []
+        for a_raw in p_raw.get("actions", []):
+            actions.append(
+                BinaryAction(
+                    name=a_raw.get("name", ""),
+                    type=ActionType(a_raw.get("type", "read")),
+                    description=a_raw.get("description", ""),
+                )
+            )
+        protocols.append(
+            BinaryProtocol(
+                name=p_raw.get("name", ""),
+                transport=p_raw.get("transport", ""),
+                description=p_raw.get("description", ""),
+                bypasses_l7=p_raw.get("bypasses_l7", False),
+                actions=actions,
+            )
+        )
+
+    return BinaryCapability(
+        path=raw.get("binary", ""),
+        description=raw.get("description", ""),
+        protocols=protocols,
+        spawns=raw.get("spawns", []),
+        can_exfiltrate=raw.get("can_exfiltrate", False),
+        exfil_mechanism=raw.get("exfil_mechanism", ""),
+        can_construct_http=raw.get("can_construct_http", False),
+    )
+
+
+def load_binary_registry(registry_dir: Path) -> BinaryRegistry:
+    """Load all binary capability descriptors from a registry directory."""
+    binaries = {}
+    binaries_dir = registry_dir / "binaries"
+    if binaries_dir.is_dir():
+        for yaml_file in binaries_dir.glob("*.yaml"):
+            cap = load_binary_capability(yaml_file)
+            binaries[cap.path] = cap
+    return BinaryRegistry(binaries=binaries)

python/openshell/prover/cli.py

@@ -0,0 +1,184 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""CLI entry point for OpenShell Policy Prover (OPP)."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from rich.console import Console
+
+from .binary_registry import load_binary_registry
+from .credential_loader import load_credential_set
+from .policy_parser import parse_policy
+from .queries import run_all_queries
+from .report import render_report
+from .z3_model import build_model
+
+_DEFAULT_REGISTRY = Path(__file__).parent / "registry"
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="opp",
+        description="OpenShell Policy Prover — formal verification for sandbox policies",
+    )
+    subparsers = parser.add_subparsers(dest="command")
+
+    verify_parser = subparsers.add_parser(
+        "prove",
+        help="Prove properties of a sandbox policy — or find counterexamples",
+    )
+    verify_parser.add_argument(
+        "--policy",
+        required=True,
+        type=Path,
+        help="Path to OpenShell sandbox policy YAML",
+    )
+    verify_parser.add_argument(
+        "--credentials",
+        required=True,
+        type=Path,
+        help="Path to credential descriptor YAML",
+    )
+    verify_parser.add_argument(
+        "--registry",
+        type=Path,
+        default=_DEFAULT_REGISTRY,
+        help="Path to capability registry directory (default: bundled registry)",
+    )
+    verify_parser.add_argument(
+        "--accepted-risks",
+        type=Path,
+        default=None,
+        help="Path to accepted risks YAML (findings matching these are marked accepted)",
+    )
+    verify_parser.add_argument(
+        "--compact",
+        action="store_true",
+        help="One-line-per-finding output (for demos and CI)",
+    )
+    verify_parser.add_argument(
+        "--color",
+        action="store_true",
+        help="Force colored output even when piped",
+    )
+    verify_parser.add_argument(
+        "--mermaid",
+        action="store_true",
+        help="Output Mermaid findings graph (risk paths by query type)",
+    )
+    verify_parser.add_argument(
+        "--mermaid-topology",
+        action="store_true",
+        help="Output Mermaid policy topology map (all endpoints, binaries, enforcement levels)",
+    )
+    verify_parser.add_argument(
+        "--html",
+        type=Path,
+        default=None,
+        metavar="PATH",
+        help="Output self-contained HTML report with interactive diagrams (e.g., --html report.html)",
+    )
+
+    args = parser.parse_args()
+
+    if args.command is None:
+        parser.print_help()
+        sys.exit(1)
+
+    if args.command == "prove":
+        exit_code = cmd_prove(args)
+        sys.exit(exit_code)
+
+
+def cmd_prove(args) -> int:
+    """Execute the prove command."""
+    console = Console(force_terminal=args.color) if args.color else Console()
+
+    # Load inputs
+    try:
+        policy = parse_policy(args.policy)
+    except Exception as e:
+        console.print(f"[red]Error loading policy: {e}[/red]")
+        return 2
+
+    try:
+        credential_set = load_credential_set(args.credentials, args.registry)
+    except Exception as e:
+        console.print(f"[red]Error loading credentials: {e}[/red]")
+        return 2
+
+    try:
+        binary_registry = load_binary_registry(args.registry)
+    except Exception as e:
+        console.print(f"[red]Error loading binary registry: {e}[/red]")
+        return 2
+
+    # Build model
+    model = build_model(policy, credential_set, binary_registry)
+
+    # Run queries
+    findings = run_all_queries(model)
+
+    # Apply accepted risks
+    if args.accepted_risks:
+        try:
+            from .accepted_risks import apply_accepted_risks, load_accepted_risks
+
+            accepted = load_accepted_risks(args.accepted_risks)
+            findings = apply_accepted_risks(findings, accepted)
+        except Exception as e:
+            console.print(f"[red]Error loading accepted risks: {e}[/red]")
+            return 2
+
+    # Render
+    has_critical_high = any(
+        f.risk.value in ("critical", "high") and not f.accepted for f in findings
+    )
+    if args.html:
+        from .report import render_html
+
+        exit_code = render_html(
+            findings,
+            str(args.policy),
+            str(args.credentials),
+            policy,
+            binary_registry,
+            args.html,
+        )
+        console.print(f"Report written to {args.html}")
+        return exit_code
+    elif args.mermaid_topology:
+        from .report import generate_mermaid_topology
+
+        print(generate_mermaid_topology(policy, binary_registry, findings))
+        return 1 if has_critical_high else 0
+    elif args.mermaid:
+        from .report import generate_mermaid
+
+        print(generate_mermaid(findings))
+        return 1 if has_critical_high else 0
+    elif args.compact:
+        from .report import render_compact
+
+        return render_compact(
+            findings,
+            str(args.policy),
+            str(args.credentials),
+            console=console,
+        )
+    else:
+        return render_report(
+            findings,
+            str(args.policy),
+            str(args.credentials),
+            console=console,
+        )
+
+
+if __name__ == "__main__":
+    main()

python/openshell/prover/credential_loader.py

@@ -0,0 +1,141 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Parse credential descriptors and API capability registries."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+import yaml
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@dataclass
+class Credential:
+    name: str
+    type: str
+    scopes: list[str] = field(default_factory=list)
+    injected_via: str = ""
+    target_hosts: list[str] = field(default_factory=list)
+
+
+@dataclass
+class ApiAction:
+    method: str
+    path: str
+    action: str
+
+
+@dataclass
+class ApiCapability:
+    api: str
+    host: str
+    port: int = 443
+    credential_type: str = ""
+    scope_capabilities: dict[str, list[ApiAction]] = field(default_factory=dict)
+    action_risk: dict[str, str] = field(default_factory=dict)
+
+    def actions_for_scopes(self, scopes: list[str]) -> list[ApiAction]:
+        """Return all API actions enabled by the given scopes."""
+        result = []
+        for scope in scopes:
+            result.extend(self.scope_capabilities.get(scope, []))
+        return result
+
+    def write_actions_for_scopes(self, scopes: list[str]) -> list[ApiAction]:
+        """Return API actions that involve writing (POST, PUT, PATCH, DELETE)."""
+        return [
+            a
+            for a in self.actions_for_scopes(scopes)
+            if a.method.upper() in {"POST", "PUT", "PATCH", "DELETE"}
+        ]
+
+    def destructive_actions_for_scopes(self, scopes: list[str]) -> list[ApiAction]:
+        """Return actions classified as high or critical risk."""
+        actions = self.actions_for_scopes(scopes)
+        return [
+            a
+            for a in actions
+            if self.action_risk.get(a.action, "low") in {"high", "critical"}
+        ]
+
+
+@dataclass
+class CredentialSet:
+    credentials: list[Credential] = field(default_factory=list)
+    api_registries: dict[str, ApiCapability] = field(default_factory=dict)
+
+    def credentials_for_host(self, host: str) -> list[Credential]:
+        return [c for c in self.credentials if host in c.target_hosts]
+
+    def api_for_host(self, host: str) -> ApiCapability | None:
+        for api in self.api_registries.values():
+            if api.host == host:
+                return api
+        return None
+
+
+def load_credentials(path: Path) -> list[Credential]:
+    """Load credential descriptors from YAML."""
+    with open(path) as f:  # noqa: PTH123
+        raw = yaml.safe_load(f)
+
+    credentials = []
+    for c_raw in raw.get("credentials", []):
+        credentials.append(
+            Credential(
+                name=c_raw.get("name", ""),
+                type=c_raw.get("type", ""),
+                scopes=c_raw.get("scopes", []),
+                injected_via=c_raw.get("injected_via", ""),
+                target_hosts=c_raw.get("target_hosts", []),
+            )
+        )
+    return credentials
+
+
+def load_api_registry(path: Path) -> ApiCapability:
+    """Load an API capability registry from YAML."""
+    with open(path) as f:  # noqa: PTH123
+        raw = yaml.safe_load(f)
+
+    scope_caps = {}
+    for scope, actions_raw in (raw.get("scope_capabilities") or {}).items():
+        scope_caps[scope] = [
+            ApiAction(
+                method=a.get("method", ""),
+                path=a.get("path", ""),
+                action=a.get("action", ""),
+            )
+            for a in actions_raw
+        ]
+
+    return ApiCapability(
+        api=raw.get("api", ""),
+        host=raw.get("host", ""),
+        port=raw.get("port", 443),
+        credential_type=raw.get("credential_type", ""),
+        scope_capabilities=scope_caps,
+        action_risk=raw.get("action_risk", {}),
+    )
+
+
+def load_credential_set(
+    credentials_path: Path,
+    registry_dir: Path,
+) -> CredentialSet:
+    """Load credentials and all API registries from a registry directory."""
+    creds = load_credentials(credentials_path)
+
+    api_registries = {}
+    apis_dir = registry_dir / "apis"
+    if apis_dir.is_dir():
+        for api_file in apis_dir.glob("*.yaml"):
+            api = load_api_registry(api_file)
+            api_registries[api.api] = api
+
+    return CredentialSet(credentials=creds, api_registries=api_registries)

python/openshell/prover/policy_parser.py

@@ -0,0 +1,219 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Parse OpenShell sandbox policy YAML into typed dataclasses."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING
+
+import yaml
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class PolicyIntent(Enum):
+    L4_ONLY = "l4_only"
+    READ_ONLY = "read_only"
+    READ_WRITE = "read_write"
+    FULL = "full"
+    CUSTOM = "custom"
+
+
+@dataclass
+class L7Rule:
+    method: str
+    path: str
+    command: str = ""
+
+
+@dataclass
+class Endpoint:
+    host: str
+    port: int
+    ports: list[int] = field(default_factory=list)
+    protocol: str = ""
+    tls: str = ""
+    enforcement: str = "audit"
+    access: str = ""
+    rules: list[L7Rule] = field(default_factory=list)
+    allowed_ips: list[str] = field(default_factory=list)
+
+    @property
+    def is_l7_enforced(self) -> bool:
+        return bool(self.protocol)
+
+    @property
+    def intent(self) -> PolicyIntent:
+        if not self.protocol:
+            return PolicyIntent.L4_ONLY
+        if self.access == "read-only":
+            return PolicyIntent.READ_ONLY
+        if self.access == "read-write":
+            return PolicyIntent.READ_WRITE
+        if self.access == "full":
+            return PolicyIntent.FULL
+        if self.rules:
+            methods = {r.method.upper() for r in self.rules}
+            if methods <= {"GET", "HEAD", "OPTIONS"}:
+                return PolicyIntent.READ_ONLY
+            if "DELETE" not in methods:
+                return PolicyIntent.READ_WRITE
+            return PolicyIntent.FULL
+        return PolicyIntent.CUSTOM
+
+    @property
+    def effective_ports(self) -> list[int]:
+        if self.ports:
+            return self.ports
+        if self.port:
+            return [self.port]
+        return []
+
+    @property
+    def allowed_methods(self) -> set[str]:
+        """Return the set of HTTP methods this endpoint allows. Empty means all (L4-only)."""
+        if not self.protocol:
+            return set()  # L4-only: all traffic passes
+        if self.access == "read-only":
+            return {"GET", "HEAD", "OPTIONS"}
+        if self.access == "read-write":
+            return {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH"}
+        if self.access == "full":
+            return {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE"}
+        if self.rules:
+            methods = set()
+            for r in self.rules:
+                m = r.method.upper()
+                if m == "*":
+                    return {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE"}
+                methods.add(m)
+            return methods
+        return set()
+
+
+WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
+
+
+@dataclass
+class Binary:
+    path: str
+
+
+@dataclass
+class NetworkPolicyRule:
+    name: str
+    endpoints: list[Endpoint] = field(default_factory=list)
+    binaries: list[Binary] = field(default_factory=list)
+
+
+@dataclass
+class FilesystemPolicy:
+    include_workdir: bool = True
+    read_only: list[str] = field(default_factory=list)
+    read_write: list[str] = field(default_factory=list)
+
+    @property
+    def readable_paths(self) -> list[str]:
+        return self.read_only + self.read_write
+
+
+@dataclass
+class PolicyModel:
+    version: int = 1
+    filesystem_policy: FilesystemPolicy = field(default_factory=FilesystemPolicy)
+    network_policies: dict[str, NetworkPolicyRule] = field(default_factory=dict)
+
+    @property
+    def all_endpoints(self) -> list[tuple[str, Endpoint]]:
+        """Return all (policy_name, endpoint) pairs."""
+        result = []
+        for name, rule in self.network_policies.items():
+            for ep in rule.endpoints:
+                result.append((name, ep))
+        return result
+
+    @property
+    def all_binaries(self) -> list[Binary]:
+        """Return deduplicated list of all binaries across all policies."""
+        seen = set()
+        result = []
+        for rule in self.network_policies.values():
+            for b in rule.binaries:
+                if b.path not in seen:
+                    seen.add(b.path)
+                    result.append(b)
+        return result
+
+    @property
+    def binary_endpoint_pairs(self) -> list[tuple[Binary, str, Endpoint]]:
+        """Return all (binary, policy_name, endpoint) triples."""
+        result = []
+        for name, rule in self.network_policies.items():
+            for b in rule.binaries:
+                for ep in rule.endpoints:
+                    result.append((b, name, ep))
+        return result
+
+
+def parse_policy(path: Path) -> PolicyModel:
+    """Parse an OpenShell policy YAML file into a PolicyModel."""
+    with open(path) as f:  # noqa: PTH123
+        raw = yaml.safe_load(f)
+
+    if not raw:
+        return PolicyModel()
+
+    fs_raw = raw.get("filesystem_policy", {}) or {}
+    fs = FilesystemPolicy(
+        include_workdir=fs_raw.get("include_workdir", True),
+        read_only=fs_raw.get("read_only", []),
+        read_write=fs_raw.get("read_write", []),
+    )
+
+    network_policies = {}
+    for key, rule_raw in (raw.get("network_policies") or {}).items():
+        endpoints = []
+        for ep_raw in rule_raw.get("endpoints", []):
+            rules = []
+            for r in ep_raw.get("rules", []):
+                allow = r.get("allow", {})
+                rules.append(
+                    L7Rule(
+                        method=allow.get("method", ""),
+                        path=allow.get("path", ""),
+                        command=allow.get("command", ""),
+                    )
+                )
+            endpoints.append(
+                Endpoint(
+                    host=ep_raw.get("host", ""),
+                    port=ep_raw.get("port", 0),
+                    ports=ep_raw.get("ports", []),
+                    protocol=ep_raw.get("protocol", ""),
+                    tls=ep_raw.get("tls", ""),
+                    enforcement=ep_raw.get("enforcement", "audit"),
+                    access=ep_raw.get("access", ""),
+                    rules=rules,
+                    allowed_ips=ep_raw.get("allowed_ips", []),
+                )
+            )
+
+        binaries = []
+        for b_raw in rule_raw.get("binaries", []):
+            binaries.append(Binary(path=b_raw.get("path", "")))
+
+        network_policies[key] = NetworkPolicyRule(
+            name=rule_raw.get("name", key),
+            endpoints=endpoints,
+            binaries=binaries,
+        )
+
+    return PolicyModel(
+        version=raw.get("version", 1),
+        filesystem_policy=fs,
+        network_policies=network_policies,
+    )

python/openshell/prover/prover_test.py

@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the OpenShell Policy Prover."""
+
+from pathlib import Path
+
+from openshell.prover.binary_registry import load_binary_registry
+from openshell.prover.credential_loader import load_credential_set
+from openshell.prover.policy_parser import PolicyIntent, parse_policy
+from openshell.prover.queries import RiskLevel, run_all_queries
+from openshell.prover.z3_model import build_model
+
+_TESTDATA = Path(__file__).parent / "testdata"
+_REGISTRY = Path(__file__).parent / "registry"
+
+
+def _run_prover(policy_name: str, credentials_name: str):
+    policy = parse_policy(_TESTDATA / policy_name)
+    creds = load_credential_set(_TESTDATA / credentials_name, _REGISTRY)
+    binaries = load_binary_registry(_REGISTRY)
+    model = build_model(policy, creds, binaries)
+    return run_all_queries(model)
+
+
+def test_parse_policy():
+    """Verify policy parsing produces correct structure."""
+    policy = parse_policy(_TESTDATA / "policy.yaml")
+    assert policy.version == 1
+    assert "github_readonly" in policy.network_policies
+
+    rule = policy.network_policies["github_readonly"]
+    assert len(rule.endpoints) == 2
+    assert len(rule.binaries) == 3
+
+    # First endpoint: L7 enforced read-only
+    ep0 = rule.endpoints[0]
+    assert ep0.host == "api.github.com"
+    assert ep0.is_l7_enforced
+    assert ep0.intent == PolicyIntent.READ_ONLY
+
+    # Second endpoint: L4 only
+    ep1 = rule.endpoints[1]
+    assert ep1.host == "github.com"
+    assert not ep1.is_l7_enforced
+    assert ep1.intent == PolicyIntent.L4_ONLY
+
+
+def test_filesystem_policy():
+    """Verify filesystem policy parsing."""
+    policy = parse_policy(_TESTDATA / "policy.yaml")
+    fs = policy.filesystem_policy
+    assert "/usr" in fs.read_only
+    assert "/sandbox" in fs.read_write
+    assert "/usr" in fs.readable_paths
+    assert "/sandbox" in fs.readable_paths
+
+
+def test_git_push_bypass_findings():
+    """End-to-end: detect git push bypass in L4-only + L7 policy."""
+    findings = _run_prover("policy.yaml", "credentials.yaml")
+    risks = {f.risk for f in findings}
+    queries = {f.query for f in findings}
+
+    assert RiskLevel.CRITICAL in risks, "Should detect critical exfil paths"
+    assert RiskLevel.HIGH in risks, "Should detect write bypass"
+    assert "data_exfiltration" in queries
+    assert "write_bypass" in queries
+    assert "binary_inheritance" in queries
+
+    # Verify git bypass specifically detected
+    write_findings = [f for f in findings if f.query == "write_bypass"]
+    assert len(write_findings) == 1
+    bypass_binaries = {p.binary for p in write_findings[0].paths}
+    assert "/usr/bin/git" in bypass_binaries, "Git should be flagged for L7 bypass"
+
+
+def test_empty_policy_advisory_only():
+    """Deny-all policy should only produce the inference relay advisory."""
+    findings = _run_prover("empty_policy.yaml", "empty_credentials.yaml")
+    queries = {f.query for f in findings}
+    risks = {f.risk for f in findings}
+
+    assert "inference_relay" in queries
+    assert RiskLevel.CRITICAL not in risks
+    assert RiskLevel.HIGH not in risks
+    assert len(findings) == 1
+    assert findings[0].risk == RiskLevel.ADVISORY

python/openshell/prover/queries.py

@@ -0,0 +1,608 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Verification queries that ask reachability questions against the Z3 model."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING
+
+import z3
+
+from .policy_parser import PolicyIntent
+
+if TYPE_CHECKING:
+    from .z3_model import ReachabilityModel
+
+
+class RiskLevel(Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+    ADVISORY = "advisory"
+
+
+@dataclass
+class ExfilPath:
+    """A concrete path through which data can be exfiltrated."""
+
+    binary: str
+    endpoint_host: str
+    endpoint_port: int
+    mechanism: str
+    policy_name: str
+    l7_status: str  # "l4_only", "l7_allows_write", "l7_bypassed"
+
+
+@dataclass
+class WriteBypassPath:
+    """A path that allows writing despite read-only intent."""
+
+    binary: str
+    endpoint_host: str
+    endpoint_port: int
+    policy_name: str
+    policy_intent: str
+    bypass_reason: str  # "l4_only", "l7_bypass_protocol", "credential_write_scope"
+    credential_actions: list[str] = field(default_factory=list)
+
+
+@dataclass
+class InheritancePath:
+    """A binary that inherits network access through the ancestor chain."""
+
+    parent: str
+    child: str
+    endpoint_host: str
+    endpoint_port: int
+    child_capabilities: list[str]
+
+
+@dataclass
+class OverpermissiveMethodPath:
+    """An endpoint where wildcard or overly broad HTTP methods are allowed."""
+
+    endpoint_host: str
+    endpoint_port: int
+    policy_name: str
+    binary: str
+    allowed_methods: str  # e.g. "*" or "GET, POST, PUT, PATCH, DELETE"
+    needed_methods: str  # e.g. "POST" — what the endpoint likely needs
+    excess_methods: str  # e.g. "DELETE, PUT, PATCH" — what's unnecessary
+    risk_detail: str  # why the excess matters
+
+
+@dataclass
+class L4PolicyGapPath:
+    """An endpoint missing L7 enforcement where HTTP-capable binaries have access."""
+
+    endpoint_host: str
+    endpoint_port: int
+    policy_name: str
+    binary: str
+    binary_can_http: bool
+    binary_bypasses_l7: bool
+    has_sibling_l7: bool  # True if other endpoints in same policy group have L7
+    detail: str
+
+
+@dataclass
+class Finding:
+    """A single verification finding."""
+
+    query: str
+    title: str
+    description: str
+    risk: RiskLevel
+    paths: list[
+        ExfilPath
+        | WriteBypassPath
+        | InheritancePath
+        | OverpermissiveMethodPath
+        | L4PolicyGapPath
+    ] = field(default_factory=list)
+    remediation: list[str] = field(default_factory=list)
+    accepted: bool = False
+    accepted_reason: str = ""
+
+
+def check_data_exfiltration(model: ReachabilityModel) -> list[Finding]:
+    """Check for data exfiltration paths from readable filesystem to writable egress channels."""
+    findings = []
+
+    if not model.policy.filesystem_policy.readable_paths:
+        return findings
+
+    exfil_paths: list[ExfilPath] = []
+
+    for bpath in model.binary_paths:
+        cap = model.binary_registry.get_or_unknown(bpath)
+        if not cap.can_exfiltrate:
+            continue
+
+        for eid in model.endpoints:
+            expr = model.can_exfil_via_endpoint(bpath, eid)
+            # Use a fresh solver to check each path independently
+            s = z3.Solver()
+            s.add(model.solver.assertions())
+            s.add(expr)
+
+            if s.check() == z3.sat:
+                # Determine the mechanism
+                ek = eid.key
+                bypass = cap.bypasses_l7
+
+                if bypass:
+                    l7_status = "l7_bypassed"
+                    mechanism = f"{cap.description} — uses non-HTTP protocol, bypasses L7 inspection"
+                elif ek not in model.l7_enforced:
+                    l7_status = "l4_only"
+                    mechanism = f"L4-only endpoint — no HTTP inspection, {bpath} can send arbitrary data"
+                else:
+                    ep_is_l7 = False
+                    for _pn, rule in model.policy.network_policies.items():
+                        for ep in rule.endpoints:
+                            if ep.host == eid.host and eid.port in ep.effective_ports:
+                                ep_is_l7 = ep.is_l7_enforced
+                    if not ep_is_l7:
+                        l7_status = "l4_only"
+                        mechanism = f"L4-only endpoint — no HTTP inspection, {bpath} can send arbitrary data"
+                    else:
+                        l7_status = "l7_allows_write"
+                        mechanism = (
+                            f"L7 allows write methods — {bpath} can POST/PUT data"
+                        )
+
+                if cap.exfil_mechanism:
+                    mechanism += f". Exfil via: {cap.exfil_mechanism}"
+
+                exfil_paths.append(
+                    ExfilPath(
+                        binary=bpath,
+                        endpoint_host=eid.host,
+                        endpoint_port=eid.port,
+                        mechanism=mechanism,
+                        policy_name=eid.policy_name,
+                        l7_status=l7_status,
+                    )
+                )
+
+    if exfil_paths:
+        readable = model.policy.filesystem_policy.readable_paths
+        has_l4_only = any(p.l7_status == "l4_only" for p in exfil_paths)
+        has_bypass = any(p.l7_status == "l7_bypassed" for p in exfil_paths)
+        risk = RiskLevel.CRITICAL if (has_l4_only or has_bypass) else RiskLevel.HIGH
+
+        remediation = []
+        if has_l4_only:
+            remediation.append(
+                "Add `protocol: rest` with specific L7 rules to L4-only endpoints "
+                "to enable HTTP inspection and restrict to safe methods/paths."
+            )
+        if has_bypass:
+            remediation.append(
+                "Binaries using non-HTTP protocols (git, ssh, nc) bypass L7 inspection. "
+                "Remove these binaries from the policy if write access is not intended, "
+                "or restrict credential scopes to read-only."
+            )
+        remediation.append(
+            "Restrict filesystem read access to only the paths the agent needs."
+        )
+
+        findings.append(
+            Finding(
+                query="data_exfiltration",
+                title="Data Exfiltration Paths Detected",
+                description=(
+                    f"{len(exfil_paths)} exfiltration path(s) found from "
+                    f"{len(readable)} readable filesystem path(s) to external endpoints."
+                ),
+                risk=risk,
+                paths=exfil_paths,
+                remediation=remediation,
+            )
+        )
+
+    return findings
+
+
+def check_write_bypass(model: ReachabilityModel) -> list[Finding]:
+    """Check for write capabilities that bypass read-only policy intent."""
+    findings = []
+    bypass_paths: list[WriteBypassPath] = []
+
+    for policy_name, rule in model.policy.network_policies.items():
+        for ep in rule.endpoints:
+            if ep.intent not in (PolicyIntent.READ_ONLY, PolicyIntent.L4_ONLY):
+                continue
+
+            for port in ep.effective_ports:
+                for b in rule.binaries:
+                    cap = model.binary_registry.get_or_unknown(b.path)
+
+                    # Check: binary bypasses L7 and can write
+                    if cap.bypasses_l7 and cap.can_write:
+                        creds = model.credentials.credentials_for_host(ep.host)
+                        api = model.credentials.api_for_host(ep.host)
+                        cred_actions = []
+                        for cred in creds:
+                            if api:
+                                for wa in api.write_actions_for_scopes(cred.scopes):
+                                    cred_actions.append(
+                                        f"{wa.method} {wa.path} ({wa.action})"
+                                    )
+                            else:
+                                cred_actions.append(
+                                    f"credential '{cred.name}' has scopes: {cred.scopes}"
+                                )
+
+                        if cred_actions or not creds:
+                            bypass_paths.append(
+                                WriteBypassPath(
+                                    binary=b.path,
+                                    endpoint_host=ep.host,
+                                    endpoint_port=port,
+                                    policy_name=policy_name,
+                                    policy_intent=ep.intent.value,
+                                    bypass_reason="l7_bypass_protocol",
+                                    credential_actions=cred_actions,
+                                )
+                            )
+
+                    # Check: L4-only endpoint + binary can construct HTTP + credential has write
+                    if not ep.is_l7_enforced and cap.can_construct_http:
+                        creds = model.credentials.credentials_for_host(ep.host)
+                        api = model.credentials.api_for_host(ep.host)
+                        cred_actions = []
+                        for cred in creds:
+                            if api:
+                                for wa in api.write_actions_for_scopes(cred.scopes):
+                                    cred_actions.append(
+                                        f"{wa.method} {wa.path} ({wa.action})"
+                                    )
+                            else:
+                                cred_actions.append(
+                                    f"credential '{cred.name}' has scopes: {cred.scopes}"
+                                )
+
+                        if cred_actions:
+                            bypass_paths.append(
+                                WriteBypassPath(
+                                    binary=b.path,
+                                    endpoint_host=ep.host,
+                                    endpoint_port=port,
+                                    policy_name=policy_name,
+                                    policy_intent=ep.intent.value,
+                                    bypass_reason="l4_only",
+                                    credential_actions=cred_actions,
+                                )
+                            )
+
+    if bypass_paths:
+        findings.append(
+            Finding(
+                query="write_bypass",
+                title="Write Bypass Detected — Read-Only Intent Violated",
+                description=(
+                    f"{len(bypass_paths)} path(s) allow write operations despite "
+                    f"read-only policy intent."
+                ),
+                risk=RiskLevel.HIGH,
+                paths=bypass_paths,
+                remediation=[
+                    "For L4-only endpoints: add `protocol: rest` with `access: read-only` "
+                    "to enable HTTP method filtering.",
+                    "For L7-bypassing binaries (git, ssh, nc): remove them from the policy's "
+                    "binary list if write access is not intended.",
+                    "Restrict credential scopes to read-only where possible.",
+                ],
+            )
+        )
+
+    return findings
+
+
+def check_binary_inheritance(model: ReachabilityModel) -> list[Finding]:
+    """Check for binaries that inherit network access through the ancestor chain."""
+    findings = []
+    inheritance_paths: list[InheritancePath] = []
+
+    for bpath in model.binary_paths:
+        children = model.binary_registry.transitive_spawns(bpath)
+        if not children:
+            continue
+
+        for eid in model.endpoints:
+            access_key = f"{bpath}:{eid.key}"
+            if access_key not in model.policy_allows:
+                continue
+
+            for child in children:
+                child_cap = model.binary_registry.get_or_unknown(child)
+                capabilities = []
+                if child_cap.can_write:
+                    capabilities.append("can_write")
+                if child_cap.bypasses_l7:
+                    capabilities.append("bypasses_l7")
+                if child_cap.can_exfiltrate:
+                    capabilities.append("can_exfiltrate")
+                if child_cap.can_construct_http:
+                    capabilities.append("can_construct_http")
+
+                if capabilities:
+                    inheritance_paths.append(
+                        InheritancePath(
+                            parent=bpath,
+                            child=child,
+                            endpoint_host=eid.host,
+                            endpoint_port=eid.port,
+                            child_capabilities=capabilities,
+                        )
+                    )
+
+    if inheritance_paths:
+        seen = set()
+        deduped = []
+        for ip in inheritance_paths:
+            key = (ip.parent, ip.child)
+            if key not in seen:
+                seen.add(key)
+                deduped.append(ip)
+
+        findings.append(
+            Finding(
+                query="binary_inheritance",
+                title="Binary Ancestor Chain — Transitive Access Inheritance",
+                description=(
+                    f"{len(deduped)} binary pair(s) where child processes inherit "
+                    f"network access from parent binaries via ancestor matching."
+                ),
+                risk=RiskLevel.MEDIUM,
+                paths=deduped,
+                remediation=[
+                    "Review which child processes are spawned by allowed binaries.",
+                    "If a child binary (e.g., git, curl) has write or L7-bypass capabilities "
+                    "that exceed the policy intent, consider removing the parent binary or "
+                    "restricting it to a more specific binary path.",
+                ],
+            )
+        )
+
+    return findings
+
+
+def check_inference_relay(_model: ReachabilityModel) -> list[Finding]:
+    """Flag inference.local as a potential side channel for policy bypass."""
+    return [
+        Finding(
+            query="inference_relay",
+            title="Inference Relay — inference.local Bypasses Policy",
+            description=(
+                "inference.local is always reachable from within the sandbox and bypasses "
+                "OPA policy evaluation entirely. If the backing model supports tool use "
+                "or function calling, the agent may instruct it to access external services, "
+                "creating a side channel outside the policy boundary."
+            ),
+            risk=RiskLevel.ADVISORY,
+            paths=[],
+            remediation=[
+                "Verify that the inference backend does not support unrestricted tool use "
+                "or code execution.",
+                "If using tool-use models, ensure the inference router restricts which tools "
+                "are available and what external access they can perform.",
+                "Consider auditing inference.local traffic for unexpected patterns.",
+            ],
+        )
+    ]
+
+
+def check_overpermissive_methods(model: ReachabilityModel) -> list[Finding]:
+    """Flag endpoints using wildcard or overly broad HTTP method rules."""
+    endpoint_hints: dict[str, tuple[set[str], str]] = {
+        "integrate.api.nvidia.com": (
+            {"POST"},
+            "NVIDIA Cloud Functions management APIs (DELETE /v2/nvcf/assets, "
+            "DELETE /v2/nvcf/deployments) share this host",
+        ),
+        "inference-api.nvidia.com": (
+            {"POST"},
+            "Inference endpoint — only POST to completion paths is needed",
+        ),
+        "api.anthropic.com": (
+            {"POST"},
+            "Anthropic management APIs (DELETE /v1/files, DELETE /v1/skills) share this host",
+        ),
+        "statsig.anthropic.com": (
+            {"POST"},
+            "Telemetry endpoint — only POST to ingest paths is needed",
+        ),
+        "sentry.io": (
+            {"POST"},
+            "Error reporting — only POST to ingest paths is needed; "
+            "Sentry management APIs (DELETE projects, releases) share this host",
+        ),
+        "registry.npmjs.org": (
+            {"GET"},
+            "Package installation is read-only (GET); wildcard allows npm publish (PUT)",
+        ),
+    }
+
+    paths: list[OverpermissiveMethodPath] = []
+
+    for policy_name, rule in model.policy.network_policies.items():
+        for ep in rule.endpoints:
+            if not ep.is_l7_enforced:
+                continue
+
+            allowed = ep.allowed_methods
+            if not allowed:
+                continue
+
+            has_wildcard = any(r.method == "*" for r in ep.rules)
+            has_delete = "DELETE" in allowed
+
+            if not has_wildcard and not has_delete:
+                continue
+
+            hint = endpoint_hints.get(ep.host)
+            if hint:
+                needed, risk_detail = hint
+                excess = allowed - needed - {"HEAD", "OPTIONS"}
+            else:
+                if not has_delete:
+                    continue
+                needed = allowed - {"DELETE"}
+                excess = {"DELETE"}
+                risk_detail = "DELETE method allowed but likely not required"
+
+            if not excess:
+                continue
+
+            for b in rule.binaries:
+                for port in ep.effective_ports:
+                    paths.append(
+                        OverpermissiveMethodPath(
+                            endpoint_host=ep.host,
+                            endpoint_port=port,
+                            policy_name=policy_name,
+                            binary=b.path,
+                            allowed_methods="*"
+                            if has_wildcard
+                            else ", ".join(sorted(allowed)),
+                            needed_methods=", ".join(sorted(needed)),
+                            excess_methods=", ".join(sorted(excess)),
+                            risk_detail=risk_detail,
+                        )
+                    )
+
+    if not paths:
+        return []
+
+    unique_endpoints = {
+        (p.endpoint_host, p.endpoint_port, p.policy_name) for p in paths
+    }
+
+    return [
+        Finding(
+            query="overpermissive_methods",
+            title="Overpermissive HTTP Methods — Destructive Operations Allowed",
+            description=(
+                f"{len(unique_endpoints)} endpoint(s) allow HTTP methods beyond what is needed. "
+                f'Wildcard (method: "*") or explicit DELETE on endpoints that share hosts with '
+                f"management APIs enables destructive operations."
+            ),
+            risk=RiskLevel.HIGH,
+            paths=paths,
+            remediation=[
+                'Replace `method: "*"` with explicit method lists matching actual usage.',
+                "Inference endpoints: restrict to `POST` on `/v1/chat/completions`, "
+                "`/v1/messages`, `/v1/responses`.",
+                "Telemetry endpoints: restrict to `POST` on known ingest paths.",
+                "Package registries: restrict to `GET` for installation (no publish).",
+            ],
+        )
+    ]
+
+
+def check_l4_policy_gaps(model: ReachabilityModel) -> list[Finding]:
+    """Flag endpoints without L7 enforcement where HTTP-capable binaries have access."""
+    paths: list[L4PolicyGapPath] = []
+
+    for policy_name, rule in model.policy.network_policies.items():
+        has_any_l7 = any(ep.is_l7_enforced for ep in rule.endpoints)
+
+        for ep in rule.endpoints:
+            if ep.is_l7_enforced:
+                continue
+
+            for b in rule.binaries:
+                cap = model.binary_registry.get_or_unknown(b.path)
+                if not (
+                    cap.can_construct_http or cap.bypasses_l7 or cap.can_exfiltrate
+                ):
+                    continue
+
+                detail_parts = []
+                if has_any_l7:
+                    l7_siblings = [
+                        f"{sib.host}:{sib.effective_ports[0]}"
+                        for sib in rule.endpoints
+                        if sib.is_l7_enforced and sib.effective_ports
+                    ]
+                    detail_parts.append(
+                        f"Inconsistent: sibling endpoint(s) {', '.join(l7_siblings)} "
+                        f"in '{policy_name}' have L7, but this one does not"
+                    )
+
+                if cap.can_construct_http:
+                    detail_parts.append(
+                        f"{b.path} can construct arbitrary HTTP — all methods/paths "
+                        f"pass uninspected"
+                    )
+                if cap.bypasses_l7:
+                    detail_parts.append(
+                        f"{b.path} uses non-HTTP protocol — even with L7, "
+                        f"traffic would bypass inspection"
+                    )
+
+                for port in ep.effective_ports:
+                    paths.append(
+                        L4PolicyGapPath(
+                            endpoint_host=ep.host,
+                            endpoint_port=port,
+                            policy_name=policy_name,
+                            binary=b.path,
+                            binary_can_http=cap.can_construct_http,
+                            binary_bypasses_l7=cap.bypasses_l7,
+                            has_sibling_l7=has_any_l7,
+                            detail="; ".join(detail_parts),
+                        )
+                    )
+
+    if not paths:
+        return []
+
+    has_inconsistent = any(p.has_sibling_l7 for p in paths)
+    unique_endpoints = {(p.endpoint_host, p.endpoint_port) for p in paths}
+
+    return [
+        Finding(
+            query="l4_policy_gaps",
+            title="L4-Only Endpoints — No HTTP Inspection",
+            description=(
+                f"{len(unique_endpoints)} endpoint(s) lack `protocol: rest` — all traffic "
+                f"passes without HTTP method or path enforcement. "
+                + (
+                    "Some share a policy group with L7-enforced siblings, indicating "
+                    "inconsistent enforcement."
+                    if has_inconsistent
+                    else "HTTP-capable binaries have unrestricted access."
+                )
+            ),
+            risk=RiskLevel.HIGH if has_inconsistent else RiskLevel.MEDIUM,
+            paths=paths,
+            remediation=[
+                "Add `protocol: rest` with `enforcement: enforce` and `tls: terminate` "
+                "to enable HTTP inspection.",
+                'Define explicit method and path rules (e.g., `allow: { method: GET, path: "/**" }`) '
+                "to restrict what traffic is permitted.",
+                "If L4-only is intentional (e.g., for binary protocols like git), "
+                "document the rationale and restrict the binary list tightly.",
+            ],
+        )
+    ]
+
+
+def run_all_queries(model: ReachabilityModel) -> list[Finding]:
+    """Run all verification queries and return findings."""
+    findings = []
+    findings.extend(check_data_exfiltration(model))
+    findings.extend(check_write_bypass(model))
+    findings.extend(check_overpermissive_methods(model))
+    findings.extend(check_l4_policy_gaps(model))
+    findings.extend(check_binary_inheritance(model))
+    findings.extend(check_inference_relay(model))
+    return findings

python/openshell/prover/registry/apis/github.yaml

@@ -0,0 +1,62 @@
+api: github
+host: api.github.com
+port: 443
+credential_type: github-pat
+
+scope_capabilities:
+  repo:
+    - { method: GET, path: "/repos/**", action: read_repo }
+    - { method: GET, path: "/user/repos", action: list_repos }
+    - { method: POST, path: "/repos/*/issues", action: create_issue }
+    - { method: POST, path: "/repos/*/issues/*/comments", action: comment_issue }
+    - { method: POST, path: "/repos/*/pulls", action: create_pr }
+    - { method: PUT, path: "/repos/*/contents/**", action: write_file }
+    - { method: DELETE, path: "/repos/*/contents/**", action: delete_file }
+    - { method: DELETE, path: "/repos/*/*", action: delete_repo }
+    - { method: POST, path: "/repos/*/git/**", action: write_git_data }
+    - { method: POST, path: "/repos/*/forks", action: fork_repo }
+    - { method: PATCH, path: "/repos/*/*", action: update_repo }
+    - { method: POST, path: "/repos/*/releases", action: create_release }
+    - { method: POST, path: "/repos/*/dispatches", action: trigger_workflow }
+  "read:org":
+    - { method: GET, path: "/orgs/**", action: read_org }
+    - { method: GET, path: "/user/orgs", action: list_orgs }
+  "write:org":
+    - { method: GET, path: "/orgs/**", action: read_org }
+    - { method: POST, path: "/orgs/*/repos", action: create_org_repo }
+    - { method: PUT, path: "/orgs/*/memberships/**", action: manage_members }
+  gist:
+    - { method: GET, path: "/gists/**", action: read_gists }
+    - { method: POST, path: "/gists", action: create_gist }
+    - { method: PATCH, path: "/gists/*", action: update_gist }
+    - { method: DELETE, path: "/gists/*", action: delete_gist }
+  "admin:repo_hook":
+    - { method: GET, path: "/repos/*/hooks/**", action: read_hooks }
+    - { method: POST, path: "/repos/*/hooks", action: create_hook }
+    - { method: DELETE, path: "/repos/*/hooks/*", action: delete_hook }
+
+action_risk:
+  read_repo: low
+  list_repos: low
+  create_issue: medium
+  comment_issue: medium
+  create_pr: medium
+  write_file: high
+  delete_file: high
+  delete_repo: critical
+  write_git_data: high
+  fork_repo: medium
+  update_repo: high
+  create_release: high
+  trigger_workflow: critical
+  read_org: low
+  list_orgs: low
+  create_org_repo: high
+  manage_members: critical
+  read_gists: low
+  create_gist: medium
+  update_gist: medium
+  delete_gist: medium
+  read_hooks: low
+  create_hook: critical
+  delete_hook: high

python/openshell/prover/registry/binaries/claude.yaml

@@ -0,0 +1,27 @@
+binary: /usr/local/bin/claude
+description: "Claude Code CLI — AI coding agent"
+protocols:
+  - name: anthropic-api
+    transport: https
+    description: "Anthropic API calls for inference"
+    bypasses_l7: false
+    actions:
+      - { name: inference, type: write, description: "Send prompts and receive completions" }
+  - name: http-via-tools
+    transport: https
+    description: "Claude can make HTTP requests via tool use (Bash, WebFetch)"
+    bypasses_l7: false
+    actions:
+      - { name: get, type: read }
+      - { name: post, type: write }
+      - { name: put, type: write }
+spawns:
+  - /usr/local/bin/node
+  - /usr/local/bin/python3.13
+  - /usr/bin/git
+  - /usr/bin/curl
+  - /usr/bin/ssh
+  - /usr/bin/nc
+can_exfiltrate: true
+exfil_mechanism: "Can read files and send contents via tool use, API calls, or spawned subprocesses"
+can_construct_http: true

python/openshell/prover/registry/binaries/curl.yaml

@@ -0,0 +1,17 @@
+binary: /usr/bin/curl
+description: "Command-line HTTP client"
+protocols:
+  - name: http
+    transport: https
+    description: "HTTP/HTTPS requests — can construct arbitrary method, path, headers, body"
+    bypasses_l7: false
+    actions:
+      - { name: get, type: read, description: "HTTP GET request" }
+      - { name: post, type: write, description: "HTTP POST with arbitrary body" }
+      - { name: put, type: write, description: "HTTP PUT with arbitrary body" }
+      - { name: delete, type: destructive, description: "HTTP DELETE request" }
+      - { name: upload, type: write, description: "File upload via multipart POST" }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "POST/PUT file contents to any reachable endpoint, or encode in URL query parameters"
+can_construct_http: true

python/openshell/prover/registry/binaries/dig.yaml

@@ -0,0 +1,13 @@
+binary: /usr/bin/dig
+description: "DNS lookup utility"
+protocols:
+  - name: dns
+    transport: udp
+    description: "DNS queries — can encode data in query names for DNS exfiltration"
+    bypasses_l7: true
+    actions:
+      - { name: query, type: read, description: "DNS lookup" }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "Encode data in DNS query labels (e.g., base64.data.attacker.com) — DNS exfiltration"
+can_construct_http: false

python/openshell/prover/registry/binaries/gh.yaml

@@ -0,0 +1,16 @@
+binary: /usr/bin/gh
+description: "GitHub CLI — REST API client for GitHub"
+protocols:
+  - name: github-rest
+    transport: https
+    description: "GitHub REST API via gh cli — uses standard HTTP, subject to L7 inspection"
+    bypasses_l7: false
+    actions:
+      - { name: api_read, type: read, description: "GET requests to GitHub API" }
+      - { name: api_write, type: write, description: "POST/PUT/PATCH requests (create issues, PRs, etc.)" }
+      - { name: api_delete, type: destructive, description: "DELETE requests (delete repos, branches, etc.)" }
+spawns:
+  - /usr/bin/git
+can_exfiltrate: true
+exfil_mechanism: "Create gists, issues, or PRs with file contents"
+can_construct_http: true

python/openshell/prover/registry/binaries/git.yaml

@@ -0,0 +1,25 @@
+binary: /usr/bin/git
+description: "Distributed version control system"
+protocols:
+  - name: git-smart-http
+    transport: https
+    description: "Git smart HTTP protocol over HTTPS CONNECT tunnel — not REST, bypasses L7 HTTP inspection"
+    bypasses_l7: true
+    actions:
+      - { name: clone, type: read, description: "Clone/fetch repository" }
+      - { name: push, type: write, description: "Push commits to remote" }
+      - { name: force_push, type: destructive, description: "Force push, rewriting remote history" }
+  - name: git-ssh
+    transport: ssh
+    description: "Git over SSH"
+    bypasses_l7: true
+    actions:
+      - { name: clone, type: read }
+      - { name: push, type: write }
+      - { name: force_push, type: destructive }
+spawns:
+  - /usr/lib/git-core/git-remote-https
+  - /usr/bin/ssh
+can_exfiltrate: true
+exfil_mechanism: "Commit data to repo and push, or encode in branch/tag names"
+can_construct_http: false

python/openshell/prover/registry/binaries/nano.yaml

@@ -0,0 +1,6 @@
+binary: /usr/bin/nano
+description: "Text editor"
+protocols: []
+spawns: []
+can_exfiltrate: false
+can_construct_http: false

python/openshell/prover/registry/binaries/nc.yaml

@@ -0,0 +1,15 @@
+binary: /usr/bin/nc
+description: "Netcat — arbitrary TCP/UDP connections"
+protocols:
+  - name: raw-tcp
+    transport: tcp
+    description: "Raw TCP connection — can send/receive arbitrary data"
+    bypasses_l7: true
+    actions:
+      - { name: connect, type: read, description: "Establish TCP connection" }
+      - { name: send, type: write, description: "Send arbitrary data over TCP" }
+      - { name: listen, type: read, description: "Listen for incoming connections" }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "Stream arbitrary data over raw TCP connection"
+can_construct_http: false

python/openshell/prover/registry/binaries/node.yaml

@@ -0,0 +1,18 @@
+binary: /usr/local/bin/node
+description: "Node.js runtime — general-purpose JavaScript runtime"
+protocols:
+  - name: http-programmatic
+    transport: https
+    description: "Node can construct arbitrary HTTP requests via fetch, axios, http module"
+    bypasses_l7: false
+    actions:
+      - { name: get, type: read }
+      - { name: post, type: write }
+      - { name: put, type: write }
+      - { name: delete, type: destructive }
+spawns:
+  - /usr/bin/curl
+  - /usr/bin/git
+can_exfiltrate: true
+exfil_mechanism: "Construct arbitrary HTTP requests, spawn subprocesses"
+can_construct_http: true

python/openshell/prover/registry/binaries/npm.yaml

@@ -0,0 +1,16 @@
+binary: /usr/local/bin/npm
+description: "Node Package Manager — installs and manages JavaScript packages"
+protocols:
+  - name: npm-registry
+    transport: https
+    description: "npm registry protocol for package installation and publishing"
+    bypasses_l7: false
+    actions:
+      - { name: install, type: read, description: "Download packages from registry" }
+      - { name: publish, type: write, description: "Publish packages to registry" }
+spawns:
+  - /usr/local/bin/node
+  - /usr/bin/git
+can_exfiltrate: true
+exfil_mechanism: "Can publish arbitrary package contents to npm registry; postinstall scripts can exfiltrate"
+can_construct_http: false

python/openshell/prover/registry/binaries/nslookup.yaml

@@ -0,0 +1,13 @@
+binary: /usr/bin/nslookup
+description: "DNS lookup utility (legacy)"
+protocols:
+  - name: dns
+    transport: udp
+    description: "DNS queries"
+    bypasses_l7: true
+    actions:
+      - { name: query, type: read }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "DNS exfiltration via crafted query names"
+can_construct_http: false

python/openshell/prover/registry/binaries/openclaw.yaml

@@ -0,0 +1,16 @@
+binary: /usr/local/bin/openclaw
+description: "OpenClaw agent CLI — manages agent sessions, plugins, and ClawHub integration"
+protocols:
+  - name: http-programmatic
+    transport: https
+    description: "OpenClaw makes HTTP requests to ClawHub, openclaw.ai, and plugin registries"
+    bypasses_l7: false
+    actions:
+      - { name: get, type: read }
+      - { name: post, type: write }
+spawns:
+  - /usr/local/bin/node
+  - /usr/local/bin/npm
+can_exfiltrate: true
+exfil_mechanism: "Can send data to ClawHub or plugin endpoints via HTTP"
+can_construct_http: true

python/openshell/prover/registry/binaries/ping.yaml

@@ -0,0 +1,12 @@
+binary: /usr/bin/ping
+description: "ICMP echo utility"
+protocols:
+  - name: icmp
+    transport: icmp
+    description: "ICMP echo — limited data channel"
+    bypasses_l7: true
+    actions:
+      - { name: ping, type: read }
+spawns: []
+can_exfiltrate: false
+can_construct_http: false

python/openshell/prover/registry/binaries/python3.yaml

@@ -0,0 +1,21 @@
+binary: /usr/local/bin/python3.13
+description: "Python interpreter — general-purpose runtime"
+protocols:
+  - name: http-programmatic
+    transport: https
+    description: "Python can construct arbitrary HTTP requests via urllib, httpx, requests"
+    bypasses_l7: false
+    actions:
+      - { name: get, type: read }
+      - { name: post, type: write }
+      - { name: put, type: write }
+      - { name: delete, type: destructive }
+spawns:
+  - /usr/bin/curl
+  - /usr/bin/git
+  - /usr/bin/ssh
+  - /usr/bin/nc
+  - /usr/bin/wget
+can_exfiltrate: true
+exfil_mechanism: "Construct arbitrary HTTP requests, write to files, spawn subprocesses"
+can_construct_http: true

python/openshell/prover/registry/binaries/ssh.yaml

@@ -0,0 +1,15 @@
+binary: /usr/bin/ssh
+description: "OpenSSH client"
+protocols:
+  - name: ssh
+    transport: ssh
+    description: "SSH protocol — encrypted tunnel, can forward ports and transfer files"
+    bypasses_l7: true
+    actions:
+      - { name: connect, type: read, description: "SSH shell connection" }
+      - { name: tunnel, type: write, description: "Port forwarding / tunnel" }
+      - { name: scp, type: write, description: "File transfer via SCP" }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "SCP files or pipe data through SSH tunnel"
+can_construct_http: false

python/openshell/prover/registry/binaries/traceroute.yaml

@@ -0,0 +1,10 @@
+binary: /usr/bin/traceroute
+description: "Network path tracing utility"
+protocols:
+  - name: icmp
+    transport: icmp
+    actions:
+      - { name: trace, type: read }
+spawns: []
+can_exfiltrate: false
+can_construct_http: false

python/openshell/prover/registry/binaries/vim.yaml

@@ -0,0 +1,6 @@
+binary: /usr/bin/vim
+description: "Text editor"
+protocols: []
+spawns: []
+can_exfiltrate: false
+can_construct_http: false

python/openshell/prover/registry/binaries/wget.yaml

@@ -0,0 +1,14 @@
+binary: /usr/bin/wget
+description: "Non-interactive network downloader"
+protocols:
+  - name: http
+    transport: https
+    description: "HTTP/HTTPS downloads — can also POST with --post-data"
+    bypasses_l7: false
+    actions:
+      - { name: download, type: read }
+      - { name: post, type: write, description: "POST via --post-data or --post-file" }
+spawns: []
+can_exfiltrate: true
+exfil_mechanism: "POST file contents via --post-file to any reachable endpoint"
+can_construct_http: true

python/openshell/prover/testdata/credentials.yaml

@@ -0,0 +1,9 @@
+credentials:
+  - name: github-pat
+    type: github-pat
+    scopes:
+      - repo
+    injected_via: GITHUB_TOKEN
+    target_hosts:
+      - api.github.com
+      - github.com

python/openshell/prover/testdata/empty_credentials.yaml

@@ -0,0 +1 @@
+credentials: []

python/openshell/prover/testdata/empty_policy.yaml

@@ -0,0 +1,14 @@
+version: 1
+
+filesystem_policy:
+  include_workdir: true
+  read_only:
+    - /usr
+    - /lib
+    - /etc
+  read_write:
+    - /sandbox
+    - /tmp
+
+# Empty network policies — deny all external. Only inference.local reachable.
+network_policies: {}

python/openshell/prover/testdata/policy.yaml

@@ -0,0 +1,33 @@
+version: 1
+
+filesystem_policy:
+  include_workdir: true
+  read_only:
+    - /usr
+    - /lib
+    - /proc
+    - /etc
+  read_write:
+    - /sandbox
+    - /tmp
+
+process:
+  run_as_user: sandbox
+  run_as_group: sandbox
+
+network_policies:
+  github_readonly:
+    name: github_readonly
+    endpoints:
+      - host: api.github.com
+        port: 443
+        protocol: rest
+        enforcement: enforce
+        access: read-only
+      - host: github.com
+        port: 443
+        # No protocol field — L4 only! Git wire protocol passes through.
+    binaries:
+      - { path: /usr/local/bin/claude }
+      - { path: /usr/bin/git }
+      - { path: /usr/bin/curl }

python/openshell/prover/z3_model.py

@@ -0,0 +1,296 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Encode OpenShell policy + credentials + binary capabilities as Z3 SMT constraints."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+import z3
+
+from .policy_parser import WRITE_METHODS, PolicyModel
+
+if TYPE_CHECKING:
+    from .binary_registry import BinaryRegistry
+    from .credential_loader import CredentialSet
+
+
+@dataclass
+class EndpointId:
+    """Unique identifier for an endpoint in the model."""
+
+    policy_name: str
+    host: str
+    port: int
+
+    @property
+    def key(self) -> str:
+        return f"{self.policy_name}:{self.host}:{self.port}"
+
+
+@dataclass
+class ReachabilityModel:
+    """Z3-backed reachability model for an OpenShell sandbox policy."""
+
+    policy: PolicyModel
+    credentials: CredentialSet
+    binary_registry: BinaryRegistry
+
+    # Indexed facts derived from the inputs
+    endpoints: list[EndpointId] = field(default_factory=list)
+    binary_paths: list[str] = field(default_factory=list)
+
+    # Z3 boolean variables — indexed by string keys
+    # policy_allows[binary_key:endpoint_key] = True if policy allows this binary on this endpoint
+    policy_allows: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # l7_enforced[endpoint_key] = True if endpoint has protocol field
+    l7_enforced: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # l7_allows_write[endpoint_key] = True if L7 rules permit write methods
+    l7_allows_write: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # binary_bypasses_l7[binary_path] = True if binary uses non-HTTP protocol
+    binary_bypasses_l7: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # binary_can_write[binary_path] = True if binary can perform write actions
+    binary_can_write: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # binary_can_exfil[binary_path] = True if binary can exfiltrate data
+    binary_can_exfil: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # binary_can_construct_http[binary_path] = True if binary can make arbitrary HTTP requests
+    binary_can_construct_http: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # credential_has_write[host] = True if any credential for this host has write capabilities
+    credential_has_write: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # credential_has_destructive[host] = True if any credential has destructive capabilities
+    credential_has_destructive: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # spawns[parent:child] = True if parent spawns child
+    spawns_rel: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # inherits_access[child:endpoint_key] = True if child inherits access to endpoint
+    inherits_access: dict[str, z3.BoolRef] = field(default_factory=dict)
+    # filesystem_readable[path] = True
+    filesystem_readable: dict[str, z3.BoolRef] = field(default_factory=dict)
+
+    solver: z3.Solver = field(default_factory=z3.Solver)
+
+    def __post_init__(self):
+        self._build()
+
+    def _build(self):
+        """Build all Z3 constraints from the inputs."""
+        self._index_endpoints()
+        self._index_binaries()
+        self._encode_policy_allows()
+        self._encode_l7_enforcement()
+        self._encode_binary_capabilities()
+        self._encode_credentials()
+        self._encode_spawn_chains()
+        self._encode_filesystem()
+
+    def _index_endpoints(self):
+        """Collect all unique endpoints from the policy."""
+        for policy_name, rule in self.policy.network_policies.items():
+            for ep in rule.endpoints:
+                for port in ep.effective_ports:
+                    eid = EndpointId(policy_name, ep.host, port)
+                    self.endpoints.append(eid)
+
+    def _index_binaries(self):
+        """Collect all unique binary paths from the policy."""
+        seen = set()
+        for rule in self.policy.network_policies.values():
+            for b in rule.binaries:
+                if b.path not in seen:
+                    seen.add(b.path)
+                    self.binary_paths.append(b.path)
+
+    def _encode_policy_allows(self):
+        """Encode which (binary, endpoint) pairs the policy allows."""
+        for policy_name, rule in self.policy.network_policies.items():
+            for ep in rule.endpoints:
+                for port in ep.effective_ports:
+                    eid = EndpointId(policy_name, ep.host, port)
+                    for b in rule.binaries:
+                        key = f"{b.path}:{eid.key}"
+                        var = z3.Bool(f"policy_allows_{key}")
+                        self.policy_allows[key] = var
+                        self.solver.add(var == True)  # noqa: E712 — Z3 constraint
+
+    def _encode_l7_enforcement(self):
+        """Encode which endpoints have L7 enforcement and what they allow."""
+        for policy_name, rule in self.policy.network_policies.items():
+            for ep in rule.endpoints:
+                for port in ep.effective_ports:
+                    eid = EndpointId(policy_name, ep.host, port)
+                    ek = eid.key
+
+                    # L7 enforced?
+                    l7_var = z3.Bool(f"l7_enforced_{ek}")
+                    self.l7_enforced[ek] = l7_var
+                    self.solver.add(l7_var == ep.is_l7_enforced)
+
+                    # L7 allows write?
+                    allowed = ep.allowed_methods
+                    has_write = bool(allowed & WRITE_METHODS) if allowed else True
+                    l7w_var = z3.Bool(f"l7_allows_write_{ek}")
+                    self.l7_allows_write[ek] = l7w_var
+                    if ep.is_l7_enforced:
+                        self.solver.add(l7w_var == has_write)
+                    else:
+                        # L4-only: all methods pass through
+                        self.solver.add(l7w_var == True)  # noqa: E712
+
+    def _encode_binary_capabilities(self):
+        """Encode binary capabilities from the registry."""
+        for bpath in self.binary_paths:
+            cap = self.binary_registry.get_or_unknown(bpath)
+
+            bypass_var = z3.Bool(f"binary_bypasses_l7_{bpath}")
+            self.binary_bypasses_l7[bpath] = bypass_var
+            self.solver.add(bypass_var == cap.bypasses_l7)
+
+            write_var = z3.Bool(f"binary_can_write_{bpath}")
+            self.binary_can_write[bpath] = write_var
+            self.solver.add(write_var == cap.can_write)
+
+            exfil_var = z3.Bool(f"binary_can_exfil_{bpath}")
+            self.binary_can_exfil[bpath] = exfil_var
+            self.solver.add(exfil_var == cap.can_exfiltrate)
+
+            http_var = z3.Bool(f"binary_can_construct_http_{bpath}")
+            self.binary_can_construct_http[bpath] = http_var
+            self.solver.add(http_var == cap.can_construct_http)
+
+    def _encode_credentials(self):
+        """Encode credential capabilities per host."""
+        hosts = {eid.host for eid in self.endpoints}
+
+        for host in hosts:
+            creds = self.credentials.credentials_for_host(host)
+            api = self.credentials.api_for_host(host)
+
+            has_write = False
+            has_destructive = False
+
+            for cred in creds:
+                if api:
+                    write_actions = api.write_actions_for_scopes(cred.scopes)
+                    destructive_actions = api.destructive_actions_for_scopes(
+                        cred.scopes
+                    )
+                    if write_actions:
+                        has_write = True
+                    if destructive_actions:
+                        has_destructive = True
+                elif cred.scopes:
+                    # No API registry — conservatively assume credential enables writes
+                    has_write = True
+
+            cw_var = z3.Bool(f"credential_has_write_{host}")
+            self.credential_has_write[host] = cw_var
+            self.solver.add(cw_var == has_write)
+
+            cd_var = z3.Bool(f"credential_has_destructive_{host}")
+            self.credential_has_destructive[host] = cd_var
+            self.solver.add(cd_var == has_destructive)
+
+    def _encode_spawn_chains(self):
+        """Encode binary spawn relationships and transitive access inheritance."""
+        for bpath in self.binary_paths:
+            cap = self.binary_registry.get_or_unknown(bpath)
+            for child in cap.spawns:
+                key = f"{bpath}:{child}"
+                var = z3.Bool(f"spawns_{key}")
+                self.spawns_rel[key] = var
+                self.solver.add(var == True)  # noqa: E712
+
+        # Encode transitive access inheritance
+        for bpath in self.binary_paths:
+            all_children = self.binary_registry.transitive_spawns(bpath)
+            for eid in self.endpoints:
+                parent_key = f"{bpath}:{eid.key}"
+                if parent_key in self.policy_allows:
+                    for child in all_children:
+                        inherit_key = f"{child}:{eid.key}"
+                        var = z3.Bool(f"inherits_access_{inherit_key}")
+                        self.inherits_access[inherit_key] = var
+                        self.solver.add(var == True)  # noqa: E712
+
+    def _encode_filesystem(self):
+        """Encode filesystem readability."""
+        for path in self.policy.filesystem_policy.readable_paths:
+            var = z3.Bool(f"fs_readable_{path}")
+            self.filesystem_readable[path] = var
+            self.solver.add(var == True)  # noqa: E712
+
+    # --- Query helpers ---
+
+    def _has_access(self, bpath: str, ek: str) -> z3.BoolRef | None:
+        """Return Z3 expression for access (direct or inherited), or None."""
+        access_key = f"{bpath}:{ek}"
+        direct = self.policy_allows.get(access_key)
+        inherited = self.inherits_access.get(access_key)
+        if direct is not None and inherited is not None:
+            return z3.Or(direct, inherited)
+        if direct is not None:
+            return direct
+        return inherited  # may be None
+
+    def can_write_to_endpoint(self, bpath: str, eid: EndpointId) -> z3.BoolRef:
+        """Return a Z3 expression for whether a binary can write to an endpoint."""
+        ek = eid.key
+
+        has_access = self._has_access(bpath, ek)
+        if has_access is None:
+            return z3.BoolVal(False)
+
+        bypass = self.binary_bypasses_l7.get(bpath, z3.BoolVal(False))
+        l7_enforced = self.l7_enforced.get(ek, z3.BoolVal(False))
+        l7_write = self.l7_allows_write.get(ek, z3.BoolVal(False))
+        binary_write = self.binary_can_write.get(bpath, z3.BoolVal(False))
+        cred_write = self.credential_has_write.get(eid.host, z3.BoolVal(False))
+
+        return z3.And(
+            has_access,
+            binary_write,
+            z3.Or(
+                z3.Not(l7_enforced),  # L4-only
+                l7_write,  # L7 allows write
+                bypass,  # binary bypasses L7
+            ),
+            cred_write,
+        )
+
+    def can_exfil_via_endpoint(self, bpath: str, eid: EndpointId) -> z3.BoolRef:
+        """Return a Z3 expression for whether data can be exfiltrated via this path."""
+        ek = eid.key
+
+        has_access = self._has_access(bpath, ek)
+        if has_access is None:
+            return z3.BoolVal(False)
+
+        exfil = self.binary_can_exfil.get(bpath, z3.BoolVal(False))
+        bypass = self.binary_bypasses_l7.get(bpath, z3.BoolVal(False))
+        l7_enforced = self.l7_enforced.get(ek, z3.BoolVal(False))
+        l7_write = self.l7_allows_write.get(ek, z3.BoolVal(False))
+        http = self.binary_can_construct_http.get(bpath, z3.BoolVal(False))
+
+        return z3.And(
+            has_access,
+            exfil,
+            z3.Or(
+                z3.And(z3.Not(l7_enforced), http),  # L4-only + HTTP
+                z3.And(l7_write, http),  # L7 write + HTTP
+                bypass,  # non-HTTP protocol bypass
+            ),
+        )
+
+
+def build_model(
+    policy: PolicyModel,
+    credentials: CredentialSet,
+    binary_registry: BinaryRegistry,
+) -> ReachabilityModel:
+    """Build a Z3 reachability model from policy, credentials, and binary registry."""
+    return ReachabilityModel(
+        policy=policy,
+        credentials=credentials,
+        binary_registry=binary_registry,
+    )