diff --git a/README.md b/README.md index 54dbe26..58a49f8 100644 --- a/README.md +++ b/README.md @@ -45,13 +45,20 @@ Then open http://127.0.0.1:8123. ## Tests ```sh -.venv/bin/python -m tests.test_api +.venv/bin/python -m tests.test_api # 122 checks, in-process +.venv/bin/python -m tests.test_concurrency # 12 checks, against a real uvicorn ``` -Exercises the API end to end against a throwaway database — the auth gate, -nested categories and locations, search across every indexed field, filter -rollups, stock adjustment and history, patch semantics, and cascade behaviour -when a category or location is deleted. +`test_api` exercises the API end to end against a throwaway database — the auth +gate and session-token signing, nested categories and locations, search across +every indexed field, filter rollups, stock adjustment and history, patch +semantics including explicit nulls, taxonomy cycle rejection, security headers +and asset versioning, and login throttling. + +`test_concurrency` needs a real server process, because a lost update only shows +up when two requests genuinely overlap inside SQLite. It fires overlapping +adjustments, patches and creates at one part and asserts the stock log always +sums to the stored quantity. ## Configuration @@ -62,6 +69,20 @@ when a category or location is deleted. | `PARTS_SESSION_DAYS` | Session lifetime, default 30. | | `PARTS_AUTH` | `off` disables the login gate (LAN-only use). | | `PARTS_DB` | SQLite path. `/data/parts.db` in the container. | +| `PARTS_LOGIN_MAX_FAILURES` | Failed logins allowed per window, default 10. | +| `PARTS_LOGIN_WINDOW` | Throttle window in seconds, default 300. | +| `PARTS_SECURE_COOKIE` | `auto` (default) trusts `X-Forwarded-Proto`; `on`/`off` force it. | + +Rotating `PARTS_PASSWORD` alone does **not** invalidate existing sessions when +`PARTS_SECRET` is set independently — the cookie is signed with the secret. +Rotate both to revoke every outstanding cookie. + +Failed logins are throttled on a **global** window rather than per source +address. Caddy appends to `X-Forwarded-For` instead of replacing it, so the +client-supplied end of that header is forgeable and a per-IP bucket would be +trivially evaded. There is one legitimate user, so a global cap costs nothing +real; the trade-off is that a guessing spray can block the login form for the +window. Existing sessions keep working throughout. ## Deploying @@ -75,13 +96,23 @@ cd ~/srv/parts && sudo docker compose up -d --build ``` The database is in the `parts_parts_data` docker volume, which survives -rebuilds. To back it up: +rebuilds. + +**Back it up with SQLite's backup API, not `cp`.** The database runs in WAL +mode, so recently committed rows may still live in `parts.db-wal` and copying +`parts.db` alone can silently lose them. `VACUUM INTO` takes a consistent +snapshot of a live database: ```sh -sudo docker run --rm -v parts_parts_data:/d -v "$PWD":/out alpine \ - sh -c 'cp /d/parts.db /out/parts-backup.db' +docker exec parts python -c \ + "import sqlite3; sqlite3.connect('/data/parts.db').execute(\"VACUUM INTO '/data/backup.db'\")" +docker cp parts:/data/backup.db ./parts-backup-$(date +%F).db +docker exec parts rm /data/backup.db ``` +Restore by stopping the container, copying the file back over `/data/parts.db` +and deleting any leftover `-wal`/`-shm` alongside it. + ## API Everything under `/api` is JSON and cookie-authenticated. `GET /healthz` is open. @@ -103,3 +134,18 @@ GET /api/stats The "what could I build with what I'm holding?" idea is meant to arrive as another consumer of these endpoints — the arbiter on `.8` already has the lane routing and typed-action machinery for it — rather than as a fork of them. + +## Hardening notes + +Responses carry `X-Content-Type-Options`, `X-Frame-Options`, `Referrer-Policy` +and a `Content-Security-Policy` that keeps the page to same-origin scripts and +no framing; HSTS is added when `X-Forwarded-Proto` says the original request was +HTTPS. The session cookie is HttpOnly, SameSite=Lax and Secure, and its +signature is a fixed-width HMAC appended to the payload rather than delimited — +a delimiter byte can occur inside a raw digest, which previously invalidated +about 12% of issued tokens. + +Static assets are served under content-hashed URLs (`app.js?v=`), so +Cloudflare caching them for hours is harmless: a deploy changes the URL. The +bare, unhashed paths are served `no-cache` so nothing can pin stale frontend +code against a newer API. diff --git a/app/auth.py b/app/auth.py index c396030..24aac2c 100644 --- a/app/auth.py +++ b/app/auth.py @@ -9,17 +9,26 @@ import base64 import hashlib import hmac import os +import threading import time -from fastapi import Cookie, HTTPException +from fastapi import Cookie, HTTPException, Request COOKIE_NAME = "parts_session" +DIGEST_SIZE = hashlib.sha256().digest_size def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() +def _env_int(name: str, default: int, minimum: int = 1) -> int: + try: + return max(minimum, int(_env(name, str(default)))) + except ValueError: + return default + + def auth_enabled() -> bool: return _env("PARTS_AUTH", "on").lower() not in ("off", "0", "false", "no") @@ -34,10 +43,7 @@ def _secret() -> bytes: def session_days() -> int: - try: - return max(1, int(_env("PARTS_SESSION_DAYS", "30"))) - except ValueError: - return 30 + return _env_int("PARTS_SESSION_DAYS", 30) def check_password(candidate: str) -> bool: @@ -47,17 +53,27 @@ def check_password(candidate: str) -> bool: return hmac.compare_digest(candidate.encode(), expected.encode()) +# --- session tokens --------------------------------------------------------- +# +# The signature is raw HMAC bytes, which can contain any byte value including +# the one for ".". An earlier version joined payload and signature with "." and +# split on the last occurrence, so roughly 12% of issued tokens (1 - (255/256)^32) +# split in the wrong place and failed their own validator. The digest is a fixed +# 32 bytes, so slice by length instead of looking for a delimiter. + def issue_token() -> str: expires = int(time.time()) + session_days() * 86400 payload = str(expires).encode() sig = hmac.new(_secret(), payload, hashlib.sha256).digest() - return base64.urlsafe_b64encode(payload + b"." + sig).decode() + return base64.urlsafe_b64encode(payload + sig).decode() def token_valid(token: str) -> bool: try: raw = base64.urlsafe_b64decode(token.encode()) - payload, sig = raw.rsplit(b".", 1) + if len(raw) <= DIGEST_SIZE: + return False + payload, sig = raw[:-DIGEST_SIZE], raw[-DIGEST_SIZE:] expected = hmac.new(_secret(), payload, hashlib.sha256).digest() if not hmac.compare_digest(sig, expected): return False @@ -66,6 +82,73 @@ def token_valid(token: str) -> bool: return False +# --- failed-login throttling ------------------------------------------------ +# +# Deliberately a GLOBAL window rather than per-IP. Caddy appends to +# X-Forwarded-For rather than replacing it, so the client-supplied end of that +# header is attacker-controlled and a per-IP bucket would be trivially evaded by +# forging it. There is exactly one legitimate user, so a global cap costs +# nothing in practice and cannot be side-stepped by rotating source addresses. +# +# The trade-off: someone spraying guesses can lock the login form for the +# cooldown. Existing sessions keep working (they carry a valid cookie), so that +# is an annoyance rather than a lockout. + +_failures: list[float] = [] +_failures_lock = threading.Lock() + + +def _window() -> int: + return _env_int("PARTS_LOGIN_WINDOW", 300) + + +def _max_failures() -> int: + return _env_int("PARTS_LOGIN_MAX_FAILURES", 10) + + +def login_retry_after() -> int: + """Seconds until the next login attempt is allowed, or 0 if allowed now.""" + now = time.time() + window = _window() + with _failures_lock: + recent = [t for t in _failures if now - t < window] + _failures[:] = recent + if len(recent) < _max_failures(): + return 0 + return max(1, int(window - (now - recent[0]))) + + +def record_failure(): + with _failures_lock: + _failures.append(time.time()) + + +def clear_failures(): + with _failures_lock: + _failures.clear() + + +# --- request helpers -------------------------------------------------------- + +def is_https(request: Request) -> bool: + """Whether the *original* client request used TLS. + + Caddy terminates TLS and proxies to this app over plain HTTP, so + `request.url.scheme` is always "http" in production and can't decide whether + the session cookie may carry the Secure flag. X-Forwarded-Proto is what the + proxy records the real scheme in. + """ + override = _env("PARTS_SECURE_COOKIE", "auto").lower() + if override in ("on", "1", "true", "yes"): + return True + if override in ("off", "0", "false", "no"): + return False + forwarded = request.headers.get("x-forwarded-proto", "") + if forwarded: + return forwarded.split(",")[0].strip().lower() == "https" + return request.url.scheme == "https" + + def require_auth(parts_session: str | None = Cookie(default=None)): """FastAPI dependency guarding every data route.""" if not auth_enabled(): diff --git a/app/db.py b/app/db.py index 37b842c..fc09356 100644 --- a/app/db.py +++ b/app/db.py @@ -150,31 +150,88 @@ def get_db(): yield conn +# --- tree helpers ----------------------------------------------------------- + +def tree_paths(conn, table: str) -> dict[int, str]: + """Map id -> 'Parent / Child' display path for a self-referencing table.""" + rows = conn.execute(f"SELECT id, name, parent_id FROM {table}").fetchall() + by_id = {r["id"]: (r["name"], r["parent_id"]) for r in rows} + paths: dict[int, str] = {} + + def resolve(node_id: int, seen: set[int]) -> str: + if node_id in paths: + return paths[node_id] + name, parent = by_id[node_id] + # `seen` guards against a cycle introduced by a bad re-parent. + if parent and parent in by_id and parent not in seen: + path = resolve(parent, seen | {node_id}) + " / " + name + else: + path = name + paths[node_id] = path + return path + + for node_id in by_id: + resolve(node_id, set()) + return paths + + +def descendants(conn, table: str, root_id: int) -> list[int]: + """A node id plus every id beneath it, so filtering by 'Electronics' + catches parts filed under 'Electronics / Resistors'.""" + rows = conn.execute(f"SELECT id, parent_id FROM {table}").fetchall() + children: dict[int, list[int]] = {} + for r in rows: + children.setdefault(r["parent_id"], []).append(r["id"]) + out, stack, seen = [], [root_id], set() + while stack: + node = stack.pop() + if node in seen: + continue + seen.add(node) + out.append(node) + stack.extend(children.get(node, [])) + return out + + +def begin_immediate(conn): + """Take SQLite's write lock up front. + + Anything that reads a value, computes from it and writes it back must hold + the write lock across all three steps. Python's sqlite3 only begins its + implicit transaction at the first *write*, which leaves the preceding read + outside the transaction — two concurrent stock adjustments would then read + the same starting quantity and one would overwrite the other. + """ + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + + # --- search index ----------------------------------------------------------- -def reindex_part(conn, part_id: int): +def reindex_part(conn, part_id: int, cat_paths=None, loc_paths=None): """Rebuild one part's row in the FTS index. The index is a plain (self-contained) FTS5 table rather than an external-content one: it costs a duplicate copy of some short text, and in exchange a delete is just `DELETE ... WHERE rowid = ?` instead of the contentless table's delete-with-original-values dance. + + Full category and location *paths* are indexed, not just the leaf names, so + searching "Workshop" finds what is sitting in "Workshop / Bin A3". That is + also why renaming a node has to reindex everything beneath it. """ conn.execute("DELETE FROM parts_fts WHERE rowid = ?", (part_id,)) row = conn.execute( - """ - SELECT p.name, p.description, p.manufacturer, p.mpn, - COALESCE(c.name, '') AS category, - COALESCE(l.name, '') AS location - FROM parts p - LEFT JOIN categories c ON c.id = p.category_id - LEFT JOIN locations l ON l.id = p.location_id - WHERE p.id = ? - """, + "SELECT name, description, manufacturer, mpn, category_id, location_id " + "FROM parts WHERE id = ?", (part_id,), ).fetchone() if row is None: return + if cat_paths is None: + cat_paths = tree_paths(conn, "categories") + if loc_paths is None: + loc_paths = tree_paths(conn, "locations") specs = conn.execute( "SELECT key, value FROM part_specs WHERE part_id = ?", (part_id,) ).fetchall() @@ -191,15 +248,38 @@ def reindex_part(conn, part_id: int): """, ( part_id, row["name"], row["description"], row["manufacturer"], row["mpn"], - spec_text, tag_text, row["category"], row["location"], + spec_text, tag_text, + cat_paths.get(row["category_id"], "") if row["category_id"] else "", + loc_paths.get(row["location_id"], "") if row["location_id"] else "", ), ) +def parts_under(conn, table: str, node_id: int) -> list[int]: + """Ids of every part filed at a node or anywhere beneath it.""" + column = "category_id" if table == "categories" else "location_id" + ids = descendants(conn, table, node_id) + marks = ",".join("?" * len(ids)) + rows = conn.execute( + f"SELECT id FROM parts WHERE {column} IN ({marks})", ids + ).fetchall() + return [r["id"] for r in rows] + + +def reindex_parts(conn, part_ids): + """Reindex a batch, resolving the path tables only once.""" + part_ids = list(part_ids) + if not part_ids: + return + cat_paths = tree_paths(conn, "categories") + loc_paths = tree_paths(conn, "locations") + for part_id in part_ids: + reindex_part(conn, part_id, cat_paths, loc_paths) + + def reindex_all(conn): conn.execute("DELETE FROM parts_fts") - for (pid,) in conn.execute("SELECT id FROM parts").fetchall(): - reindex_part(conn, pid) + reindex_parts(conn, [r[0] for r in conn.execute("SELECT id FROM parts").fetchall()]) def fts_query(text: str) -> str: diff --git a/app/main.py b/app/main.py index 8d8eb54..cfcfc7d 100644 --- a/app/main.py +++ b/app/main.py @@ -6,22 +6,35 @@ the future "what could I build with this?" feature is meant to be another consumer of the same endpoints rather than a fork of them. """ +import asyncio +import hashlib import json import os -import time -from typing import Any, Literal - +import sqlite3 from contextlib import asynccontextmanager +from typing import Annotated, Any, Literal from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response -from fastapi.responses import FileResponse, JSONResponse +from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel, Field +from pydantic import AfterValidator, BaseModel, Field from . import auth, db STATIC_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "static") +CSP = ( + "default-src 'self'; " + "img-src 'self' data:; " + "style-src 'self' 'unsafe-inline'; " # the UI sets inline style attributes + "script-src 'self'; " + "connect-src 'self'; " + "frame-ancestors 'none'; " + "base-uri 'none'; " + "form-action 'self'" +) + + @asynccontextmanager async def lifespan(_app: FastAPI): db.init() @@ -31,102 +44,90 @@ async def lifespan(_app: FastAPI): app = FastAPI(title="Parts Inventory", docs_url=None, redoc_url=None, lifespan=lifespan) -# --- models ----------------------------------------------------------------- +# --- validated field types -------------------------------------------------- + +def _required(v: str) -> str: + v = v.strip() + if not v: + raise ValueError("must not be blank") + return v + + +RequiredText = Annotated[str, AfterValidator(_required)] +Text = Annotated[str, AfterValidator(lambda v: v.strip())] + class SpecIn(BaseModel): - key: str = Field(min_length=1, max_length=80) - value: str = Field(default="", max_length=400) + key: RequiredText = Field(max_length=80) + value: Text = Field(default="", max_length=400) class PartIn(BaseModel): - name: str = Field(min_length=1, max_length=200) - description: str = Field(default="", max_length=2000) + name: RequiredText = Field(max_length=200) + description: Text = Field(default="", max_length=2000) category_id: int | None = None location_id: int | None = None - manufacturer: str = Field(default="", max_length=200) - mpn: str = Field(default="", max_length=120) - quantity: float = 0 - unit: str = Field(default="pcs", max_length=20) - min_quantity: float | None = None - cost_each: float | None = None - datasheet_url: str = Field(default="", max_length=1000) - product_url: str = Field(default="", max_length=1000) - notes: str = Field(default="", max_length=4000) + manufacturer: Text = Field(default="", max_length=200) + mpn: Text = Field(default="", max_length=120) + quantity: float = Field(default=0, ge=0) + unit: RequiredText = Field(default="pcs", max_length=20) + min_quantity: float | None = Field(default=None, ge=0) + cost_each: float | None = Field(default=None, ge=0) + datasheet_url: Text = Field(default="", max_length=1000) + product_url: Text = Field(default="", max_length=1000) + notes: Text = Field(default="", max_length=4000) specs: list[SpecIn] = [] tags: list[str] = [] -class PartPatch(PartIn): - name: str | None = Field(default=None, min_length=1, max_length=200) - description: str | None = None - quantity: float | None = None - unit: str | None = None +class PartPatch(BaseModel): + name: RequiredText | None = Field(default=None, max_length=200) + description: Text | None = Field(default=None, max_length=2000) + category_id: int | None = None + location_id: int | None = None + manufacturer: Text | None = Field(default=None, max_length=200) + mpn: Text | None = Field(default=None, max_length=120) + quantity: float | None = Field(default=None, ge=0) + unit: RequiredText | None = Field(default=None, max_length=20) + min_quantity: float | None = Field(default=None, ge=0) + cost_each: float | None = Field(default=None, ge=0) + datasheet_url: Text | None = Field(default=None, max_length=1000) + product_url: Text | None = Field(default=None, max_length=1000) + notes: Text | None = Field(default=None, max_length=4000) specs: list[SpecIn] | None = None tags: list[str] | None = None + reason: Text = Field(default="edited", max_length=300) + + +# Columns a PATCH may legitimately set back to NULL. Everything else is NOT NULL +# in the schema, so an explicit null there is a client error, not a 500. +NULLABLE_COLUMNS = {"category_id", "location_id", "min_quantity", "cost_each"} +PART_COLUMNS = [ + "name", "description", "category_id", "location_id", "manufacturer", "mpn", + "quantity", "unit", "min_quantity", "cost_each", "datasheet_url", "product_url", "notes", +] class AdjustIn(BaseModel): delta: float - reason: str = Field(default="", max_length=300) + reason: Text = Field(default="", max_length=300) class NodeIn(BaseModel): - name: str = Field(min_length=1, max_length=120) + name: RequiredText = Field(max_length=120) parent_id: int | None = None - unit: str = Field(default="pcs", max_length=20) + unit: RequiredText = Field(default="pcs", max_length=20) spec_template: list[SpecIn] = [] - notes: str = Field(default="", max_length=1000) + notes: Text = Field(default="", max_length=1000) sort_order: int = 0 class LoginIn(BaseModel): - password: str + password: str = Field(max_length=500) # --- helpers ---------------------------------------------------------------- -def _tree_paths(conn, table: str) -> dict[int, str]: - """Map id -> 'Parent / Child' display path for a self-referencing table.""" - rows = conn.execute(f"SELECT id, name, parent_id FROM {table}").fetchall() - by_id = {r["id"]: (r["name"], r["parent_id"]) for r in rows} - paths: dict[int, str] = {} - - def resolve(node_id: int, seen: set[int]) -> str: - if node_id in paths: - return paths[node_id] - name, parent = by_id[node_id] - # `seen` guards against a cycle introduced by a bad re-parent. - if parent and parent in by_id and parent not in seen: - path = resolve(parent, seen | {node_id}) + " / " + name - else: - path = name - paths[node_id] = path - return path - - for node_id in by_id: - resolve(node_id, set()) - return paths - - -def _descendants(conn, table: str, root_id: int) -> list[int]: - """A node id plus every id beneath it, so filtering by 'Electronics' - catches parts filed under 'Electronics / Resistors'.""" - rows = conn.execute(f"SELECT id, parent_id FROM {table}").fetchall() - children: dict[int, list[int]] = {} - for r in rows: - children.setdefault(r["parent_id"], []).append(r["id"]) - out, stack = [], [root_id] - seen = set() - while stack: - node = stack.pop() - if node in seen: - continue - seen.add(node) - out.append(node) - stack.extend(children.get(node, [])) - return out - - def _specs_for(conn, part_ids: list[int]) -> dict[int, list[dict]]: if not part_ids: return {} @@ -177,13 +178,12 @@ def _write_specs(conn, part_id: int, specs: list[SpecIn]): seen = set() position = 0 for s in specs: - key = s.key.strip() - if not key or key.lower() in seen: + if s.key.lower() in seen: continue - seen.add(key.lower()) + seen.add(s.key.lower()) conn.execute( "INSERT INTO part_specs(part_id, key, value, position) VALUES (?,?,?,?)", - (part_id, key, s.value.strip(), position), + (part_id, s.key, s.value, position), ) position += 1 @@ -203,17 +203,69 @@ def _write_tags(conn, part_id: int, tags: list[str]): ) +def _log_stock(conn, part_id: int, delta: float, after: float, reason: str): + conn.execute( + "INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)", + (part_id, delta, after, reason), + ) + + def _fetch_part(conn, part_id: int) -> dict: row = conn.execute("SELECT * FROM parts WHERE id = ?", (part_id,)).fetchone() if row is None: raise HTTPException(404, "Part not found") - cat_paths = _tree_paths(conn, "categories") - loc_paths = _tree_paths(conn, "locations") return _serialise( - [row], _specs_for(conn, [part_id]), _tags_for(conn, [part_id]), cat_paths, loc_paths + [row], _specs_for(conn, [part_id]), _tags_for(conn, [part_id]), + db.tree_paths(conn, "categories"), db.tree_paths(conn, "locations"), )[0] +def _require_node(conn, table: str, node_id: int, label: str): + if conn.execute(f"SELECT 1 FROM {table} WHERE id = ?", (node_id,)).fetchone() is None: + raise HTTPException(404, f"{label} not found") + + +def _check_parent(conn, table: str, node_id: int, parent_id: int | None, label: str): + """Reject re-parenting a node under itself or any of its own descendants. + + A direct self-parent is the obvious case; the indirect one (A under B, where + B is already a child of A) is the one that actually corrupts the tree. + """ + if parent_id is None: + return + if parent_id == node_id: + raise HTTPException(400, f"A {label} cannot be its own parent") + _require_node(conn, table, parent_id, f"Parent {label}") + if parent_id in db.descendants(conn, table, node_id): + raise HTTPException(400, f"A {label} cannot be moved beneath itself") + + +# --- middleware ------------------------------------------------------------- + +@app.middleware("http") +async def security_headers(request: Request, call_next): + response = await call_next(request) + headers = response.headers + headers.setdefault("X-Content-Type-Options", "nosniff") + headers.setdefault("X-Frame-Options", "DENY") + headers.setdefault("Referrer-Policy", "no-referrer") + headers.setdefault("Content-Security-Policy", CSP) + if auth.is_https(request): + headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains") + if request.url.path.startswith("/static/"): + # Asset URLs carry a content hash (see _index_html), so a versioned URL + # is safe to cache forever and a deploy simply changes the URL. Without + # the hash — someone hitting the bare path — it must not be cached, or + # an intermediary can pin stale frontend code against a new API. + if request.query_params.get("v"): + headers["Cache-Control"] = "public, max-age=31536000, immutable" + else: + headers["Cache-Control"] = "no-cache" + elif request.url.path.startswith("/api/"): + headers["Cache-Control"] = "no-store" + return response + + # --- auth routes ------------------------------------------------------------ @app.get("/api/me") @@ -228,20 +280,31 @@ def me(request: Request): @app.post("/api/login") -def login(body: LoginIn, response: Response, request: Request): +async def login(body: LoginIn, response: Response, request: Request): if not auth.auth_enabled(): return {"authenticated": True, "auth_required": False} + + retry_after = auth.login_retry_after() + if retry_after: + raise HTTPException( + 429, "Too many failed attempts", headers={"Retry-After": str(retry_after)} + ) + if not auth.check_password(body.password): - # A flat delay blunts online guessing without needing a rate-limit store. - time.sleep(1.0) + auth.record_failure() + # Async sleep, so a burst of guesses can't tie up the worker threadpool + # that real requests need. + await asyncio.sleep(0.5) raise HTTPException(401, "Incorrect password") + + auth.clear_failures() response.set_cookie( auth.COOKIE_NAME, auth.issue_token(), max_age=auth.session_days() * 86400, httponly=True, samesite="lax", - secure=request.url.scheme == "https", + secure=auth.is_https(request), path="/", ) return {"authenticated": True, "auth_required": True} @@ -283,12 +346,12 @@ def list_parts( order = "parts_fts.rank ASC, p.name COLLATE NOCASE ASC" if category_id is not None: - ids = _descendants(conn, "categories", category_id) + ids = db.descendants(conn, "categories", category_id) where.append(f"p.category_id IN ({','.join('?' * len(ids))})") params.extend(ids) if location_id is not None: - ids = _descendants(conn, "locations", location_id) + ids = db.descendants(conn, "locations", location_id) where.append(f"p.location_id IN ({','.join('?' * len(ids))})") params.extend(ids) @@ -323,37 +386,35 @@ def list_parts( ids = [r["id"] for r in rows] items = _serialise( - rows, - _specs_for(conn, ids), - _tags_for(conn, ids), - _tree_paths(conn, "categories"), - _tree_paths(conn, "locations"), + rows, _specs_for(conn, ids), _tags_for(conn, ids), + db.tree_paths(conn, "categories"), db.tree_paths(conn, "locations"), ) return {"total": total, "limit": limit, "offset": offset, "items": items} @app.post("/api/parts", dependencies=[Depends(auth.require_auth)], status_code=201) def create_part(body: PartIn, conn=Depends(db.get_db)): + if body.category_id is not None: + _require_node(conn, "categories", body.category_id, "Category") + if body.location_id is not None: + _require_node(conn, "locations", body.location_id, "Location") cur = conn.execute( """INSERT INTO parts(name, description, category_id, location_id, manufacturer, mpn, quantity, unit, min_quantity, cost_each, datasheet_url, product_url, notes) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( - body.name.strip(), body.description.strip(), body.category_id, body.location_id, - body.manufacturer.strip(), body.mpn.strip(), body.quantity, body.unit.strip() or "pcs", - body.min_quantity, body.cost_each, body.datasheet_url.strip(), - body.product_url.strip(), body.notes.strip(), + body.name, body.description, body.category_id, body.location_id, + body.manufacturer, body.mpn, body.quantity, body.unit, + body.min_quantity, body.cost_each, body.datasheet_url, + body.product_url, body.notes, ), ) part_id = cur.lastrowid _write_specs(conn, part_id, body.specs) _write_tags(conn, part_id, body.tags) if body.quantity: - conn.execute( - "INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)", - (part_id, body.quantity, body.quantity, "initial stock"), - ) + _log_stock(conn, part_id, body.quantity, body.quantity, "initial stock") db.reindex_part(conn, part_id) return _fetch_part(conn, part_id) @@ -365,28 +426,40 @@ def get_part(part_id: int, conn=Depends(db.get_db)): @app.patch("/api/parts/{part_id}", dependencies=[Depends(auth.require_auth)]) def update_part(part_id: int, body: PartPatch, conn=Depends(db.get_db)): - if conn.execute("SELECT 1 FROM parts WHERE id = ?", (part_id,)).fetchone() is None: - raise HTTPException(404, "Part not found") - fields = body.model_dump(exclude_unset=True) fields.pop("specs", None) fields.pop("tags", None) - columns = [ - "name", "description", "category_id", "location_id", "manufacturer", "mpn", - "quantity", "unit", "min_quantity", "cost_each", "datasheet_url", "product_url", "notes", - ] + reason = fields.pop("reason", "edited") or "edited" + + for col, value in fields.items(): + if value is None and col not in NULLABLE_COLUMNS: + raise HTTPException(422, f"'{col}' cannot be null") + if fields.get("category_id") is not None: + _require_node(conn, "categories", fields["category_id"], "Category") + if fields.get("location_id") is not None: + _require_node(conn, "locations", fields["location_id"], "Location") + + # Setting quantity is a read-modify-write like any adjustment, and has to be + # logged the same way — "every change is recorded" has to mean every change, + # or the history is worse than useless. + db.begin_immediate(conn) + row = conn.execute("SELECT quantity FROM parts WHERE id = ?", (part_id,)).fetchone() + if row is None: + raise HTTPException(404, "Part not found") + sets, params = [], [] - for col in columns: + for col in PART_COLUMNS: if col in fields: - value = fields[col] - if isinstance(value, str): - value = value.strip() sets.append(f"{col} = ?") - params.append(value) + params.append(fields[col]) if sets: sets.append("updated_at = datetime('now')") conn.execute(f"UPDATE parts SET {', '.join(sets)} WHERE id = ?", params + [part_id]) + if "quantity" in fields and fields["quantity"] != row["quantity"]: + new_qty = fields["quantity"] + _log_stock(conn, part_id, new_qty - row["quantity"], new_qty, reason) + if body.specs is not None: _write_specs(conn, part_id, body.specs) if body.tags is not None: @@ -406,20 +479,25 @@ def delete_part(part_id: int, conn=Depends(db.get_db)): @app.post("/api/parts/{part_id}/adjust", dependencies=[Depends(auth.require_auth)]) def adjust_part(part_id: int, body: AdjustIn, conn=Depends(db.get_db)): + # Read and write under one write lock. Without it, concurrent adjustments + # read the same starting quantity and silently overwrite each other, leaving + # the stock log disagreeing with the stock. + db.begin_immediate(conn) row = conn.execute("SELECT quantity FROM parts WHERE id = ?", (part_id,)).fetchone() if row is None: raise HTTPException(404, "Part not found") - after = round(row["quantity"] + body.delta, 4) - if after < 0: - after = 0.0 + + before = row["quantity"] + after = round(max(0.0, before + body.delta), 4) + # Log what actually happened. Asking for -999 against a stock of 7 removes + # 7, and recording -999 would make the ledger unsummable. + applied = round(after - before, 4) + conn.execute( "UPDATE parts SET quantity = ?, updated_at = datetime('now') WHERE id = ?", (after, part_id), ) - conn.execute( - "INSERT INTO stock_log(part_id, delta, quantity_after, reason) VALUES (?,?,?,?)", - (part_id, body.delta, after, body.reason.strip()), - ) + _log_stock(conn, part_id, applied, after, body.reason) return _fetch_part(conn, part_id) @@ -434,7 +512,7 @@ def part_history(part_id: int, conn=Depends(db.get_db), limit: int = Query(defau # --- categories & locations ------------------------------------------------- def _node_payload(conn, table: str) -> list[dict]: - paths = _tree_paths(conn, table) + paths = db.tree_paths(conn, table) rows = conn.execute( f"SELECT * FROM {table} ORDER BY sort_order, name COLLATE NOCASE" ).fetchall() @@ -475,6 +553,10 @@ def _node_payload(conn, table: str) -> list[dict]: return out +def _template_json(specs: list[SpecIn]) -> str: + return json.dumps([{"key": s.key} for s in specs]) + + @app.get("/api/categories", dependencies=[Depends(auth.require_auth)]) def list_categories(conn=Depends(db.get_db)): return {"items": _node_payload(conn, "categories")} @@ -482,33 +564,44 @@ def list_categories(conn=Depends(db.get_db)): @app.post("/api/categories", dependencies=[Depends(auth.require_auth)], status_code=201) def create_category(body: NodeIn, conn=Depends(db.get_db)): - template = json.dumps([{"key": s.key.strip()} for s in body.spec_template if s.key.strip()]) + if body.parent_id is not None: + _require_node(conn, "categories", body.parent_id, "Parent category") try: cur = conn.execute( "INSERT INTO categories(name, parent_id, unit, spec_template, sort_order) VALUES (?,?,?,?,?)", - (body.name.strip(), body.parent_id, body.unit.strip() or "pcs", template, body.sort_order), + (body.name, body.parent_id, body.unit, _template_json(body.spec_template), body.sort_order), ) - except Exception: + except sqlite3.IntegrityError: raise HTTPException(409, "A category with that name already exists here") return {"id": cur.lastrowid} @app.patch("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)]) def update_category(cat_id: int, body: NodeIn, conn=Depends(db.get_db)): - template = json.dumps([{"key": s.key.strip()} for s in body.spec_template if s.key.strip()]) - if body.parent_id == cat_id: - raise HTTPException(400, "A category cannot be its own parent") - conn.execute( - "UPDATE categories SET name=?, parent_id=?, unit=?, spec_template=?, sort_order=? WHERE id=?", - (body.name.strip(), body.parent_id, body.unit.strip() or "pcs", template, body.sort_order, cat_id), - ) + _require_node(conn, "categories", cat_id, "Category") + _check_parent(conn, "categories", cat_id, body.parent_id, "category") + affected = db.parts_under(conn, "categories", cat_id) + try: + conn.execute( + "UPDATE categories SET name=?, parent_id=?, unit=?, spec_template=?, sort_order=? WHERE id=?", + (body.name, body.parent_id, body.unit, _template_json(body.spec_template), + body.sort_order, cat_id), + ) + except sqlite3.IntegrityError: + raise HTTPException(409, "A category with that name already exists here") + # The category path is part of every affected part's search text. + db.reindex_parts(conn, affected) return {"id": cat_id} @app.delete("/api/categories/{cat_id}", dependencies=[Depends(auth.require_auth)]) def delete_category(cat_id: int, conn=Depends(db.get_db)): - # ON DELETE SET NULL means parts survive as uncategorised rather than vanish. + _require_node(conn, "categories", cat_id, "Category") + # Capture the parts first: ON DELETE SET NULL means they survive as + # uncategorised rather than vanish, but their indexed path must be rebuilt. + affected = db.parts_under(conn, "categories", cat_id) conn.execute("DELETE FROM categories WHERE id = ?", (cat_id,)) + db.reindex_parts(conn, affected) return {"deleted": cat_id} @@ -519,30 +612,40 @@ def list_locations(conn=Depends(db.get_db)): @app.post("/api/locations", dependencies=[Depends(auth.require_auth)], status_code=201) def create_location(body: NodeIn, conn=Depends(db.get_db)): + if body.parent_id is not None: + _require_node(conn, "locations", body.parent_id, "Parent location") try: cur = conn.execute( "INSERT INTO locations(name, parent_id, notes, sort_order) VALUES (?,?,?,?)", - (body.name.strip(), body.parent_id, body.notes.strip(), body.sort_order), + (body.name, body.parent_id, body.notes, body.sort_order), ) - except Exception: + except sqlite3.IntegrityError: raise HTTPException(409, "A location with that name already exists here") return {"id": cur.lastrowid} @app.patch("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)]) def update_location(loc_id: int, body: NodeIn, conn=Depends(db.get_db)): - if body.parent_id == loc_id: - raise HTTPException(400, "A location cannot be its own parent") - conn.execute( - "UPDATE locations SET name=?, parent_id=?, notes=?, sort_order=? WHERE id=?", - (body.name.strip(), body.parent_id, body.notes.strip(), body.sort_order, loc_id), - ) + _require_node(conn, "locations", loc_id, "Location") + _check_parent(conn, "locations", loc_id, body.parent_id, "location") + affected = db.parts_under(conn, "locations", loc_id) + try: + conn.execute( + "UPDATE locations SET name=?, parent_id=?, notes=?, sort_order=? WHERE id=?", + (body.name, body.parent_id, body.notes, body.sort_order, loc_id), + ) + except sqlite3.IntegrityError: + raise HTTPException(409, "A location with that name already exists here") + db.reindex_parts(conn, affected) return {"id": loc_id} @app.delete("/api/locations/{loc_id}", dependencies=[Depends(auth.require_auth)]) def delete_location(loc_id: int, conn=Depends(db.get_db)): + _require_node(conn, "locations", loc_id, "Location") + affected = db.parts_under(conn, "locations", loc_id) conn.execute("DELETE FROM locations WHERE id = ?", (loc_id,)) + db.reindex_parts(conn, affected) return {"deleted": loc_id} @@ -581,10 +684,37 @@ def healthz(): app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") +_index_cache: str | None = None + + +def _index_html() -> str: + """index.html with a content hash appended to each asset URL. + + Cloudflare sits in front of this and will happily serve a cached app.js for + hours. Versioned URLs mean a deploy changes the URL itself, so a stale copy + can never be paired with a newer API. + """ + global _index_cache + if _index_cache is None: + with open(os.path.join(STATIC_DIR, "index.html"), encoding="utf-8") as fh: + html = fh.read() + for asset in ("app.css", "app.js"): + with open(os.path.join(STATIC_DIR, asset), "rb") as fh: + digest = hashlib.sha256(fh.read()).hexdigest()[:10] + html = html.replace(f"/static/{asset}", f"/static/{asset}?v={digest}") + _index_cache = html + return _index_cache + + +def _index_response(status_code: int = 200) -> HTMLResponse: + return HTMLResponse( + _index_html(), status_code=status_code, headers={"Cache-Control": "no-store"} + ) + @app.get("/") def index(): - return FileResponse(os.path.join(STATIC_DIR, "index.html")) + return _index_response() @app.exception_handler(404) @@ -592,4 +722,4 @@ def not_found(request: Request, exc): path = request.url.path if path.startswith("/api/") or path.startswith("/static/") or path == "/healthz": return JSONResponse({"detail": "Not found"}, status_code=404) - return FileResponse(os.path.join(STATIC_DIR, "index.html"), status_code=200) + return _index_response() diff --git a/requirements.txt b/requirements.txt index 9d1ad89..3414f51 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,8 @@ -fastapi==0.115.6 -uvicorn[standard]==0.34.0 -pydantic==2.10.4 -python-multipart==0.0.20 +# starlette is pinned explicitly, not just inherited from fastapi's range: the +# versions fastapi would otherwise accept include ones with published advisories +# (notably a FileResponse Range-header CPU DoS, which matters because / and the +# static assets are publicly reachable). +fastapi==0.141.1 +starlette==1.6.0 +uvicorn[standard]==0.52.4 +pydantic==2.13.4 diff --git a/tests/test_api.py b/tests/test_api.py index a02abbe..fc389ea 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -23,6 +23,22 @@ def check(label, condition, detail=""): 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): + payload = str(exp).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) @@ -189,7 +205,127 @@ with TestClient(app) as client: s = client.get("/api/stats").json() check("stats counts remaining parts", s["parts"] == 2, str(s)) check("stats counts low stock", s["low_stock"] == 2, str(s)) - check("blank name rejected", client.post("/api/parts", json={"name": " "}).status_code in (201, 422)) + 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) @@ -200,6 +336,21 @@ with TestClient(app) as client: 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) + print() if failures: print(f"{len(failures)} FAILED: " + "; ".join(failures)) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..bacfd82 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,135 @@ +"""Concurrency checks against a real uvicorn process. + +The in-process test client is not enough here: a lost update needs two requests +genuinely overlapping inside SQLite, which means a real server and real threads. +""" + +import json +import os +import shutil +import signal +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor + +PORT = 8137 +BASE = f"http://127.0.0.1:{PORT}" +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +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) + + +def req(path, method="GET", body=None): + data = json.dumps(body).encode() if body is not None else None + r = urllib.request.Request(BASE + path, data=data, method=method, + headers={"Content-Type": "application/json"}) + return json.load(urllib.request.urlopen(r, timeout=60)) + + +def main(): + tmp = tempfile.mkdtemp() + env = { + **os.environ, + "PARTS_DB": os.path.join(tmp, "conc.db"), + "PARTS_AUTH": "off", + } + server = subprocess.Popen( + [sys.executable, "-m", "uvicorn", "app.main:app", "--port", str(PORT), "--log-level", "warning"], + cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + try: + for _ in range(120): + try: + req("/healthz") + break + except Exception: + time.sleep(0.25) + else: + raise RuntimeError("server never started") + + # --- concurrent decrements must not lose updates --- + pid = req("/api/parts", "POST", {"name": "race target", "quantity": 100})["id"] + with ThreadPoolExecutor(max_workers=50) as ex: + codes = list(ex.map( + lambda _: req(f"/api/parts/{pid}/adjust", "POST", {"delta": -1, "reason": "race"}) and 200, + range(50))) + check("all 50 concurrent adjustments succeeded", codes.count(200) == 50) + final = req(f"/api/parts/{pid}")["quantity"] + check("50 concurrent -1 adjustments land at 50", final == 50, f"got {final}") + hist = req(f"/api/parts/{pid}/history?limit=500")["items"] + check("history has one row per change", len(hist) == 51, f"got {len(hist)}") + total = sum(h["delta"] for h in hist) + check("stock log sums to the stored quantity", total == final, f"log={total} stored={final}") + + # --- mixed concurrent increments and decrements --- + # Start high enough that no intermediate ordering can hit the floor at + # zero: with the floor in play the net is legitimately order-dependent, + # which would make this assertion about clamping rather than atomicity. + start = 100.0 + deltas = [5] * 30 + [-3] * 30 + pid2 = req("/api/parts", "POST", {"name": "mixed target", "quantity": start})["id"] + with ThreadPoolExecutor(max_workers=30) as ex: + list(ex.map(lambda d: req(f"/api/parts/{pid2}/adjust", "POST", {"delta": d}), deltas)) + got = req(f"/api/parts/{pid2}")["quantity"] + check("mixed concurrent adjustments net out correctly", + got == start + sum(deltas), f"got {got}, expected {start + sum(deltas)}") + h2 = req(f"/api/parts/{pid2}/history?limit=500")["items"] + check("mixed adjustment log sums to the stored quantity", + sum(x["delta"] for x in h2) == got, f"log={sum(x['delta'] for x in h2)} stored={got}") + + # --- the floor at zero is still honoured under contention --- + pid_floor = req("/api/parts", "POST", {"name": "floor race", "quantity": 10})["id"] + with ThreadPoolExecutor(max_workers=20) as ex: + list(ex.map(lambda _: req(f"/api/parts/{pid_floor}/adjust", "POST", {"delta": -1}), range(20))) + fq = req(f"/api/parts/{pid_floor}")["quantity"] + fh = req(f"/api/parts/{pid_floor}/history?limit=500")["items"] + check("20 concurrent -1 on a stock of 10 floors at zero", fq == 0, f"got {fq}") + check("floored concurrent log still sums to the stored quantity", + sum(x["delta"] for x in fh) == fq, f"log={sum(x['delta'] for x in fh)} stored={fq}") + + # --- concurrent PATCH quantity is logged exactly once per change --- + pid3 = req("/api/parts", "POST", {"name": "patch target", "quantity": 0})["id"] + with ThreadPoolExecutor(max_workers=20) as ex: + list(ex.map(lambda i: req(f"/api/parts/{pid3}", "PATCH", {"quantity": float(i + 1)}), range(20))) + h3 = req(f"/api/parts/{pid3}/history?limit=500")["items"] + stored = req(f"/api/parts/{pid3}")["quantity"] + check("concurrent PATCHes each logged a row", len(h3) == 20, f"got {len(h3)}") + check("PATCH log's final quantity_after matches stored", + h3[0]["quantity_after"] == stored, f"log={h3[0]['quantity_after']} stored={stored}") + + # --- concurrent creates don't collide --- + with ThreadPoolExecutor(max_workers=25) as ex: + ids = list(ex.map( + lambda i: req("/api/parts", "POST", {"name": f"bulk {i}", "quantity": 1})["id"], range(25))) + check("25 concurrent creates produced 25 distinct parts", len(set(ids)) == 25) + check("all concurrent creates are searchable", + req("/api/parts?q=bulk&limit=100")["total"] == 25, + str(req("/api/parts?q=bulk&limit=100")["total"])) + finally: + server.send_signal(signal.SIGINT) + try: + server.wait(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + shutil.rmtree(tmp, ignore_errors=True) + + print() + if failures: + print(f"{len(failures)} FAILED: " + "; ".join(failures)) + return 1 + print("all concurrency checks passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())