feat: add OPA eval CLI commands

Author: rhettsies-os

README.md

@@ -19,6 +19,7 @@ MVP Python implementation of an open control plane for agent tool governance. It
 - Registry persistence: registry saved to `registry.json` by default (override with `OCPA_REGISTRY_PATH`), loaded on startup; create via `/registry/tools` and `/registry/agents` (validates side_effect_level and tool references). Mutations require JWT bearer token (HS256 via `OCPA_JWT_SECRET`) and roles header `X-OCPA-Roles` containing `OCPA_ADMIN_ROLE`. Server uses FastAPI lifespan hooks for startup.
 - Auth: Bearer JWTs validated via HS256 shared secret (`OCPA_JWT_SECRET`) or JWKS (`OCPA_JWKS_URL`) with caching; optional audience/issuer checks; mutation RBAC enforced via required admin role. JWKS cache honors `OCPA_JWKS_CACHE_TTL` and refreshes on missing `kid` (rotation).
 - Policy RBAC: Rego now checks `allowed_roles` against actor `roles`/`actor_roles` in context (with `actor_id` forwarded); read-only/low/high rules still apply and high side effects require ticket + non-prod. Registry mutations are also enforced via OPA (`policies/registry_allow.rego`).
+- CLI: `scripts/ocpa_cli.py` provides `list-tools`, `list-agents`, `eval-tool-policy`, and `eval-registry-policy` using the OPA HTTP API.
 
 ## Getting Started
 Requirements: Python 3.9+ and Docker (for OPA/demo stack).

scripts/ocpa_cli.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 import json
+from typing import Optional
 
 import typer
 
+from ocpa.opa import OPAClient
 from ocpa.registry import registry
+from ocpa.types import ToolMetadata
 
-app = typer.Typer(help="OCPA registry CLI")
+app = typer.Typer(help="OCPA registry and policy CLI")
 
 
 @app.command()
@@ -23,5 +26,55 @@ def list_agents():
     typer.echo(json.dumps([agent.__dict__ for agent in agents], indent=2))
 
 
+@app.command()
+def eval_tool_policy(
+    tool_id: str = typer.Option(..., help="Tool id"),
+    side_effect_level: str = typer.Option(..., help="none|low|high"),
+    data_classes: str = typer.Option("", help="Comma-separated data classes"),
+    env: str = typer.Option("dev", help="Environment (env in context)"),
+    roles: str = typer.Option("", help="Comma-separated actor roles"),
+    opa_url: str = typer.Option("http://localhost:8181", help="OPA base URL"),
+    ticket_id: Optional[str] = typer.Option(None, help="Ticket id for high side-effect"),
+):
+    """Evaluate tool policy via OPA HTTP API."""
+    client = OPAClient(opa_url)
+    metadata = ToolMetadata(
+        tool_id=tool_id,
+        description="cli-eval",
+        side_effect_level=side_effect_level,
+        data_classes=[r for r in data_classes.split(",") if r] if data_classes else [],
+    )
+    context = {
+        "env": env,
+        "roles": [r for r in roles.split(",") if r] if roles else [],
+        "ticket_id": ticket_id,
+    }
+    result = client.evaluate_tool(tool_metadata=metadata, tool_input={}, context=context)
+    typer.echo(json.dumps(result.__dict__, indent=2))
+
+
+@app.command()
+def eval_registry_policy(
+    action: str = typer.Option(..., help="create/update/delete"),
+    resource_type: str = typer.Option(..., help="tool|agent"),
+    env: str = typer.Option("dev", help="Environment"),
+    roles: str = typer.Option("", help="Comma-separated actor roles"),
+    required_admin_role: str = typer.Option("admin", help="Admin role required"),
+    ticket_id: Optional[str] = typer.Option(None, help="Ticket id for prod"),
+    opa_url: str = typer.Option("http://localhost:8181", help="OPA base URL"),
+):
+    """Evaluate registry mutation policy via OPA HTTP API."""
+    client = OPAClient(opa_url)
+    context = {
+        "env": env,
+        "roles": [r for r in roles.split(",") if r] if roles else [],
+        "actor_roles": [r for r in roles.split(",") if r] if roles else [],
+        "required_admin_role": required_admin_role,
+        "ticket_id": ticket_id,
+    }
+    result = client.evaluate_registry(action=action, resource_type=resource_type, context=context, payload={})
+    typer.echo(json.dumps(result.__dict__, indent=2))
+
+
 if __name__ == "__main__":
     app()

tests/test_cli.py

@@ -0,0 +1,71 @@
+import json
+
+from typer.testing import CliRunner
+
+from scripts.ocpa_cli import app
+
+
+def test_eval_tool_policy_cli(monkeypatch):
+    runner = CliRunner()
+
+    def ok_post(url, json=None, timeout=None):
+        return DummyResponse({"result": {"allow": True, "reason": "ok", "redact_paths": []}})
+
+    monkeypatch.setattr("ocpa.opa.requests.post", ok_post)
+    result = runner.invoke(
+        app,
+        [
+            "eval-tool-policy",
+            "--tool-id",
+            "t1",
+            "--side-effect-level",
+            "none",
+            "--roles",
+            "admin",
+            "--opa-url",
+            "http://opa",
+        ],
+    )
+    assert result.exit_code == 0
+    data = json.loads(result.stdout)
+    assert data["allowed"] is True
+
+
+def test_eval_registry_policy_cli(monkeypatch):
+    runner = CliRunner()
+
+    def ok_post(url, json=None, timeout=None):
+        return DummyResponse({"result": {"allow": False, "reason": "deny_test"}})
+
+    monkeypatch.setattr("ocpa.opa.requests.post", ok_post)
+    result = runner.invoke(
+        app,
+        [
+            "eval-registry-policy",
+            "--action",
+            "create",
+            "--resource-type",
+            "tool",
+            "--roles",
+            "admin",
+            "--opa-url",
+            "http://opa",
+        ],
+    )
+    assert result.exit_code == 0
+    data = json.loads(result.stdout)
+    assert data["allowed"] is False
+    assert data["reason"] == "deny_test"
+
+
+class DummyResponse:
+    def __init__(self, payload, status_code: int = 200):
+        self.payload = payload
+        self.status_code = status_code
+
+    def json(self):
+        return self.payload
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise Exception("status error")