feat: enforce registry update/delete via OPA and extend CLI

Author: rhettsies-os

example_app/api.py

@@ -43,6 +43,20 @@ class AgentCreate(BaseModel):
     description: str
     tools: list[str]
 
+
+class ToolUpdate(BaseModel):
+    description: Optional[str] = None
+    side_effect_level: Optional[str] = None
+    data_classes: Optional[list[str]] = None
+    tags: Optional[list[str]] = None
+    redaction_paths: Optional[list[str]] = None
+    allowed_roles: Optional[list[str]] = None
+
+
+class AgentUpdate(BaseModel):
+    description: Optional[str] = None
+    tools: Optional[list[str]] = None
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     register_example_tools()
@@ -152,6 +166,90 @@ def create_agent(
     return {"status": "ok", "agent_id": agent.agent_id}
 
 
+@app.patch("/registry/tools/{tool_id}")
+def update_tool(
+    tool_id: str,
+    payload: ToolUpdate,
+    admin_token: Optional[str] = Header(None, alias="Authorization"),
+    roles_header: Optional[str] = Header(None, alias="X-OCPA-Roles"),
+) -> Dict[str, Any]:
+    _require_admin(admin_token, roles_header)
+    existing = registry.get_tool(tool_id)
+    if not existing:
+        raise HTTPException(status_code=404, detail="tool not found")
+    updates = {k: v for k, v in payload.model_dump(exclude_none=True).items()}
+    updated = existing.__dict__ | updates
+    if updated.get("side_effect_level") not in {"none", "low", "high"}:
+        raise HTTPException(status_code=400, detail="invalid side_effect_level")
+    _require_registry_policy("update", "tool", {"tool_id": tool_id, **updates}, roles_header)
+    registry.register_tool(ToolMetadata(**updated))
+    emit_validated(
+        lambda: build_registry_audit_event(
+            action="update",
+            resource_type="tool",
+            resource_id=tool_id,
+            actor=_actor_from_headers(roles_header),
+            env=os.environ.get("OCPA_ENV", "dev"),
+            request=updates,
+        )
+    )
+    return {"status": "ok", "tool_id": tool_id}
+
+
+@app.delete("/registry/tools/{tool_id}")
+def delete_tool(
+    tool_id: str,
+    admin_token: Optional[str] = Header(None, alias="Authorization"),
+    roles_header: Optional[str] = Header(None, alias="X-OCPA-Roles"),
+) -> Dict[str, Any]:
+    _require_admin(admin_token, roles_header)
+    _require_registry_policy("delete", "tool", {"tool_id": tool_id}, roles_header)
+    if not registry.get_tool(tool_id):
+        raise HTTPException(status_code=404, detail="tool not found")
+    # remove tool and any references in agents
+    registry._tools.pop(tool_id, None)  # noqa: SLF001
+    for agent in registry._agents.values():  # noqa: SLF001
+        if tool_id in agent.tools:
+            agent.tools = [t for t in agent.tools if t != tool_id]
+    registry._save()
+    emit_validated(
+        lambda: build_registry_audit_event(
+            action="delete",
+            resource_type="tool",
+            resource_id=tool_id,
+            actor=_actor_from_headers(roles_header),
+            env=os.environ.get("OCPA_ENV", "dev"),
+            request={"tool_id": tool_id},
+        )
+    )
+    return {"status": "deleted", "tool_id": tool_id}
+
+
+@app.delete("/registry/agents/{agent_id}")
+def delete_agent(
+    agent_id: str,
+    admin_token: Optional[str] = Header(None, alias="Authorization"),
+    roles_header: Optional[str] = Header(None, alias="X-OCPA-Roles"),
+) -> Dict[str, Any]:
+    _require_admin(admin_token, roles_header)
+    _require_registry_policy("delete", "agent", {"agent_id": agent_id}, roles_header)
+    if not registry.get_agent(agent_id):
+        raise HTTPException(status_code=404, detail="agent not found")
+    registry._agents.pop(agent_id, None)  # noqa: SLF001
+    registry._save()
+    emit_validated(
+        lambda: build_registry_audit_event(
+            action="delete",
+            resource_type="agent",
+            resource_id=agent_id,
+            actor=_actor_from_headers(roles_header),
+            env=os.environ.get("OCPA_ENV", "dev"),
+            request={"agent_id": agent_id},
+        )
+    )
+    return {"status": "deleted", "agent_id": agent_id}
+
+
 def _require_admin(auth_header: Optional[str], roles_header: Optional[str]) -> None:
     required_role = os.environ.get("OCPA_ADMIN_ROLE", "admin")
     verify_bearer_token(auth_header, required_role, role_claim=os.environ.get("OCPA_ROLE_CLAIM", "roles"))

policies/registry_allow.rego

@@ -40,6 +40,30 @@ allow = result {
   result := {"allow": true, "reason": "registry_create_agent_allow"}
 }
 
+allow = result {
+  role_ok
+  input.action == "update"
+  input.resource_type == "tool"
+  not deny_prod_without_ticket
+  result := {"allow": true, "reason": "registry_update_tool_allow"}
+}
+
+allow = result {
+  role_ok
+  input.action == "delete"
+  input.resource_type == "tool"
+  not deny_prod_without_ticket
+  result := {"allow": true, "reason": "registry_delete_tool_allow"}
+}
+
+allow = result {
+  role_ok
+  input.action == "delete"
+  input.resource_type == "agent"
+  not deny_prod_without_ticket
+  result := {"allow": true, "reason": "registry_delete_agent_allow"}
+}
+
 deny_prod_without_ticket {
   input.context.env == "prod"
   not input.context.ticket_id

tests/test_registry_api.py

@@ -4,7 +4,7 @@
 from tests.test_opa_policy import DummyResponse
 
 
-def test_registry_api_create_tool_and_agent(tmp_path, monkeypatch):
+def test_registry_api_create_update_delete(tmp_path, monkeypatch):
     monkeypatch.setenv("OCPA_REGISTRY_PATH", str(tmp_path / "registry.json"))
     monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
     monkeypatch.setenv("OCPA_ADMIN_ROLE", "admin")
@@ -61,3 +61,17 @@ def ok_post(url, json=None, timeout=None):
     assert len(captured_events) == 2
     assert captured_events[0]["resource_type"] == "tool"
     assert captured_events[1]["resource_type"] == "agent"
+
+    # Update tool
+    resp_update = client.patch(
+        "/registry/tools/t1",
+        json={"description": "Updated"},
+        headers=headers,
+    )
+    assert resp_update.status_code == 200
+
+    # Delete agent and tool
+    resp_delete_agent = client.delete("/registry/agents/a1", headers=headers)
+    assert resp_delete_agent.status_code == 200
+    resp_delete_tool = client.delete("/registry/tools/t1", headers=headers)
+    assert resp_delete_tool.status_code == 200