cc-ci/tests/lasuite-docs/custom/_oidc_session.py

"""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)