🔒 Security fix: Enforce JWT audience verification

libs/kest-core/python/src/kest/core/framework/ext.py

@@ -149,10 +149,15 @@ class KestIdentityMiddleware:
     """
 
     def __init__(
-        self, app, jwks_uri: str | None = None, user_claim: str = "preferred_username"
+        self,
+        app,
+        jwks_uri: str | None = None,
+        user_claim: str = "preferred_username",
+        audience: str | list[str] | None = None,
     ):
         self.app = app
         self.user_claim = user_claim
+        self.audience = audience
         self._jwks_client = None
         if jwks_uri:
             try:
@@ -191,7 +196,7 @@ def _decode_token(self, token: str) -> dict:
                     token,
                     signing_key.key,
                     algorithms=["RS256", "ES256"],
-                    options={"verify_aud": False},
+                    audience=self.audience,
                 )
             except Exception as exc:
                 print(f"[Kest.IdentityMiddleware] JWT verification failed: {exc}")

showcase/kest-lab/agent.py

@@ -6,6 +6,7 @@
 its own SPIFFE identity + the OBO JWT, which propagates alice's identity
 through the entire hop chain via KestIdentityMiddleware + OTel baggage.
 """
+
 import os
 import traceback
 import httpx
@@ -58,7 +59,9 @@ def _load_cedar_policies(policy_dir: str = "/workspace/app/cedar/policies") -> d
                 with open(os.path.join(policy_dir, fname)) as f:
                     policies[policy_id] = f.read()
     if not policies:
-        policies["fallback"] = 'permit(principal, action, resource) when { context["trust_score"] >= 10 };'
+        policies["fallback"] = (
+            'permit(principal, action, resource) when { context["trust_score"] >= 10 };'
+        )
     return policies
 
 
@@ -92,6 +95,7 @@ def _load_cedar_policies(policy_dir: str = "/workspace/app/cedar/policies") -> d
     KestIdentityMiddleware,
     jwks_uri=JWKS_URI,
     user_claim="preferred_username",
+    audience="account",
 )
 app.add_middleware(KestMiddleware)
 
@@ -173,7 +177,9 @@ async def call_hop1_as_agent(obo_token: str):
     # Read kest.user from OTel baggage (set by KestIdentityMiddleware from OBO JWT)
     # Spec-compliant key: SPEC-v0.3.0 §8.4
     resolved_user = baggage.get_baggage("kest.user")
