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

#!/usr/bin/env python3
"""Does domain authentication hold its guarantees? — D13 / T10.7.

Covers the four things `docs/waves/decisions-2026-08-21.md` calls non-negotiable,
plus the two provisioning rules that decide whether an existing admin survives the
switch. These are the failure modes where the app still *looks* fine:

  1. An empty password must not reach bind(). In LDAP a simple bind with an empty
     password is an ANONYMOUS bind and it SUCCEEDS — so without the guard, a blank
     password authenticates as whatever username was submitted.
  2. TLS must be CERT_REQUIRED with an explicit CA file. CERT_NONE still encrypts,
     so it fails silently; what it loses is the ability to tell a real DC from
     someone harvesting domain passwords.
  3. Group membership must be evaluated through AD's nested-group matching rule.
     Plain memberOf is direct membership only and wrongly refuses real people.
  4. The fake directory must be impossible to select against a real database.
  5. A refused sign-in must create no account.
  6. An existing admin must still be an admin afterwards.

Self-contained: throwaway SQLite + its own uvicorn. No browser, no domain — the
fake directory (server/ldap_fake.py) stands in for the DC.
Exit 0 all passed, 1 a failure, 2 could not run.
"""
import json
import os
import ssl
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request

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 = [], []
PW = "CorrectHorseBattery9"
GROUP = "WP-Suite-Users"
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 post(base, path, payload):
    req = urllib.request.Request(base + path, method="POST",
                                 data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return r.status, json.loads(r.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode() or "{}")
        except Exception:
            return e.code, {}


def start(port, db_path, fake, required_group=""):
    env = dict(os.environ)
    env["DATABASE_URL"] = "sqlite:///" + db_path.replace("\\", "/")
    env["AUTH_SECRET_KEY"] = "ldap-auth-check-not-for-production"
    env["WP_LDAP_FAKE_DIRECTORY"] = json.dumps(fake)
    if required_group:
        env["LDAP_REQUIRED_GROUP"] = required_group
    else:
        env.pop("LDAP_REQUIRED_GROUP", 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 users_in(db_path):
    """Read the users table straight out of the given file.

    Deliberately NOT via server.db.SessionLocal: that engine is bound from
    DATABASE_URL when the module is first imported, so setting the env var later
    keeps reading whichever database was configured first. Two assertions in this
    file passed against the wrong file before that was noticed.
    """
    import sqlite3
    con = sqlite3.connect(db_path)
    try:
        try:
            return {r[0]: r[1] for r in con.execute("select username, role from users")}
        except sqlite3.OperationalError:
            return {}          # no users table yet
    finally:
        con.close()


def main():
    print("1. the guards that do not need a server")

    os.environ["DATABASE_URL"] = "sqlite:///./_ldapcheck_unit.db"
    os.environ["WP_LDAP_FAKE_DIRECTORY"] = json.dumps(
        {"root": {"password": PW, "groups": [GROUP]}})
    from server import ldap_auth

    # 1 — the anonymous-bind guard. Connection is nulled so ANY call to bind()
    # would raise: this proves the guard returns before the transport is touched,
    # rather than merely that the result is a failure.
    saved, ldap_auth.Connection = ldap_auth.Connection, None
    try:
        for label, u, p in [("an empty password", "root", ""),
                            ("a whitespace-only password", "root", "   "),
                            ("an empty username", "", PW)]:
            r = ldap_auth.verify(u, p, required_group="")
            chk(f"{label} is refused without reaching bind()",
                (not r.ok) and r.reason == ldap_auth.EMPTY_INPUT, r.reason)
    finally:
        ldap_auth.Connection = saved

    # 2 — TLS configuration.
    tls = ldap_auth._tls()
    chk("TLS validate is CERT_REQUIRED", tls.validate == ssl.CERT_REQUIRED, tls.validate)
    chk("...with an explicit CA file, not the system store",
        bool(tls.ca_certs_file) and os.path.isfile(tls.ca_certs_file), tls.ca_certs_file)
    src = open(os.path.join(ROOT, "server", "ldap_auth.py"), encoding="utf-8").read()
    # Parse the module rather than grep it: the docstring names validate=ssl.CERT_NONE
    # in order to explain why it must never be used, and a text search cannot tell
    # that apart from an actual call. Walk every keyword argument called `validate`
    # and check what it is really set to.
    import ast as _ast
    bad = []
    for node in _ast.walk(_ast.parse(src)):
        if isinstance(node, _ast.Call):
            for kw in node.keywords:
                if kw.arg == "validate":
                    name = (kw.value.attr if isinstance(kw.value, _ast.Attribute)
                            else getattr(kw.value, "id", ""))
                    if name != "CERT_REQUIRED":
                        bad.append(f"line {node.lineno}: validate={name or '?'}")
    chk("every validate= in the module is CERT_REQUIRED (AST, not grep)",
        not bad, "; ".join(bad))

    # 3 — nested groups. The fake cannot model AD nesting (it is a string list), so
    # what is asserted is that the REAL path builds AD's transitive matching rule
    # into its filter. A test that truly exercises nesting needs a real directory.
    chk("membership uses AD's nested matching rule, not plain memberOf",
        ldap_auth.NESTED_MEMBER_RULE == "1.2.840.113556.1.4.1941"
        and "memberOf:{NESTED_MEMBER_RULE}:=" in src,
        "the transitive matching rule is not in the search filter")

    # 4 — the production guard on the fake, in a subprocess because DATABASE_URL is
    # read at import time.
    out = subprocess.run(
        [sys.executable, "-c",
         "from server import ldap_fake; print(ldap_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_LDAP_FAKE_DIRECTORY": json.dumps({"root": {"password": PW}})})
    chk("the fake directory REFUSES to work against a non-SQLite database",
        out.stdout.strip() == "False", out.stdout.strip() or out.stderr[-200:])

    print("\n2. sign-in, against a server")
    db_fd, db_path = tempfile.mkstemp(suffix=".db"); os.close(db_fd)
    port = free_port()
    fake = {"root": {"password": PW, "mail": "root@example.test",
                     "full_name": "Root Person", "groups": [GROUP]},
            "outsider": {"password": PW, "mail": "outsider@example.test",
                         "full_name": "Out Sider", "groups": ["SomeOtherGroup"]}}
    server = start(port, db_path, fake, required_group=GROUP)
    if server is None:
        print("the test server would not start.")
        return 2
    base = f"http://127.0.0.1:{port}"
    try:
        st, _ = post(base, "/api/auth/login", {"username": "root", "password": PW})
        chk("a correct password signs in", st == 200, st)
        chk("...and provisioned the account at project_user",
            users_in(db_path).get("root") == "project_user", users_in(db_path))

        st, body = post(base, "/api/auth/login", {"username": "root", "password": "wrong-" + PW})
        chk("a wrong password is refused", st == 401, st)
        chk("...with a message that does not say why",
            "password" in (body.get("detail") or "").lower()
            and "expired" not in (body.get("detail") or "").lower(), body)

        st, _ = post(base, "/api/auth/login", {"username": "root", "password": ""})
        chk("a blank password is refused at the endpoint too", st == 401, st)

        before = set(users_in(db_path))
        st, _ = post(base, "/api/auth/login", {"username": "outsider", "password": PW})
        chk("a correct password OUTSIDE the required group is refused", st == 401, st)
        chk("...and no account was created for them",
            set(users_in(db_path)) == before, set(users_in(db_path)) - before)

        st, _ = post(base, "/api/auth/login", {"username": "nobody", "password": PW})
        chk("an unknown account is refused and creates nothing",
            st == 401 and "nobody" not in users_in(db_path), st)
    finally:
        server.kill()
        try:
            server.wait(timeout=10)
        except subprocess.TimeoutExpired:
            pass

    print("\n3. an existing admin survives the switch")
    db_fd2, db2 = tempfile.mkstemp(suffix=".db"); os.close(db_fd2)
    # Build the schema with a NEW engine bound to this file — see users_in().
    import sqlalchemy as _sa
    from server.db import Base
    from server import models                                          # noqa: F401
    eng2 = _sa.create_engine("sqlite:///" + db2.replace("\\", "/"))
    Base.metadata.create_all(bind=eng2)
    with eng2.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'))"))
    eng2.dispose()
    port2 = free_port()
    server2 = start(port2, db2, fake, required_group=GROUP)
    if server2 is None:
        print("the second test server would not start.")
        return 2
    try:
        st, body = post(f"http://127.0.0.1:{port2}", "/api/auth/login",
                        {"username": "root", "password": PW})
        chk("the pre-existing admin signs in", st == 200, st)
        chk("...and is STILL an admin (D13 criterion 4)",
            users_in(db2).get("root") == "admin", users_in(db2))
        chk("...and their locally-set name was not overwritten by the directory",
            (body.get("user") or {}).get("full_name") == "Set By Hand", body.get("user"))
        chk("...and no duplicate account appeared",
            len(users_in(db2)) == 1, 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)