Project-SDE-WP-Suite/tests/okta_auth_check.py

#!/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)