-    print(f"[{SERVICE_NAME}] Calling hop1 as agent, delegating for user={resolved_user}.")
+    print(
+        f"[{SERVICE_NAME}] Calling hop1 as agent, delegating for user={resolved_user}."
+    )
     async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
         response = await client.get(
             HOP1_URL,
@@ -237,18 +243,22 @@ async def _delegate_to_gateway_logic(obo_token: str):
       2. Extracts the task token from the response
       3. Calls kest-gateway /execute-task with the task token
     """
-    resolved_user = baggage.get_baggage("kest.user")  # spec-compliant (SPEC-v0.3.0 §8.4)
-    print(f"[{SERVICE_NAME}] /delegate-to-gateway — delegating for user={resolved_user!r}")
+    resolved_user = baggage.get_baggage(
+        "kest.user"
+    )  # spec-compliant (SPEC-v0.3.0 §8.4)
+    print(
+        f"[{SERVICE_NAME}] /delegate-to-gateway — delegating for user={resolved_user!r}"
+    )
 
     # Inject W3C traceparent and baggage so gateway continues the trace
     span_ctx = trace.get_current_span().get_span_context()
     traceparent = ""
     if span_ctx and span_ctx.is_valid:
         traceparent = f"00-{span_ctx.trace_id:032x}-{span_ctx.span_id:016x}-{span_ctx.trace_flags:02x}"
-        
+
     all_baggage = baggage.get_all()
     baggage_header = ",".join(f"{k}={v}" for k, v in all_baggage.items())
-    
+
     base_headers = {}
     if traceparent:
         base_headers["traceparent"] = traceparent
@@ -271,12 +281,14 @@ async def _delegate_to_gateway_logic(obo_token: str):
         auth_data = auth_resp.json()
         task_token = auth_data.get("task_token", "")
         if not task_token:
-            raise HTTPException(status_code=500, detail="kest-gateway did not return a task_token")
+            raise HTTPException(
+                status_code=500, detail="kest-gateway did not return a task_token"
+            )
 
         # Step 2: Submit the narrow task token to /execute-task
         exec_headers = dict(base_headers)
         exec_headers["Content-Type"] = "application/json"
-        
+
         exec_resp = await client.post(
             f"{GATEWAY_URL}/execute-task",
             headers=exec_headers,

showcase/kest-lab/app.py

@@ -50,6 +50,7 @@
     KestIdentityMiddleware,
     jwks_uri=JWKS_URI,
     user_claim="preferred_username",  # Use username, not UUID, for policy/audit matching
+    audience="account",
 )
 
 
@@ -61,6 +62,7 @@ async def permission_error_handler(request: Request, exc: PermissionError):
     """
     return JSONResponse(status_code=403, content={"detail": str(exc)})
 
+
 # --- Configure Kest ---
 NEXT_HOP = os.getenv("NEXT_HOP_URL", "none")
 

showcase/kest-lab/gateway.py

@@ -20,6 +20,7 @@
 Security note: if this gateway is compromised, the attacker can mint arbitrary
 task tokens. Mitigations are documented in docs/GATEWAY_E2E.md §"What If Compromised?".
 """
+
 import os
 import json
 import time
@@ -118,6 +119,7 @@ def _load_cedar_policies(policy_dir: str = "/workspace/app/cedar/policies") -> d
     KestIdentityMiddleware,
     jwks_uri=JWKS_URI,
     user_claim="preferred_username",
+    audience="account",
 )
 app.add_middleware(KestMiddleware)
 
@@ -176,12 +178,16 @@ def _mint_task_token(
 
     # Build a minimal JWS using the gateway's identity provider
     header = {"alg": "EdDSA", "typ": "JWT", "kid": SERVICE_NAME}
-    header_b64 = base64.urlsafe_b64encode(
-        json.dumps(header, separators=(",", ":")).encode()
-    ).rstrip(b"=").decode()
-    payload_b64 = base64.urlsafe_b64encode(
-        json.dumps(payload, separators=(",", ":")).encode()
-    ).rstrip(b"=").decode()
+    header_b64 = (
+        base64.urlsafe_b64encode(json.dumps(header, separators=(",", ":")).encode())
+        .rstrip(b"=")
+        .decode()
+    )
+    payload_b64 = (
+        base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode())
+        .rstrip(b"=")
+        .decode()
+    )
 
     signing_input = f"{header_b64}.{payload_b64}"
     signature = identity_provider.sign(signing_input.encode())
@@ -204,6 +210,7 @@ def _decode_task_token(token: str) -> dict:
 # Endpoints
 # ---------------------------------------------------------------------------
 
+
 @app.post("/authorise")
 async def authorise_handler(request: Request):
     """
@@ -239,18 +246,24 @@ async def _authorise_logic():
       - context["scope"] contains "read:data"
     """
     try:
-        user = str(baggage.get_baggage("kest.user") or "")    # spec §8.4
-        agent = str(baggage.get_baggage("kest.agent") or "")   # spec §8.4
-        scope = str(baggage.get_baggage("kest.scope") or "")   # impl extension (raw OAuth scope)
+        user = str(baggage.get_baggage("kest.user") or "")  # spec §8.4
+        agent = str(baggage.get_baggage("kest.agent") or "")  # spec §8.4
+        scope = str(
+            baggage.get_baggage("kest.scope") or ""
+        )  # impl extension (raw OAuth scope)
 
         print(
             f"[{SERVICE_NAME}] /authorise — user={user!r}, agent={agent!r}, scope={scope!r}"
         )
 
         if not user:
-            raise HTTPException(status_code=403, detail="No delegated user identity in OBO token")
+            raise HTTPException(
+                status_code=403, detail="No delegated user identity in OBO token"
+            )
         if not agent:
-            raise HTTPException(status_code=403, detail="No agent identity in OBO token")
+            raise HTTPException(
+                status_code=403, detail="No agent identity in OBO token"
+            )
 
         # Mint a narrow task token
         task_token = _mint_task_token(
@@ -292,7 +305,9 @@ async def execute_task_handler(request: Request):
     body = await request.json()
     task_token = body.get("task_token", "")
     if not task_token:
-        raise HTTPException(status_code=400, detail="task_token required in request body")
+        raise HTTPException(
+            status_code=400, detail="task_token required in request body"
+        )
 
     # Decode the task token and inject its scope into baggage so
     # task_policy can read context["principal_scope"] == "task:process-data"
@@ -307,10 +322,14 @@ async def execute_task_handler(request: Request):
     from opentelemetry import baggage as otel_baggage
 
     ctx = otel_context.get_current()
-    ctx = otel_baggage.set_baggage("kest.task", task_scope, context=ctx)          # spec key
-    ctx = otel_baggage.set_baggage("kest.scope", task_scope, context=ctx)         # impl extension
-    ctx = otel_baggage.set_baggage("kest.user", delegated_user, context=ctx)     # spec key
-    ctx = otel_baggage.set_baggage("kest.agent", delegated_agent, context=ctx)   # spec key
+    ctx = otel_baggage.set_baggage("kest.task", task_scope, context=ctx)  # spec key
+    ctx = otel_baggage.set_baggage(
+        "kest.scope", task_scope, context=ctx
+    )  # impl extension
+    ctx = otel_baggage.set_baggage("kest.user", delegated_user, context=ctx)  # spec key
+    ctx = otel_baggage.set_baggage(
+        "kest.agent", delegated_agent, context=ctx
+    )  # spec key
     token = otel_context.attach(ctx)
     try:
         return await _execute_task_logic(task_token, delegated_user, delegated_agent)
@@ -322,7 +341,9 @@ async def execute_task_handler(request: Request):
     policy="task_policy",
     origin="internal",
 )
-async def _execute_task_logic(task_token: str, delegated_user: str, delegated_agent: str):
+async def _execute_task_logic(
+    task_token: str, delegated_user: str, delegated_agent: str
+):
     """
     @kest_verified-decorated logic for /execute-task.
 
@@ -380,11 +401,13 @@ async def _execute_task_logic(task_token: str, delegated_user: str, delegated_ag
 # Health
 # ---------------------------------------------------------------------------
 
+
 @app.get("/health")
 async def health():
     return {"status": "ok", "service": SERVICE_NAME}
 
 
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run(app, host="0.0.0.0", port=8002)