Prefer header JWT when RBAC role-sync enabled

When RBAC/JWT-role sync is enabled, read the end-user JWT from the gateway-forwarded header named by get_jwt_auth_header() instead of the ibm-openrag-session cookie; preserve the existing cookie flow when RBAC is off. Strip a leading "Bearer " prefix if present and fall back to the cookie when role-sync is disabled. Added jwt_roles_enabled and get_jwt_auth_header imports and adjusted token selection logic in src/dependencies.py. Added unit tests (tests/unit/dependencies/test_ibm_header_auth.py) that cover RBAC-on header usage, RBAC-off cookie usage, and missing-header authentication failure.

Author: edwinjosechittilappilly

src/dependencies.py

@@ -432,6 +432,9 @@ async def _get_ibm_user(request: Request, required: bool) -> Optional["User"]:
        for OpenSearch. Decoded and persisted to connections.json per user.
     1. ibm-openrag-session cookie — set by Traefik after validating credentials
        with AMS. JWT is decoded without re-validation (Traefik already validated).
+       When RBAC/JWT-role sync is enabled, this JWT is instead read from the
+       gateway-forwarded header named by ``get_jwt_auth_header()`` (the cookie is
+       used only when RBAC is off); identity and roles both come from that token.
     2. ibm-auth-basic cookie — local dev fallback set by our ibm_login endpoint
        when Traefik is not present.
 
