test: cover the Keycloak issuer migration and cross-app token handoff

tests/accessibility/test_accessibility.py

@@ -73,9 +73,13 @@ def test_login_page_has_no_critical_axe_violations(self, page: Page):
         """Login page must pass axe WCAG 2.1 AA scan with no critical/serious issues.
 
         Currently xfailed: see docs/app_bugs.md #5. The Keycloak login page
-        (opub-kc.civicdatalab.in/auth/realms/DataSpace/...) has serious
-        color-contrast and link-name violations. Owned by the Keycloak team,
-        not the Parakh frontend. Confirmed via Playwright MCP 2026-05-08.
+        (auth.civicdatalab.in/realms/DataSpace/... — was
+        opub-kc.civicdatalab.in/auth/realms/DataSpace/... before the
+        2026 Keycloak migration moved every CivicDataLab product to the new
+        host, where Keycloak serves from the domain root with no `/auth`
+        prefix) has serious color-contrast and link-name violations. Owned by
+        the Keycloak team, not the Parakh frontend. Confirmed via Playwright
+        MCP 2026-05-08.
         """
         _require_axe()
         home = HomePage(page)

tests/api/test_keycloak_migration.py

@@ -0,0 +1,240 @@
+"""
+Keycloak migration regression coverage — ParakhAI side.
+
+Every CivicDataLab product moved off `opub-kc.civicdatalab.in`, which served
+each realm under an `/auth` path prefix, onto `auth.civicdatalab.in`, which
+serves from the domain root. Two distinct things changed: the host AND the
+disappearance of the `/auth` segment.
+
+Two failures this guards, both of which happened in production:
+
+1. ParakhAI's sign-in must hand users to the NEW issuer. A stale issuer mints
+   tokens the CivicDataSpace backend rejects.
+
+2. ParakhAI's backend does not verify Keycloak tokens itself. Its
+   `DataSpaceAuthMiddleware` forwards the user's bearer token to the
+   CivicDataSpace backend (`/api/auth/keycloak/login/`) via `dataspace_sdk`
+   and adopts the user it returns. When CivicDataSpace moved to the new
+   Keycloak and ParakhAI had not, every such call failed with
+   `DataSpaceAuthError: Invalid or expired token` — ParakhAI broke because a
+   *different application's* identity provider moved. Separately, the SDK
+   hardcoded `/auth` into its Keycloak URLs and could not reach a root-path
+   Keycloak at all (fixed in dataspace-sdk 0.5.5).
+
+Why this file does NOT assert on a GraphQL query
+------------------------------------------------
+The obvious test — call `{ myAssignments { id } }` authenticated and check it
+succeeds — is worthless here. Verified against dev: the SAME query
+unauthenticated returns the SAME body, `{"data": {"myAssignments": []}}`.
+`DataSpaceAuthMiddleware` swallows a rejected/expired token and silently sets
+`request.user = AnonymousUser()`, returning HTTP 200 with empty data and no
+error. `audits` and `auditorAssignments` behave identically. So no GraphQL
+response distinguishes "authenticated" from "silently anonymous", and any
+assertion built on one would pass even with authentication completely broken.
+
+Instead, test 2 exercises the cross-application handoff directly: take a real
+Keycloak token and present it to the CivicDataSpace endpoint the middleware
+actually calls. That returns 401 when the issuers disagree, which is exactly
+the regression, with no ambiguity.
+
+Markers: api, regression (+ auth for the test needing a real login).
+"""
+
+import os
+import time
+from urllib.parse import unquote
+
+import pytest
+import requests
+
+from utils.config import Config
+
+pytestmark = [pytest.mark.api, pytest.mark.regression]
+
+KEYCLOAK_BASE = "https://auth.civicdatalab.in"
+KEYCLOAK_REALM = "DataSpace"
+EXPECTED_ISSUER = f"{KEYCLOAK_BASE}/realms/{KEYCLOAK_REALM}"
+AUTH_ENDPOINT = f"{EXPECTED_ISSUER}/protocol/openid-connect/auth"
+EXPECTED_CLIENT_ID = "dataspace"
+
+# The endpoint DataSpaceAuthMiddleware calls through dataspace_sdk.
+CDS_BASE = os.getenv("CDS_URL", "https://dev.civicdataspace.in").rstrip("/")
+CDS_API_BASE = CDS_BASE.replace("://dev.", "://dev.api.", 1)
+CDS_KEYCLOAK_LOGIN = f"{CDS_API_BASE}/api/auth/keycloak/login/"
+
+# Dev nginx 403s non-browser User-Agents.
+BROWSER_UA = (
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
+)
+TIMEOUT = 30
+
+
+def _browser_session() -> requests.Session:
+    s = requests.Session()
+    s.headers["User-Agent"] = BROWSER_UA
+    return s
+
+
+class TestSignInIssuer:
+    """ParakhAI must hand sign-in to the migrated Keycloak."""
+
+    def test_signin_redirects_to_migrated_issuer(self):
+        """/api/auth/signin/keycloak must 302 to the new issuer, not opub-kc."""
+        base = Config.BASE_URL.rstrip("/")
+        session = _browser_session()
+
+        csrf = session.get(f"{base}/api/auth/csrf", timeout=TIMEOUT)
+        assert csrf.status_code == 200, (
+            f"CSRF endpoint {base}/api/auth/csrf returned {csrf.status_code}: "
+            f"{csrf.text[:300]}"
+        )
+        token = csrf.json().get("csrfToken")
+        assert token, f"No csrfToken in {csrf.text[:300]}"
+
+        resp = session.post(
+            f"{base}/api/auth/signin/keycloak",
+            data={"csrfToken": token, "callbackUrl": base},
+            allow_redirects=False,
+            timeout=TIMEOUT,
+        )
+        assert resp.status_code == 302, (
+            f"Expected 302 to Keycloak from {base}/api/auth/signin/keycloak, "
+            f"got {resp.status_code}: {resp.text[:300]}"
+        )
+
+        location = resp.headers.get("Location", "")
+        assert AUTH_ENDPOINT in location, (
+            f"Sign-in did not hand off to the migrated authorization endpoint "
+            f"{AUTH_ENDPOINT}. Location: {location!r}"
+        )
+        assert f"client_id={EXPECTED_CLIENT_ID}" in location, (
+            f"Expected client_id={EXPECTED_CLIENT_ID}. Location: {location!r}"
+        )
+        assert f"{base}/api/auth/callback/keycloak" in unquote(location), (
+            f"Expected redirect_uri back to {base}/api/auth/callback/keycloak. "
+            f"Location: {location!r}"
+        )
+
+    def test_signin_does_not_use_decommissioned_keycloak(self):
+        """The old host and its /auth path must not appear in the hand-off."""
+        base = Config.BASE_URL.rstrip("/")
+        session = _browser_session()
+
+        csrf = session.get(f"{base}/api/auth/csrf", timeout=TIMEOUT)
+        token = csrf.json().get("csrfToken")
+        resp = session.post(
+            f"{base}/api/auth/signin/keycloak",
+            data={"csrfToken": token, "callbackUrl": base},
+            allow_redirects=False,
+            timeout=TIMEOUT,
+        )
+        location = resp.headers.get("Location", "")
+
+        assert "opub-kc" not in location, (
+            "Sign-in still points at the decommissioned Keycloak host "
+            f"(opub-kc). Location: {location!r}"
+        )
+        assert "/auth/realms/" not in location, (
+            "Sign-in uses the pre-migration /auth/realms/ path; Keycloak now "
+            f"serves realms from the domain root. Location: {location!r}"
+        )
+
+
+class TestCrossApplicationTokenHandoff:
+    """
+    The ParakhAI -> CivicDataSpace token exchange that DataSpaceAuthMiddleware
+    depends on. This is the path that broke in production (ParakhAI-Backend#107).
+    """
+
+    pytestmark = [pytest.mark.api, pytest.mark.regression, pytest.mark.auth]
+
+    def test_parakh_token_is_accepted_by_civicdataspace(self, authenticated_page):
+        """
+        A token minted for a logged-in ParakhAI user must be accepted by the
+        CivicDataSpace backend endpoint the middleware forwards it to.
+
+        A 401 here is the exact production regression: the two applications
+        trusting different issuers. Asserted directly rather than through a
+        GraphQL response, because the middleware degrades to AnonymousUser
+        silently and every GraphQL query returns 200 with empty data either
+        way (see module docstring).
+        """
+        session_blob = authenticated_page.evaluate(
+            "async () => await (await fetch('/api/auth/session')).json()"
+        )
+        access_token = (session_blob or {}).get("access_token")
+        if not access_token:
+            pytest.skip(
+                "No access_token on the NextAuth session — login did not "
+                "complete, so there is no token to exchange."
+            )
+
+        # The endpoint intermittently exceeds nginx's 60s proxy timeout on dev
+        # (measured: 504, 200 in 7s, 200 in 42s, 504 across four calls). Retry
+        # only that gateway timeout, so a slow backend does not read as a
+        # rejected token - and so a persistently dead endpoint still fails.
+        resp = None
+        last_error = None
+        for attempt in range(3):
+            try:
+                resp = _browser_session().post(
+                    CDS_KEYCLOAK_LOGIN,
+                    json={"token": access_token},
+                    # Longer than nginx's own 60s proxy timeout, so a slow
+                    # response arrives as a status code we can reason about
+                    # instead of a client-side ReadTimeout.
+                    timeout=90,
+                )
+            except requests.RequestException as exc:
+                last_error = exc
+                resp = None
+            else:
+                last_error = None
+                if resp.status_code not in (502, 503, 504):
+                    break
+            if attempt < 2:
+                time.sleep(3)
+
+        # Distinguish "the token was rejected" from "the endpoint could not
+        # answer". Only the first is the #107 regression this test exists to
+        # catch; the second is a separate, known defect on dev - the exchange
+        # endpoint intermittently exceeds nginx's 60s proxy timeout (measured
+        # at roughly half of calls, with successes taking up to 42s).
+        #
+        # Skipping there is deliberate. Failing would report an issuer
+        # mismatch that has not been shown, and retrying harder would just
+        # dress a broken endpoint up as a passing test.
+        if resp is None or resp.status_code in (502, 503, 504):
+            detail = (
+                f"HTTP {resp.status_code}" if resp is not None
+                else f"{type(last_error).__name__}: {last_error}"
+            )
+            pytest.skip(
+                f"{CDS_KEYCLOAK_LOGIN} did not respond after 3 attempts "
+                f"({detail}). The token exchange is timing out on dev, so "
+                "cross-application acceptance cannot be evaluated. This is an "
+                "endpoint availability problem, not an issuer mismatch - a "
+                "401 would still fail this test."
+            )
+
+        # Asserted as == 200, not != 401. A 5xx also satisfies "not 401", so
+        # the weaker form passed while the endpoint was timing out entirely -
+        # proving nothing about whether the token is accepted.
+        # == 200, not != 401: a 5xx also satisfies "not 401", so the weaker
+        # form passed while the endpoint was timing out entirely and proved
+        # nothing about whether the token is accepted.
+        assert resp.status_code == 200, (
+            f"CivicDataSpace did not accept a live ParakhAI token at "
+            f"{CDS_KEYCLOAK_LOGIN} (HTTP {resp.status_code}). ParakhAI's "
+            "DataSpaceAuthMiddleware forwards user tokens here, so every "
+            "authenticated ParakhAI request degrades to anonymous when this "
+            "fails - the silent failure in ParakhAI-Backend#107. "
+            f"Body: {resp.text[:300]}"
+        )
+
+        body = resp.json()
+        assert body.get("access"), (
+            "Token exchange succeeded but returned no 'access' token, so "
+            f"DataSpaceAuthMiddleware could not adopt a user. Body: {resp.text[:300]}"
+        )

