feat: add JWT auth for registry mutations

Author: rhettsies-os

README.md

@@ -15,7 +15,7 @@ MVP Python implementation of an open control plane for agent tool governance. It
 - Metrics and events: Prometheus metrics at `/metrics` and SIEM-friendly JSON events for every tool call (decision, reason, redacted request/response).
 - Registry/RBAC: in-memory registry tracks tools/agents; tools can declare `allowed_roles` and the SDK enforces actor roles from context (`roles`/`actor_roles`) before OPA.
 - Registry API: list tools at `/registry/tools` and agents at `/registry/agents`.
-- 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 bearer token `Authorization: Bearer $OCPA_ADMIN_TOKEN` and roles header `X-OCPA-Roles` containing `OCPA_ADMIN_ROLE`. Server uses FastAPI lifespan hooks for startup.
+- 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.
 - Policy RBAC: Rego now checks `allowed_roles` against actor `roles` in context; read-only/low/high rules still apply.
 
 ## Getting Started

example_app/api.py

@@ -8,6 +8,7 @@
 from pydantic import BaseModel, Field
 
 from ocpa import AgentMetadata, ToolMetadata
+from ocpa.auth import verify_bearer_token
 from ocpa.sdk import PolicyDenied, invoke_tool, write_abom
 from ocpa.registry import registry
 from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
@@ -126,15 +127,5 @@ def create_agent(
 
 
 def _require_admin(auth_header: Optional[str], roles_header: Optional[str]) -> None:
-    if not auth_header or not auth_header.startswith("Bearer "):
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token")
-    token = auth_header.removeprefix("Bearer ").strip()
-    expected_token = os.environ.get("OCPA_ADMIN_TOKEN")
     required_role = os.environ.get("OCPA_ADMIN_ROLE", "admin")
-    if not expected_token:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="admin token not configured")
-    if token != expected_token:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="invalid admin token")
-    roles_list = [role.strip() for role in (roles_header or "").split(",") if role.strip()]
-    if required_role and required_role not in roles_list:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="missing required admin role")
+    verify_bearer_token(auth_header, required_role, role_claim=os.environ.get("OCPA_ROLE_CLAIM", "roles"))

ocpa/auth.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import os
+from typing import List, Optional
+
+import jwt
+from fastapi import HTTPException, status
+
+
+def verify_bearer_token(
+    auth_header: Optional[str],
+    required_role: Optional[str],
+    role_claim: str = "roles",
+) -> None:
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token")
+    token = auth_header.removeprefix("Bearer ").strip()
+    secret = os.environ.get("OCPA_JWT_SECRET")
+    if not secret:
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="jwt secret not configured")
+    audience = os.environ.get("OCPA_JWT_AUDIENCE")
+    issuer = os.environ.get("OCPA_JWT_ISSUER")
+    options = {"verify_aud": bool(audience)}
+    try:
+        payload = jwt.decode(
+            token,
+            secret,
+            algorithms=["HS256"],
+            audience=audience,
+            issuer=issuer,
+            options=options,
+        )
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token expired") from None
+    except jwt.InvalidTokenError as exc:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"invalid token: {exc}") from None
+    if required_role:
+        roles = payload.get(role_claim, [])
+        if isinstance(roles, str):
+            roles = [roles]
+        if required_role not in roles:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="missing required admin role")

pyproject.toml

@@ -19,7 +19,8 @@ dependencies = [
     "uvicorn[standard]>=0.23.0",
     "prometheus-client>=0.20.0",
     "typer>=0.12.0",
-    "httpx>=0.27.0"
+    "httpx>=0.27.0",
+    "PyJWT>=2.8.0"
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -10,3 +10,4 @@ uvicorn[standard]>=0.23.0
 prometheus-client>=0.20.0
 typer>=0.12.0
 httpx>=0.27.0
+PyJWT>=2.8.0

tests/test_auth.py

@@ -0,0 +1,84 @@
+import os
+
+import pytest
+from fastapi import HTTPException
+
+from ocpa.auth import verify_bearer_token
+
+
+def test_verify_bearer_token_with_role(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    monkeypatch.setenv("OCPA_ADMIN_ROLE", "admin")
+    import jwt
+
+    token = jwt.encode({"roles": ["admin"]}, "secret", algorithm="HS256")
+    auth_header = f"Bearer {token}"
+
+    verify_bearer_token(auth_header, "admin")
+
+
+def test_verify_bearer_token_missing(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    with pytest.raises(HTTPException):
+        verify_bearer_token(None, "admin")
+
+
+def test_verify_bearer_token_expired(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    import jwt, datetime
+
+    token = jwt.encode(
+        {"roles": ["admin"], "exp": datetime.datetime.utcnow() - datetime.timedelta(seconds=1)},
+        "secret",
+        algorithm="HS256",
+    )
+    auth_header = f"Bearer {token}"
+    with pytest.raises(HTTPException):
+        verify_bearer_token(auth_header, "admin")
+
+
+def test_verify_bearer_token_with_audience_and_issuer(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    monkeypatch.setenv("OCPA_JWT_AUDIENCE", "ocpa")
+    monkeypatch.setenv("OCPA_JWT_ISSUER", "https://issuer.example.com")
+    import jwt
+
+    token = jwt.encode(
+        {"roles": ["admin"], "aud": "ocpa", "iss": "https://issuer.example.com"},
+        "secret",
+        algorithm="HS256",
+    )
+    auth_header = f"Bearer {token}"
+    verify_bearer_token(auth_header, "admin")
+
+
+def test_verify_bearer_token_missing_role(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    import jwt
+
+    token = jwt.encode({"roles": ["user"]}, "secret", algorithm="HS256")
+    auth_header = f"Bearer {token}"
+    with pytest.raises(HTTPException):
+        verify_bearer_token(auth_header, "admin")
+
+
+def test_verify_bearer_token_invalid_token(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    # malformed token
+    with pytest.raises(HTTPException):
+        verify_bearer_token("Bearer not-a-jwt", "admin")
+
+
+def test_verify_bearer_token_missing_secret(monkeypatch):
+    monkeypatch.delenv("OCPA_JWT_SECRET", raising=False)
+    with pytest.raises(HTTPException):
+        verify_bearer_token("Bearer abc", "admin")
+
+
+def test_verify_bearer_token_role_as_string(monkeypatch):
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
+    import jwt
+
+    token = jwt.encode({"roles": "admin"}, "secret", algorithm="HS256")
+    auth_header = f"Bearer {token}"
+    verify_bearer_token(auth_header, "admin")

tests/test_registry_api.py

@@ -6,8 +6,9 @@
 
 def test_registry_api_create_tool_and_agent(tmp_path, monkeypatch):
     monkeypatch.setenv("OCPA_REGISTRY_PATH", str(tmp_path / "registry.json"))
-    monkeypatch.setenv("OCPA_ADMIN_TOKEN", "secret")
+    monkeypatch.setenv("OCPA_JWT_SECRET", "secret")
     monkeypatch.setenv("OCPA_ADMIN_ROLE", "admin")
+    import jwt
     import ocpa.registry as reg_module
 
     importlib.reload(reg_module)
@@ -17,7 +18,8 @@ def test_registry_api_create_tool_and_agent(tmp_path, monkeypatch):
 
     client = TestClient(api.app)
 
-    headers = {"Authorization": "Bearer secret", "X-OCPA-Roles": "admin"}
+    token = jwt.encode({"roles": ["admin"]}, "secret", algorithm="HS256")
+    headers = {"Authorization": f"Bearer {token}", "X-OCPA-Roles": "admin"}
 
     resp = client.post(
         "/registry/tools",