feat(auth): control schema, local accounts and sign-in sessions (M2a, #92)

alembic/versions/0004_control_schema.py

@@ -0,0 +1,62 @@
+"""Control schema: tenants, users, memberships and sessions (M2a, #92, BR-36).
+
+Shared by every tenant. Roles are the four from ADR-0003. Session tokens are
+stored only as SHA-256 hashes, so a database read does not yield usable tokens.
+
+Revision ID: 0004
+Revises: 0003
+Create Date: 2026-10-26
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0004"
+down_revision = "0003"
+branch_labels = None
+depends_on = None
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.CONTROL_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    op.execute(f"CREATE SCHEMA IF NOT EXISTS {s}")
+    op.execute(f"""
+        CREATE TABLE {s}.tenants (
+            id BIGSERIAL PRIMARY KEY,
+            slug TEXT NOT NULL UNIQUE CHECK (slug ~ '^[a-z][a-z0-9_]{{1,40}}$'),
+            name TEXT NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )""")
+    op.execute(f"""
+        CREATE TABLE {s}.users (
+            id BIGSERIAL PRIMARY KEY,
+            username TEXT NOT NULL UNIQUE,
+            password_hash TEXT NOT NULL,
+            is_active BOOLEAN NOT NULL DEFAULT true,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )""")
+    op.execute(f"""
+        CREATE TABLE {s}.memberships (
+            user_id BIGINT NOT NULL REFERENCES {s}.users(id) ON DELETE CASCADE,
+            tenant_id BIGINT NOT NULL REFERENCES {s}.tenants(id) ON DELETE CASCADE,
+            role TEXT NOT NULL CHECK (role IN ('uploader', 'reviewer', 'tenant_admin', 'auditor')),
+            PRIMARY KEY (user_id, tenant_id)
+        )""")
+    op.execute(f"""
+        CREATE TABLE {s}.sessions (
+            token_hash TEXT PRIMARY KEY,
+            user_id BIGINT NOT NULL REFERENCES {s}.users(id) ON DELETE CASCADE,
+            tenant_id BIGINT NOT NULL REFERENCES {s}.tenants(id) ON DELETE CASCADE,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            expires_at TIMESTAMPTZ NOT NULL,
+            revoked_at TIMESTAMPTZ
+        )""")
+
+
+def downgrade() -> None:
+    op.execute(f"DROP SCHEMA IF EXISTS {_schema()} CASCADE")

api/auth.py

@@ -13,9 +13,11 @@
 
 import secrets
 
-from fastapi import Header, HTTPException
+from fastapi import Depends, Header, HTTPException
 
 from api.config import settings
+from api.db import Conn
+from api.services.accounts import session_for
 
 
 def require_api_key(x_api_key: str = Header(default="")) -> None:
@@ -24,3 +26,24 @@ def require_api_key(x_api_key: str = Header(default="")) -> None:
         return
     if not secrets.compare_digest(x_api_key, settings.API_KEY):
         raise HTTPException(status_code=401, detail="Missing or invalid X-API-Key header")
+
+
+def bearer_token(authorization: str = Header(default="")) -> str:
+    """The token from an ``Authorization: Bearer <token>`` header, or 401."""
+    scheme, _, token = authorization.partition(" ")
+    if scheme.lower() != "bearer" or not token.strip():
+        raise HTTPException(status_code=401, detail="Sign in required")
+    return token.strip()
+
+
+def require_session(token: str = Depends(bearer_token)) -> dict:
+    """The caller's live session (user, tenant, role), or 401 (M2a, #92).
+
+    Not yet applied to the data routers; tenant routing (#93) and the role
+    matrix (#94) switch them over.
+    """
+    with Conn() as conn, conn.cursor() as cur:
+        session = session_for(cur, token)
+    if session is None:
+        raise HTTPException(status_code=401, detail="Session expired or invalid; sign in again")
+    return session

api/config.py

@@ -12,6 +12,10 @@ class Settings:
 
     # Schema where dynamically created per-CSV tables + the registry live.
     UPLOADS_SCHEMA: str = os.environ.get("CSV_UPLOADS_SCHEMA", "csv_uploads")
+    # M2 (#92): tenants, users, memberships and sessions, shared by all tenants.
+    CONTROL_SCHEMA: str = os.environ.get("CONTROL_SCHEMA", "control")
+    # Lifetime of a sign-in session, in hours.
+    SESSION_TTL_HOURS: int = int(os.environ.get("SESSION_TTL_HOURS", "12"))
 
     # Schema holding the fixed T&E tables (deploy_all.sh dev creates te_dev).
     TE_SCHEMA: str = os.environ.get("TE_SCHEMA", "te_dev")

api/main.py

@@ -18,7 +18,7 @@
 from api import db
 from api.config import settings
 from api.ratelimit import rate_limit_middleware
-from api.routers import audit_routes, csv_routes, te_routes
+from api.routers import audit_routes, auth_routes, csv_routes, te_routes
 
 logger = logging.getLogger(__name__)
 
@@ -86,14 +86,15 @@ async def lifespan(app: FastAPI):
     CORSMiddleware,
     allow_origins=settings.CORS_ORIGINS,
     allow_methods=["GET", "POST", "DELETE"],
-    allow_headers=["Content-Type", "X-API-Key"],
+    allow_headers=["Content-Type", "X-API-Key", "Authorization"],
 )
 
 # O-03: rate limiting. Registered after CORSMiddleware so it sits outside it in
 # Starlette's onion (last added = outermost) and a 429 still carries CORS headers,
 # which the browser needs in order to read the status at all.
 app.middleware("http")(rate_limit_middleware)
 
+app.include_router(auth_routes.router)
 app.include_router(csv_routes.router)
 app.include_router(te_routes.router)
 app.include_router(audit_routes.router)

api/routers/auth_routes.py

@@ -0,0 +1,48 @@
+"""Sign-in with local accounts (M2a, #92, BR-36).
+
+Sessions are bearer tokens: ``Authorization: Bearer <token>``. Login failures
+all return the same 401 so a caller cannot probe for valid usernames or tenants.
+The global rate limiter's write budget applies to login attempts.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+
+from api.auth import bearer_token, require_session
+from api.db import Conn
+from api.services import accounts
+
+router = APIRouter(prefix="/api/auth", tags=["auth"])
+
+
+class LoginRequest(BaseModel):
+    tenant: str = Field(min_length=1, max_length=64)
+    username: str = Field(min_length=1, max_length=128)
+    password: str = Field(min_length=1, max_length=1024)
+
+
+@router.post("/login")
+def login(req: LoginRequest) -> dict:
+    with Conn() as conn:
+        with conn.cursor() as cur:
+            session = accounts.login(cur, req.tenant, req.username, req.password)
+        conn.commit()
+    if session is None:
+        raise HTTPException(status_code=401, detail="Invalid tenant, username or password")
+    return session
+
+
+@router.post("/logout")
+def logout(token: str = Depends(bearer_token)) -> dict:
+    with Conn() as conn:
+        with conn.cursor() as cur:
+            accounts.revoke(cur, token)
+        conn.commit()
+    return {"status": "ok"}
+
+
+@router.get("/me")
+def me(session: dict = Depends(require_session)) -> dict:
+    return session

api/services/accounts.py

@@ -0,0 +1,114 @@
+"""Local accounts and sign-in sessions (M2a, #92, BR-36).
+
+Passwords use stdlib scrypt, so air-gapped installs need no extra package.
+Session tokens are random, returned once, and stored only as SHA-256 hashes.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import secrets
+
+from psycopg2 import sql
+
+from api.config import settings
+
+ROLES = ("uploader", "reviewer", "tenant_admin", "auditor")
+_N, _R, _P = 2**14, 8, 1
+
+
+def hash_password(password: str) -> str:
+    salt = secrets.token_bytes(16)
+    dk = hashlib.scrypt(password.encode(), salt=salt, n=_N, r=_R, p=_P)
+    return f"scrypt${_N}${_R}${_P}${salt.hex()}${dk.hex()}"
+
+
+def verify_password(password: str, stored: str) -> bool:
+    try:
+        algo, n, r, p, salt, want = stored.split("$")
+        if algo != "scrypt":
+            return False
+        dk = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), n=int(n), r=int(r), p=int(p))
+    except (ValueError, TypeError):
+        return False
+    return hmac.compare_digest(dk.hex(), want)
+
+
+# Checked when the username is unknown, so a miss costs the same time as a hit.
+_DUMMY_HASH = hash_password(secrets.token_hex(8))
+
+
+def _token_hash(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+def _q(query: str) -> sql.Composed:
+    return sql.SQL(query).format(c=sql.Identifier(settings.CONTROL_SCHEMA))
+
+
+def create_tenant(cur, slug: str, name: str) -> int:
+    cur.execute(_q("INSERT INTO {c}.tenants (slug, name) VALUES (%s, %s) RETURNING id"), (slug, name))
+    return cur.fetchone()[0]
+
+
+def create_user(cur, username: str, password: str) -> int:
+    cur.execute(_q("INSERT INTO {c}.users (username, password_hash) VALUES (%s, %s) RETURNING id"),
+                (username, hash_password(password)))
+    return cur.fetchone()[0]
+
+
+def add_membership(cur, user_id: int, tenant_id: int, role: str) -> None:
+    if role not in ROLES:
+        raise ValueError(f"Unknown role {role!r}; expected one of {ROLES}")
+    cur.execute(_q("INSERT INTO {c}.memberships (user_id, tenant_id, role) VALUES (%s, %s, %s)"),
+                (user_id, tenant_id, role))
+
+
+def login(cur, tenant_slug: str, username: str, password: str) -> dict | None:
+    """Return a new session for an active member of the tenant, else None.
+
+    The caller commits. Wrong tenant, unknown user, wrong password and inactive
+    user all return None, so a caller cannot tell which one failed.
+    """
+    cur.execute(_q(
+        "SELECT u.id, u.password_hash, u.is_active, t.id, m.role "
+        "FROM {c}.users u "
+        "LEFT JOIN {c}.tenants t ON t.slug = %s "
+        "LEFT JOIN {c}.memberships m ON m.user_id = u.id AND m.tenant_id = t.id "
+        "WHERE u.username = %s"), (tenant_slug, username))
+    row = cur.fetchone()
+    ok = verify_password(password, row[1] if row else _DUMMY_HASH)
+    if not (row and ok and row[2] and row[4]):
+        return None
+    user_id, tenant_id, role = row[0], row[3], row[4]
+    token = secrets.token_urlsafe(32)
+    cur.execute(_q(
+        "INSERT INTO {c}.sessions (token_hash, user_id, tenant_id, expires_at) "
+        "VALUES (%s, %s, %s, now() + make_interval(hours => %s)) RETURNING expires_at"),
+        (_token_hash(token), user_id, tenant_id, settings.SESSION_TTL_HOURS))
+    expires_at = cur.fetchone()[0]
+    return {"token": token, "expiresAt": expires_at.isoformat(), "username": username,
+            "tenant": tenant_slug, "role": role}
+
+
+def session_for(cur, token: str) -> dict | None:
+    """The live session for a token, with its user, tenant and role, else None."""
+    cur.execute(_q(
+        "SELECT u.id, u.username, t.id, t.slug, m.role, s.expires_at "
+        "FROM {c}.sessions s "
+        "JOIN {c}.users u ON u.id = s.user_id AND u.is_active "
+        "JOIN {c}.tenants t ON t.id = s.tenant_id "
+        "JOIN {c}.memberships m ON m.user_id = u.id AND m.tenant_id = t.id "
+        "WHERE s.token_hash = %s AND s.revoked_at IS NULL AND s.expires_at > now()"),
+        (_token_hash(token),))
+    row = cur.fetchone()
+    if not row:
+        return None
+    return {"userId": row[0], "username": row[1], "tenantId": row[2], "tenant": row[3],
+            "role": row[4], "expiresAt": row[5].isoformat()}
+
+
+def revoke(cur, token: str) -> None:
+    cur.execute(_q("UPDATE {c}.sessions SET revoked_at = now() "
+                   "WHERE token_hash = %s AND revoked_at IS NULL"), (_token_hash(token),))

docs/ENV_VARIABLES.md

@@ -39,6 +39,8 @@ Read by `api/config.py`. Set these in the shell before running
 | `API_POOL_GETCONN_TIMEOUT` | `5.0` | No | Seconds a request waits to borrow a database connection from the pool. Exceeded = 503 response. |
 | `CORS_ORIGINS` | `http://localhost:8080,http://localhost:5173,http://localhost:3000` | Yes | Comma-separated list of allowed CORS origins. Set to your frontend URL in production. |
 | `CSV_UPLOADS_SCHEMA` | `csv_uploads` | No | Schema where per-CSV tables and the file registry live. |
+| `CONTROL_SCHEMA` | `control` | No | Schema for tenants, users, memberships and sign-in sessions (M2, #92). |
+| `SESSION_TTL_HOURS` | `12` | No | How long a sign-in session lasts before the user must sign in again. |
 | `TE_SCHEMA` | `te_dev` | No | Schema holding the 12 fixed T&E tables. Switch to `te_test`, `te_staging`, or `te_prod` per environment. |
 | `MAX_UPLOAD_BYTES` | `52428800` (50 MB) | No | Maximum file size accepted by the upload endpoint. |
 

scripts/create_tenant_admin.py

@@ -0,0 +1,50 @@
+"""Create a tenant and its first tenant admin (M2a, #92).
+
+Run from the repo root with the API's PG* environment set:
+    python scripts/create_tenant_admin.py --tenant acme --name "Acme Pty Ltd" --username alice
+
+The password is read from TE_ADMIN_PASSWORD if set, otherwise prompted for.
+Brings the database to the current migration first.
+"""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from api import db  # noqa: E402
+from api.services import accounts  # noqa: E402
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--tenant", required=True, help="tenant slug: lowercase letters, digits, underscore")
+    ap.add_argument("--name", required=True, help="tenant display name")
+    ap.add_argument("--username", required=True)
+    args = ap.parse_args()
+    password = os.environ.get("TE_ADMIN_PASSWORD") or getpass.getpass("Password for the new admin: ")
+    if len(password) < 12:
+        print("Password must be at least 12 characters.", file=sys.stderr)
+        return 2
+    db.init_pool()
+    db.bootstrap()
+    try:
+        with db.Conn() as conn:
+            with conn.cursor() as cur:
+                tenant_id = accounts.create_tenant(cur, args.tenant, args.name)
+                user_id = accounts.create_user(cur, args.username, password)
+                accounts.add_membership(cur, user_id, tenant_id, "tenant_admin")
+            conn.commit()
+    finally:
+        db.close_pool()
+    print(f"Created tenant '{args.tenant}' with tenant admin '{args.username}'.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/lint.sh

@@ -17,6 +17,11 @@ PY_SRC=(
     alembic/versions/0001_baseline.py
     alembic/versions/0002_registry_unique.py
     alembic/versions/0003_import_counts.py
+    alembic/versions/0004_control_schema.py
+    api/auth.py
+    api/services/accounts.py
+    api/routers/auth_routes.py
+    scripts/create_tenant_admin.py
 )
 
 PY_TESTS=(
@@ -30,6 +35,7 @@ PY_TESTS=(
     tests/test_evals_tier_e_x.py
     tests/test_migrations.py
     tests/test_import_counts.py
+    tests/test_accounts.py
 )
 
 # Collect scripts/*.py dynamically