@@ -440,11 +443,13 @@ async def _get_ibm_user(request: Request, required: bool) -> Optional["User"]:
     """
     import auth.ibm_auth as ibm_auth
     from auth.ibm_auth import extract_ibm_credentials
+    from auth.jwt_roles import jwt_roles_enabled
     from config.settings import (
         IBM_CREDENTIALS_HEADER,
         IBM_SESSION_COOKIE_NAME,
         PLATFORM_PASSWORD,
         PLATFORM_USERNAME,
+        get_jwt_auth_header,
     )
 
     # ── Option -1: Environment variable override (local dev/calls) ───────
@@ -471,7 +476,16 @@ async def _get_ibm_user(request: Request, required: bool) -> Optional["User"]:
     # ── Option 0: Configurable credentials header (Traefik production) ───
 
     lh_credentials = request.headers.get(IBM_CREDENTIALS_HEADER, "")
-    ibm_token = request.cookies.get(IBM_SESSION_COOKIE_NAME)
+    # When RBAC/JWT-role sync is on, the gateway forwards the end-user JWT in the
+    # configured header; use it as the source of identity and roles. When RBAC is
+    # off, preserve the existing ibm-openrag-session cookie flow.
+    if jwt_roles_enabled():
+        raw_jwt = request.headers.get(get_jwt_auth_header(), "")
+        ibm_token = (
+            raw_jwt[7:].strip() if raw_jwt.startswith("Bearer ") else raw_jwt.strip()
+        ) or None
+    else:
+        ibm_token = request.cookies.get(IBM_SESSION_COOKIE_NAME)
     user_id = None
     email = None
     name = None

tests/unit/dependencies/test_ibm_header_auth.py

@@ -0,0 +1,99 @@
+"""IBM AMS auth: JWT source switches from cookie to header under RBAC.
+
+Covers the RBAC-gated token source in ``_get_ibm_user`` (``src/dependencies.py``):
+when JWT-role sync is enabled the end-user JWT is read from the gateway-forwarded
+header named by ``get_jwt_auth_header()``; when RBAC is off the existing
+``ibm-openrag-session`` cookie flow is preserved. ``decode_ibm_jwt`` is
+monkeypatched so no real token is needed.
+"""
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+
+ROOT = Path(__file__).resolve().parent.parent.parent.parent
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+import auth.ibm_auth as ibm_auth  # noqa: E402
+import config.settings as app_settings  # noqa: E402
+from dependencies import _get_ibm_user  # noqa: E402
+
+COOKIE_NAME = "ibm-openrag-session"
+
+
+class _FakeRequest:
+    """Minimal stand-in for starlette Request used by ``_get_ibm_user``."""
+
+    def __init__(self, headers: dict | None = None, cookies: dict | None = None):
+        self.headers = headers or {}
+        self.cookies = cookies or {}
+        self.state = SimpleNamespace()
+        # Empty services -> the connector lookup in Option 1 no-ops.
+        self.app = SimpleNamespace(state=SimpleNamespace(services={}))
+
+
+@pytest.fixture(autouse=True)
+def _env(monkeypatch):
+    monkeypatch.setenv("OPENRAG_JWT_ROLES_CLAIM", "openrag_roles")
+    monkeypatch.setenv("OPENRAG_ROLE_CLAIM_ADMIN", "admin")
+    monkeypatch.setenv("OPENRAG_ROLE_CLAIM_DEVELOPER", "manager")
+    monkeypatch.setenv("OPENRAG_ROLE_CLAIM_USER", "user")
+    monkeypatch.delenv("OPENRAG_ROLE_CLAIM_VIEWER", raising=False)
+    monkeypatch.setenv("OPENRAG_JWT_AUTH_HEADER", "X-OpenRAG-JWT")
+    # Keep the Option -1 env override from short-circuiting the function.
+    monkeypatch.setattr(app_settings, "PLATFORM_USERNAME", None)
+    monkeypatch.setattr(app_settings, "PLATFORM_PASSWORD", None)
+
+
+@pytest.mark.asyncio
+async def test_rbac_on_reads_jwt_from_header(monkeypatch):
+    monkeypatch.setenv("OPENRAG_RBAC_ENFORCE", "true")
+    claims = {"sub": "s1", "username": "alice", "display_name": "Alice", "openrag_roles": ["admin"]}
+    monkeypatch.setattr(ibm_auth, "decode_ibm_jwt", lambda tok: claims if tok == "tok" else None)
+    # Cookie present too — it must be ignored when RBAC is on.
+    req = _FakeRequest(headers={"X-OpenRAG-JWT": "Bearer tok"}, cookies={COOKIE_NAME: "COOKIE"})
+
+    user = await _get_ibm_user(req, required=True)
+
+    assert user is not None
+    assert user.user_id == "alice"
+    assert user.name == "Alice"
+    assert user.jwt_token == "Bearer tok"  # built from the header token, not the cookie
+    assert req.state.jwt_roles == ["admin"]
+
+
+@pytest.mark.asyncio
+async def test_rbac_off_reads_jwt_from_cookie(monkeypatch):
+    monkeypatch.setenv("OPENRAG_RBAC_ENFORCE", "false")
+    claims = {"sub": "s2", "username": "bob"}
+    monkeypatch.setattr(ibm_auth, "decode_ibm_jwt", lambda tok: claims if tok == "ctok" else None)
+    # Header present too — it must be ignored when RBAC is off.
+    req = _FakeRequest(headers={"X-OpenRAG-JWT": "Bearer HEADERTOK"}, cookies={COOKIE_NAME: "ctok"})
+
+    user = await _get_ibm_user(req, required=True)
+
+    assert user is not None
+    assert user.user_id == "bob"
+    assert user.jwt_token == "Bearer ctok"
+    assert req.state.jwt_roles is None  # legacy default-role path under RBAC off
+
+
+@pytest.mark.asyncio
+async def test_rbac_on_missing_header_401(monkeypatch):
+    monkeypatch.setenv("OPENRAG_RBAC_ENFORCE", "true")
+
+    def _boom(tok):  # decode must not run when no header token is present
+        raise AssertionError("decode_ibm_jwt should not be called without a header token")
+
+    monkeypatch.setattr(ibm_auth, "decode_ibm_jwt", _boom)
+    # Cookie present but ignored under RBAC -> no token -> unauthenticated.
+    req = _FakeRequest(headers={}, cookies={COOKIE_NAME: "ctok"})
+
+    with pytest.raises(HTTPException) as exc:
+        await _get_ibm_user(req, required=True)
+    assert exc.value.status_code == 401