tests/e2e/test_add_model_flow.py

@@ -43,6 +43,28 @@ def skip_if_cds_unreachable() -> None:
         pytest.skip(f"CivicDataSpace ({CDS_BASE}) unreachable — skipping CDS tests")
 
 
+# Login-page indicators — same keyword idiom used in tests/e2e/test_auth.py and
+# test_homepage.py. Kept host-agnostic on purpose: every CivicDataLab product
+# migrated from `opub-kc.civicdatalab.in/auth/realms/DataSpace` to
+# `auth.civicdatalab.in/realms/DataSpace` (different host AND the `/auth` path
+# segment is gone — Keycloak now serves from the domain root). The previous
+# guard matched the literal strings "opub-kc" and "auth/realms", so after the
+# migration it stopped firing entirely and a session bounce surfaced as a
+# confusing assertion failure instead of a clean skip.
+_LOGIN_URL_KEYWORDS = ("login", "auth", "keycloak", "sso", "signin", "sign-in")
+
+
+def _is_login_redirect(url: str) -> bool:
+    """True when *url* looks like a Keycloak/SSO login page rather than the app.
+
+    Matches either Keycloak's realm path (`/realms/...`, present on both the old
+    `/auth/realms/...` and the new root-path `/realms/...` deployments) or any of
+    the generic login keywords, so it works against either server.
+    """
+    lowered = url.lower()
+    return "/realms/" in lowered or any(kw in lowered for kw in _LOGIN_URL_KEYWORDS)
+
+
 class TestAddModelRedirect:
     """Tests for the ParakhAI side of the Add Model cross-platform redirect."""
 
