thejayman77
d76642f85d
Two session-management bugs from the audit: - /api/password recorded failed attempts against the login budget but never checked it, so a borrowed session could guess the current password without limit while still locking the owner out of /api/login. Verified: thirteen consecutive wrong guesses all returned 403 and none returned 429. It now spends from the same budget it was topping up. - bump_epoch read the epoch and wrote it back without the write lock, so concurrent revocations lost increments and sessions that should have been cut off survived. Verified: twenty concurrent bumps advanced the counter from 2 to 6, and twenty concurrent "sign out other devices" calls left four sessions authenticated. It takes BEGIN IMMEDIATE now; set_password hashes before locking, so scrypt doesn't serialise unrelated writes. Removing either fix makes its test fail with exactly that symptom. README corrections: deployment is rsync, not git pull — gitea on .8 cannot serve a clone to .8 itself, which the deploy section now documents — and the concurrency check count was understated. Checks go from 182 to 197. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
512 lines
29 KiB
Python
512 lines
29 KiB
Python
"""End-to-end exercise of the API against a throwaway database."""
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
TMP = tempfile.mkdtemp()
|
|
os.environ["PARTS_DB"] = os.path.join(TMP, "test.db")
|
|
os.environ["PARTS_PASSWORD"] = "hunter2"
|
|
os.environ["PARTS_SECRET"] = "test-secret"
|
|
os.environ["PARTS_AUTH"] = "on"
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from app.main import app # noqa: E402
|
|
|
|
failures = []
|
|
|
|
|
|
def check(label, condition, detail=""):
|
|
print((" PASS " if condition else " FAIL ") + label + (f" [{detail}]" if detail and not condition else ""))
|
|
if not condition:
|
|
failures.append(label)
|
|
|
|
|
|
with TestClient(app) as client:
|
|
# --- session tokens survive every signature byte value ---
|
|
import app.auth as _auth
|
|
import base64 as _b64, hashlib as _hl, hmac as _hm
|
|
|
|
def _issue_at(exp, epoch=0):
|
|
payload = f"{exp}:{epoch}".encode()
|
|
return _b64.urlsafe_b64encode(
|
|
payload + _hm.new(_auth._secret(), payload, _hl.sha256).digest()).decode()
|
|
|
|
_base = int(__import__("time").time()) + 90 * 86400
|
|
_bad = [e for e in range(_base, _base + 3000) if not _auth.token_valid(_issue_at(e))]
|
|
check("no issued token fails its own validator", not _bad, f"{len(_bad)}/3000 failed")
|
|
check("tampered token rejected", not _auth.token_valid(_issue_at(_base)[:-2] + "AA"))
|
|
check("expired token rejected", not _auth.token_valid(_issue_at(int(__import__("time").time()) - 5)))
|
|
check("garbage token rejected", not _auth.token_valid("not-a-token"))
|
|
|
|
# --- auth gate ---
|
|
check("unauthenticated list is 401", client.get("/api/parts").status_code == 401)
|
|
check("wrong password rejected", client.post("/api/login", json={"password": "nope"}).status_code == 401)
|
|
check("healthz open", client.get("/healthz").json() == {"ok": True})
|
|
|
|
r = client.post("/api/login", json={"password": "hunter2"})
|
|
check("login succeeds", r.status_code == 200, r.text)
|
|
check("session cookie set", "parts_session" in client.cookies)
|
|
check("me reports authenticated", client.get("/api/me").json()["authenticated"] is True)
|
|
|
|
# --- seeded taxonomy ---
|
|
cats = client.get("/api/categories").json()["items"]
|
|
check("categories seeded", len(cats) > 30, str(len(cats)))
|
|
filament = next(c for c in cats if c["name"] == "Filament")
|
|
check("filament nests under 3D Printing", filament["path"] == "3D Printing / Filament", filament["path"])
|
|
check("filament default unit is grams", filament["unit"] == "g")
|
|
check("filament has spec template", any(s["key"] == "Diameter" for s in filament["spec_template"]))
|
|
resistors = next(c for c in cats if c["name"] == "Resistors")
|
|
electronics = next(c for c in cats if c["name"] == "Electronics")
|
|
|
|
locs = client.get("/api/locations").json()["items"]
|
|
workshop = next(l for l in locs if l["name"] == "Workshop")
|
|
|
|
r = client.post("/api/locations", json={"name": "Bin A3", "parent_id": workshop["id"]})
|
|
check("nested location created", r.status_code == 201, r.text)
|
|
bin_a3 = r.json()["id"]
|
|
paths = {l["id"]: l["path"] for l in client.get("/api/locations").json()["items"]}
|
|
check("location path nests", paths[bin_a3] == "Workshop / Bin A3", paths[bin_a3])
|
|
|
|
check("duplicate location rejected",
|
|
client.post("/api/locations", json={"name": "Bin A3", "parent_id": workshop["id"]}).status_code == 409)
|
|
|
|
# --- create parts ---
|
|
r = client.post("/api/parts", json={
|
|
"name": "10k resistor 0603",
|
|
"category_id": resistors["id"],
|
|
"location_id": bin_a3,
|
|
"manufacturer": "Yageo",
|
|
"mpn": "RC0603FR-0710KL",
|
|
"quantity": 480, "unit": "pcs", "min_quantity": 50, "cost_each": 0.01,
|
|
"specs": [{"key": "Resistance", "value": "10k"}, {"key": "Package", "value": "0603"},
|
|
{"key": "Tolerance", "value": "1%"}],
|
|
"tags": ["smd", "passives"],
|
|
})
|
|
check("part created", r.status_code == 201, r.text)
|
|
resistor = r.json()
|
|
check("specs round-trip", len(resistor["specs"]) == 3)
|
|
check("tags round-trip", resistor["tags"] == ["passives", "smd"], str(resistor["tags"]))
|
|
check("category path resolved", resistor["category_path"] == "Electronics / Resistors", str(resistor["category_path"]))
|
|
check("not low stock at 480/50", resistor["low_stock"] is False)
|
|
|
|
r = client.post("/api/parts", json={
|
|
"name": "PLA Black",
|
|
"category_id": filament["id"],
|
|
"location_id": workshop["id"],
|
|
"manufacturer": "Prusament",
|
|
"quantity": 640, "unit": "g", "min_quantity": 200,
|
|
"specs": [{"key": "Material", "value": "PLA"}, {"key": "Colour", "value": "Black"},
|
|
{"key": "Diameter", "value": "1.75mm"}],
|
|
"tags": ["3d-printing"],
|
|
})
|
|
check("filament created with gram unit", r.status_code == 201 and r.json()["unit"] == "g", r.text)
|
|
pla = r.json()
|
|
|
|
r = client.post("/api/parts", json={
|
|
"name": "M3x8 socket cap screw",
|
|
"location_id": bin_a3,
|
|
"quantity": 12, "unit": "pcs", "min_quantity": 20,
|
|
"specs": [{"key": "Thread", "value": "M3"}, {"key": "Length", "value": "8mm"}],
|
|
})
|
|
screw = r.json()
|
|
check("low stock flagged at 12/20", screw["low_stock"] is True)
|
|
|
|
# --- search ---
|
|
def ids(q, **kw):
|
|
params = {"q": q, **kw}
|
|
return [p["name"] for p in client.get("/api/parts", params=params).json()["items"]]
|
|
|
|
check("search by name", "10k resistor 0603" in ids("10k"))
|
|
check("search by mpn", "10k resistor 0603" in ids("RC0603"))
|
|
check("search by manufacturer", "PLA Black" in ids("prusament"))
|
|
check("search by spec value", "PLA Black" in ids("1.75mm"))
|
|
check("search by tag", "10k resistor 0603" in ids("smd"))
|
|
check("prefix search while typing", "PLA Black" in ids("prus"))
|
|
check("search by location name", "10k resistor 0603" in ids("Bin A3"))
|
|
check("nonsense search returns nothing", ids("zzzznope") == [])
|
|
check("punctuation does not crash FTS", isinstance(ids('0.1uF "quoted" AND OR *'), list))
|
|
|
|
# --- filters ---
|
|
parent_filtered = client.get("/api/parts", params={"category_id": electronics["id"]}).json()
|
|
check("parent category catches children", parent_filtered["total"] == 1, str(parent_filtered["total"]))
|
|
check("location filter includes descendants",
|
|
client.get("/api/parts", params={"location_id": workshop["id"]}).json()["total"] == 3)
|
|
check("low stock filter", [p["name"] for p in
|
|
client.get("/api/parts", params={"low_stock": "true"}).json()["items"]] == ["M3x8 socket cap screw"])
|
|
check("tag filter", client.get("/api/parts", params={"tag": "passives"}).json()["total"] == 1)
|
|
|
|
# --- rolled-up counts match what the filter returns ---
|
|
cats_now = {c["name"]: c for c in client.get("/api/categories").json()["items"]}
|
|
check("parent count rolls up children", cats_now["Electronics"]["part_count"] == 1,
|
|
str(cats_now["Electronics"]["part_count"]))
|
|
check("parent has no direct parts", cats_now["Electronics"]["direct_count"] == 0)
|
|
check("leaf count is its own", cats_now["Resistors"]["part_count"] == 1)
|
|
locs_now = {l["name"]: l for l in client.get("/api/locations").json()["items"]}
|
|
check("location count rolls up", locs_now["Workshop"]["part_count"] == 3,
|
|
str(locs_now["Workshop"]["part_count"]))
|
|
|
|
# --- sorting & paging ---
|
|
names = [p["name"] for p in client.get("/api/parts", params={"sort": "name"}).json()["items"]]
|
|
check("sort by name", names == sorted(names, key=str.lower), str(names))
|
|
check("sort by quantity ascending",
|
|
[p["quantity"] for p in client.get("/api/parts", params={"sort": "quantity"}).json()["items"]] == [12, 480, 640])
|
|
page = client.get("/api/parts", params={"sort": "name", "limit": 2, "offset": 0}).json()
|
|
check("paging returns total plus page", page["total"] == 3 and len(page["items"]) == 2)
|
|
|
|
# --- stock adjustments ---
|
|
r = client.post(f"/api/parts/{pla['id']}/adjust", json={"delta": -140, "reason": "bracket print"})
|
|
check("adjust decrements", r.json()["quantity"] == 500, r.text)
|
|
r = client.post(f"/api/parts/{pla['id']}/adjust", json={"delta": -400, "reason": "big print"})
|
|
check("adjust flags low stock", r.json()["low_stock"] is True and r.json()["quantity"] == 100)
|
|
r = client.post(f"/api/parts/{screw['id']}/adjust", json={"delta": -999})
|
|
check("quantity floors at zero", r.json()["quantity"] == 0)
|
|
history = client.get(f"/api/parts/{pla['id']}/history").json()["items"]
|
|
check("history records initial stock plus both adjustments", len(history) == 3, str(len(history)))
|
|
check("history is newest first", history[0]["reason"] == "big print")
|
|
|
|
# --- updates ---
|
|
r = client.patch(f"/api/parts/{resistor['id']}", json={"name": "10k resistor 0603 1%",
|
|
"specs": [{"key": "Resistance", "value": "10 kilohm"}]})
|
|
check("patch renames", r.json()["name"] == "10k resistor 0603 1%")
|
|
check("patch replaces specs", len(r.json()["specs"]) == 1)
|
|
check("patch leaves quantity alone", r.json()["quantity"] == 480)
|
|
check("reindex picks up new spec text", "10k resistor 0603 1%" in ids("kilohm"))
|
|
check("old spec value no longer matches", ids("0603") != [] and "kilohm" not in str(ids("10k")))
|
|
|
|
r = client.patch(f"/api/parts/{resistor['id']}", json={"quantity": 5})
|
|
check("patch quantity alone works", r.json()["quantity"] == 5)
|
|
|
|
# --- taxonomy edits keep parts ---
|
|
r = client.post("/api/categories", json={"name": "Salvage", "unit": "pcs",
|
|
"spec_template": [{"key": "Source", "value": ""}]})
|
|
check("custom category created", r.status_code == 201, r.text)
|
|
salvage_id = r.json()["id"]
|
|
check("custom template stored",
|
|
any(c["id"] == salvage_id and c["spec_template"] == [{"key": "Source"}]
|
|
for c in client.get("/api/categories").json()["items"]))
|
|
client.delete(f"/api/categories/{salvage_id}")
|
|
|
|
client.delete(f"/api/locations/{bin_a3}")
|
|
survivor = client.get(f"/api/parts/{resistor['id']}").json()
|
|
check("deleting a location keeps the part", survivor["location_id"] is None)
|
|
check("part still searchable after location delete",
|
|
"10k resistor 0603 1%" in ids("10k") or "10k resistor 0603 1%" in ids("resistor"))
|
|
|
|
# --- deletion ---
|
|
check("delete part", client.delete(f"/api/parts/{screw['id']}").status_code == 200)
|
|
check("deleted part is gone", client.get(f"/api/parts/{screw['id']}").status_code == 404)
|
|
check("deleted part leaves the index", "M3x8 socket cap screw" not in ids("M3x8"))
|
|
check("missing part 404s", client.get("/api/parts/999999").status_code == 404)
|
|
check("history of a missing part 404s",
|
|
client.get("/api/parts/999999/history").status_code == 404)
|
|
# httpx refuses to serialise inf/nan, so these go as raw bodies — which is
|
|
# exactly how a real client would smuggle them in: Python's json.loads
|
|
# accepts the bare `Infinity` and `NaN` tokens.
|
|
JSONH = {"Content-Type": "application/json"}
|
|
|
|
def raw_post(path, body):
|
|
return client.post(path, content=body, headers=JSONH)
|
|
|
|
def raw_patch(path, body):
|
|
return client.patch(path, content=body, headers=JSONH)
|
|
|
|
check("Infinity quantity rejected",
|
|
raw_post("/api/parts", '{"name":"inf","quantity":Infinity}').status_code == 422)
|
|
check("-Infinity quantity rejected",
|
|
raw_post("/api/parts", '{"name":"ninf","quantity":-Infinity}').status_code == 422)
|
|
check("NaN quantity rejected",
|
|
raw_post("/api/parts", '{"name":"nan","quantity":NaN}').status_code == 422)
|
|
check("Infinity cost rejected",
|
|
raw_post("/api/parts", '{"name":"inf2","cost_each":Infinity}').status_code == 422)
|
|
check("Infinity min_quantity rejected",
|
|
raw_post("/api/parts", '{"name":"inf3","min_quantity":Infinity}').status_code == 422)
|
|
_infp = client.post("/api/parts", json={"name": "inf target", "quantity": 5}).json()["id"]
|
|
check("Infinity adjustment delta rejected",
|
|
raw_post(f"/api/parts/{_infp}/adjust", '{"delta":Infinity}').status_code == 422)
|
|
check("NaN adjustment delta rejected",
|
|
raw_post(f"/api/parts/{_infp}/adjust", '{"delta":NaN}').status_code == 422)
|
|
check("Infinity PATCH quantity rejected",
|
|
raw_patch(f"/api/parts/{_infp}", '{"quantity":Infinity}').status_code == 422)
|
|
check("the part is untouched after a rejected Infinity",
|
|
client.get(f"/api/parts/{_infp}").json()["quantity"] == 5)
|
|
check("adjusting a missing part 404s",
|
|
client.post("/api/parts/999999/adjust", json={"delta": 1}).status_code == 404)
|
|
|
|
# --- stats & validation ---
|
|
# Derived from the listing rather than a magic number, so adding a fixture
|
|
# part earlier in the file can't silently invalidate it.
|
|
s = client.get("/api/stats").json()
|
|
listed = client.get("/api/parts", params={"limit": 1000}).json()
|
|
check("stats part count matches the listing", s["parts"] == listed["total"], str(s))
|
|
check("stats low-stock count matches the listing",
|
|
s["low_stock"] == sum(1 for p in listed["items"] if p["low_stock"]), str(s))
|
|
check("stats value matches the listing",
|
|
s["estimated_value"] == round(sum(p["quantity"] * (p["cost_each"] or 0)
|
|
for p in listed["items"]), 2), str(s))
|
|
check("whitespace-only name rejected",
|
|
client.post("/api/parts", json={"name": " "}).status_code == 422)
|
|
check("negative quantity rejected",
|
|
client.post("/api/parts", json={"name": "neg", "quantity": -5}).status_code == 422)
|
|
check("negative min_quantity rejected",
|
|
client.post("/api/parts", json={"name": "neg2", "min_quantity": -1}).status_code == 422)
|
|
check("blank unit rejected",
|
|
client.post("/api/parts", json={"name": "u", "unit": " "}).status_code == 422)
|
|
check("unknown category on create 404s",
|
|
client.post("/api/parts", json={"name": "ghost", "category_id": 999999}).status_code == 404)
|
|
|
|
# --- regression: quantity changes are always logged ---
|
|
r = client.post("/api/parts", json={"name": "log target", "quantity": 10})
|
|
logged = r.json()["id"]
|
|
before = len(client.get(f"/api/parts/{logged}/history").json()["items"])
|
|
client.patch(f"/api/parts/{logged}", json={"quantity": 3, "reason": "stocktake"})
|
|
hist = client.get(f"/api/parts/{logged}/history").json()["items"]
|
|
check("PATCH quantity writes history", len(hist) == before + 1, f"{before} -> {len(hist)}")
|
|
check("PATCH history records the signed delta", hist[0]["delta"] == -7, str(hist[0]["delta"]))
|
|
check("PATCH history keeps the reason", hist[0]["reason"] == "stocktake")
|
|
n_before = len(client.get(f"/api/parts/{logged}/history").json()["items"])
|
|
client.patch(f"/api/parts/{logged}", json={"name": "log target renamed"})
|
|
check("PATCH without quantity writes no history",
|
|
len(client.get(f"/api/parts/{logged}/history").json()["items"]) == n_before)
|
|
client.patch(f"/api/parts/{logged}", json={"quantity": 3})
|
|
check("PATCH to the same quantity writes no history",
|
|
len(client.get(f"/api/parts/{logged}/history").json()["items"]) == n_before)
|
|
|
|
# --- regression: a floored adjustment logs what was applied, not requested ---
|
|
r = client.post("/api/parts", json={"name": "floor target", "quantity": 7})
|
|
floored = r.json()["id"]
|
|
client.post(f"/api/parts/{floored}/adjust", json={"delta": -999, "reason": "floor"})
|
|
fh = client.get(f"/api/parts/{floored}/history").json()["items"]
|
|
check("floored adjustment logs the applied delta", fh[0]["delta"] == -7, str(fh[0]["delta"]))
|
|
check("floored adjustment lands at zero", fh[0]["quantity_after"] == 0)
|
|
check("stock log sums to current quantity",
|
|
sum(h["delta"] for h in fh) == client.get(f"/api/parts/{floored}").json()["quantity"])
|
|
|
|
# --- regression: renaming taxonomy rebuilds the search index ---
|
|
shelf = client.post("/api/locations", json={"name": "Old Shelf"}).json()["id"]
|
|
client.post("/api/parts", json={"name": "shelf widget", "location_id": shelf, "quantity": 1})
|
|
check("part found by its location name", client.get("/api/parts", params={"q": "Old Shelf"}).json()["total"] == 1)
|
|
client.patch(f"/api/locations/{shelf}", json={"name": "New Shelf"})
|
|
check("rename drops the stale location term",
|
|
client.get("/api/parts", params={"q": "Old Shelf"}).json()["total"] == 0)
|
|
check("rename indexes the new location term",
|
|
client.get("/api/parts", params={"q": "New Shelf"}).json()["total"] == 1)
|
|
|
|
# Nested: renaming a PARENT has to reindex everything beneath it, because
|
|
# the full path is what gets indexed.
|
|
outer = client.post("/api/locations", json={"name": "Old Room"}).json()["id"]
|
|
inner = client.post("/api/locations", json={"name": "Inner Bin", "parent_id": outer}).json()["id"]
|
|
client.post("/api/parts", json={"name": "nested widget", "location_id": inner, "quantity": 1})
|
|
check("part found by its parent location", client.get("/api/parts", params={"q": "Old Room"}).json()["total"] == 1)
|
|
client.patch(f"/api/locations/{outer}", json={"name": "New Room"})
|
|
check("parent rename reindexes descendants' parts",
|
|
client.get("/api/parts", params={"q": "Old Room"}).json()["total"] == 0
|
|
and client.get("/api/parts", params={"q": "New Room"}).json()["total"] == 1)
|
|
client.delete(f"/api/locations/{outer}")
|
|
check("deleting a location clears its term from the index",
|
|
client.get("/api/parts", params={"q": "New Room"}).json()["total"] == 0)
|
|
check("the part itself survives the delete",
|
|
client.get("/api/parts", params={"q": "nested widget"}).json()["total"] == 1)
|
|
|
|
# --- regression: PATCH null handling ---
|
|
check("explicit null name is a 422, not a 500",
|
|
client.patch(f"/api/parts/{logged}", json={"name": None}).status_code == 422)
|
|
check("explicit null unit is a 422",
|
|
client.patch(f"/api/parts/{logged}", json={"unit": None}).status_code == 422)
|
|
check("explicit null category_id is allowed",
|
|
client.patch(f"/api/parts/{logged}", json={"category_id": None}).status_code == 200)
|
|
check("explicit null cost_each is allowed",
|
|
client.patch(f"/api/parts/{logged}", json={"cost_each": None}).status_code == 200)
|
|
|
|
# --- regression: taxonomy cycles and missing ids ---
|
|
a = client.post("/api/categories", json={"name": "CycleA"}).json()["id"]
|
|
b = client.post("/api/categories", json={"name": "CycleB", "parent_id": a}).json()["id"]
|
|
c = client.post("/api/categories", json={"name": "CycleC", "parent_id": b}).json()["id"]
|
|
check("direct self-parent rejected",
|
|
client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": a}).status_code == 400)
|
|
check("indirect cycle rejected",
|
|
client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": b}).status_code == 400)
|
|
check("deep indirect cycle rejected",
|
|
client.patch(f"/api/categories/{a}", json={"name": "CycleA", "parent_id": c}).status_code == 400)
|
|
check("legitimate re-parent still allowed",
|
|
client.patch(f"/api/categories/{c}", json={"name": "CycleC", "parent_id": a}).status_code == 200)
|
|
check("PATCH missing category 404s",
|
|
client.patch("/api/categories/999999", json={"name": "ghost"}).status_code == 404)
|
|
check("DELETE missing category 404s", client.delete("/api/categories/999999").status_code == 404)
|
|
check("PATCH missing location 404s",
|
|
client.patch("/api/locations/999999", json={"name": "ghost"}).status_code == 404)
|
|
check("DELETE missing location 404s", client.delete("/api/locations/999999").status_code == 404)
|
|
check("unknown parent 404s",
|
|
client.post("/api/categories", json={"name": "orphan", "parent_id": 999999}).status_code == 404)
|
|
check("blank category name rejected",
|
|
client.post("/api/categories", json={"name": " "}).status_code == 422)
|
|
dup = client.post("/api/categories", json={"name": "DupTarget"}).json()["id"]
|
|
client.post("/api/categories", json={"name": "DupOther"})
|
|
check("PATCH into a duplicate name is a 409, not a 500",
|
|
client.patch(f"/api/categories/{dup}", json={"name": "DupOther"}).status_code == 409)
|
|
|
|
# --- security headers and asset caching ---
|
|
r = client.get("/")
|
|
check("nosniff header", r.headers.get("x-content-type-options") == "nosniff")
|
|
check("frame-options header", r.headers.get("x-frame-options") == "DENY")
|
|
check("referrer-policy header", r.headers.get("referrer-policy") == "no-referrer")
|
|
check("content-security-policy header", "default-src 'self'" in r.headers.get("content-security-policy", ""))
|
|
check("index is not cacheable", "no-store" in r.headers.get("cache-control", ""))
|
|
check("api responses are not cacheable",
|
|
"no-store" in client.get("/api/stats").headers.get("cache-control", ""))
|
|
check("index references content-hashed assets",
|
|
"/static/app.js?v=" in r.text and "/static/app.css?v=" in r.text)
|
|
import re as _re
|
|
digest = _re.search(r"/static/app\.js\?v=([0-9a-f]+)", r.text).group(1)
|
|
check("versioned asset is immutably cacheable",
|
|
"immutable" in client.get(f"/static/app.js?v={digest}").headers.get("cache-control", ""))
|
|
check("unversioned asset is not cached",
|
|
client.get("/static/app.js").headers.get("cache-control") == "no-cache")
|
|
check("hsts only when the original request was https",
|
|
"strict-transport-security" not in r.headers
|
|
and "strict-transport-security" in client.get("/", headers={"X-Forwarded-Proto": "https"}).headers)
|
|
|
|
# --- frontend & logout ---
|
|
check("index served", client.get("/").status_code == 200)
|
|
check("css served", client.get("/static/app.css").status_code == 200)
|
|
check("missing static 404s", client.get("/static/nope.css").status_code == 404)
|
|
check("unknown api path 404s json", client.get("/api/nope").status_code == 404)
|
|
|
|
client.post("/api/logout")
|
|
check("logout clears session", client.get("/api/parts").status_code == 401)
|
|
|
|
# --- failed-login throttling (last: it trips global state deliberately) ---
|
|
auth_mod = __import__("app.auth", fromlist=["auth"])
|
|
auth_mod.clear_failures()
|
|
codes = [client.post("/api/login", json={"password": "wrong"}).status_code for _ in range(12)]
|
|
check("repeated wrong passwords start returning 429", 429 in codes, str(codes))
|
|
check("throttle kicks in only after the configured budget",
|
|
codes.index(429) == auth_mod._max_failures(), str(codes))
|
|
blocked = client.post("/api/login", json={"password": "hunter2"})
|
|
check("the correct password is refused while throttled", blocked.status_code == 429)
|
|
check("429 tells the client when to retry", blocked.headers.get("retry-after", "").isdigit())
|
|
auth_mod.clear_failures()
|
|
check("a successful login works again once the window clears",
|
|
client.post("/api/login", json={"password": "hunter2"}).status_code == 200)
|
|
check("a successful login resets the failure budget", auth_mod.login_retry_after() == 0)
|
|
|
|
# --- password management through the API ---
|
|
check("bootstrap password is flagged", client.get("/api/me").json()["using_bootstrap_password"] is True)
|
|
check("min length is advertised", client.get("/api/me").json()["min_password_length"] == 8)
|
|
|
|
check("wrong current password is refused",
|
|
client.post("/api/password",
|
|
json={"current_password": "nope", "new_password": "brand-new-secret"}).status_code == 403)
|
|
|
|
# The change-password route verifies the same credential as the login form,
|
|
# so it has to spend from the same budget instead of only topping it up.
|
|
auth_mod.clear_failures()
|
|
pw_codes = [client.post("/api/password",
|
|
json={"current_password": "wrong", "new_password": "brand-new-secret"}).status_code
|
|
for _ in range(auth_mod._max_failures() + 3)]
|
|
check("guessing through change-password eventually 429s", 429 in pw_codes, str(pw_codes))
|
|
check("change-password throttles on the same budget as login",
|
|
pw_codes.index(429) == auth_mod._max_failures(), str(pw_codes))
|
|
check("a throttled change-password blocks even the right password",
|
|
client.post("/api/password",
|
|
json={"current_password": "hunter2",
|
|
"new_password": "brand-new-secret"}).status_code == 429)
|
|
check("login is throttled too once the budget is spent",
|
|
client.post("/api/login", json={"password": "hunter2"}).status_code == 429)
|
|
auth_mod.clear_failures()
|
|
check("clearing the budget restores change-password",
|
|
client.post("/api/password",
|
|
json={"current_password": "wrong", "new_password": "x"}).status_code in (403, 422))
|
|
auth_mod.clear_failures()
|
|
check("short new password rejected",
|
|
client.post("/api/password",
|
|
json={"current_password": "hunter2", "new_password": "short"}).status_code == 422)
|
|
check("reusing the current password rejected",
|
|
client.post("/api/password",
|
|
json={"current_password": "hunter2", "new_password": "hunter2"}).status_code == 422)
|
|
check("still logged in after failed attempts", client.get("/api/parts").status_code == 200)
|
|
|
|
# A second, independent session that should be cut off by the change.
|
|
other = TestClient(app)
|
|
other.post("/api/login", json={"password": "hunter2"})
|
|
check("second session works before the change", other.get("/api/parts").status_code == 200)
|
|
|
|
r = client.post("/api/password",
|
|
json={"current_password": "hunter2", "new_password": "correct-horse-battery"})
|
|
check("password change succeeds", r.status_code == 200, r.text)
|
|
check("change reports other sessions signed out", r.json()["other_sessions_signed_out"] is True)
|
|
check("the changing session stays signed in", client.get("/api/parts").status_code == 200)
|
|
check("other sessions are signed out", other.get("/api/parts").status_code == 401)
|
|
check("bootstrap flag clears", client.get("/api/me").json()["using_bootstrap_password"] is False)
|
|
|
|
fresh = TestClient(app)
|
|
check("the old password no longer works",
|
|
fresh.post("/api/login", json={"password": "hunter2"}).status_code == 401)
|
|
auth_mod.clear_failures()
|
|
check("the bootstrap env password is ignored once one is set",
|
|
fresh.post("/api/login", json={"password": os.environ["PARTS_PASSWORD"]}).status_code == 401)
|
|
auth_mod.clear_failures()
|
|
check("the new password works",
|
|
fresh.post("/api/login", json={"password": "correct-horse-battery"}).status_code == 200)
|
|
check("the new session can read data", fresh.get("/api/parts").status_code == 200)
|
|
|
|
# --- sign out other devices, without signing out this one ---
|
|
third = TestClient(app)
|
|
third.post("/api/login", json={"password": "correct-horse-battery"})
|
|
check("third session works", third.get("/api/parts").status_code == 200)
|
|
check("revoke succeeds", client.post("/api/sessions/revoke").status_code == 200)
|
|
check("revoke keeps the calling session", client.get("/api/parts").status_code == 200)
|
|
check("revoke cuts off the other session", third.get("/api/parts").status_code == 401)
|
|
|
|
# --- epoch bumps must not lose increments ---
|
|
from concurrent.futures import ThreadPoolExecutor as _Pool
|
|
import app.db as _dbmod
|
|
|
|
def _bump_once(_):
|
|
with _dbmod.session() as _c:
|
|
auth_mod.bump_epoch(_c)
|
|
|
|
with _dbmod.session() as _c:
|
|
_epoch_before = auth_mod.current_epoch(_c)
|
|
with _Pool(max_workers=20) as _ex:
|
|
list(_ex.map(_bump_once, range(20)))
|
|
with _dbmod.session() as _c:
|
|
_epoch_after = auth_mod.current_epoch(_c)
|
|
check("20 concurrent epoch bumps advance it by exactly 20",
|
|
_epoch_after - _epoch_before == 20, f"{_epoch_before} -> {_epoch_after}")
|
|
|
|
# --- the hash itself ---
|
|
import app.db as _db
|
|
with _db.session() as _conn:
|
|
stored = auth_mod.stored_hash(_conn)
|
|
check("password is stored hashed, not in the clear",
|
|
stored is not None and stored.startswith("scrypt$") and "correct-horse-battery" not in stored)
|
|
check("the stored hash verifies", auth_mod.verify_hash("correct-horse-battery", stored))
|
|
check("the stored hash rejects a wrong password", not auth_mod.verify_hash("wrong", stored))
|
|
check("two hashes of the same password differ (salted)",
|
|
auth_mod.hash_password("same") != auth_mod.hash_password("same"))
|
|
|
|
# --- recovery CLI ---
|
|
import app.admin as _admin
|
|
check("admin set-password works", _admin.main(["set-password", "cli-set-password"]) == 0)
|
|
cli = TestClient(app)
|
|
check("CLI-set password logs in",
|
|
cli.post("/api/login", json={"password": "cli-set-password"}).status_code == 200)
|
|
auth_mod.clear_failures()
|
|
check("CLI change signed out the API session", client.get("/api/parts").status_code == 401)
|
|
check("admin rejects a short password", _admin.main(["set-password", "abc"]) == 1)
|
|
check("admin show-status works", _admin.main(["show-status"]) == 0)
|
|
check("admin rejects an unknown command", _admin.main(["nonsense"]) == 2)
|
|
check("admin clear-password works", _admin.main(["clear-password"]) == 0)
|
|
back = TestClient(app)
|
|
check("clearing restores the env password",
|
|
back.post("/api/login", json={"password": os.environ["PARTS_PASSWORD"]}).status_code == 200)
|
|
check("bootstrap flag returns after clearing",
|
|
back.get("/api/me").json()["using_bootstrap_password"] is True)
|
|
|
|
print()
|
|
if failures:
|
|
print(f"{len(failures)} FAILED: " + "; ".join(failures))
|
|
raise SystemExit(1)
|
|
print("all checks passed")
|