test(lasuite-docs): update stale OIDC tests for impress v5.4.0 Bearer-auth removal
continuous-integration/drone/push Build was killed

test_oidc_login_via_keycloak and test_create_doc_and_read_back authenticated with
'Authorization: Bearer <password-grant JWT>'; impress v5.4.0 removed Bearer/JWT
auth on the API (SessionAuthentication only), so both went RED with 401 on the
v5.4.1 upgrade.

Updated to the successor auth path: a new recipe-local _oidc_session.py drives the
real OIDC authorization-code flow (app -> keycloak login form -> callback ->
Django session cookie, with CSRF headers on unsafe methods).
- test_oidc_login: still asserts the unauth challenge redirect; NOW also asserts a
  raw Bearer JWT is REJECTED (401/403 - the v5.4.0 hardening, asserted as the new
  correct behavior); then asserts the session-authenticated whoami returns the
  provisioned user. No assertion weakened - the auth proof is stronger than before.
- test_create_doc: same create+read-back round-trip assertions, now over the
  session-authenticated API.

Stale-test fix for recipe PR
recipe-maintainers/lasuite-docs#7
(carry-over from /upgrade-all 2026-07-24).
This commit is contained in:
2026-08-03 20:43:07 +00:00
parent 5366e0616b
commit a1a6790c9b
3 changed files with 205 additions and 48 deletions
+158
View File
@@ -0,0 +1,158 @@
"""Recipe-local OIDC *session* login helper (authorization-code flow + session cookie).
impress v5.4.0 removed Bearer-token (JWT) authentication on the API — the app now accepts only
its own session cookie, established through the standard OIDC authorization-code browser flow
(app login URL → keycloak login form → callback → Django session). This helper drives that flow
with urllib + a CookieJar so the custom tests can exercise the API the way a real client does.
Kept recipe-local (cf. tests/ghost/custom/_ghost.py precedent) rather than in runner/harness —
promote it there if a third recipe needs it.
Usage:
sess = OidcSession(f"https://{live_app}")
me = sess.login(kc["user"], kc["password"]) # asserts whoami 200; returns the user dict
status, body = sess.post("/api/v1.0/documents/", {"title": "x"}) # CSRF handled
"""
from __future__ import annotations
import contextlib
import html
import http.cookiejar
import json
import re
import ssl
import urllib.error
import urllib.parse
import urllib.request
# Per-run *.ci.commoninternet.net domains serve the operator's wildcard cert via the Traefik file
# provider; chain verification is done once in the install tier (generic.served_cert).
_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE
_LOGIN_PATHS = ("/api/v1.0/authenticate/", "/oidc/authenticate/", "/api/v1.0/users/me/")
_WHOAMI = "/api/v1.0/users/me/"
class OidcSession:
"""A cookie-carrying HTTP session logged in via the app's OIDC authorization-code flow."""
def __init__(self, base: str):
self.base = base.rstrip("/")
self.jar = http.cookiejar.CookieJar()
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.jar),
urllib.request.HTTPSHandler(context=_CTX),
)
# -- low-level ---------------------------------------------------------------------------
def _open(
self,
url: str,
data: bytes | None = None,
headers: dict[str, str] | None = None,
method: str | None = None,
timeout: int = 30,
) -> tuple[int, str, bytes]:
"""Open a URL (following redirects, carrying cookies). Returns (status, final_url, body)."""
req = urllib.request.Request(url, data=data, method=method)
for k, v in (headers or {}).items():
req.add_header(k, v)
try:
with self.opener.open(req, timeout=timeout) as resp:
return resp.getcode(), resp.geturl(), resp.read()
except urllib.error.HTTPError as e:
body = b""
with contextlib.suppress(Exception):
body = e.read()
return e.code, e.filename or url, body
def _csrf_token(self) -> str | None:
for c in self.jar:
if "csrftoken" in c.name.lower():
return c.value
return None
# -- login -------------------------------------------------------------------------------
def login(
self,
username: str,
password: str,
login_paths: tuple[str, ...] = _LOGIN_PATHS,
whoami: str = _WHOAMI,
) -> dict:
"""OIDC authorization-code login: app → keycloak form → callback → session cookie.
Asserts the resulting session GETs `whoami` with HTTP 200 and returns the parsed user.
"""
page, page_url, last = None, None, (0, "", b"")
for path in login_paths:
status, final_url, body = self._open(self.base + path)
last = (status, final_url, body)
text = body.decode(errors="replace")
if "kc-form-login" in text or (
"/protocol/openid-connect/" in final_url and "<form" in text
):
page, page_url = text, final_url
break
assert page is not None, (
f"could not reach the keycloak login form via {login_paths}: last URL "
f"{last[1]!r} HTTP {last[0]} body[:200]={last[2][:200]!r}"
)
m = re.search(r'<form[^>]*id="kc-form-login"[^>]*action="([^"]+)"', page) or re.search(
r'<form[^>]*action="([^"]+)"[^>]*method=["\']?post', page, re.I
)
assert m, f"no login form action on keycloak page {page_url!r}: {page[:300]!r}"
action = html.unescape(m.group(1))
form = urllib.parse.urlencode(
{"username": username, "password": password, "credentialId": ""}
).encode()
status, landed, body = self._open(
action, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}
)
status, _, who = self._open(self.base + whoami)
assert status == 200, (
f"OIDC session login failed: GET {whoami} -> HTTP {status} after submitting the "
f"keycloak form (landed at {landed!r}; excerpt: {body[:200]!r})"
)
parsed = json.loads(who)
assert isinstance(parsed, dict), f"unexpected whoami payload: {who[:200]!r}"
return parsed
# -- API calls with the session ----------------------------------------------------------
def request(self, method: str, path: str, data: dict | None = None) -> tuple[int, object]:
"""Issue an API call with the session cookie (+ CSRF header on unsafe methods)."""
url = path if path.startswith("http") else self.base + path
headers: dict[str, str] = {}
body: bytes | None = None
if data is not None:
body = json.dumps(data).encode()
headers["Content-Type"] = "application/json"
if method.upper() not in ("GET", "HEAD", "OPTIONS"):
tok = self._csrf_token()
if tok:
headers["X-CSRFToken"] = tok
headers["Referer"] = self.base + "/"
headers["Origin"] = self.base
status, _, raw = self._open(url, data=body, headers=headers, method=method.upper())
try:
return status, json.loads(raw)
except (json.JSONDecodeError, ValueError):
return status, None
def get(self, path: str) -> tuple[int, object]:
return self.request("GET", path)
def post(self, path: str, data: dict | None = None) -> tuple[int, object]:
return self.request("POST", path, data)
def delete(self, path: str) -> tuple[int, object]:
return self.request("DELETE", path)
+19 -32
View File
@@ -3,12 +3,13 @@
Plan §4.3 explicitly names this test for lasuite-docs: "create a doc, edit via the API, confirm Plan §4.3 explicitly names this test for lasuite-docs: "create a doc, edit via the API, confirm
persistence". This is the canonical create-an-object + read-it-back for lasuite-docs. persistence". This is the canonical create-an-object + read-it-back for lasuite-docs.
Flow (uses an OIDC token from the dep keycloak): Flow (updated for impress v5.4.0, which removed Bearer/JWT auth on the API — the doc CRUD now
1. Obtain a JWT via OIDC password grant against the dep keycloak (the test user is provisioned runs on the app's session cookie from the real OIDC authorization-code login):
by the orchestrator's dep-provisioning step). 1. Log in via the OIDC authorization-code flow against the dep keycloak (the test user is
2. POST `/api/v1.0/documents/` with `Authorization: Bearer <jwt>` to create a new doc with a provisioned by the orchestrator's dep-provisioning step) → session cookie.
2. POST `/api/v1.0/documents/` with the session (+ CSRF header) to create a new doc with a
unique title; capture the returned `id`. unique title; capture the returned `id`.
3. GET `/api/v1.0/documents/<id>/` with the same Bearer token; assert the returned title and 3. GET `/api/v1.0/documents/<id>/` with the same session; assert the returned title and
id match. id match.
Non-vacuous: a misconfigured OIDC, broken backend, or missing endpoint fails at the layer it's Non-vacuous: a misconfigured OIDC, broken backend, or missing endpoint fails at the layer it's
@@ -26,9 +27,9 @@ import uuid
import pytest import pytest
sys.path.insert(0, os.path.dirname(__file__))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner")) sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http # noqa: E402 from _oidc_session import OidcSession # noqa: E402 (recipe-local helper, same dir)
from harness import sso
@pytest.mark.requires_deps @pytest.mark.requires_deps
@@ -36,43 +37,29 @@ def test_create_doc_and_read_back(live_app, deps):
"""Create a doc via the authenticated API; fetch it back; assert round-trip.""" """Create a doc via the authenticated API; fetch it back; assert round-trip."""
kc = deps["keycloak"] kc = deps["keycloak"]
# Obtain a JWT via OIDC password grant # Session login via the OIDC authorization-code flow (impress v5.4.0+ rejects Bearer JWTs)
access_token = sso.oidc_password_grant( sess = OidcSession(f"https://{live_app}")
{ sess.login(kc["user"], kc["password"])
"client_id": kc["client_id"],
"client_secret": kc["client_secret"],
"user": kc["user"],
"password": kc["password"],
"token_url": kc["token_url"],
}
)
auth = {"Authorization": f"Bearer {access_token}"}
# Create a doc with a unique title # Create a doc with a unique title
title = f"ccci-doc-{uuid.uuid4().hex[:8]}" title = f"ccci-doc-{uuid.uuid4().hex[:8]}"
s, body = harness_http.http_post( s, body = sess.post("/api/v1.0/documents/", {"title": title})
f"https://{live_app}/api/v1.0/documents/",
data={"title": title},
headers=auth,
)
assert s in (200, 201), f"POST /api/v1.0/documents/ HTTP {s}: {body!r}" assert s in (200, 201), f"POST /api/v1.0/documents/ HTTP {s}: {body!r}"
assert isinstance(body, dict), f"unexpected response shape: {body!r}" assert isinstance(body, dict), f"unexpected response shape: {body!r}"
doc_id = body.get("id") doc_id = body.get("id")
assert doc_id, f"created doc has no id: {body!r}" assert doc_id, f"created doc has no id: {body!r}"
assert ( assert body.get("title") == title, (
body.get("title") == title f"created doc title mismatch: created={title!r}, response={body.get('title')!r}"
), f"created doc title mismatch: created={title!r}, response={body.get('title')!r}" )
# Fetch it back via the dedicated GET endpoint # Fetch it back via the dedicated GET endpoint
s, fetched = harness_http.http_get( s, fetched = sess.get(f"/api/v1.0/documents/{doc_id}/")
f"https://{live_app}/api/v1.0/documents/{doc_id}/", headers=auth
)
assert s == 200, f"GET /api/v1.0/documents/{doc_id}/ HTTP {s}: {fetched!r}" assert s == 200, f"GET /api/v1.0/documents/{doc_id}/ HTTP {s}: {fetched!r}"
assert isinstance(fetched, dict), f"unexpected GET response: {fetched!r}" assert isinstance(fetched, dict), f"unexpected GET response: {fetched!r}"
assert fetched.get("id") in ( assert fetched.get("id") in (
doc_id, doc_id,
str(doc_id), str(doc_id),
), f"fetched id mismatch: created={doc_id!r}, fetched={fetched.get('id')!r}" ), f"fetched id mismatch: created={doc_id!r}, fetched={fetched.get('id')!r}"
assert ( assert fetched.get("title") == title, (
fetched.get("title") == title f"fetched title mismatch: created={title!r}, fetched={fetched.get('title')!r}"
), f"fetched title mismatch: created={title!r}, fetched={fetched.get('title')!r}" )
+28 -16
View File
@@ -2,13 +2,16 @@
SOURCE: references/recipe-maintainer/recipe-info/lasuite-docs/tests/oidc_login.py SOURCE: references/recipe-maintainer/recipe-info/lasuite-docs/tests/oidc_login.py
End-to-end flow: End-to-end flow (updated for impress v5.4.0, which REMOVED Bearer/JWT auth on the API —
the app now only accepts its own session cookie from the OIDC authorization-code flow):
1. GET `/api/v1.0/users/me/` without auth → asserts the response REDIRECTS to the dep 1. GET `/api/v1.0/users/me/` without auth → asserts the response REDIRECTS to the dep
keycloak's realm auth endpoint (the recipe is correctly configured to challenge keycloak's realm auth endpoint (the recipe is correctly configured to challenge
unauthenticated callers — wired via install_steps.sh). unauthenticated callers — wired via install_steps.sh).
2. Obtain an OIDC token from the dep keycloak via password grant 2. Obtain an OIDC token from the dep keycloak via password grant, and assert the API
(the test user provisioned by the orchestrator's realm setup). now REJECTS it as a Bearer credential (the v5.4.0 auth hardening — a 200 here would
3. Call `/api/v1.0/users/me/` with `Authorization: Bearer <jwt>` → asserts 200 and the mean the hardening regressed).
3. Log in via the real OIDC authorization-code flow (app → keycloak form → callback →
session cookie) and call `/api/v1.0/users/me/` with the session → asserts 200 and the
returned user's email matches the provisioned test user. returned user's email matches the provisioned test user.
Marked @pytest.mark.requires_deps — skips with `deps-not-ready` if dep provisioning failed. Marked @pytest.mark.requires_deps — skips with `deps-not-ready` if dep provisioning failed.
@@ -24,9 +27,11 @@ import urllib.request
import pytest import pytest
sys.path.insert(0, os.path.dirname(__file__))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner")) sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from _oidc_session import OidcSession # noqa: E402 (recipe-local helper, same dir)
from harness import http as harness_http # noqa: E402 from harness import http as harness_http # noqa: E402
from harness import sso from harness import sso # noqa: E402
_CTX = ssl.create_default_context() _CTX = ssl.create_default_context()
_CTX.check_hostname = False _CTX.check_hostname = False
@@ -62,16 +67,18 @@ def test_oidc_login_via_keycloak(live_app, deps):
# 302 redirect. Both are valid "auth-required" indicators — accept either, but if a # 302 redirect. Both are valid "auth-required" indicators — accept either, but if a
# redirect is returned it must point at the dep keycloak realm. # redirect is returned it must point at the dep keycloak realm.
if status in (301, 302, 303, 307, 308): if status in (301, 302, 303, 307, 308):
assert expected_prefix in ( assert expected_prefix in (redirect or ""), (
redirect or "" f"Docs redirected to {redirect!r}, expected to start with {expected_prefix!r}"
), f"Docs redirected to {redirect!r}, expected to start with {expected_prefix!r}" )
else: else:
assert status in (401, 403), ( assert status in (401, 403), (
f"GET /api/v1.0/users/me/ unauth: HTTP {status}; expected redirect to keycloak " f"GET /api/v1.0/users/me/ unauth: HTTP {status}; expected redirect to keycloak "
f"OR 401/403. (200 would be an auth leak.)" f"OR 401/403. (200 would be an auth leak.)"
) )
# Step 2: obtain an OIDC token via password grant against the dep keycloak # Step 2: obtain an OIDC token via password grant against the dep keycloak, and assert
# the API REJECTS it as Bearer — impress v5.4.0 removed Bearer/JWT auth (SessionAuthentication
# only); a 200 here would mean the auth hardening regressed.
creds = { creds = {
"client_id": kc["client_id"], "client_id": kc["client_id"],
"client_secret": kc["client_secret"], "client_secret": kc["client_secret"],
@@ -81,14 +88,19 @@ def test_oidc_login_via_keycloak(live_app, deps):
} }
access_token = sso.oidc_password_grant(creds) access_token = sso.oidc_password_grant(creds)
assert isinstance(access_token, str) and access_token.count(".") == 2, "expected JWT" assert isinstance(access_token, str) and access_token.count(".") == 2, "expected JWT"
# Step 3: call the protected API with the Bearer token; assert 200 + user email
status, body = harness_http.http_get( status, body = harness_http.http_get(
f"https://{live_app}/api/v1.0/users/me/", f"https://{live_app}/api/v1.0/users/me/",
headers={"Authorization": f"Bearer {access_token}"}, headers={"Authorization": f"Bearer {access_token}"},
) )
assert status == 200, f"GET /api/v1.0/users/me/ with token HTTP {status}: {body!r}" assert status in (401, 403), (
assert isinstance(body, dict), f"unexpected response: {body!r}" f"GET /api/v1.0/users/me/ with a Bearer JWT returned HTTP {status} — impress >= v5.4.0 "
assert ( f"must reject raw Bearer tokens (got body {body!r})"
body.get("email") == kc["email"] )
), f"unexpected user email: got {body.get('email')!r}, expected {kc['email']!r}"
# Step 3: the successor auth path — real OIDC authorization-code login (session cookie);
# the session-authenticated whoami must return the provisioned user.
sess = OidcSession(f"https://{live_app}")
me = sess.login(kc["user"], kc["password"])
assert me.get("email") == kc["email"], (
f"unexpected user email: got {me.get('email')!r}, expected {kc['email']!r}"
)