@@ -117,7 +139,7 @@ def test_cds001_no_js_syntax_error_on_editor_page_load(self, page: Page):
         page.on("pageerror", lambda e: errors.append(str(e)))
         page.goto(Config.cds_url("/en/manage/ai-models"), wait_until="domcontentloaded", timeout=20000)
         page.wait_for_timeout(3000)
-        if "opub-kc" in page.url or "auth/realms" in page.url:
+        if _is_login_redirect(page.url):
             pytest.skip(
                 "CDS editor not reached — page (uses anonymous `page` fixture, no CDS "
                 f"auth) redirected to Keycloak login before the editor loaded: {page.url}"
@@ -144,7 +166,7 @@ def test_cds001_editor_has_no_console_errors_on_load(self, page: Page):
         page.on("pageerror", lambda e: console_errors.append(str(e)))
         page.goto(Config.cds_url("/en/manage/ai-models"), wait_until="domcontentloaded", timeout=20000)
         page.wait_for_timeout(3000)
-        if "opub-kc" in page.url or "auth/realms" in page.url:
+        if _is_login_redirect(page.url):
             pytest.skip(
                 "CDS editor not reached — page (uses anonymous `page` fixture, no CDS "
                 f"auth) redirected to Keycloak login before the editor loaded: {page.url}"