diff --git a/architecture/policy-prover.md b/architecture/policy-prover.md index 623d138f9..3ee5e9ee8 100644 --- a/architecture/policy-prover.md +++ b/architecture/policy-prover.md @@ -2,9 +2,12 @@ Formal verification for AI agent sandbox policies. -OPP answers the question that no other agent security tool can: *"What can this agent actually do?"* — not what the policy says, but what is mathematically reachable given the full interaction of policy rules, credential scopes, binary capabilities, and enforcement layers. +OPP answers two yes/no security questions about any sandbox policy: -Every policy has properties. OPP proves them — or shows you exactly how they break. +1. **"Can data leave this sandbox?"** — exfiltration analysis +2. **"Can the agent write despite read-only intent?"** — write bypass detection + +Any finding = FAIL. No findings = PASS. Like AWS Zelkova for S3 bucket policies, but for agent sandboxes. ## The Problem @@ -18,78 +21,29 @@ This isn't a bug in the sandbox. Each layer works correctly in isolation. The pr ## The Approach -OPP uses **formal methods** — the same class of techniques AWS uses to verify S3 bucket policies and IAM permissions — to reason about the complete enforcement model. +OPP uses **Z3 SMT solving** — the same class of formal methods AWS uses to verify S3 bucket policies — to reason about the complete enforcement model. -The tool takes three inputs: +Three inputs: 1. **The sandbox policy** (OpenShell YAML) — what the proxy enforces 2. **Credential descriptors** — what the injected tokens can actually authorize 3. **Binary capability registry** — what each binary in the sandbox can do -It encodes these as constraints in a **Z3 SMT solver** and asks reachability questions. The solver doesn't guess or estimate — it either proves a property holds or produces a concrete counterexample showing exactly how it can be violated. - -### Formal Verification vs. Other Approaches - -| Approach | Can it find gaps? | Guarantees? | Scales? | -|----------|------------------|-------------|---------| -| **Manual review** | Sometimes | None — depends on reviewer expertise | No | -| **LLM classifier** | Often | None — probabilistic, can be bypassed | Yes | -| **Penetration testing** | If the tester thinks of it | None — absence of evidence isn't evidence of absence | No | -| **Formal verification (OPP)** | **Always, if the model covers it** | **Mathematical proof** — if it says safe, it's safe | **Yes — solver handles combinatorial explosion** | - -The key distinction: an LLM classifier says "this action *looks* safe." OPP says "there is *no possible path* to this action" — or produces the exact path if one exists. - -This is not a replacement for runtime enforcement (the proxy, OPA engine, Landlock, seccomp). It's a **design-time verification layer** that catches gaps before the agent runs. Defense in depth: enforce at runtime, prove at authoring time. +It encodes these as Z3 constraints and asks reachability questions. The solver either proves a property holds or produces a concrete counterexample showing exactly how it can be violated. ## What It Detects ### Data Exfiltration Paths -*"Can data leave the sandbox?"* +*"Can data leave this sandbox?"* -OPP maps every path from readable filesystem locations to writable egress channels: L4-only endpoints where binaries can POST freely, git push channels, raw TCP via netcat, DNS exfiltration, even inference relay side channels. If a path exists, the solver finds it. - -**Example finding:** -``` -CRITICAL Data exfiltration possible - L4-only: github.com:443; wire protocol bypass: api.github.com:443 -``` +Maps every path from readable filesystem locations (including the workdir) to writable egress channels: L4-only endpoints where binaries can POST freely, git push channels, raw TCP via netcat. If a path exists, the solver finds it. ### Write Bypass Detection -*"Can the agent modify something despite a read-only policy?"* +*"Can the agent write despite read-only intent?"* -OPP checks whether any combination of binary capabilities, credential scopes, and transport protocols allows write operations that the policy intends to block. - -**Example finding:** -``` -HIGH Write bypass — read-only intent violated - L4-only + wire protocol: api.github.com:443, github.com:443 -``` - -### Overpermissive HTTP Methods - -*"Does this policy allow destructive operations the agent doesn't need?"* - -OPP flags `method: "*"` wildcards and unnecessary `DELETE` access on endpoints that share hosts with management APIs. An inference endpoint that only needs `POST /v1/chat/completions` shouldn't also allow `DELETE /v2/nvcf/assets/{id}`. - -### L4 Policy Gaps - -*"Are there endpoints with no HTTP inspection at all?"* - -OPP flags endpoints missing `protocol: rest` where HTTP-capable binaries have access — meaning all traffic passes uninspected. Especially flagged when sibling endpoints in the same policy group DO have L7 enforcement, indicating inconsistent configuration. - -### Binary Inheritance Audit - -*"Which processes inherit network access they weren't explicitly granted?"* - -OpenShell's binary matching includes ancestor chain matching — if `claude` is allowed, any process `claude` spawns inherits that access. OPP traces the full inheritance tree and flags binaries with capabilities that exceed the policy's apparent intent. - -### Inference Relay Risk - -*"Can the agent reach external resources through inference.local?"* - -The `inference.local` virtual host always bypasses OPA policy evaluation. If the backing model supports tool use or function calling, the agent may be able to instruct it to access external services — creating a policy-free side channel. +Checks whether any combination of binary capabilities, credential scopes, and transport protocols allows write operations that the policy intends to block. ## How It Works @@ -101,38 +55,23 @@ Credentials ───┼───>│ Z3 Constraint │───>│ SAT? Binaries ──────┘ └─────────────────┘ └───────────┘ ``` -1. **Parse** the policy YAML into an enforcement model: which endpoints are L4-only vs L7-enforced, what methods/paths are allowed, which binaries have access. +1. **Parse** the policy YAML into an enforcement model +2. **Load** credential descriptors mapping token scopes to API capabilities +3. **Load** binary capability descriptors (protocols, spawn chains, exfil ability) +4. **Encode** as Z3 SMT constraints +5. **Query** the solver: exfiltration possible? Write bypass possible? +6. **Report** findings with concrete paths and remediation -2. **Load** credential descriptors that map token scopes to concrete API capabilities (e.g., GitHub PAT `repo` scope -> can push, create issues, delete repos). +### Accepted Risks -3. **Load** binary capability descriptors that describe what each binary can do: protocols it speaks, whether it bypasses L7, what child processes it spawns, whether it can exfiltrate data. - -4. **Encode** all of the above as Z3 SMT constraints — boolean variables and logical implications representing every possible action path. - -5. **Query** the solver: "Is data exfiltration possible?" "Can the agent write despite read-only intent?" "Which binaries inherit unintended access?" - -6. **Report** findings with concrete paths, risk ratings, and remediation suggestions. - -### The Knowledge Base - -OPP's accuracy depends on the quality of its capability descriptors — how well it models what each API, credential type, and binary can do. - -**Trusted sources only.** OPP restricts its knowledge base to: - -- **Published API specs** (OpenAPI/Swagger) for well-known services -- **Curated binary descriptors** for the binaries that ship in the base sandbox image (~17 binaries with well-known, stable behavior) -- **Credential scope documentation** from the credential provider (e.g., GitHub's PAT scope definitions) - -Anything not in the knowledge base is flagged conservatively — OPP assumes unknown binaries can exfiltrate and construct HTTP. This is deliberate: a prover that claims more than it can prove is worse than no prover at all. - -The knowledge base is extensible. Registry YAML files live at `python/openshell/prover/registry/`. +Teams can acknowledge known paths via an accepted-risks YAML file. For example: "we accept that api.anthropic.com is an exfil path because that's our inference provider." Accepted findings show as PASS with a note, not FAIL. ## Usage ### CLI ```bash -# Via the Rust CLI (recommended) +# Via the Rust CLI openshell policy prove \ --policy policy.yaml \ --credentials credentials.yaml @@ -143,12 +82,6 @@ openshell policy prove \ --credentials credentials.yaml \ --compact -# HTML report with interactive Mermaid diagrams -openshell policy prove \ - --policy policy.yaml \ - --credentials credentials.yaml \ - --html report.html - # With accepted risks openshell policy prove \ --policy policy.yaml \ @@ -161,7 +94,7 @@ python3 -m openshell.prover.cli prove \ --credentials credentials.yaml ``` -**Exit codes:** 0 = PASS (advisories only), 1 = FAIL (critical/high gaps), 2 = input error. +**Exit codes:** 0 = PASS, 1 = FAIL (exfiltration or write bypass found), 2 = input error. ### Python API @@ -192,58 +125,26 @@ The `harden-sandbox-policy` skill wraps OPP for iterative policy hardening: ## Architecture -### Source Layout - ``` python/openshell/prover/ - cli.py # CLI entry point (argparse) + cli.py # CLI entry point policy_parser.py # OpenShell YAML -> PolicyModel - z3_model.py # PolicyModel + credentials + binaries -> Z3 constraints - queries.py # Verification queries (6 query types) + z3_model.py # Z3 constraint encoding + queries.py # 2 verification queries binary_registry.py # Binary capability descriptors credential_loader.py # Credential + API capability descriptors - report.py # Terminal, compact, Mermaid, and HTML output + report.py # Terminal and compact output accepted_risks.py # Risk acceptance filtering - registry/ # YAML capability descriptors (binaries, APIs, credentials) + registry/ # YAML capability descriptors ``` -### Rust CLI Integration - -The `openshell policy prove` subcommand in `crates/openshell-cli/src/main.rs` delegates to the Python prover via `uv run python3 -m openshell.prover.cli`. This keeps the prover logic in Python (where Z3 bindings are most mature) while providing a unified CLI experience. +The `openshell policy prove` Rust CLI subcommand delegates to the Python prover via `uv run python3 -m openshell.prover.cli`. ### Dependencies -The prover is an optional dependency group to avoid bloating the base SDK: +Optional dependency group — does not bloat the base SDK: ```toml [project.optional-dependencies] prover = ["z3-solver>=4.12", "pyyaml>=6.0", "rich>=13.0"] ``` - -Install with: `uv pip install '.[prover]'` or include in dev dependencies via `uv sync`. - -## Differentiation - -### vs. OPA / Rego - -OPA evaluates policies at runtime on a per-request basis using Rego rules. OPP operates at design time and reasons about the *complete enforcement model* — all possible request paths, not just the one being evaluated now. - -### vs. Static Policy Linting - -Tools like `opa check` or conftest validate that a policy is syntactically correct and internally consistent. They don't model what the policy *allows in practice* given credentials, binary capabilities, and layer interactions. - -### vs. Penetration Testing - -A pentest might find the git push bypass — if the tester thinks to try it. OPP finds it automatically, systematically, and exhaustively (within its model). And it runs in seconds, not days. - -### vs. LLM Classifiers - -An LLM classifier says "this action *looks* safe." OPP says "there is *no possible path* to this action." The two are complementary — classifiers catch runtime misbehavior, OPP catches policy design flaws. - -| | LLM Classifier | Policy Prover | -|---|---|---| -| **When** | Runtime (per-action) | Design time (per-policy) | -| **Method** | LLM classification | SMT formal verification | -| **Guarantee** | Probabilistic | Mathematical proof | -| **Catches** | Agent deviating from intent | Policy gaps across layers | -| **Vulnerable to** | Prompt injection, false negatives | Incomplete capability model | diff --git a/crates/openshell-cli/src/main.rs b/crates/openshell-cli/src/main.rs index 6c572c981..173c33f01 100644 --- a/crates/openshell-cli/src/main.rs +++ b/crates/openshell-cli/src/main.rs @@ -1421,9 +1421,8 @@ enum PolicyCommands { /// Formally verify a policy using the OpenShell Policy Prover (OPP). /// - /// Runs Z3-based verification queries against a sandbox policy to detect - /// data exfiltration paths, write bypass attacks, L7 enforcement gaps, - /// and privilege escalation via binary capabilities. + /// Answers two questions: "Can data leave this sandbox?" and + /// "Can the agent write despite read-only intent?" #[command(help_template = LEAF_HELP_TEMPLATE, next_help_heading = "FLAGS")] Prove { /// Path to the sandbox policy YAML file. @@ -1445,10 +1444,6 @@ enum PolicyCommands { /// One-line-per-finding output for CI. #[arg(long)] compact: bool, - - /// Output self-contained HTML report to this path. - #[arg(long, value_hint = ValueHint::FilePath)] - html: Option, }, } @@ -1941,10 +1936,11 @@ async fn main() -> Result<()> { registry, accepted_risks, compact, - html, } => { let mut cmd = std::process::Command::new("uv"); cmd.args(["run", "python3", "-m", "openshell.prover.cli", "prove"]); + // Ensure openshell package is importable from the repo's python/ dir + cmd.env("PYTHONPATH", "python"); cmd.args(["--policy", &policy]); cmd.args(["--credentials", &credentials]); if let Some(ref reg) = registry { @@ -1956,9 +1952,6 @@ async fn main() -> Result<()> { if compact { cmd.arg("--compact"); } - if let Some(ref html_path) = html { - cmd.args(["--html", html_path]); - } let status = cmd .status() .map_err(|e| miette::miette!("failed to execute policy prover: {e} — requires 'uv' and prover deps (uv sync)"))?; diff --git a/python/openshell/__init__.py b/python/openshell/__init__.py index 8f9f8eb9d..db492ff9d 100644 --- a/python/openshell/__init__.py +++ b/python/openshell/__init__.py @@ -10,7 +10,7 @@ import contextlib # Proto stubs may not be generated yet (requires Rust build). # Suppress so subpackages like openshell.prover can still be imported. with contextlib.suppress(ImportError): - from .sandbox import ( + from .sandbox import ( # noqa: F401 — intentional re-exports ClusterInferenceConfig, ExecChunk, ExecResult, @@ -30,7 +30,7 @@ try: except Exception: __version__ = "0.0.0" -__all__ = [ +_SDK_SYMBOLS = [ "ClusterInferenceConfig", "ExecChunk", "ExecResult", @@ -41,5 +41,6 @@ __all__ = [ "SandboxRef", "SandboxSession", "TlsConfig", - "__version__", ] + +__all__ = [s for s in _SDK_SYMBOLS if s in dir()] + ["__version__"] diff --git a/python/openshell/prover/accepted_risks.py b/python/openshell/prover/accepted_risks.py index 7a6b4aad3..a474d05b7 100644 --- a/python/openshell/prover/accepted_risks.py +++ b/python/openshell/prover/accepted_risks.py @@ -20,7 +20,6 @@ if TYPE_CHECKING: class AcceptedRisk: query: str reason: str - accepted_by: str = "" binary: str = "" endpoint: str = "" @@ -34,7 +33,6 @@ def load_accepted_risks(path: Path) -> list[AcceptedRisk]: AcceptedRisk( query=r.get("query", ""), reason=r.get("reason", ""), - accepted_by=r.get("accepted_by", ""), binary=r.get("binary", ""), endpoint=r.get("endpoint", ""), ) @@ -45,7 +43,7 @@ def load_accepted_risks(path: Path) -> list[AcceptedRisk]: 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", "") + path_binary = getattr(path, "binary", "") if path_binary != risk.binary: return False @@ -77,21 +75,6 @@ def apply_accepted_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: diff --git a/python/openshell/prover/cli.py b/python/openshell/prover/cli.py index 2e983a870..29c771638 100644 --- a/python/openshell/prover/cli.py +++ b/python/openshell/prover/cli.py @@ -15,7 +15,7 @@ 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 .report import render_compact, render_report from .z3_model import build_model _DEFAULT_REGISTRY = Path(__file__).parent / "registry" @@ -59,29 +59,7 @@ def main(): 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)", + help="One-line-per-finding output (for CI)", ) args = parser.parse_args() @@ -97,9 +75,8 @@ def main(): def cmd_prove(args) -> int: """Execute the prove command.""" - console = Console(force_terminal=args.color) if args.color else Console() + console = Console() - # Load inputs try: policy = parse_policy(args.policy) except Exception as e: @@ -118,13 +95,9 @@ def cmd_prove(args) -> int: 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 @@ -135,36 +108,7 @@ def cmd_prove(args) -> int: 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 - + if args.compact: return render_compact( findings, str(args.policy), diff --git a/python/openshell/prover/policy_parser.py b/python/openshell/prover/policy_parser.py index 35b59b218..b6d6ebdb6 100644 --- a/python/openshell/prover/policy_parser.py +++ b/python/openshell/prover/policy_parser.py @@ -113,12 +113,16 @@ class NetworkPolicyRule: @dataclass class FilesystemPolicy: include_workdir: bool = True + workdir: str = "/sandbox" 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 + paths = self.read_only + self.read_write + if self.include_workdir and self.workdir not in paths: + paths = [self.workdir, *paths] + return paths @dataclass @@ -170,6 +174,7 @@ def parse_policy(path: Path) -> PolicyModel: fs_raw = raw.get("filesystem_policy", {}) or {} fs = FilesystemPolicy( include_workdir=fs_raw.get("include_workdir", True), + workdir=fs_raw.get("workdir", "/sandbox"), read_only=fs_raw.get("read_only", []), read_write=fs_raw.get("read_write", []), ) diff --git a/python/openshell/prover/prover_test.py b/python/openshell/prover/prover_test.py index 2450b5f3b..081db4d55 100644 --- a/python/openshell/prover/prover_test.py +++ b/python/openshell/prover/prover_test.py @@ -7,7 +7,7 @@ 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.policy_parser import FilesystemPolicy, PolicyIntent, parse_policy from openshell.prover.queries import RiskLevel, run_all_queries from openshell.prover.z3_model import build_model @@ -33,13 +33,11 @@ def test_parse_policy(): 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 @@ -47,13 +45,33 @@ def test_parse_policy(): def test_filesystem_policy(): - """Verify filesystem policy parsing.""" + """Verify filesystem policy parsing including workdir.""" 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 + assert "/usr" in fs.readable_paths + + +def test_include_workdir_default(): + """Workdir is included in readable_paths by default.""" + fs = FilesystemPolicy(read_only=["/usr"]) + assert "/sandbox" in fs.readable_paths + assert "/usr" in fs.readable_paths + + +def test_include_workdir_false(): + """Workdir excluded when include_workdir is False.""" + fs = FilesystemPolicy(include_workdir=False, read_only=["/usr"]) + assert "/sandbox" not in fs.readable_paths + assert "/usr" in fs.readable_paths + + +def test_include_workdir_no_duplicate(): + """Workdir not duplicated if already in read_write.""" + fs = FilesystemPolicy(read_write=["/sandbox"]) + assert fs.readable_paths.count("/sandbox") == 1 def test_git_push_bypass_findings(): @@ -66,7 +84,9 @@ def test_git_push_bypass_findings(): assert RiskLevel.HIGH in risks, "Should detect write bypass" assert "data_exfiltration" in queries assert "write_bypass" in queries - assert "binary_inheritance" in queries + + # Only 2 query types in v1 + assert len(queries) == 2 # Verify git bypass specifically detected write_findings = [f for f in findings if f.query == "write_bypass"] @@ -75,14 +95,7 @@ def test_git_push_bypass_findings(): 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.""" +def test_empty_policy_no_findings(): + """Deny-all policy should produce no findings.""" 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 + assert len(findings) == 0 diff --git a/python/openshell/prover/queries.py b/python/openshell/prover/queries.py index 0ed287a4a..a43632ee1 100644 --- a/python/openshell/prover/queries.py +++ b/python/openshell/prover/queries.py @@ -1,7 +1,11 @@ # 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.""" +"""Verification queries — two Zelkova-like yes/no security questions. + +1. Can data leave this sandbox? (exfiltration) +2. Can the agent write despite read-only intent? (write bypass) +""" from __future__ import annotations @@ -18,11 +22,8 @@ if TYPE_CHECKING: class RiskLevel(Enum): - LOW = "low" - MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" - ADVISORY = "advisory" @dataclass @@ -46,49 +47,10 @@ class WriteBypassPath: endpoint_port: int policy_name: str policy_intent: str - bypass_reason: str # "l4_only", "l7_bypass_protocol", "credential_write_scope" + bypass_reason: str # "l4_only", "l7_bypass_protocol" 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.""" @@ -97,20 +59,18 @@ class Finding: title: str description: str risk: RiskLevel - paths: list[ - ExfilPath - | WriteBypassPath - | InheritancePath - | OverpermissiveMethodPath - | L4PolicyGapPath - ] = field(default_factory=list) + paths: list[ExfilPath | WriteBypassPath] = 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.""" + """Can data leave this sandbox? + + Maps every path from readable filesystem locations to writable egress + channels: L4-only endpoints, wire protocol bypasses, and L7 write channels. + """ findings = [] if not model.policy.filesystem_policy.readable_paths: @@ -125,13 +85,11 @@ def check_data_exfiltration(model: ReachabilityModel) -> list[Finding]: 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 @@ -210,7 +168,11 @@ def check_data_exfiltration(model: ReachabilityModel) -> list[Finding]: def check_write_bypass(model: ReachabilityModel) -> list[Finding]: - """Check for write capabilities that bypass read-only policy intent.""" + """Can the agent write despite read-only intent? + + Checks whether any combination of binary capabilities, credential scopes, + and transport protocols allows write operations that the policy intends to block. + """ findings = [] bypass_paths: list[WriteBypassPath] = [] @@ -305,304 +267,9 @@ def check_write_bypass(model: ReachabilityModel) -> list[Finding]: 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 diff --git a/python/openshell/prover/report.py b/python/openshell/prover/report.py index e4638424d..04c8d2c75 100644 --- a/python/openshell/prover/report.py +++ b/python/openshell/prover/report.py @@ -1,13 +1,11 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Rich terminal report, Mermaid graph, and HTML report generation for verification findings.""" +"""Terminal report and compact CI output for verification findings.""" from __future__ import annotations -import html as html_mod from pathlib import Path -from typing import TYPE_CHECKING from rich.console import Console from rich.panel import Panel @@ -18,31 +16,18 @@ from rich.tree import Tree from .queries import ( ExfilPath, Finding, - InheritancePath, - L4PolicyGapPath, - OverpermissiveMethodPath, RiskLevel, WriteBypassPath, ) -if TYPE_CHECKING: - from .binary_registry import BinaryRegistry - from .policy_parser import PolicyModel - RISK_COLORS = { RiskLevel.CRITICAL: "bold red", RiskLevel.HIGH: "red", - RiskLevel.MEDIUM: "yellow", - RiskLevel.LOW: "green", - RiskLevel.ADVISORY: "cyan", } RISK_ICONS = { RiskLevel.CRITICAL: "CRITICAL", RiskLevel.HIGH: "HIGH", - RiskLevel.MEDIUM: "MEDIUM", - RiskLevel.LOW: "LOW", - RiskLevel.ADVISORY: "ADVISORY", } @@ -53,9 +38,6 @@ def _hl(text: str) -> str: "L7": "[bold green]L7[/bold green]", "wire protocol": "[bold red]wire protocol[/bold red]", "no HTTP inspection": "[bold red]no HTTP inspection[/bold red]", - "inference.local": "[bold cyan]inference.local[/bold cyan]", - "github.com:443": "[bold]github.com:443[/bold]", - "api.github.com:443": "[bold]api.github.com:443[/bold]", } for term, markup in highlights.items(): text = text.replace(term, markup) @@ -96,25 +78,6 @@ def _compact_detail(finding: Finding) -> str: return f"L4-only (no inspection): {ep_list}" if "l7_bypass_protocol" in reasons: return f"wire protocol bypasses L7: {ep_list}" - return "" - elif finding.query == "overpermissive_methods": - endpoints = set() - for p in finding.paths: - if hasattr(p, "endpoint_host"): - endpoints.add(f"{p.endpoint_host}:{p.endpoint_port}") - return f'method: "*" or DELETE on: {", ".join(sorted(endpoints))}' - elif finding.query == "l4_policy_gaps": - endpoints = set() - inconsistent = False - for p in finding.paths: - if hasattr(p, "endpoint_host"): - endpoints.add(f"{p.endpoint_host}:{p.endpoint_port}") - if hasattr(p, "has_sibling_l7") and p.has_sibling_l7: - inconsistent = True - prefix = "inconsistent enforcement" if inconsistent else "no protocol: rest" - return f"{prefix}: {', '.join(sorted(endpoints))}" - elif finding.query == "binary_inheritance": - return f"{len(finding.paths)} via ancestor chain" return "" @@ -124,7 +87,7 @@ def render_compact( _credentials_path: str, console: Console | None = None, ) -> int: - """Compact output for demos and CI. Two lines per finding: summary + detail.""" + """Compact output for CI. Two lines per finding: summary + detail.""" if console is None or console.width > 72: console = Console(width=72) @@ -134,10 +97,6 @@ def render_compact( compact_titles = { "data_exfiltration": "Data exfiltration possible", "write_bypass": "Write bypass — read-only intent violated", - "overpermissive_methods": "Overpermissive HTTP methods — destructive ops allowed", - "l4_policy_gaps": "L4-only endpoints — no HTTP inspection", - "binary_inheritance": "Binaries inherit unintended access", - "inference_relay": "inference.local bypasses policy", } for finding in active: @@ -171,16 +130,14 @@ def render_compact( f" [bold white on red] FAIL [/bold white on red] {n} critical/high gaps{accepted_note}" ) return 1 - elif active: - console.print( - f" [bold black on yellow] PASS [/bold black on yellow] advisories only{accepted_note}" - ) - return 0 - else: + elif accepted_findings: console.print( f" [bold white on green] PASS [/bold white on green] all findings accepted{accepted_note}" ) return 0 + else: + console.print(" [bold white on green] PASS [/bold white on green] no findings") + return 0 def render_report( @@ -189,11 +146,10 @@ def render_report( credentials_path: str, console: Console | None = None, ) -> int: - """Render findings to the terminal. Returns exit code (0 = clean, 1 = critical/high).""" + """Render findings to the terminal. Returns exit code (0 = pass, 1 = fail).""" if console is None or console.width > 80: console = Console(width=80) - # Header policy_name = Path(policy_path).name creds_name = Path(credentials_path).name @@ -211,7 +167,6 @@ def render_report( active = [f for f in findings if not f.accepted] accepted_findings = [f for f in findings if f.accepted] - # Summary table counts = {} for f in active: counts[f.risk] = counts.get(f.risk, 0) + 1 @@ -219,13 +174,7 @@ def render_report( summary = Table(title="Finding Summary", show_header=True, border_style="dim") summary.add_column("Risk Level", style="bold") summary.add_column("Count", justify="right") - for level in [ - RiskLevel.CRITICAL, - RiskLevel.HIGH, - RiskLevel.MEDIUM, - RiskLevel.LOW, - RiskLevel.ADVISORY, - ]: + for level in [RiskLevel.CRITICAL, RiskLevel.HIGH]: if level in counts: style = RISK_COLORS[level] summary.add_row( @@ -279,31 +228,15 @@ def render_report( has_high = RiskLevel.HIGH in counts accepted_note = f" ({len(accepted_findings)} accepted)" if accepted_findings else "" - if has_critical: + if has_critical or has_high: console.print( Panel( - f"[bold red]FAIL[/bold red] — Critical gaps found.{accepted_note}", + f"[bold red]FAIL[/bold red] — Critical/high gaps found.{accepted_note}", border_style="red", ) ) return 1 - elif has_high: - console.print( - Panel( - f"[bold red]FAIL[/bold red] — High-risk gaps found.{accepted_note}", - border_style="red", - ) - ) - return 1 - elif active: - console.print( - Panel( - f"[bold yellow]PASS[/bold yellow] — Advisories only.{accepted_note}", - border_style="yellow", - ) - ) - return 0 - else: + elif accepted_findings: console.print( Panel( f"[bold green]PASS[/bold green] — All findings accepted.{accepted_note}", @@ -311,25 +244,26 @@ def render_report( ) ) return 0 + else: + console.print( + Panel( + "[bold green]PASS[/bold green] — No findings.", + border_style="green", + ) + ) + return 0 def _render_paths(console: Console, paths: list) -> None: - """Render finding paths as a tree or table depending on type.""" + """Render finding paths as a table or tree depending on type.""" if not paths: return first = paths[0] - if isinstance(first, ExfilPath): _render_exfil_paths(console, paths) elif isinstance(first, WriteBypassPath): _render_write_bypass_paths(console, paths) - elif isinstance(first, OverpermissiveMethodPath): - _render_overpermissive_paths(console, paths) - elif isinstance(first, L4PolicyGapPath): - _render_l4_gap_paths(console, paths) - elif isinstance(first, InheritancePath): - _render_inheritance_paths(console, paths) def _render_exfil_paths(console: Console, paths: list[ExfilPath]) -> None: @@ -371,927 +305,3 @@ def _render_write_bypass_paths(console: Console, paths: list[WriteBypassPath]) - console.print(tree) console.print() - - -def _render_overpermissive_paths( - console: Console, paths: list[OverpermissiveMethodPath] -) -> None: - table = Table(show_header=True, border_style="dim", padding=(0, 1)) - table.add_column("Endpoint", style="bold") - table.add_column("Policy") - table.add_column("Allowed") - table.add_column("Needed") - table.add_column("Excess", style="red") - table.add_column("Risk Detail", max_width=40) - - seen = set() - for p in paths: - key = (p.endpoint_host, p.endpoint_port, p.policy_name) - if key in seen: - continue - seen.add(key) - table.add_row( - f"{p.endpoint_host}:{p.endpoint_port}", - p.policy_name, - p.allowed_methods, - p.needed_methods, - p.excess_methods, - p.risk_detail, - ) - - console.print(table) - console.print() - - -def _render_l4_gap_paths(console: Console, paths: list[L4PolicyGapPath]) -> None: - by_endpoint: dict[str, list[L4PolicyGapPath]] = {} - for p in paths: - key = f"{p.endpoint_host}:{p.endpoint_port}" - by_endpoint.setdefault(key, []).append(p) - - for endpoint, eps in by_endpoint.items(): - first = eps[0] - style = "red" if first.has_sibling_l7 else "yellow" - label = "INCONSISTENT" if first.has_sibling_l7 else "L4-ONLY" - tree = Tree( - f"[{style}]{label}[/{style}] [{style} bold]{endpoint}[/{style} bold]" - f" (policy: {first.policy_name})" - ) - binaries = sorted({p.binary for p in eps}) - tree.add(f"Binaries: {', '.join(binaries)}") - tree.add(f"[dim]{first.detail}[/dim]") - console.print(tree) - - console.print() - - -def _render_inheritance_paths(console: Console, paths: list[InheritancePath]) -> None: - by_parent: dict[str, list[InheritancePath]] = {} - for p in paths: - by_parent.setdefault(p.parent, []).append(p) - - for parent, children in by_parent.items(): - tree = Tree(f"[bold]{parent}[/bold] (allowed by policy)") - for child in children: - caps = ", ".join(child.child_capabilities) - style = "red" if "bypasses_l7" in child.child_capabilities else "yellow" - tree.add(f"[{style}]{child.child}[/{style}] — [{style}]{caps}[/{style}]") - console.print(tree) - - console.print() - - -def _m_id(prefix: str, s: str) -> str: - """Generate a Mermaid-safe node ID from a string.""" - return ( - prefix - + "_" - + s.replace("/", "_") - .replace(".", "_") - .replace(":", "_") - .replace("-", "_") - .replace("*", "star") - ) - - -def _m_escape(s: str) -> str: - """Escape a string for use in Mermaid labels.""" - return s.replace('"', "'").replace("<", "<").replace(">", ">") - - -def _m_short_binary(path: str) -> str: - """Shorten a binary path for display.""" - return path.rsplit("/", 1)[-1] - - -def generate_mermaid(findings: list[Finding]) -> str: - """Generate a Mermaid findings graph showing all risk paths across query types.""" - lines = [ - "flowchart LR", - "", - " %% Style definitions", - " classDef l4only fill:#e74c3c,stroke:#c0392b,color:#fff", - " classDef l7bypass fill:#e67e22,stroke:#d35400,color:#fff", - " classDef l7write fill:#f39c12,stroke:#e67e22,color:#fff", - " classDef l7safe fill:#27ae60,stroke:#1e8449,color:#fff", - " classDef binary fill:#3498db,stroke:#2980b9,color:#fff", - " classDef filesystem fill:#95a5a6,stroke:#7f8c8d,color:#fff", - " classDef overperm fill:#e74c3c,stroke:#c0392b,color:#fff", - " classDef inherited fill:#9b59b6,stroke:#8e44ad,color:#fff", - "", - ] - - declared_nodes: set[str] = set() - edges: list[str] = [] - - exfil_paths: list[ExfilPath] = [] - write_bypasses: list[WriteBypassPath] = [] - overperm_paths: list[OverpermissiveMethodPath] = [] - l4_gaps: list[L4PolicyGapPath] = [] - inherit_paths: list[InheritancePath] = [] - - for finding in findings: - if finding.accepted: - continue - for p in finding.paths: - if isinstance(p, ExfilPath): - exfil_paths.append(p) - elif isinstance(p, WriteBypassPath): - write_bypasses.append(p) - elif isinstance(p, OverpermissiveMethodPath): - overperm_paths.append(p) - elif isinstance(p, L4PolicyGapPath): - l4_gaps.append(p) - elif isinstance(p, InheritancePath): - inherit_paths.append(p) - - # Filesystem source node - if exfil_paths: - fs_id = "FS" - if fs_id not in declared_nodes: - lines.append(f' {fs_id}[("Filesystem — readable paths")]:::filesystem') - declared_nodes.add(fs_id) - - # Exfiltration paths - if exfil_paths: - lines.append("") - lines.append(" subgraph exfil [Data Exfiltration]") - lines.append(" direction LR") - - seen_exfil: set[tuple[str, str, int]] = set() - for p in exfil_paths: - key = (p.binary, p.endpoint_host, p.endpoint_port) - if key in seen_exfil: - continue - seen_exfil.add(key) - - b_id = _m_id("B", p.binary) - e_id = _m_id("E", f"{p.endpoint_host}_{p.endpoint_port}") - - if b_id not in declared_nodes: - lines.append(f' {b_id}["{_m_short_binary(p.binary)}"]:::binary') - declared_nodes.add(b_id) - - style_class = { - "l4_only": "l4only", - "l7_bypassed": "l7bypass", - "l7_allows_write": "l7write", - }.get(p.l7_status, "l7write") - label_text = { - "l4_only": "L4 only", - "l7_bypassed": "wire protocol", - "l7_allows_write": "L7 write", - }.get(p.l7_status, "") - - if e_id not in declared_nodes: - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}"]:::{style_class}' - ) - declared_nodes.add(e_id) - - edges.append(f" FS -.->|read| {b_id}") - edges.append(f" {b_id} -->|{label_text}| {e_id}") - - lines.append(" end") - - # Write bypass paths - if write_bypasses: - lines.append("") - lines.append(" subgraph wbypass [Write Bypass]") - lines.append(" direction LR") - - seen_wb: set[tuple[str, str, int]] = set() - for p in write_bypasses: - key = (p.binary, p.endpoint_host, p.endpoint_port) - if key in seen_wb: - continue - seen_wb.add(key) - - b_id = _m_id("B", p.binary) - e_id = _m_id("WE", f"{p.endpoint_host}_{p.endpoint_port}") - - if b_id not in declared_nodes: - lines.append(f' {b_id}["{_m_short_binary(p.binary)}"]:::binary') - declared_nodes.add(b_id) - - reason_label = { - "l4_only": "L4 bypass", - "l7_bypass_protocol": "wire protocol", - }.get(p.bypass_reason, p.bypass_reason) - if e_id not in declared_nodes: - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}' - f'
intent: {p.policy_intent}"]:::l4only' - ) - declared_nodes.add(e_id) - - edges.append(f" {b_id} ==>|{reason_label}| {e_id}") - - lines.append(" end") - - # Overpermissive methods - if overperm_paths: - lines.append("") - lines.append(" subgraph operm [Overpermissive Methods]") - lines.append(" direction LR") - - seen_op: set[tuple[str, int, str]] = set() - for p in overperm_paths: - key = (p.endpoint_host, p.endpoint_port, p.policy_name) - if key in seen_op: - continue - seen_op.add(key) - - e_id = _m_id("OP", f"{p.endpoint_host}_{p.endpoint_port}") - if e_id not in declared_nodes: - excess = ( - p.excess_methods if len(p.excess_methods) < 30 else "DELETE, ..." - ) - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}' - f'
excess: {_m_escape(excess)}"]:::overperm' - ) - declared_nodes.add(e_id) - - lines.append(" end") - - # L4 policy gaps - if l4_gaps: - lines.append("") - lines.append(" subgraph l4gap [L4 Policy Gaps]") - lines.append(" direction LR") - - seen_l4: set[tuple[str, int]] = set() - for p in l4_gaps: - key = (p.endpoint_host, p.endpoint_port) - if key in seen_l4: - continue - seen_l4.add(key) - - e_id = _m_id("L4", f"{p.endpoint_host}_{p.endpoint_port}") - label_suffix = "INCONSISTENT" if p.has_sibling_l7 else "no L7" - if e_id not in declared_nodes: - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}' - f'
{label_suffix}"]:::l4only' - ) - declared_nodes.add(e_id) - - b_id = _m_id("B", p.binary) - if b_id not in declared_nodes: - lines.append(f' {b_id}["{_m_short_binary(p.binary)}"]:::binary') - declared_nodes.add(b_id) - - edges.append(f" {b_id} -.->|no inspection| {e_id}") - - lines.append(" end") - - # Binary inheritance - if inherit_paths: - lines.append("") - lines.append(" subgraph inherit [Binary Inheritance]") - lines.append(" direction TB") - - seen_inh: set[tuple[str, str]] = set() - for p in inherit_paths: - key = (p.parent, p.child) - if key in seen_inh: - continue - seen_inh.add(key) - - parent_id = _m_id("IB", p.parent) - child_id = _m_id("IC", p.child) - - if parent_id not in declared_nodes: - lines.append(f' {parent_id}["{_m_short_binary(p.parent)}"]:::binary') - declared_nodes.add(parent_id) - if child_id not in declared_nodes: - caps = ", ".join(p.child_capabilities) - style = ( - "l7bypass" if "bypasses_l7" in p.child_capabilities else "inherited" - ) - lines.append( - f' {child_id}["{_m_short_binary(p.child)}' - f'
{_m_escape(caps)}"]:::{style}' - ) - declared_nodes.add(child_id) - - edges.append(f" {parent_id} -->|spawns| {child_id}") - - lines.append(" end") - - # Add all edges after subgraphs - if edges: - lines.append("") - lines.append(" %% Edges") - for edge in dict.fromkeys(edges): - lines.append(edge) - - if not any([exfil_paths, write_bypasses, overperm_paths, l4_gaps, inherit_paths]): - lines.append(" PASS([No active findings]):::l7safe") - - return "\n".join(lines) - - -def generate_mermaid_topology( - policy: PolicyModel, - binary_registry: BinaryRegistry, - findings: list[Finding], -) -> str: - """Generate a full policy topology map showing all endpoints, binaries, and enforcement levels.""" - lines = [ - "flowchart LR", - "", - " %% Style definitions", - " classDef l4only fill:#e74c3c,stroke:#c0392b,color:#fff", - " classDef l7enforce fill:#27ae60,stroke:#1e8449,color:#fff", - " classDef l7warn fill:#f39c12,stroke:#e67e22,color:#fff", - " classDef binary fill:#3498db,stroke:#2980b9,color:#fff", - " classDef child fill:#9b59b6,stroke:#8e44ad,color:#fff", - " classDef filesystem fill:#95a5a6,stroke:#7f8c8d,color:#fff", - " classDef finding fill:#e74c3c,stroke:#c0392b,color:#fff,stroke-width:3px", - "", - ] - - declared: set[str] = set() - edges: list[str] = [] - - problem_endpoints: set[tuple[str, int]] = set() - problem_binaries: set[str] = set() - for f in findings: - if f.accepted: - continue - if f.query in ("l4_policy_gaps", "overpermissive_methods", "write_bypass"): - for p in f.paths: - if hasattr(p, "endpoint_host"): - problem_endpoints.add( - (p.endpoint_host, getattr(p, "endpoint_port", 443)) - ) - if hasattr(p, "binary"): - problem_binaries.add(p.binary) - - # Filesystem - lines.append(' FS[("Filesystem")]:::filesystem') - declared.add("FS") - - # Endpoints by policy group - for policy_name, rule in policy.network_policies.items(): - safe_name = policy_name.replace("-", "_").replace(" ", "_") - lines.append("") - lines.append(f" subgraph {safe_name} [{policy_name}]") - lines.append(" direction TB") - - for ep in rule.endpoints: - for port in ep.effective_ports: - e_id = _m_id("E", f"{ep.host}_{port}") - if e_id in declared: - continue - declared.add(e_id) - - is_problem = (ep.host, port) in problem_endpoints - - if ep.is_l7_enforced: - methods = ep.allowed_methods - has_write = ( - bool(methods & {"POST", "PUT", "PATCH", "DELETE"}) - if methods - else True - ) - has_delete = "DELETE" in methods if methods else True - has_wildcard = any(r.method == "*" for r in ep.rules) - - if has_wildcard or has_delete: - style = "l7warn" - enforcement_label = ( - "L7 method:*" if has_wildcard else "L7 +DELETE" - ) - elif has_write: - style = "l7enforce" - enforcement_label = "L7 " + ",".join( - sorted(methods - {"HEAD", "OPTIONS"}) - ) - else: - style = "l7enforce" - enforcement_label = "L7 read-only" - else: - style = "l4only" - enforcement_label = "L4 only" - - if is_problem: - style = "finding" - - lines.append( - f' {e_id}["{_m_escape(ep.host)}:{port}
{enforcement_label}"]:::{style}' - ) - - lines.append(" end") - - # Binaries and their connections - lines.append("") - lines.append(" subgraph binaries [Binaries]") - lines.append(" direction TB") - - all_binaries: dict[str, list[tuple[str, str]]] = {} - for policy_name, rule in policy.network_policies.items(): - for b in rule.binaries: - for ep in rule.endpoints: - for port in ep.effective_ports: - e_id = _m_id("E", f"{ep.host}_{port}") - all_binaries.setdefault(b.path, []).append((policy_name, e_id)) - - for bpath in all_binaries: - b_id = _m_id("B", bpath) - if b_id not in declared: - style = "finding" if bpath in problem_binaries else "binary" - lines.append(f' {b_id}["{_m_short_binary(bpath)}"]:::{style}') - declared.add(b_id) - - # Spawn chains - for bpath in all_binaries: - children = binary_registry.transitive_spawns(bpath) - for child in children: - child_id = _m_id("C", child) - if child_id not in declared: - cap = binary_registry.get_or_unknown(child) - caps = [] - if cap.bypasses_l7: - caps.append("L7-bypass") - if cap.can_exfiltrate: - caps.append("exfil") - if cap.can_construct_http: - caps.append("http") - cap_str = f"
{', '.join(caps)}" if caps else "" - lines.append( - f' {child_id}["{_m_short_binary(child)}{cap_str}"]:::child' - ) - declared.add(child_id) - - b_id = _m_id("B", bpath) - edges.append(f" {b_id} -.->|spawns| {child_id}") - - lines.append(" end") - - # Binary -> Endpoint edges - lines.append("") - lines.append(" %% Binary -> Endpoint access") - for bpath, endpoint_ids in all_binaries.items(): - b_id = _m_id("B", bpath) - for _, e_id in endpoint_ids: - edges.append(f" {b_id} --> {e_id}") - - # Filesystem -> Binary edges (for exfil-capable binaries) - for bpath in all_binaries: - cap = binary_registry.get_or_unknown(bpath) - if cap.can_exfiltrate: - b_id = _m_id("B", bpath) - edges.append(f" FS -.->|read| {b_id}") - - # Deduplicated edges - if edges: - lines.append("") - lines.append(" %% Edges") - for edge in dict.fromkeys(edges): - lines.append(edge) - - return "\n".join(lines) - - -# Per-finding focused Mermaid diagrams - -_MERMAID_STYLES = """\ - classDef l4only fill:#e74c3c,stroke:#c0392b,color:#fff - classDef l7bypass fill:#e67e22,stroke:#d35400,color:#fff - classDef l7write fill:#f39c12,stroke:#e67e22,color:#fff - classDef l7safe fill:#27ae60,stroke:#1e8449,color:#fff - classDef binary fill:#3498db,stroke:#2980b9,color:#fff - classDef filesystem fill:#95a5a6,stroke:#7f8c8d,color:#fff - classDef overperm fill:#e74c3c,stroke:#c0392b,color:#fff - classDef inherited fill:#9b59b6,stroke:#8e44ad,color:#fff - classDef finding fill:#e74c3c,stroke:#c0392b,color:#fff,stroke-width:3px - classDef l7enforce fill:#27ae60,stroke:#1e8449,color:#fff - classDef l7warn fill:#f39c12,stroke:#e67e22,color:#fff - classDef child fill:#9b59b6,stroke:#8e44ad,color:#fff""" - - -def _mermaid_for_l4_gaps(finding: Finding) -> str: - """Focused diagram: which endpoints are missing L7 and which binaries reach them.""" - lines = ["flowchart LR", _MERMAID_STYLES, ""] - declared: set[str] = set() - - by_ep: dict[tuple[str, int], list[L4PolicyGapPath]] = {} - for p in finding.paths: - if isinstance(p, L4PolicyGapPath): - by_ep.setdefault((p.endpoint_host, p.endpoint_port), []).append(p) - - for (host, port), ep_paths in by_ep.items(): - e_id = _m_id("E", f"{host}_{port}") - first = ep_paths[0] - label = "INCONSISTENT" if first.has_sibling_l7 else "no L7" - lines.append(f' {e_id}["{_m_escape(host)}:{port}
{label}"]:::l4only') - declared.add(e_id) - - for p in ep_paths: - b_id = _m_id("B", p.binary) - if b_id not in declared: - lines.append(f' {b_id}["{_m_short_binary(p.binary)}"]:::binary') - declared.add(b_id) - lines.append(f" {b_id} -->|no inspection| {e_id}") - - return "\n".join(lines) - - -def _mermaid_for_overperm(finding: Finding) -> str: - """Focused diagram: endpoints with wildcard/excess methods.""" - lines = ["flowchart LR", _MERMAID_STYLES, ""] - seen: set[tuple[str, int]] = set() - - for p in finding.paths: - if not isinstance(p, OverpermissiveMethodPath): - continue - key = (p.endpoint_host, p.endpoint_port) - if key in seen: - continue - seen.add(key) - - e_id = _m_id("E", f"{p.endpoint_host}_{p.endpoint_port}") - needed_id = _m_id("N", f"{p.endpoint_host}_{p.endpoint_port}") - excess_id = _m_id("X", f"{p.endpoint_host}_{p.endpoint_port}") - - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}' - f'
policy: {p.policy_name}"]:::binary' - ) - lines.append(f' {needed_id}(["{_m_escape(p.needed_methods)}"]):::l7safe') - lines.append(f' {excess_id}(["{_m_escape(p.excess_methods)}"]):::overperm') - lines.append(f" {e_id} -->|needed| {needed_id}") - lines.append(f" {e_id} -->|excess| {excess_id}") - - return "\n".join(lines) - - -def _mermaid_for_write_bypass(finding: Finding) -> str: - """Focused diagram: how writes bypass read-only intent.""" - lines = ["flowchart LR", _MERMAID_STYLES, ""] - declared: set[str] = set() - - seen: set[tuple[str, str, int]] = set() - for p in finding.paths: - if not isinstance(p, WriteBypassPath): - continue - key = (p.binary, p.endpoint_host, p.endpoint_port) - if key in seen: - continue - seen.add(key) - - b_id = _m_id("B", p.binary) - e_id = _m_id("E", f"{p.endpoint_host}_{p.endpoint_port}") - - if b_id not in declared: - lines.append(f' {b_id}["{_m_short_binary(p.binary)}"]:::binary') - declared.add(b_id) - if e_id not in declared: - lines.append( - f' {e_id}["{_m_escape(p.endpoint_host)}:{p.endpoint_port}' - f'
intent: {p.policy_intent}"]:::l4only' - ) - declared.add(e_id) - - reason = { - "l4_only": "L4 bypass", - "l7_bypass_protocol": "wire protocol", - }.get(p.bypass_reason, p.bypass_reason) - lines.append(f" {b_id} ==>|{reason}| {e_id}") - - if p.credential_actions: - c_id = _m_id("CR", f"{p.endpoint_host}_{p.endpoint_port}") - if c_id not in declared: - acts = p.credential_actions[:3] - if len(p.credential_actions) > 3: - acts.append(f"+{len(p.credential_actions) - 3} more") - label = "
".join(_m_escape(a) for a in acts) - lines.append(f' {c_id}["{label}"]:::overperm') - declared.add(c_id) - lines.append(f" {e_id} -.->|credential enables| {c_id}") - - return "\n".join(lines) - - -def _mermaid_for_exfil(finding: Finding) -> str: - """Focused diagram: exfiltration paths grouped by mechanism type.""" - lines = ["flowchart LR", _MERMAID_STYLES, ""] - declared: set[str] = set() - - lines.append(' FS[("Filesystem — readable paths")]:::filesystem') - declared.add("FS") - - by_status: dict[str, list[ExfilPath]] = {} - seen: set[tuple[str, str, int]] = set() - for p in finding.paths: - if not isinstance(p, ExfilPath): - continue - key = (p.binary, p.endpoint_host, p.endpoint_port) - if key in seen: - continue - seen.add(key) - by_status.setdefault(p.l7_status, []).append(p) - - status_labels = { - "l4_only": "L4-only — no inspection", - "l7_bypassed": "Wire Protocol Bypass", - "l7_allows_write": "L7 Write Channels", - } - - for status, status_paths in by_status.items(): - label = status_labels.get(status, status) - safe = "sg_" + status.replace("_", "") - lines.append(f" subgraph {safe} [{label}]") - lines.append(" direction LR") - - style_class = { - "l4_only": "l4only", - "l7_bypassed": "l7bypass", - "l7_allows_write": "l7write", - }.get(status, "l7write") - - by_binary: dict[str, list[str]] = {} - for p in status_paths: - by_binary.setdefault(p.binary, []).append( - f"{p.endpoint_host}:{p.endpoint_port}" - ) - - for bpath, endpoints in by_binary.items(): - b_id = _m_id("B", f"{bpath}_{status}") - if b_id not in declared: - lines.append(f' {b_id}["{_m_short_binary(bpath)}"]:::binary') - declared.add(b_id) - - shown = endpoints[:4] - for ep_str in shown: - e_id = _m_id("E", f"{ep_str}_{status}") - if e_id not in declared: - lines.append(f' {e_id}["{_m_escape(ep_str)}"]:::{style_class}') - declared.add(e_id) - lines.append(f" {b_id} --> {e_id}") - - if len(endpoints) > 4: - more_id = _m_id("M", f"{bpath}_{status}") - lines.append( - f' {more_id}["+{len(endpoints) - 4} more"]:::{style_class}' - ) - lines.append(f" {b_id} --> {more_id}") - - lines.append(" end") - for bpath in by_binary: - b_id = _m_id("B", f"{bpath}_{status}") - lines.append(f" FS -.-> {b_id}") - - return "\n".join(lines) - - -def _mermaid_for_inheritance(finding: Finding) -> str: - """Focused diagram: spawn tree from allowed binaries to inherited children.""" - lines = ["flowchart TB", _MERMAID_STYLES, ""] - declared: set[str] = set() - - seen_pairs: set[tuple[str, str]] = set() - for p in finding.paths: - if not isinstance(p, InheritancePath): - continue - key = (p.parent, p.child) - if key in seen_pairs: - continue - seen_pairs.add(key) - - parent_id = _m_id("P", p.parent) - child_id = _m_id("C", p.child) - - if parent_id not in declared: - lines.append( - f' {parent_id}["{_m_short_binary(p.parent)}
allowed by policy"]:::binary' - ) - declared.add(parent_id) - - if child_id not in declared: - caps = ", ".join(p.child_capabilities) - style = "l7bypass" if "bypasses_l7" in p.child_capabilities else "inherited" - lines.append( - f' {child_id}["{_m_short_binary(p.child)}
{_m_escape(caps)}"]:::{style}' - ) - declared.add(child_id) - - lines.append(f" {parent_id} -->|spawns| {child_id}") - - return "\n".join(lines) - - -# HTML Report - -_FINDING_DIAGRAM_GENERATORS = { - "l4_policy_gaps": _mermaid_for_l4_gaps, - "overpermissive_methods": _mermaid_for_overperm, - "write_bypass": _mermaid_for_write_bypass, - "data_exfiltration": _mermaid_for_exfil, - "binary_inheritance": _mermaid_for_inheritance, -} - -_RISK_COLORS_HTML = { - RiskLevel.CRITICAL: "#e74c3c", - RiskLevel.HIGH: "#e67e22", - RiskLevel.MEDIUM: "#f39c12", - RiskLevel.LOW: "#27ae60", - RiskLevel.ADVISORY: "#3498db", -} - -_RISK_BG_COLORS_HTML = { - RiskLevel.CRITICAL: "#fdedec", - RiskLevel.HIGH: "#fdebd0", - RiskLevel.MEDIUM: "#fef9e7", - RiskLevel.LOW: "#eafaf1", - RiskLevel.ADVISORY: "#ebf5fb", -} - - -def render_html( - findings: list[Finding], - policy_path: str, - _credentials_path: str, - policy: PolicyModel, - binary_registry: BinaryRegistry, - output_path: Path, -) -> int: - """Render findings as a self-contained HTML file with per-finding Mermaid diagrams.""" - policy_name = Path(policy_path).name - active = [f for f in findings if not f.accepted] - accepted_findings = [f for f in findings if f.accepted] - - counts: dict[RiskLevel, int] = {} - for f in active: - counts[f.risk] = counts.get(f.risk, 0) + 1 - - has_critical = RiskLevel.CRITICAL in counts - has_high = RiskLevel.HIGH in counts - verdict_pass = not (has_critical or has_high) - - finding_sections = [] - for _i, finding in enumerate(active, 1): - color = _RISK_COLORS_HTML[finding.risk] - bg = _RISK_BG_COLORS_HTML[finding.risk] - risk_label = finding.risk.value.upper() - - diagram_html = "" - gen = _FINDING_DIAGRAM_GENERATORS.get(finding.query) - if gen and finding.paths: - mermaid_code = gen(finding) - diagram_html = f'
\n{mermaid_code}\n
' - - remediation_html = "" - if finding.remediation: - items = "".join( - f"
  • {html_mod.escape(r)}
  • " for r in finding.remediation - ) - remediation_html = f"

    Remediation

    " - - finding_sections.append(f""" -
    -
    - {risk_label} - {html_mod.escape(finding.title)} -
    -

    {html_mod.escape(finding.description)}

    - {diagram_html} - {remediation_html} -
    """) - - accepted_html = "" - if accepted_findings: - items = [] - for f in accepted_findings: - items.append( - f"
  • {f.risk.value.upper()} {html_mod.escape(f.title)}" - f"
    {html_mod.escape(f.accepted_reason)}
  • " - ) - accepted_html = f""" -
    -

    Accepted Risks

    - -
    """ - - summary_badges = [] - for level in [ - RiskLevel.CRITICAL, - RiskLevel.HIGH, - RiskLevel.MEDIUM, - RiskLevel.LOW, - RiskLevel.ADVISORY, - ]: - if level in counts: - color = _RISK_COLORS_HTML[level] - summary_badges.append( - f'' - f"{counts[level]} {level.value.upper()}" - ) - if accepted_findings: - summary_badges.append( - f'' - f"{len(accepted_findings)} ACCEPTED" - ) - - verdict_color = "#27ae60" if verdict_pass else "#e74c3c" - verdict_text = "PASS" if verdict_pass else "FAIL" - if verdict_pass and accepted_findings: - verdict_detail = "All findings accepted" - elif verdict_pass and active: - verdict_detail = "Advisories only" - elif not verdict_pass: - n = counts.get(RiskLevel.CRITICAL, 0) + counts.get(RiskLevel.HIGH, 0) - verdict_detail = f"{n} critical/high gap{'s' if n != 1 else ''}" - else: - verdict_detail = "No findings" - - topology_mermaid = generate_mermaid_topology(policy, binary_registry, findings) - - page = f""" - - - - -OPP Report — {html_mod.escape(policy_name)} - - - - -

    OpenShell Policy Prover

    -
    Policy: {html_mod.escape(policy_name)}
    - -
    - {verdict_text} - {html_mod.escape(verdict_detail)} -
    -
    {"".join(summary_badges)}
    - -

    Findings

    -{"".join(finding_sections) if finding_sections else '

    No active findings.

    '} - -{accepted_html} - -
    -

    Policy Topology

    -

    - Full map of all endpoints, binaries, and enforcement levels. - Red nodes have structural findings. Green nodes are properly configured. -

    -
    - L7 enforced - L7 warn (wildcard) - L4 only / finding - Binary - Inherited child -
    -
    -{topology_mermaid} -
    -
    - - - - - - -""" - - output_path.write_text(page) - return 1 if not verdict_pass else 0