"""A fake Okta, for tests only — T10.7. `server/okta_auth.py` normally hands the browser off to a real Okta authorize endpoint and exchanges the code with Okta's token endpoint over the network (Authlib discovers both from `/.well-known/openid-configuration`). Tests cannot reach any of that: there is no live Okta tenant in CI, and the browser checks launch the app as a SUBPROCESS (`start_server` in tests/browser_check.py), so a monkeypatch in the test process would never reach the code doing the authenticating. The seam has to be configurable from the ENVIRONMENT — same discipline as `server/ldap_fake.py`, D13/T10.7's LDAP predecessor. Set `WP_OKTA_FAKE_DIRECTORY` to a JSON object and this module stands in for the whole round trip — the authorize redirect, a stand-in "sign in at Okta" screen, and the token exchange — with no network call anywhere: {"root": {"email": "root@example.test", "name": "Root Person"}} The key is the identity value a real Okta ID token would carry in whichever claim `OKTA_IDENTITY_CLAIM` names (default `preferred_username`) — the fake reads `okta_auth.IDENTITY_CLAIM` at request time, so it exercises whatever claim name is actually configured rather than a hard-coded one. WHAT THIS DOES AND DOES NOT REPLACE Only the OAuth-protocol plumbing that talks to Okta over the network is faked — `authorize_redirect` and `authorize_access_token`, both owned by Authlib, a third-party library, not this app's security logic. Everything this app itself decides stays real and untouched in `app.py`'s `okta_login()`/`okta_callback()`: the `?next=` open-redirect guard (`_safe_next_path`), the disabled-account check, JIT provisioning, and which claim carries identity. A fake run exercises the actual code for all of that, not a re-implementation of it — the same boundary `ldap_fake.py` drew around the anonymous-bind guard. THE PRODUCTION GUARD IS THE POINT OF THIS FILE. An environment variable that lets anyone "sign in" as any identity by visiting a picker page is exactly the kind of thing that must never be reachable outside a test process — D16 leaves no local password fallback and no break-glass, so a fake provider silently active in production would be a total authentication bypass with a friendlier UI than most. `is_active()` refuses whenever a real database is configured, using the same test `auth._load_secret` and `ldap_fake.is_active()` already use: a non-SQLite `DATABASE_URL` means production, full stop. `okta_auth.describe()` also shouts when the fake is live, and `app.py` registers the picker/consent routes only when the fake is active at import time — in production they do not exist, not merely refuse. """ import json import logging import os import secrets import time from html import escape from typing import Optional from urllib.parse import quote from starlette.responses import RedirectResponse log = logging.getLogger("wpsuite.okta.fake") ENV_VAR = "WP_OKTA_FAKE_DIRECTORY" # One-time authorization codes, in-process only. The login and the callback that # redeems the code both happen inside the SAME uvicorn process within one test # run, so this needs no more durability than that — the server restart every # check does between runs clears it for free. Not a cache: entries are popped on # first use (below) and expire on their own otherwise. _CODE_TTL_SECONDS = 120 _PENDING_CODES: dict = {} def _raw() -> str: return os.getenv(ENV_VAR, "").strip() def is_active() -> bool: """Whether the fake should answer. False in anything resembling production.""" if not _raw(): return False # Imported lazily: server.db reads DATABASE_URL at import, and this module is # imported from okta_auth (and app.py), which must stay importable on its own. from .db import DATABASE_URL if not str(DATABASE_URL).startswith("sqlite"): log.error( "%s is set but a non-SQLite DATABASE_URL is configured. REFUSING to use " "the fake Okta provider — this looks like production, and D16 leaves no " "other way in, so honouring it would be an authentication bypass. " "Unset %s.", ENV_VAR, ENV_VAR) return False return True def directory() -> dict: try: data = json.loads(_raw()) if not isinstance(data, dict): raise ValueError("top level must be an object") return data except Exception as exc: # noqa: BLE001 — a malformed fake must not look auth-shaped log.error("%s is not valid JSON (%s); the fake directory is empty", ENV_VAR, exc) return {} def new_code(claims: dict) -> str: code = secrets.token_urlsafe(24) _PENDING_CODES[code] = {"claims": claims, "expires": time.time() + _CODE_TTL_SECONDS} return code def consume_code(code: str) -> Optional[dict]: """Pop and return the claims for a code, or None if unknown/expired/reused. Popping makes the code single-use, matching a real authorization code.""" entry = _PENDING_CODES.pop(code, None) if not entry or entry["expires"] < time.time(): return None return entry["claims"] def picker_page(state: str, redirect_uri: str) -> str: """The fake's stand-in for Okta's own sign-in screen — a plain list of the identities `WP_OKTA_FAKE_DIRECTORY` defines, so a browser check can click through a real page rather than skip the round trip with a minted cookie. Deliberately plain: nothing here is styled to resemble a real Okta page.""" from . import okta_auth rows = [] for username in directory(): href = (f"/api/auth/okta/_fake_provider/consent?state={quote(state)}" f"&redirect_uri={quote(redirect_uri, safe='')}&username={quote(username)}") rows.append( f'
  • ' f'Continue as {escape(username)}
    • ') deny_href = (f"/api/auth/okta/_fake_provider/consent?state={quote(state)}" f"&redirect_uri={quote(redirect_uri, safe='')}&deny=1") deny_href = escape(deny_href, quote=True) return ( "FAKE Okta — tests only" "

    *** FAKE OKTA PROVIDER — TESTS ONLY ***

    " f"

    Identity claim in use: {escape(okta_auth.IDENTITY_CLAIM)}

    " "" f'

    Deny access

    ' ) class _FakeOktaClient: """Stands in for Authlib's `oauth.okta` — the same two methods `app.py` calls, the same async signatures, zero network calls.""" async def authorize_redirect(self, request, redirect_uri): state = secrets.token_urlsafe(24) # The only session write this fake makes. authorize_access_token below is # the only read — mirrors exactly what real Authlib does with `state`, # which is what T10.2's missing-SessionMiddleware bug was about: this # round trip is a genuine test of the same plumbing. request.session["_okta_fake_state"] = state target = redirect_uri or "/api/auth/okta/callback" url = (f"/api/auth/okta/_fake_provider?state={quote(state)}" f"&redirect_uri={quote(target, safe='')}") return RedirectResponse(url=url, status_code=302) async def authorize_access_token(self, request): from authlib.integrations.base_client import OAuthError expected = request.session.pop("_okta_fake_state", None) given = request.query_params.get("state", "") if not expected or given != expected: raise OAuthError( error="invalid_state", description="fake Okta: state did not match the session (T10.7 seam)") err = request.query_params.get("error") if err: raise OAuthError( error=err, description=request.query_params.get("error_description", "denied")) code = request.query_params.get("code", "") claims = consume_code(code) if claims is None: raise OAuthError(error="invalid_grant", description="fake Okta: unknown or expired code") return {"userinfo": claims} class FakeOAuth: """Stands in for Authlib's `OAuth()` registry. The real one exposes each registered client as an attribute by name; `app.py` only ever touches `.okta`, so that is the only attribute this needs.""" def __init__(self): self.okta = _FakeOktaClient() def build() -> "FakeOAuth": return FakeOAuth()