Project-SDE-WP-Suite/server/okta_fake.py

"""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 `<issuer>/.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'<li><a id="okta-fake-identity-{escape(username)}" href="{escape(href, quote=True)}">'
            f'Continue as {escape(username)}</a></li>')
    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 (
        "<!doctype html><title>FAKE Okta — tests only</title>"
        "<h1>*** FAKE OKTA PROVIDER — TESTS ONLY ***</h1>"
        f"<p>Identity claim in use: <code>{escape(okta_auth.IDENTITY_CLAIM)}</code></p>"
        "<ul>" + "".join(rows) + "</ul>"
        f'<p><a id="okta-fake-deny" href="{deny_href}">Deny access</a></p>'
    )


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