T10.7 D13 - the suite runs without a domain controller
Far smaller than estimated, because the premise was wrong. I had said four checks sign in and would each need a fake directory. They do not: seed() mints a session token with auth.create_token() and sets the cookie directly - browser_check's own docstring says so - and the only breakage was a leftover password_hash= kwarg on a model that no longer has the column. Deleting that one line in browser_check.seed() unblocked 39 files that import seed/start_server from it. launcher_check needed the same. console_dialogs_check's password-reset half is deleted rather than ported. Its docstring now records what went and where the prompt kit is still covered (wpPromptDialog has five callers left in wp-creation-app.js; creator_dialogs_check exercises them, validation included - verified, 20/20). Nothing was left skipped in place of the removed section. Exactly one check genuinely needed a seam: url_state_check drives the real login form to prove a deep link's ?next= survives authentication. That cannot be faked by minting a cookie, because the login round trip is the thing under test. The seam is env-driven because it has to be: start_server launches the app as a SUBPROCESS, so a monkeypatch in the test process would never reach the code doing the authenticating. server/ldap_fake.py reads WP_LDAP_FAKE_DIRECTORY and ldap_auth dispatches to it AFTER the empty-input guard, so the anonymous-bind guard covers the fake path too - a fake that reimplemented it would let the real one rot unnoticed. The production guard is the point of that module. An env var that makes any password work is exactly the kind of thing that escapes into production, and D13 left no other way in. is_active() refuses whenever a non-SQLite DATABASE_URL is configured - the same test auth._load_secret uses - and describe() shouts in capitals so a fake run can never be mistaken for a real one in the startup log. Two things found on the way, neither of them the app's fault: - url_state_check's "signing in continues to the requested page" asserted `"wp-creation-index.html" in location.href`. That string is in the ?next= parameter too, so it passed while sitting on login.html with the sign-in rejected. It would have passed with login entirely broken. Tightened to assert we actually left the login page. - Two assertions in my own new ldap_auth_check read the WRONG database: server/db.py binds its engine from DATABASE_URL at import, so setting the env var afterwards keeps reading whichever file was configured first. users_in() now opens the file it is asked about with sqlite3. The CERT_NONE check also had to become an AST walk - the module docstring names validate=ssl.CERT_NONE in order to explain why it is banned, and a text search cannot tell that apart from a real call. tests/ldap_auth_check.py is new coverage rather than repair: the anonymous-bind guard, CERT_REQUIRED by AST, the nested matching rule in the filter, the production refusal, a refused sign-in creating no account, and an existing admin still being an admin with their locally-set name intact. 20/20. Run so far, all green: browser_check 71/71, launcher_check 58/58, console_dialogs 12/12, url_state 23/23, qa_gate 41/41, critical_reopen 11/11, creator_dialogs 20/20, a11y 22/22, kitting_notify 17/17, ldap_auth 20/20. A full sweep of the remaining ~30 is running; its box stays unticked until it reports. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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"],
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()}
|
||||
|
||||
|
||||
267
tests/ldap_auth_check.py
Normal file
267
tests/ldap_auth_check.py
Normal file
@@ -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)
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user