fix auth trust boundary and enforce async job ownership

deploy/compose/init-db.sql

@@ -11,6 +11,7 @@
 --
 -- Tables in aiq_jobs:
 --   - job_info      — NAT JobStore metadata (status, timestamps, expiry)
+--   - job_access    — AIQ-owned job ownership/access control metadata
 --   - job_events    — SSE streaming events and job event persistence
 --   - summaries     — Document summaries (collection + filename keyed)
 --
@@ -51,6 +52,16 @@ CREATE TABLE IF NOT EXISTS job_info (
 CREATE INDEX IF NOT EXISTS idx_job_info_status ON job_info(status);
 CREATE INDEX IF NOT EXISTS idx_job_info_created_at ON job_info(created_at);
 
+CREATE TABLE IF NOT EXISTS job_access (
+    job_id VARCHAR PRIMARY KEY,
+    owner_auth_type VARCHAR NOT NULL,
+    owner_subject VARCHAR NOT NULL,
+    owner_email VARCHAR,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_job_access_owner ON job_access(owner_auth_type, owner_subject);
+
 -- Job events table (SSE streaming, event persistence)
 CREATE TABLE IF NOT EXISTS job_events (
     id SERIAL PRIMARY KEY,

deploy/helm/helm-charts-k8s/aiq/files/init-db.sql

@@ -12,6 +12,7 @@
 --
 -- Tables in aiq_jobs:
 --   - job_info      — NAT JobStore metadata (status, timestamps, expiry)
+--   - job_access    — AIQ-owned job ownership/access control metadata
 --   - job_events    — SSE streaming events and job event persistence
 --   - summaries     — Document summaries (collection + filename keyed)
 --
@@ -52,6 +53,16 @@ CREATE TABLE IF NOT EXISTS job_info (
 CREATE INDEX IF NOT EXISTS idx_job_info_status ON job_info(status);
 CREATE INDEX IF NOT EXISTS idx_job_info_created_at ON job_info(created_at);
 
+CREATE TABLE IF NOT EXISTS job_access (
+    job_id VARCHAR PRIMARY KEY,
+    owner_auth_type VARCHAR NOT NULL,
+    owner_subject VARCHAR NOT NULL,
+    owner_email VARCHAR,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_job_access_owner ON job_access(owner_auth_type, owner_subject);
+
 -- Job events table (SSE streaming, event persistence)
 CREATE TABLE IF NOT EXISTS job_events (
     id SERIAL PRIMARY KEY,

frontends/aiq_api/src/aiq_api/auth/middleware.py

@@ -32,7 +32,7 @@
 can read the caller type without framework coupling.
 
     from aiq_api.auth.middleware import get_current_user
-    user = get_current_user()          # {"type": "internal"|"anonymous", ...}
+    user = get_current_user()          # {"type": "jwt"|"internal"|"anonymous"|...}
     skip = user.get("skip_clarifier")  # True / False
 
 ENVIRONMENT VARIABLES
@@ -60,6 +60,7 @@
 import json
 import logging
 import os
+from contextlib import contextmanager
 from contextvars import ContextVar
 from typing import Any
 
@@ -88,6 +89,16 @@ def get_current_user() -> dict[str, Any]:
     return _current_user.get()
 
 
+@contextmanager
+def user_context(user: dict[str, Any]):
+    """Temporarily bind a resolved caller identity in the auth ContextVar."""
+    token = _current_user.set(user)
+    try:
+        yield
+    finally:
+        _current_user.reset(token)
+
+
 # ---------------------------------------------------------------------------
 # Path configuration
 # ---------------------------------------------------------------------------
@@ -119,6 +130,86 @@ def _load_external_hostnames() -> set[str]:
     return names
 
 
+def is_external_request(headers: dict[bytes, bytes], external_hostnames: set[str] | None = None) -> bool:
+    """Return ``True`` when the Host header matches an external-facing hostname."""
+    host = headers.get(b"host", b"").decode().split(":")[0]
+    return host in (external_hostnames or _load_external_hostnames())
+
+
+def is_headless_request(headers: dict[bytes, bytes]) -> bool:
+    """Return ``True`` for headless callers that should skip the clarifier."""
+    return headers.get(b"x-aiq-mode", b"").decode().lower() == "headless"
+
+
+def extract_auth_token(headers: dict[bytes, bytes]) -> str | None:
+    """Extract a bearer token or idToken cookie from ASGI headers."""
+    auth = headers.get(b"authorization", b"").decode()
+    if auth.startswith("Bearer "):
+        return auth[7:]
+
+    cookie = headers.get(b"cookie", b"").decode()
+    for part in cookie.split(";"):
+        part = part.strip()
+        if part.startswith("idToken="):
+            return part[8:]
+    return None
+
+
+async def validate_token_with_validators(token: str, validators: list) -> dict[str, Any] | None:
+    """Try validators in order and return the first successful identity dict."""
+    for validator in validators:
+        if validator.can_handle(token):
+            user = await validator.validate(token)
+            if user is not None:
+                return user
+    logger.debug("Token rejected by all %d configured validator(s)", len(validators))
+    return None
+
+
+def detect_internal_caller(headers: dict[bytes, bytes]) -> dict[str, Any]:
+    """Classify an internal request without validating any presented token."""
+    token = extract_auth_token(headers)
+    headless = is_headless_request(headers)
+    if token:
+        return {"type": "unverified_jwt", "token": token, "skip_clarifier": headless}
+    return {"type": "internal", "skip_clarifier": headless}
+
+
+async def resolve_request_user(
+    headers: dict[bytes, bytes],
+    *,
+    validators: list,
+    require_auth: bool,
+    external_hostnames: set[str] | None = None,
+) -> tuple[dict[str, Any] | None, int | None, bool]:
+    """Resolve request identity from validated credentials when present.
+
+    Returns a tuple of:
+      - resolved user dict, or None when the request should be rejected
+      - HTTP status code to use on rejection, or None on success
+      - whether the request was classified as external
+    """
+    is_external = is_external_request(headers, external_hostnames)
+    token = extract_auth_token(headers)
+
+    if token:
+        user = await validate_token_with_validators(token, validators)
+        if user is not None:
+            if is_headless_request(headers):
+                user["skip_clarifier"] = True
+            return user, None, is_external
+
+        if is_external and require_auth:
+            return None, 401, is_external
+
+    if is_external:
+        if not require_auth:
+            return {"type": "anonymous", "skip_clarifier": True}, None, is_external
+        return None, 401, is_external
+
+    return detect_internal_caller(headers), None, is_external
+
+
 # ---------------------------------------------------------------------------
 # Middleware
 # ---------------------------------------------------------------------------
@@ -178,44 +269,29 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         headers = dict(scope.get("headers", []))
         path: str = scope["path"]
+        is_external = self._is_external(headers)
 
-        # 1. Internal traffic
-        if not self._is_external(headers):
-            user = self._detect_internal_caller(headers)
-            await self._call_app(scope, receive, send, user)
-            return
-
-        # 2. External — enforce path allowlist
-        if not self._path_allowed(path):
+        # External requests still honor the public path allowlist before auth.
+        if is_external and not self._path_allowed(path):
             await self._send_json(send, 404, {"detail": "Not found"})
             return
 
-        # 3. Auth-exempt paths
-        if path in AUTH_EXEMPT_PATHS:
-            user = {"type": "anonymous", "skip_clarifier": True}
-            await self._call_app(scope, receive, send, user)
-            return
-
-        # 4. Auth disabled — allow as anonymous
-        if not self.require_auth:
+        if is_external and path in AUTH_EXEMPT_PATHS:
             user = {"type": "anonymous", "skip_clarifier": True}
             await self._call_app(scope, receive, send, user)
             return
 
-        # 5. Auth enabled — require a valid JWT Bearer token
-        token = self._extract_token(headers)
-        if not token:
-            await self._send_json(send, 401, {"detail": "Missing auth token"})
-            return
-
-        user = await self._validate_token(token)
+        user, error_status, _ = await resolve_request_user(
+            headers,
+            validators=self._validators,
+            require_auth=self.require_auth,
+            external_hostnames=self._external_hostnames,
+        )
         if user is None:
-            await self._send_json(send, 401, {"detail": "Invalid or expired auth token"})
+            detail = "Invalid or expired auth token" if self._extract_token(headers) else "Missing auth token"
+            await self._send_json(send, error_status or 401, {"detail": detail})
             return
 
-        if self._is_headless(headers):
-            user["skip_clarifier"] = True
-
         await self._call_app(scope, receive, send, user)
 
     # ------------------------------------------------------------------
@@ -233,15 +309,11 @@ async def _call_app(
             scope["state"] = {}
         scope["state"]["user"] = user
 
-        token = _current_user.set(user)
-        try:
+        with user_context(user):
             await self.app(scope, receive, send)
-        finally:
-            _current_user.reset(token)
 
     def _is_external(self, headers: dict[bytes, bytes]) -> bool:
-        host = headers.get(b"host", b"").decode().split(":")[0]
-        return host in self._external_hostnames
+        return is_external_request(headers, self._external_hostnames)
 
     def _path_allowed(self, path: str) -> bool:
         for allowed in EXTERNAL_ALLOWED_PATHS:
@@ -253,47 +325,17 @@ def _path_allowed(self, path: str) -> bool:
         return False
 
     def _extract_token(self, headers: dict[bytes, bytes]) -> str | None:
-        auth = headers.get(b"authorization", b"").decode()
-        if auth.startswith("Bearer "):
-            return auth[7:]
-        # Fall back to idToken cookie (UI / browser callers)
-        cookie = headers.get(b"cookie", b"").decode()
-        for part in cookie.split(";"):
-            part = part.strip()
-            if part.startswith("idToken="):
-                return part[8:]
-        return None
+        return extract_auth_token(headers)
 
     def _is_headless(self, headers: dict[bytes, bytes]) -> bool:
-        return headers.get(b"x-aiq-mode", b"").decode().lower() == "headless"
+        return is_headless_request(headers)
 
     async def _validate_token(self, token: str) -> dict[str, Any] | None:
-        for validator in self._validators:
-            if validator.can_handle(token):
-                user = await validator.validate(token)
-                if user is not None:
-                    return user
-        logger.debug("Token rejected by all %d configured validator(s)", len(self._validators))
-        return None
+        return await validate_token_with_validators(token, self._validators)
 
     def _detect_internal_caller(self, headers: dict[bytes, bytes]) -> dict[str, Any]:
         """Classify an internal request without validating the token."""
-        auth = headers.get(b"authorization", b"").decode()
-        if auth.startswith("Bearer eyJ"):
-            token = auth[7:]
-            headless = self._is_headless(headers)
-            return {"type": "jwt", "token": token, "skip_clarifier": headless}
-
-        cookie = headers.get(b"cookie", b"").decode()
-        for part in cookie.split(";"):
-            part = part.strip()
-            if part.startswith("idToken="):
-                token = part[8:]
-                headless = self._is_headless(headers)
-                return {"type": "jwt", "token": token, "skip_clarifier": headless}
-
-        headless = self._is_headless(headers)
-        return {"type": "internal", "skip_clarifier": headless}
+        return detect_internal_caller(headers)
 
     @staticmethod
     async def _send_json(send: Send, status: int, body: dict) -> None:

frontends/aiq_api/src/aiq_api/jobs/__init__.py

@@ -13,46 +13,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""
-Async job infrastructure for AI-Q agents.
-
-This module provides agent-agnostic job execution infrastructure:
-- EventStore: SQLAlchemy-based event persistence for SSE streaming
-- AgentEventCallback: LangChain callback handler for event emission
-- run_agent_job: Dask task function for running any registered agent
-- submit_agent_job: Submit jobs from application code
-"""
-
-from .callbacks import AgentEventCallback
-from .callbacks import ArtifactType
-from .callbacks import EventCategory
-from .callbacks import EventData
-from .callbacks import EventState
-from .callbacks import IntermediateStepEvent
-from .callbacks import ToolArtifactMapping
-from .connection_manager import SSEConnectionManager
-from .connection_manager import get_connection_manager
-from .connection_manager import reset_connection_manager
-from .event_store import EventStore
-from .runner import CancellationMonitor
-from .runner import run_agent_job
-from .runner import run_with_cancellation
-from .submit import submit_agent_job
+"""Async job submodules for AI-Q."""
 
 __all__ = [
-    "AgentEventCallback",
-    "ArtifactType",
-    "CancellationMonitor",
-    "EventCategory",
-    "EventData",
-    "EventState",
-    "EventStore",
-    "IntermediateStepEvent",
-    "SSEConnectionManager",
-    "ToolArtifactMapping",
-    "get_connection_manager",
-    "reset_connection_manager",
-    "run_agent_job",
-    "run_with_cancellation",
-    "submit_agent_job",
+    "access",
+    "callbacks",
+    "connection_manager",
+    "event_store",
+    "runner",
+    "submit",
 ]