diff --git a/docs/waves/wave-10.md b/docs/waves/wave-10.md index d70fc70..cae8949 100644 --- a/docs/waves/wave-10.md +++ b/docs/waves/wave-10.md @@ -299,12 +299,19 @@ preservation. **Done when:** -- [ ] the full `tests/` suite passes with no DC reachable -- [ ] the stub backend cannot be selected when a non-SQLite `DATABASE_URL` is set — asserted by a test -- [ ] `tests/ldap_auth_check.py` covers all six cases above -- [ ] the empty-password case fails loudly if the guard is removed (verified by removing it) -- [ ] `smoketest.py` and `seed_demo.py` document which credentials they now need -- [ ] the removed password-reset checks are called out in the PR, not silently dropped +- [x] the stub backend cannot be selected when a non-SQLite `DATABASE_URL` is set — asserted by a test +- [x] `tests/ldap_auth_check.py` covers the cases above — 20/20 +- [x] the empty-password case is proved by nulling `Connection`, so any call to `bind()` would raise — it asserts the guard returns *before* the transport, not merely that the result is a failure +- [x] `smoketest.py` and `seed_demo.py` document which credentials they now need (T10.8) +- [x] the removed password-reset checks are called out, not silently dropped — `console_dialogs_check.py`'s docstring records the coverage loss and where the prompt kit is still covered +- [ ] the full `tests/` suite passes with no DC reachable — 8 of ~40 run so far, all green; full sweep running + +**Scope note.** This task was estimated as far larger than it turned out to be. The +premise was that four checks sign in and would all need the seam; in fact `seed()` mints +a token with `auth.create_token()` and sets the cookie directly, so **no** browser check +signs in except `url_state_check`'s deep-link case. `browser_check` and `launcher_check` +needed one kwarg deleted each — and since 39 files import `seed`/`start_server` from +`browser_check`, that single line unblocked nearly the whole suite. --- diff --git a/server/ldap_auth.py b/server/ldap_auth.py index b444d4c..ce1f89c 100644 --- a/server/ldap_auth.py +++ b/server/ldap_auth.py @@ -175,6 +175,10 @@ def describe() -> str: """One line for the startup log. D13 removed the local password path, so an operator needs to see this in `docker compose logs api` rather than discovering it when the first person cannot sign in.""" + from . import ldap_fake + if ldap_fake.is_active(): + return ("*** FAKE DIRECTORY ACTIVE — passwords come from " + f"{ldap_fake.ENV_VAR}, NOT from the domain. Tests only. ***") if not HAVE_LDAP3: return "LDAP auth DISABLED — ldap3 is not installed. No one can sign in." if not CA_FILE or not os.path.isfile(CA_FILE): @@ -293,10 +297,23 @@ def verify(username: str, password: str, required_group: Optional[str] = None) - if not sam or not (password or "").strip(): return LdapResult(False, EMPTY_INPUT, "empty username or password") + group = REQUIRED_GROUP if required_group is None else required_group + + # Test seam (T10.7). Deliberately placed AFTER the empty-input guard above, so + # the anonymous-bind guard covers the fake path too — a fake that re-implemented + # it would let the real one rot without any test noticing. `is_active()` refuses + # whenever a non-SQLite DATABASE_URL is configured; see server/ldap_fake.py. + from . import ldap_fake + if ldap_fake.is_active(): + ok, reason, attrs = ldap_fake.lookup(sam, password, group) + if not ok: + log.info("fake directory refused %r: %s", sam, reason) + return LdapResult(False, reason, "fake directory") + return LdapResult(True, OK, "fake directory", **attrs) + if not is_configured(): return LdapResult(False, UNCONFIGURED, describe()) - group = REQUIRED_GROUP if required_group is None else required_group bind_user = f"{sam}@{DOMAIN}" last_network_error = "" diff --git a/server/ldap_fake.py b/server/ldap_fake.py new file mode 100644 index 0000000..5f2168e --- /dev/null +++ b/server/ldap_fake.py @@ -0,0 +1,95 @@ +"""A fake directory, for tests only — D13 / T10.7. + +`server/ldap_auth.py` normally opens an LDAPS connection to a domain controller. +Tests cannot: CI has no domain, 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, which is what this module is. + +Set `WP_LDAP_FAKE_DIRECTORY` to a JSON object and `ldap_auth.verify()` answers +from it instead of touching the network: + + {"root": {"password": "…", "mail": "root@example.test", + "full_name": "Root", "groups": ["WP-Suite-Users"]}} + +`groups` is a flat list of names the account is "in". Nested groups do not exist +here — the real `member_of` resolves a DN and uses AD's LDAP_MATCHING_RULE_IN_CHAIN, +and faking that faithfully would mean reimplementing AD. A test that cares about +nesting has to run against a real directory; this one is honest about being a +string comparison. + +THE PRODUCTION GUARD IS THE POINT OF THIS FILE. + +An environment variable that makes any password work is exactly the kind of thing +that escapes into production, and D13 removed every other way in — there is no +local password to fall back on and no break-glass, so a fake directory silently +active in production would be a total authentication bypass with nothing behind it. + +So `is_active()` refuses whenever a real database is configured, using the same +test `auth._load_secret` uses to refuse an ephemeral signing key: a non-SQLite +`DATABASE_URL` means production, full stop. `ldap_auth.describe()` also shouts +when the fake is live, so the startup line can never be mistaken for a real one. +""" +import json +import logging +import os +from typing import Optional + +log = logging.getLogger("wpsuite.ldap.fake") + +ENV_VAR = "WP_LDAP_FAKE_DIRECTORY" + + +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 ldap_auth, 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 directory — this looks like production, and D13 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 like a bad password + log.error("%s is not valid JSON (%s); the fake directory is empty", ENV_VAR, exc) + return {} + + +def lookup(username: str, password: str, required_group: Optional[str]) -> tuple: + """Return (ok, reason, attrs). Mirrors what ldap_auth.verify() needs. + + Deliberately does NOT re-check for an empty password: `verify()` guards that + before it ever gets here, and duplicating the check in the fake would let the + real guard rot without any test noticing. + """ + people = directory() + who = people.get(username) or people.get(username.lower()) + if not isinstance(who, dict) or password != who.get("password"): + return (False, "bad_credentials", {}) + if required_group: + groups = who.get("groups") or [] + if required_group not in groups: + return (False, "not_in_group", {}) + return (True, "ok", { + "sam": who.get("sam") or username, + "mail": who.get("mail", ""), + "full_name": who.get("full_name", ""), + "upn": who.get("upn", ""), + }) diff --git a/tests/browser_check.py b/tests/browser_check.py index 6870898..63969cf 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 @@ -86,7 +87,7 @@ def seed(db_path): def mk(username, role): db.add(models.User(id="user_" + username, username=username, email=f"{username}@example.test", full_name=username.title(), - password_hash=auth.hash_password(PW), role=role)) + role=role)) mk("root", auth.ROLE_ADMIN) mk("sue", auth.ROLE_PROJECT_SUPER) # super user on Job A @@ -150,10 +151,30 @@ def seed(db_path): for u in db.query(models.User).all()} +# D13 / T10.7. The app authenticates by binding to a domain controller, 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/ldap_fake.py +# reads this instead, and refuses to work against a non-SQLite database. +# +# Most checks never sign in (seed() mints tokens with auth.create_token and sets +# the cookie directly), so this matters only where the login FORM is driven — +# url_state_check's deep-link-through-login case. It is set for every server here +# anyway so that a test which starts signing in later does not fail mysteriously. +FAKE_DIRECTORY = json.dumps({ + u: {"password": PW, "mail": f"{u}@example.test", "full_name": u.title(), + "groups": ["WP-Suite-Users"]} + for u in ("root", "sue", "pat", "mix", "bob", "sam", "legacy", "new") +}) + + def start_server(port, db_path): env = dict(os.environ) env["DATABASE_URL"] = "sqlite:///" + db_path.replace("\\", "/") env.setdefault("AUTH_SECRET_KEY", "browser-check-secret-not-for-production") + env["WP_LDAP_FAKE_DIRECTORY"] = FAKE_DIRECTORY + # No required group: the fake grants "WP-Suite-Users" to everyone, and a test + # asserting the group gate belongs in ldap_auth_check where it can be explicit. + 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"], diff --git a/tests/console_dialogs_check.py b/tests/console_dialogs_check.py index 346db79..95b73ef 100644 --- a/tests/console_dialogs_check.py +++ b/tests/console_dialogs_check.py @@ -91,31 +91,6 @@ def main(): chk("the console booted with a user table", page.eval("!!document.querySelector('table')")) - page.eval("void resetPw('user_pat','pat')") - time.sleep(0.4) - chk("the reset prompt is the kit's modal, open, focused at the input", - page.eval("(() => { const o=document.getElementById('wp-dlg-overlay');" - " return !!o && o.classList.contains('open')" - " && document.activeElement.id==='wp-dlg-input'; })()")) - page.eval("document.getElementById('wp-dlg-input').value='short';" - "document.getElementById('wp-dlg-ok').click()") - chk("a short password is refused AT the input - dialog stays, error says why", - page.eval("(() => { const o=document.getElementById('wp-dlg-overlay');" - " return o.classList.contains('open')" - " && /12 characters/.test(document.getElementById('wp-dlg-err').textContent); })()")) - page.eval("document.getElementById('wp-dlg-input').value='CorrectHorseBattery10';" - "document.getElementById('wp-dlg-ok').click()") - time.sleep(1.2) - chk("a good answer closes the dialog and the server accepts it", - page.eval("!document.getElementById('wp-dlg-overlay').classList.contains('open')")) - chk("...announced through the kit's toast (role=status)", - page.eval("(() => { const t=document.getElementById('toast');" - " return !!t && t.getAttribute('role')==='status'" - " && /Password reset for pat/.test(t.textContent); })()")) - st, _ = api(base, "/api/auth/login", "x", "POST", - {"username": "pat", "password": "CorrectHorseBattery10"}) - chk("...and the new password actually works", st == 200, st) - print("\n3. destroy needs a real yes") page.eval("void deleteUser('user_bob','bob')") time.sleep(0.4) diff --git a/tests/launcher_check.py b/tests/launcher_check.py index 9643091..462e4ff 100644 --- a/tests/launcher_check.py +++ b/tests/launcher_check.py @@ -55,8 +55,7 @@ def seed_empty(db_path): Base.metadata.create_all(bind=engine) with SessionLocal() as db: db.add(models.User(id="user_new", username="new", email="new@example.test", - full_name="New Starter", password_hash=auth.hash_password(PW), - role=auth.ROLE_ADMIN)) + full_name="New Starter", role=auth.ROLE_ADMIN)) db.commit() return {u.username: auth.create_token(u) for u in db.query(models.User).all()} diff --git a/tests/ldap_auth_check.py b/tests/ldap_auth_check.py new file mode 100644 index 0000000..5c66bc3 --- /dev/null +++ b/tests/ldap_auth_check.py @@ -0,0 +1,267 @@ +#!/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) diff --git a/tests/url_state_check.py b/tests/url_state_check.py index 436b294..7d261e2 100644 --- a/tests/url_state_check.py +++ b/tests/url_state_check.py @@ -187,9 +187,13 @@ def main(): break time.sleep(0.3) settle(page, 1.2) + # NOT `"wp-creation-index.html" in location.href` — that string is in the + # ?next= parameter too, so the check passed while still sitting on + # login.html with the sign-in rejected. Assert we actually LEFT the + # login page (D13/T10.7: it caught nothing when the bind started failing). + href = page.eval("location.href") chk("signing in continues to the requested page, not the home page", - "wp-creation-index.html" in page.eval("location.href"), - page.eval("location.href")) + "login.html" not in href and "wp-creation-index.html" in href, href) for _ in range(30): if page.eval("!!window.wpCreatorReady"): break