diff --git a/docs/waves/wave-10.md b/docs/waves/wave-10.md index 4365b61..160fdcc 100644 --- a/docs/waves/wave-10.md +++ b/docs/waves/wave-10.md @@ -63,6 +63,28 @@ Depends only on `main` as it stands after `D15`. Not sequenced behind any other - **T10.7 — Test coverage without a live Okta dependency.** A fake-OIDC-provider test seam, mirroring `ldap_fake.py`, so the suite runs with no live Okta tenant reachable. + Built: `server/okta_fake.py` (env-driven, `WP_OKTA_FAKE_DIRECTORY`, production-refusing + the same way `ldap_fake.py` does), dispatched from `okta_auth._build_oauth()` before + the real Authlib client is considered. Only the two Authlib calls that touch the + network — `authorize_redirect` / `authorize_access_token` — are faked; `app.py`'s + `okta_login()`/`okta_callback()` (the `?next=` guard, the disabled-account check, JIT + provisioning, the identity-claim lookup) run unmodified against the fake, same + boundary the LDAP predecessor drew around the anonymous-bind guard. Two fake-only + routes (`/_fake_provider`, `/_fake_provider/consent`) stand in for Okta's own sign-in + screen and are registered in `app.py` only when the fake is active at import time — + in production they do not exist, not merely refuse. `tests/browser_check.py`'s + `start_server()` now takes an optional `extra_env` and sets + `WP_OKTA_FAKE_DIRECTORY` unconditionally (same reasoning `ldap_fake`'s equivalent + used: almost nothing signs in, but the one check that does should not fail + mysteriously). `tests/url_state_check.py` scenario 2 is un-skipped and drives the + real round trip — login.html's button, the fake picker page, the fake consent + redirect, `okta_callback()` — proving `?next=` survives it, same tightened + "actually left login.html" assertion the LDAP predecessor's own bug fix used. + `tests/okta_auth_check.py` is new: the production guard, single-use/replay on the + authorization code, an unsolicited callback hit, a tampered state, a denied consent, + an unknown identity, a disabled account, JIT provisioning, an existing admin + surviving unchanged, same-site vs. off-site `?next=`, and `OKTA_IDENTITY_CLAIM` + genuinely working under a non-default claim name — 22/22. - **T10.8 — Verification.** 390px and 1440px, full suite, done-when checks per task, matching the rigor D13 was held to. diff --git a/server/app.py b/server/app.py index e7774b5..3df79e3 100644 --- a/server/app.py +++ b/server/app.py @@ -27,7 +27,7 @@ from sqlalchemy.orm import Session from starlette.middleware.sessions import SessionMiddleware from .db import Base, engine, get_db -from . import models, auth, notify, assets_db, okta_auth +from . import models, auth, notify, assets_db, okta_auth, okta_fake log = logging.getLogger("wpsuite.app") @@ -766,6 +766,46 @@ async def okta_callback(request: Request, db: Session = Depends(get_db)): return redirect +# ── Fake Okta test seam (T10.7) ────────────────────────────────────────────── +# Registered ONLY when the fake is active — checked once, at import time, same +# timing okta_auth.oauth itself is built at. In production these two routes do +# not exist at all, not merely refuse a request: see server/okta_fake.py's +# docstring for why that distinction matters given D16 leaves no other way in. +if okta_fake.is_active(): + + @app.get("/api/auth/okta/_fake_provider") + async def okta_fake_provider(request: Request): + """Stands in for Okta's own sign-in screen. A plain list of the + identities WP_OKTA_FAKE_DIRECTORY defines, so a browser check drives a + real page through a real round trip rather than skipping it.""" + state = request.query_params.get("state", "") + redirect_uri = request.query_params.get("redirect_uri") or "/api/auth/okta/callback" + from fastapi.responses import HTMLResponse + return HTMLResponse(okta_fake.picker_page(state, redirect_uri)) + + @app.get("/api/auth/okta/_fake_provider/consent") + async def okta_fake_consent(request: Request): + """What clicking an identity (or Deny) on the fake picker does: hands + back an authorization code (or an error) at okta_callback, exactly the + shape a real Okta redirect would carry. Everything after this — the + state check, JIT provisioning, the disabled-account and open-redirect + guards — is the real okta_callback() above, unmodified.""" + state = request.query_params.get("state", "") + redirect_uri = request.query_params.get("redirect_uri") or "/api/auth/okta/callback" + if request.query_params.get("deny"): + return RedirectResponse( + url=f"{redirect_uri}?error=access_denied&error_description=denied+by+fake+user&state={state}", + status_code=303) + username = request.query_params.get("username", "") + entry = okta_fake.directory().get(username) + if not isinstance(entry, dict): + return RedirectResponse( + url=f"{redirect_uri}?error=invalid_request&error_description=unknown+fake+identity&state={state}", + status_code=303) + claims = {okta_auth.IDENTITY_CLAIM: username, + "email": entry.get("email", ""), "name": entry.get("name", "")} + code = okta_fake.new_code(claims) + return RedirectResponse(url=f"{redirect_uri}?code={code}&state={state}", status_code=303) @app.get("/api/auth/me") diff --git a/server/okta_auth.py b/server/okta_auth.py index 4bbcc9d..f12f41e 100644 --- a/server/okta_auth.py +++ b/server/okta_auth.py @@ -63,6 +63,10 @@ def describe() -> str: """One line for the startup log, matching the LDAPS module's discipline: an unconfigured deploy must be visible in `docker compose logs api`, not discovered at the login button.""" + from . import okta_fake + if okta_fake.is_active(): + return (f"*** FAKE OKTA PROVIDER ACTIVE — identities come from " + f"{okta_fake.ENV_VAR}, NOT from Okta. Tests only. ***") if not HAVE_AUTHLIB: return "Okta auth DISABLED — authlib is not installed. No one can sign in." missing = [name for name, val in ( @@ -75,9 +79,18 @@ def describe() -> str: f"identity claim {IDENTITY_CLAIM!r}") -def _build_oauth() -> Optional["OAuth"]: +def _build_oauth(): """Register the Okta client. Returns None when unconfigured so the caller (T10.2's - routes) can fail loudly instead of Authlib raising deep inside a request.""" + routes) can fail loudly instead of Authlib raising deep inside a request. + + Checked before `is_configured()`: the fake (T10.7) needs none of the four real + Okta settings, and must win whenever it is legitimately active so a test run + never has to also fill in placeholder OKTA_ISSUER/CLIENT_ID/etc. `is_active()` + already refuses outside a SQLite-backed test database — see okta_fake.py.""" + from . import okta_fake + if okta_fake.is_active(): + log.warning("*** FAKE OKTA PROVIDER ACTIVE (%s) — tests only ***", okta_fake.ENV_VAR) + return okta_fake.build() if not is_configured(): return None oauth = OAuth() diff --git a/server/okta_fake.py b/server/okta_fake.py new file mode 100644 index 0000000..09824fc --- /dev/null +++ b/server/okta_fake.py @@ -0,0 +1,190 @@ +"""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() diff --git a/tests/browser_check.py b/tests/browser_check.py index 99c7246..8212657 100644 --- a/tests/browser_check.py +++ b/tests/browser_check.py @@ -26,6 +26,7 @@ browser found, or the server would not start). 2 is distinct on purpose: "I coul not test this" is not the same answer as "this is broken". """ import argparse +import json import os import subprocess import sys @@ -149,10 +150,29 @@ def seed(db_path): for u in db.query(models.User).all()} -def start_server(port, db_path): +# T10.7. The app authenticates through Okta, which no test can reach, and +# start_server launches it as a SUBPROCESS — so a monkeypatch here would never +# reach the code doing the authenticating. server/okta_fake.py reads this +# instead, and refuses to work against a non-SQLite database. +# +# Almost no check ever signs in (seed() mints tokens with auth.create_token and +# sets the cookie directly), so this matters only where the sign-in ROUND TRIP +# is driven — url_state_check's deep-link case. It is set for every server here +# anyway so that a test which starts signing in later does not fail +# mysteriously — the same reasoning the LDAP predecessor (D13/T10.7) used. +FAKE_DIRECTORY = json.dumps({ + u: {"email": f"{u}@example.test", "name": u.title()} + for u in ("root", "sue", "pat", "mix", "bob", "sam", "legacy", "new") +}) + + +def start_server(port, db_path, extra_env=None): env = dict(os.environ) env["DATABASE_URL"] = "sqlite:///" + db_path.replace("\\", "/") env.setdefault("AUTH_SECRET_KEY", "browser-check-secret-not-for-production") + env["WP_OKTA_FAKE_DIRECTORY"] = FAKE_DIRECTORY + if extra_env: + env.update(extra_env) proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "server.app:app", "--host", "127.0.0.1", "--port", str(port), "--log-level", "warning"], diff --git a/tests/okta_auth_check.py b/tests/okta_auth_check.py new file mode 100644 index 0000000..73e2fe6 --- /dev/null +++ b/tests/okta_auth_check.py @@ -0,0 +1,327 @@ +#!/usr/bin/env python3 +"""Does Okta sign-in hold its guarantees? — D15/D16 / T10.7. + +The Okta equivalent of tests/ldap_auth_check.py (D13's predecessor). Covers the +things that would still let the app *look* fine while quietly failing: + + 1. The fake provider must be impossible to select against a real database. + 2. A one-time authorization code cannot be redeemed twice (replay). + 3. An unsolicited hit on the callback (no real login ever started) is refused, + not a 500 — and creates no account. + 4. A denied ("Cancel") consent is refused cleanly (?error=cancelled) and + creates no account. + 5. An unknown fake identity at consent is refused and creates no account. + 6. A correct sign-in works, and ?next= carries through to the real target — + but only when it is a same-site path; an off-site next= is ignored. + 7. A disabled local account is refused even though Okta itself approved it — + deprovisioning stays local (D15). + 8. An unrecognized identity is JIT-provisioned at the lowest role. + 9. An existing admin signs in and is STILL an admin, with their locally-set + name intact — Okta never overwrites what this app already knows. + 10. OKTA_IDENTITY_CLAIM is genuinely configurable: sign-in still works with a + non-default claim name, proving T10.3's "no hard-coded claim" promise. + +Self-contained: throwaway SQLite + its own uvicorn. No browser, no live Okta — +server/okta_fake.py stands in for the provider. Two layers, like its LDAP +predecessor: guards that need no server (direct calls into okta_fake), then +sign-in checks against a real running app. +Exit 0 all passed, 1 a failure, 2 could not run. +""" +import json +import os +import re +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from http.cookiejar import CookieJar +from urllib.parse import quote, urlparse + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from cdp import free_port # noqa: E402 + +_PASS, _FAIL = [], [] +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +def chk(label, ok, detail=""): + (_PASS if ok else _FAIL).append(label) + print(f" {'PASS' if ok else 'FAIL'} {label}" + ("" if ok else f" {detail}")) + + +def users_in(db_path): + """Read the users table straight out of the given file. Deliberately NOT + via server.db.SessionLocal — that engine binds from DATABASE_URL at import, + so setting the env var later keeps reading whichever file came first. The + LDAP predecessor lost two assertions to exactly this before it was noticed.""" + import sqlite3 + con = sqlite3.connect(db_path) + try: + try: + return {r[0]: (r[1], r[2], bool(r[3])) + for r in con.execute( + "select username, role, full_name, is_active from users")} + except sqlite3.OperationalError: + return {} + finally: + con.close() + + +def start(port, db_path, fake, identity_claim=None): + env = dict(os.environ) + env["DATABASE_URL"] = "sqlite:///" + db_path.replace("\\", "/") + env["AUTH_SECRET_KEY"] = "okta-auth-check-not-for-production" + env["WP_OKTA_FAKE_DIRECTORY"] = json.dumps(fake) + if identity_claim: + env["OKTA_IDENTITY_CLAIM"] = identity_claim + else: + env.pop("OKTA_IDENTITY_CLAIM", None) + proc = subprocess.Popen( + [sys.executable, "-m", "uvicorn", "server.app:app", "--host", "127.0.0.1", + "--port", str(port), "--log-level", "warning"], + env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=ROOT) + for _ in range(160): + try: + with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/health", timeout=1): + return proc + except Exception: + if proc.poll() is not None: + return None + time.sleep(0.25) + proc.kill() + return None + + +def opener(): + return urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(CookieJar())) + + +def fetch(op, url): + """GET, auto-following redirects (urllib's default) — matches what a real + browser does across the login -> fake-provider -> callback -> target chain. + Returns (final_url, status, body).""" + try: + with op.open(url, timeout=10) as r: + return r.geturl(), r.status, r.read().decode("utf-8", "replace") + except urllib.error.HTTPError as e: + return e.geturl(), e.code, e.read().decode("utf-8", "replace") + + +def sign_in(base, op, username, next_path=None): + """Drive the real round trip: /api/auth/okta/login -> the fake provider's + picker page -> consent as `username` -> okta_callback(). Every hop except + the fake provider itself is the app's own unmodified code.""" + login_url = base + "/api/auth/okta/login" + if next_path: + login_url += "?next=" + quote(next_path, safe="") + final_url, status, body = fetch(op, login_url) + if "_fake_provider" not in final_url: + return final_url, status, body # never reached the fake at all + m = re.search( + r'id="okta-fake-identity-%s" href="([^"]+)"' % re.escape(username), body) + if not m: + return final_url, status, body # identity not offered + consent_url = base + m.group(1).replace("&", "&") + return fetch(op, consent_url) + + +def deny(base, op): + _, _, body = fetch(op, base + "/api/auth/okta/login") + m = re.search(r'id="okta-fake-deny" href="([^"]+)"', body) + if not m: + return None, None, body + return fetch(op, base + m.group(1).replace("&", "&")) + + +def consent_as(base, op, username, state_override=None): + """Reach the consent endpoint directly with an arbitrary `username` (which + need not be one the picker actually offered) and, optionally, a `state` + that does not match the one the login step stashed in the session — so + 'unknown identity' and 'tampered state' can be tested as the server's own + refusal, not merely as absence from the picker's list.""" + _, _, body = fetch(op, base + "/api/auth/okta/login") + m = re.search(r'id="okta-fake-deny" href="([^"]+)"', body) + if not m: + return None, None, body + href = m.group(1).replace("&", "&").replace("deny=1", "username=" + quote(username)) + if state_override is not None: + href = re.sub(r"state=[^&]*", "state=" + quote(state_override), href) + return fetch(op, base + href) + + +def main(): + print("1. the guards that do not need a server") + + os.environ["DATABASE_URL"] = "sqlite:///./_oktacheck_unit.db" + os.environ["WP_OKTA_FAKE_DIRECTORY"] = json.dumps({"root": {"email": "r@x.test", "name": "R"}}) + from server import okta_fake + + chk("the fake is active against a SQLite database", okta_fake.is_active()) + + out = subprocess.run( + [sys.executable, "-c", + "from server import okta_fake; print(okta_fake.is_active())"], + cwd=ROOT, capture_output=True, text=True, + env={**os.environ, + "DATABASE_URL": "postgresql+psycopg://u:p@localhost:5432/db", + "AUTH_SECRET_KEY": "x", + "WP_OKTA_FAKE_DIRECTORY": json.dumps({"root": {"email": "r@x.test"}})}) + chk("the fake provider REFUSES to work against a non-SQLite database", + out.stdout.strip() == "False", out.stdout.strip() or out.stderr[-200:]) + + code = okta_fake.new_code({"preferred_username": "root"}) + first = okta_fake.consume_code(code) + second = okta_fake.consume_code(code) + chk("a fresh code redeems once", first == {"preferred_username": "root"}, first) + chk("...and a REPLAYED code is refused the second time", second is None, second) + + print("\n2. sign-in, against a server") + db_fd, db_path = tempfile.mkstemp(suffix=".db"); os.close(db_fd) + port = free_port() + fake = {"root": {"email": "root@example.test", "name": "Root Person"}, + "newperson": {"email": "newperson@example.test", "name": "New Person"}} + server = start(port, db_path, fake) + if server is None: + print("the test server would not start.") + return 2 + base = f"http://127.0.0.1:{port}" + try: + # Seed one pre-existing admin and one pre-existing but disabled account, + # the same way manage_users.py / the admin console would have left them. + import sqlalchemy as _sa + from server.db import Base + from server import models # noqa: F401 + eng = _sa.create_engine("sqlite:///" + db_path.replace("\\", "/")) + Base.metadata.create_all(bind=eng) + with eng.begin() as con: + con.execute(_sa.text( + "insert into users (id,username,email,full_name,role,is_active," + "failed_attempts,token_version,project_role,locale,timezone," + "auto_add_projects,auto_add_role,created_at,updated_at) values " + "('user_root','root','','Set By Hand','admin',1,0,0,'','','',0,''," + "datetime('now'),datetime('now'))")) + con.execute(_sa.text( + "insert into users (id,username,email,full_name,role,is_active," + "failed_attempts,token_version,project_role,locale,timezone," + "auto_add_projects,auto_add_role,created_at,updated_at) values " + "('user_shelved','shelved','','Shelved Person','project_user',0,0,0," + "'','','',0,'',datetime('now'),datetime('now'))")) + eng.dispose() + fake["shelved"] = {"email": "shelved@example.test", "name": "Shelved Person"} + # Restart so the running process picks up the augmented directory. + server.kill(); server.wait(timeout=10) + server = start(port, db_path, fake) + if server is None: + print("the test server would not restart with the augmented directory.") + return 2 + + final_url, status, _ = sign_in(base, opener(), "root") + chk("a seeded identity signs in", status == 200 and urlparse(final_url).path == "/index.html", + (final_url, status)) + chk("...and the existing admin is STILL an admin", + users_in(db_path).get("root", ("", "", None))[0] == "admin", users_in(db_path)) + chk("...with their locally-set name untouched by Okta", + users_in(db_path).get("root", (None, None))[1] == "Set By Hand", users_in(db_path)) + + before = set(users_in(db_path)) + final_url, status, _ = sign_in(base, opener(), "newperson") + chk("an unrecognized identity is JIT-provisioned", status == 200, (final_url, status)) + chk("...at the lowest role", users_in(db_path).get("newperson", ("", "", None))[0] + == "project_user", users_in(db_path).get("newperson")) + chk("...and only one new row appeared", + set(users_in(db_path)) - before == {"newperson"}, set(users_in(db_path)) - before) + + final_url, status, _ = sign_in(base, opener(), "shelved") + chk("a disabled local account is refused despite Okta approving it", + "error=disabled" in final_url, final_url) + + before = set(users_in(db_path)) + final_url, status, _ = consent_as(base, opener(), "nobody-such-identity") + # okta_callback()'s `except OAuthError` is deliberately generic (T10.5: + # a plain-language ?error= for whatever Authlib/the provider rejected, + # not a code-by-code breakdown) — so this lands on the same ?error= + # cancelled as every other refusal, not a distinct "invalid_request". + # The thing actually under test is the SERVER-side refusal, verified by + # checking no account got created — not the display string. + chk("consenting as an identity the fake never offered is refused BY THE SERVER" + " (not just absent from the picker)", + "login.html" in final_url and "error=cancelled" in final_url, final_url) + chk("...and creates nothing", set(users_in(db_path)) == before, + set(users_in(db_path)) - before) + + before = set(users_in(db_path)) + final_url, status, _ = consent_as(base, opener(), "root", state_override="tampered-state") + chk("a consent hit whose state does not match the session is refused", + "login.html" in final_url and "error=cancelled" in final_url, final_url) + chk("...and creates nothing", set(users_in(db_path)) == before, + set(users_in(db_path)) - before) + + before = set(users_in(db_path)) + final_url, status, _ = deny(base, opener()) + chk("denying consent lands back on login with a plain message", + "login.html" in final_url and "error=cancelled" in final_url, final_url) + chk("...and creates nothing", set(users_in(db_path)) == before, + set(users_in(db_path)) - before) + + cold = opener() + final_url, status, _ = fetch( + cold, base + "/api/auth/okta/callback?code=forged&state=forged") + chk("an unsolicited hit on the callback (no login ever started) is refused, not a 500", + status == 200 and "login.html" in final_url and "error=cancelled" in final_url, + (final_url, status)) + + final_url, status, _ = sign_in(base, opener(), "root", next_path="/wp-creation-index.html?wp=x") + chk("a same-site ?next= survives the round trip", + urlparse(final_url).path == "/wp-creation-index.html" + and "wp=x" in urlparse(final_url).query, final_url) + + final_url, status, _ = sign_in(base, opener(), "root", next_path="https://evil.example.com/phish") + chk("an off-site ?next= is ignored, not honoured", + urlparse(final_url).path == "/index.html", final_url) + finally: + server.kill() + try: + server.wait(timeout=10) + except subprocess.TimeoutExpired: + pass + + print("\n3. the identity claim name is genuinely configurable (T10.3)") + db_fd2, db2 = tempfile.mkstemp(suffix=".db"); os.close(db_fd2) + port2 = free_port() + server2 = start(port2, db2, {"root": {"email": "root@example.test", "name": "Root"}}, + identity_claim="upn") + if server2 is None: + print("the identity-claim test server would not start.") + return 2 + try: + base2 = f"http://127.0.0.1:{port2}" + final_url, status, _ = sign_in(base2, opener(), "root") + chk("sign-in works with a non-default OKTA_IDENTITY_CLAIM (upn)", + status == 200 and urlparse(final_url).path == "/index.html", (final_url, status)) + chk("...and JIT-provisioned the account under that identity", + users_in(db2).get("root", ("", "", None))[0] == "project_user", users_in(db2)) + finally: + server2.kill() + try: + server2.wait(timeout=10) + except subprocess.TimeoutExpired: + pass + + print("\n" + "-" * 54) + print(f"{len(_PASS)}/{len(_PASS) + len(_FAIL)} checks passed.") + for f in _FAIL: + print(" - " + f) + return 1 if _FAIL else 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Exception as exc: # noqa: BLE001 + print(f"could not run: {type(exc).__name__}: {exc}") + sys.exit(2) diff --git a/tests/url_state_check.py b/tests/url_state_check.py index 82dcd7f..1bf7bf6 100644 --- a/tests/url_state_check.py +++ b/tests/url_state_check.py @@ -7,11 +7,13 @@ the suite, so no work package had an address. This checks the promise those emai will rest on. 1. a URL identifying a work package opens that work package - 2. the same URL works for a SIGNED-OUT user, via login, landing on the target - — SKIPPED as of T10.4: local login is gone (D15/D16), and the redirect- - through-Okta replacement doesn't exist until T10.5 rebuilds login.html. - Re-test this once that lands. Signing in via a minted token stands in as - setup only, so scenarios 3-6 below still get a signed-in page to run on. + 2. the same URL works for a SIGNED-OUT user, via the real Okta sign-in round + trip (T10.7's fake provider stands in for Okta itself — see + server/okta_fake.py — but the app's own login.html, the redirect to + /api/auth/okta/login, the state round trip through SessionMiddleware, and + okta_callback()'s handling of ?next= are all real, unmodified code). + Landing on the requested target, not the home page, doubles as setup for + scenarios 3-6 below. 3. refresh preserves project, package, tab and view 4. Back and Forward move through states without a reload or a broken view 5. the URL survives being copied to a second browsing context @@ -172,9 +174,7 @@ def main(): finally: page2.close() - print("\n2. the same URL works for a signed-out user, via login") - print(" SKIPPED: local login is gone (D15/D16); the Okta-redirect replacement") - print(" doesn't exist until T10.5. Re-test the next= round trip once it lands.") + print("\n2. the same URL works for a signed-out user, via the real Okta round trip") page.clear_cookies() page.goto(deep) settle(page, 1.6) @@ -183,14 +183,44 @@ def main(): nxt = page.eval("new URLSearchParams(location.search).get('next')||''") chk("...carrying the requested target, package id and all", "wp-creation-index.html" in nxt and "wp=wpA1" in nxt, "next=%r" % nxt) - # Setup only for scenarios 3-6 below, NOT a re-test of "signing in continues - # to the requested page" — that promise is specific to the login FORM this - # task removed, and can't be honestly re-proven until T10.5 rebuilds it as an - # Okta redirect. A minted-token cookie gets `page` to the same signed-in, - # on-target state those later scenarios need, without claiming to have - # exercised the (currently nonexistent) sign-in flow itself. - page.set_cookie("wp_session", tok["root"]) - page.goto(deep) + + # Drive the actual button, not a shortcut to it — its href already + # carries ?next= (login.js's safeNext()); this is the same click a + # person makes. + signin_href = page.eval( + "(document.getElementById('okta-signin')||{}).getAttribute('href')||''") + chk("the sign-in link itself carries ?next=", "next=" in signin_href, signin_href) + page.goto(base + signin_href) + for _ in range(30): + if "_fake_provider" in page.eval("location.href"): + break + time.sleep(0.3) + chk("the app hands off to the (fake) Okta provider", + "_fake_provider" in page.eval("location.href"), page.eval("location.href")) + + # The fake provider's own picker page — a real page, not a shortcut. + # See server/okta_fake.py: only the network-touching Authlib calls are + # faked, not app.py's own login/callback/JIT/guard code. + identity_href = page.eval( + "(document.getElementById('okta-fake-identity-root')||{}).getAttribute('href')||''") + chk("the fake provider offers the seeded 'root' identity", bool(identity_href), + page.eval("document.body.innerHTML")) + page.goto(base + identity_href) + for _ in range(30): + href = page.eval("location.href") + if "login.html" not in href and "_fake_provider" not in href: + break + time.sleep(0.3) + settle(page, 1.0) + # NOT `"wp-creation-index.html" in location.href` alone — that string + # is in the ?next= parameter too, so this would pass while still + # sitting on login.html with the sign-in rejected (the same mistake + # D13/T10.7's LDAP predecessor caught and fixed here). Assert we + # actually LEFT the login page. + href = page.eval("location.href") + chk("signing in continues to the requested page, not the home page", + "login.html" not in href and "wp-creation-index.html" in href + and "wp=wpA1" in href, href) for _ in range(30): if page.eval("!!window.wpCreatorReady"): break