Phase 0 milestone 0.3: auth + JWT + iam.assurance_level claim

Lands local password auth with JWT access + refresh tokens, and starts
carrying YPMS-specific claims (`uid`, `roles`, `iam.assurance_level`,
`mfa_verified`) so downstream authorization can gate high-risk actions
above base RBAC without a DB round-trip. Follows the OWC auth pattern
verbatim except where noted.

Added:

- src/ypms/auth/security.py    hash_password / verify_password
                               (pbkdf2_sha256), create_access_token
                               with YPMS-specific claims,
                               create_refresh_token, decode_token
- src/ypms/auth/rate_limit.py  SlidingWindow (vendored from OWC),
                               login_limiter tuned to NIST AC-7 /
                               source RFP C.5.7.10: 5 failed attempts
                               per IP per 15 minutes → 429 lockout
- src/ypms/auth/schemas.py     LoginResponse, RefreshRequest,
                               RefreshResponse, ChangePasswordRequest,
                               UserRead
- src/ypms/auth/deps.py        get_current_user replaces the milestone
                               0.2 stub; rejects refresh tokens on
                               resource endpoints; refuses inactive users
- src/ypms/auth/router.py      POST /auth/login  (OAuth2 form; audit-logs
                               successful auth; DOES_SYSTEM_ADMIN role
                               upgrades assurance_level to `admin` in
                               the token)
                               POST /auth/refresh (issues fresh access;
                               refuses non-refresh tokens)
                               GET  /auth/me
                               POST /auth/change-password (self-service;
                               ≥12 char floor; audit-logged)
- src/ypms/main.py             Registers auth_router

Tests (43 passing, 93.47% coverage — up from 26 / 93.10%):

- tests/auth/test_security.py  Token claims shape, refresh-vs-access
                               separation, invalid-token handling
                               (5 tests)
- tests/auth/test_router.py    End-to-end via TestClient with a
                               dependency-overridden DB: login happy
                               path + wrong password + inactive user,
                               admin assurance upgrade, /me auth-gate,
                               refresh happy path + reject access token,
                               change-password full flow + length gate,
                               login lockout after 5 attempts (12 tests)

Test-infra fixes:

- tests/conftest.py            Env defaults moved from a session-scoped
                               fixture to module-import time so
                               `ypms.main` (which builds the app at
                               import) can be transitively imported by
                               test modules without ValidationError on
                               Settings.
                               `db` fixture switched to StaticPool +
                               check_same_thread=False so TestClient's
                               worker thread sees the same in-memory
                               SQLite DB the fixture set up.

Notes:

- OWC includes an /auth/register for external users; that's out of scope
  here — Phase 1 lands parental co-consent and youth/host registration,
  which needs its own PIA-driven flow. Keeping /auth minimal in Phase 0.
- MFA is captured in `User.mfa_verified` and JWT claim but not enforced
  yet; enforcement lands as an assurance-level gate in a later phase.
- Refresh flow returns only a new access token; refresh-token rotation
  is a later hardening step.

Advances YPMS-POAM-040 and unblocks milestone 0.4 (RBAC seed) and 0.5
(Registrant vertical slice).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/ypms/auth/deps.py

@@ -1,28 +1,45 @@
 """Authentication dependencies.
 
-Phase 0 milestone 0.2: placeholder `get_current_user` that always raises 401.
-Milestone 0.3 replaces this with a JWT-verifying implementation that also
-issues the `iam.assurance_level` claim on the token.
+Milestone 0.3: real `get_current_user` — validates the JWT, refuses refresh
+tokens on resource endpoints, loads the User and refuses if inactive.
 
-Kept as its own module so the RBAC engine and router factory can import it now
-and milestone 0.3's swap is a single-file change.
+Downstream authorization (RBAC and assurance-level gating) reads user.roles
+and user.assurance_level, so the returned User carries live values from the
+DB rather than the token snapshot — a role revocation or account
+deactivation takes effect on the next request, not on token expiry.
 """
 
 from __future__ import annotations
 
-from fastapi import HTTPException, status
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
 
+from ypms.auth.security import decode_token
+from ypms.db import get_db
 from ypms.users.models import User
 
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login", auto_error=False)
 
-def get_current_user() -> User:
-    """Placeholder. Milestone 0.3 lands real JWT verification.
 
-    Raises 401 for any route that goes through it. Tests that need to exercise
-    RBAC or routes without a real auth stack override this dependency with a
-    fixture-provided User via `app.dependency_overrides`.
-    """
+def get_current_user(
+    token: str | None = Depends(oauth2_scheme),
+    db: Session = Depends(get_db),
+) -> User:
+    if not token:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated")
+    try:
+        payload = decode_token(token)
+    except ValueError as e:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(e)) from e
+    if payload.get("type") == "refresh":
         raise HTTPException(
-        status.HTTP_401_UNAUTHORIZED,
-        detail={"error": "unauthenticated", "phase": "milestone 0.3 lands JWT auth"},
+            status.HTTP_401_UNAUTHORIZED, "refresh token cannot access resources"
         )
+    username = payload.get("sub")
+    if not username:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")
+    user = db.query(User).filter(User.username == username).one_or_none()
+    if not user or not user.is_active:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found or inactive")
+    return user

src/ypms/auth/rate_limit.py

@@ -0,0 +1,47 @@
+"""Tiny in-memory rate limiter.
+
+Vendored from OWC `auth/rate_limit.py` (snapshot 2026-04-20); unchanged.
+
+Not suitable for multi-process prod. Swap to Redis-backed once we're
+horizontally scaled (Phase 9+ / Azure deploy). Used today to cap login
+attempts per IP (NIST AC-7 / source RFP C.5.7.10).
+"""
+
+from __future__ import annotations
+
+import time
+from collections import deque
+from threading import Lock
+
+
+class SlidingWindow:
+    def __init__(self, limit: int, window_seconds: int):
+        self.limit = limit
+        self.window = window_seconds
+        self._hits: dict[str, deque[float]] = {}
+        self._lock = Lock()
+
+    def allow(self, key: str) -> bool:
+        now = time.time()
+        with self._lock:
+            q = self._hits.setdefault(key, deque())
+            # Evict entries outside the window.
+            while q and now - q[0] > self.window:
+                q.popleft()
+            if len(q) >= self.limit:
+                return False
+            q.append(now)
+            return True
+
+    def reset(self, key: str | None = None) -> None:
+        """Test helper: clear a single key or the whole window."""
+        with self._lock:
+            if key is None:
+                self._hits.clear()
+            else:
+                self._hits.pop(key, None)
+
+
+# NIST AC-7 / source RFP `C.5.7.10`: 5 failed login attempts triggers lockout.
+# We track by IP and use a longer window than OWC to keep the block honest.
+login_limiter = SlidingWindow(limit=5, window_seconds=900)  # 15 minutes

src/ypms/auth/router.py

@@ -0,0 +1,145 @@
+"""Auth surface: /auth/login, /auth/refresh, /auth/me, /auth/change-password.
+
+Source RFP requirements addressed here:
+- C.5.7.1  role-based authN/authZ (roles baked into JWT claim)
+- C.5.7.3  session timeout — enforced by access token TTL
+- C.5.7.7  min 12-alphanumeric identifier — pydantic constraint on register
+           lands in Phase 1; here we only accept credentials, not create them
+- C.5.7.8  user-established and changeable passwords → /change-password
+- C.5.7.9  FIPS 140-2/3 crypto — pbkdf2_sha256 + Key Vault (0.3 uses env-var
+           key; Key Vault swap-in is Phase 1)
+- C.5.7.10 5 failed attempts → lockout — enforced by login_limiter
+- C.5.7.38 unique user ID per user — enforced by User.username uniqueness
+
+NIST controls: IA-2, IA-5, IA-5(1), IA-8 (organizational side), AC-7, AC-11.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+
+from ypms.audit import service as audit
+from ypms.auth.deps import get_current_user
+from ypms.auth.rate_limit import login_limiter
+from ypms.auth.schemas import (
+    ChangePasswordRequest,
+    LoginResponse,
+    RefreshRequest,
+    RefreshResponse,
+    UserRead,
+)
+from ypms.auth.security import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+    hash_password,
+    verify_password,
+)
+from ypms.db import get_db
+from ypms.security.rbac import ADMIN_ROLE
+from ypms.users.models import (
+    ASSURANCE_ADMIN,
+    User,
+)
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+
+
+def _issue_access_token(user: User) -> str:
+    role_codes = [r.code for r in user.roles]
+    # DOES_SYSTEM_ADMIN implies admin assurance regardless of the persisted
+    # value — the super-role bypass already grants blanket authority and the
+    # token should reflect that so downstream assurance-level gates behave.
+    level = ASSURANCE_ADMIN if ADMIN_ROLE in role_codes else user.assurance_level
+    return create_access_token(
+        sub=user.username,
+        uid=user.id,
+        roles=role_codes,
+        assurance_level=level,
+        mfa_verified=user.mfa_verified,
+    )
+
+
+@router.post("/login", response_model=LoginResponse)
+def login(
+    request: Request,
+    form: OAuth2PasswordRequestForm = Depends(),
+    db: Session = Depends(get_db),
+) -> LoginResponse:
+    ip = request.client.host if request.client else "unknown"
+    if not login_limiter.allow(ip):
+        # NIST AC-7: cap unsuccessful logon attempts.
+        raise HTTPException(
+            status.HTTP_429_TOO_MANY_REQUESTS,
+            "too many login attempts; try again shortly",
+            headers={"Retry-After": "900"},
+        )
+    user = db.query(User).filter(User.username == form.username).one_or_none()
+    if not user or not verify_password(form.password, user.password_hash):
+        # Deliberately identical error surface so we don't leak whether the
+        # username exists (IA-6 authenticator feedback discipline).
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials")
+    if not user.is_active:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user inactive")
+
+    audit.record(
+        db, actor=user, action="login", entity="User", entity_id=user.id, ip=ip
+    )
+    db.commit()
+
+    return LoginResponse(
+        access_token=_issue_access_token(user),
+        refresh_token=create_refresh_token(sub=user.username, uid=user.id),
+    )
+
+
+@router.post("/refresh", response_model=RefreshResponse)
+def refresh(
+    body: RefreshRequest,
+    db: Session = Depends(get_db),
+) -> RefreshResponse:
+    try:
+        payload = decode_token(body.refresh_token)
+    except ValueError as e:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(e)) from e
+    if payload.get("type") != "refresh":
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not a refresh token")
+    username = payload.get("sub")
+    user = db.query(User).filter(User.username == username).one_or_none()
+    if not user or not user.is_active:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found or inactive")
+    return RefreshResponse(access_token=_issue_access_token(user))
+
+
+@router.get("/me", response_model=UserRead)
+def me(current: User = Depends(get_current_user)) -> User:
+    return current
+
+
+@router.post("/change-password", status_code=status.HTTP_204_NO_CONTENT)
+def change_password(
+    body: ChangePasswordRequest,
+    request: Request,
+    db: Session = Depends(get_db),
+    current: User = Depends(get_current_user),
+) -> None:
+    if not verify_password(body.current_password, current.password_hash):
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "current password incorrect")
+    if len(body.new_password) < 12:
+        # Source RFP C.5.7.45 leaves the exact policy to the org. 12 chars is
+        # the floor from C.5.7.7 (min identifier length) — treat the same for
+        # secrets. Complexity + rotation policy lands with the admin UI.
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "new password must be at least 12 characters")
+
+    current.password_hash = hash_password(body.new_password)
+    db.flush()
+
+    ip = request.client.host if request.client else "unknown"
+    audit.record(
+        db, actor=current, action="change_password",
+        entity="User", entity_id=current.id, ip=ip,
+    )
+    db.commit()
+    return None

src/ypms/auth/schemas.py

@@ -0,0 +1,40 @@
+"""Pydantic schemas for the auth surface."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class LoginResponse(BaseModel):
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+
+
+class RefreshRequest(BaseModel):
+    refresh_token: str
+
+
+class RefreshResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+
+
+class ChangePasswordRequest(BaseModel):
+    current_password: str
+    new_password: str
+
+
+class UserRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    username: str
+    email: str | None = None
+    full_name: str | None = None
+    is_active: bool
+    mfa_verified: bool
+    assurance_level: str
+    created_at: datetime

src/ypms/auth/security.py

@@ -0,0 +1,82 @@
+"""Password hashing + JWT create/verify.
+
+Vendored from OWC `auth/security.py` (snapshot 2026-04-20). YPMS additions:
+`create_access_token()` bakes in YPMS-specific claims — `uid`, `roles`,
+`iam.assurance_level`, `mfa_verified` — so downstream authorization can gate
+high-risk actions above base RBAC (see ARCHITECTURE.md §4.5, §8.2).
+
+`iam.assurance_level` values are defined in `ypms.users.models` and follow
+NIST 800-63 IAL/AAL semantics: `pre_application` < `standard` <
+`payroll_authorized` < `staff` < `admin`. Payroll-touching endpoints will
+require `payroll_authorized` or higher (Phase 7+).
+
+pbkdf2_sha256 not bcrypt — see ARCHITECTURE.md §4.6.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+from ypms.settings import get_settings
+
+_pwd = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
+
+
+def hash_password(pw: str) -> str:
+    h: str = _pwd.hash(pw)
+    return h
+
+
+def verify_password(pw: str, hashed: str) -> bool:
+    return bool(_pwd.verify(pw, hashed))
+
+
+def create_access_token(
+    *,
+    sub: str,
+    uid: int,
+    roles: list[str],
+    assurance_level: str,
+    mfa_verified: bool,
+) -> str:
+    s = get_settings()
+    now = datetime.now(UTC)
+    payload: dict[str, Any] = {
+        "sub": sub,
+        "uid": uid,
+        "roles": roles,
+        "iam.assurance_level": assurance_level,
+        "mfa_verified": mfa_verified,
+        "iat": int(now.timestamp()),
+        "exp": int((now + timedelta(minutes=s.access_token_ttl_minutes)).timestamp()),
+        "type": "access",
+    }
+    token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg)
+    return token
+
+
+def create_refresh_token(*, sub: str, uid: int) -> str:
+    s = get_settings()
+    now = datetime.now(UTC)
+    payload: dict[str, Any] = {
+        "sub": sub,
+        "uid": uid,
+        "iat": int(now.timestamp()),
+        "exp": int((now + timedelta(days=s.refresh_token_ttl_days)).timestamp()),
+        "type": "refresh",
+    }
+    token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg)
+    return token
+
+
+def decode_token(token: str) -> dict[str, Any]:
+    s = get_settings()
+    try:
+        payload: dict[str, Any] = jwt.decode(token, s.jwt_secret, algorithms=[s.jwt_alg])
+    except JWTError as e:
+        raise ValueError("invalid token") from e
+    return payload

src/ypms/main.py

@@ -12,6 +12,7 @@ from fastapi import FastAPI, status
 from fastapi.middleware.cors import CORSMiddleware
 from sqlalchemy import text
 
+from ypms.auth.router import router as auth_router
 from ypms.db import engine
 from ypms.logging import configure_logging, get_logger
 from ypms.middleware import RequestIdMiddleware
@@ -40,6 +41,8 @@ def create_app() -> FastAPI:
         expose_headers=["X-Request-ID"],
     )
 
+    app.include_router(auth_router)
+
     @app.get("/healthz", tags=["system"])
     def healthz() -> dict[str, str]:
         """Liveness — process is up. No dependency checks."""

tests/auth/test_router.py

@@ -0,0 +1,253 @@
+"""End-to-end auth flow via the FastAPI TestClient.
+
+Uses the shared SQLite in-memory DB by wiring the `get_db` dependency to the
+per-test session, then seeding a user and driving login/refresh/me/
+change-password.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from ypms.auth.rate_limit import login_limiter
+from ypms.auth.security import (
+    create_access_token,
+    create_refresh_token,
+    hash_password,
+)
+from ypms.db import get_db
+from ypms.main import create_app
+from ypms.security.rbac import ADMIN_ROLE
+from ypms.users.models import (
+    ASSURANCE_ADMIN,
+    ASSURANCE_STANDARD,
+    Role,
+    User,
+    UserRole,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_rate_limits() -> Iterator[None]:
+    login_limiter.reset()
+    yield
+    login_limiter.reset()
+
+
+@pytest.fixture
+def staff_user(db: Session) -> User:
+    user = User(
+        username="staff.user",
+        email="staff@dc.gov",
+        full_name="Staff User",
+        password_hash=hash_password("correct-horse-battery"),
+        is_active=True,
+        assurance_level=ASSURANCE_STANDARD,
+    )
+    db.add(user)
+    db.flush()
+    db.refresh(user)
+    return user
+
+
+@pytest.fixture
+def admin_user(db: Session) -> User:
+    role = Role(code=ADMIN_ROLE, description="super-role")
+    db.add(role)
+    db.flush()
+    user = User(
+        username="chris",
+        email="chris@dc.gov",
+        full_name="Chris Tonjes",
+        password_hash=hash_password("correct-horse-battery"),
+        is_active=True,
+        assurance_level=ASSURANCE_STANDARD,  # role bump upgrades to admin
+    )
+    db.add(user)
+    db.flush()
+    db.add(UserRole(user_id=user.id, role_id=role.id))
+    db.flush()
+    db.refresh(user)
+    return user
+
+
+@pytest.fixture
+def app_client(db: Session) -> Iterator[TestClient]:
+    """A TestClient wired to the per-test SQLite session via dep override."""
+    app = create_app()
+
+    def _override_get_db() -> Iterator[Session]:
+        yield db
+
+    app.dependency_overrides[get_db] = _override_get_db
+    with TestClient(app) as c:
+        yield c
+    app.dependency_overrides.clear()
+
+
+def test_login_wrong_password_returns_401(app_client: TestClient, staff_user: User) -> None:
+    r = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "wrong"},
+    )
+    assert r.status_code == 401
+    assert r.json()["detail"] == "invalid credentials"
+
+
+def test_login_success_returns_tokens(app_client: TestClient, staff_user: User) -> None:
+    r = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["token_type"] == "bearer"
+    assert body["access_token"]
+    assert body["refresh_token"]
+
+
+def test_admin_login_upgrades_assurance_level(
+    app_client: TestClient, admin_user: User
+) -> None:
+    from ypms.auth.security import decode_token
+
+    r = app_client.post(
+        "/auth/login",
+        data={"username": admin_user.username, "password": "correct-horse-battery"},
+    )
+    assert r.status_code == 200
+    claims = decode_token(r.json()["access_token"])
+    assert claims["iam.assurance_level"] == ASSURANCE_ADMIN
+    assert ADMIN_ROLE in claims["roles"]
+
+
+def test_me_requires_authentication(app_client: TestClient) -> None:
+    r = app_client.get("/auth/me")
+    assert r.status_code == 401
+
+
+def test_me_returns_current_user(app_client: TestClient, staff_user: User) -> None:
+    login = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    access = login.json()["access_token"]
+    r = app_client.get("/auth/me", headers={"Authorization": f"Bearer {access}"})
+    assert r.status_code == 200
+    assert r.json()["username"] == staff_user.username
+    assert r.json()["email"] == "staff@dc.gov"
+
+
+def test_refresh_returns_new_access_token(app_client: TestClient, staff_user: User) -> None:
+    login = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    refresh = login.json()["refresh_token"]
+    r = app_client.post("/auth/refresh", json={"refresh_token": refresh})
+    assert r.status_code == 200
+    assert r.json()["access_token"]
+    assert "refresh_token" not in r.json()
+
+
+def test_refresh_rejects_access_token(app_client: TestClient, staff_user: User) -> None:
+    access = create_access_token(
+        sub=staff_user.username,
+        uid=staff_user.id,
+        roles=[],
+        assurance_level=ASSURANCE_STANDARD,
+        mfa_verified=False,
+    )
+    r = app_client.post("/auth/refresh", json={"refresh_token": access})
+    assert r.status_code == 401
+
+
+def test_me_rejects_refresh_token(app_client: TestClient, staff_user: User) -> None:
+    refresh = create_refresh_token(sub=staff_user.username, uid=staff_user.id)
+    r = app_client.get("/auth/me", headers={"Authorization": f"Bearer {refresh}"})
+    assert r.status_code == 401
+
+
+def test_change_password_flow(app_client: TestClient, staff_user: User) -> None:
+    login = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    access = login.json()["access_token"]
+    r = app_client.post(
+        "/auth/change-password",
+        headers={"Authorization": f"Bearer {access}"},
+        json={
+            "current_password": "correct-horse-battery",
+            "new_password": "totally-different-passphrase",
+        },
+    )
+    assert r.status_code == 204
+
+    # Old password no longer works.
+    r2 = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    assert r2.status_code == 401
+
+    # New password does.
+    r3 = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "totally-different-passphrase"},
+    )
+    assert r3.status_code == 200
+
+
+def test_change_password_rejects_short_new_password(
+    app_client: TestClient, staff_user: User
+) -> None:
+    login = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    access = login.json()["access_token"]
+    r = app_client.post(
+        "/auth/change-password",
+        headers={"Authorization": f"Bearer {access}"},
+        json={"current_password": "correct-horse-battery", "new_password": "short"},
+    )
+    assert r.status_code == 400
+
+
+def test_change_password_rejects_wrong_current(
+    app_client: TestClient, staff_user: User
+) -> None:
+    login = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    access = login.json()["access_token"]
+    r = app_client.post(
+        "/auth/change-password",
+        headers={"Authorization": f"Bearer {access}"},
+        json={"current_password": "wrong", "new_password": "long-enough-password"},
+    )
+    assert r.status_code == 401
+
+
+def test_login_lockout_after_five_failed_attempts(
+    app_client: TestClient, staff_user: User
+) -> None:
+    for _ in range(5):
+        r = app_client.post(
+            "/auth/login",
+            data={"username": staff_user.username, "password": "wrong"},
+        )
+        assert r.status_code == 401
+    # 6th attempt (any credentials) is rate-limited.
+    r = app_client.post(
+        "/auth/login",
+        data={"username": staff_user.username, "password": "correct-horse-battery"},
+    )
+    assert r.status_code == 429
+    assert r.headers["Retry-After"] == "900"

tests/auth/test_security.py

@@ -0,0 +1,59 @@
+"""Password + JWT primitives."""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from ypms.auth.security import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+    hash_password,
+    verify_password,
+)
+
+
+def test_hash_password_returns_pbkdf2_prefix() -> None:
+    h = hash_password("hunter2hunter2")
+    assert h.startswith("$pbkdf2-sha256$")
+
+
+def test_verify_password_true_on_match_false_on_mismatch() -> None:
+    h = hash_password("correcthorsebattery")
+    assert verify_password("correcthorsebattery", h) is True
+    assert verify_password("wrong", h) is False
+
+
+def test_access_token_carries_ypms_claims() -> None:
+    tok = create_access_token(
+        sub="chris",
+        uid=1,
+        roles=["DOES_SYSTEM_ADMIN"],
+        assurance_level="admin",
+        mfa_verified=False,
+    )
+    claims = decode_token(tok)
+    assert claims["sub"] == "chris"
+    assert claims["uid"] == 1
+    assert claims["roles"] == ["DOES_SYSTEM_ADMIN"]
+    assert claims["iam.assurance_level"] == "admin"
+    assert claims["mfa_verified"] is False
+    assert claims["type"] == "access"
+    assert claims["exp"] > claims["iat"] >= int(time.time()) - 5
+
+
+def test_refresh_token_has_type_refresh_and_carries_uid() -> None:
+    tok = create_refresh_token(sub="chris", uid=42)
+    claims = decode_token(tok)
+    assert claims["type"] == "refresh"
+    assert claims["sub"] == "chris"
+    assert claims["uid"] == 42
+    assert "roles" not in claims
+    assert "iam.assurance_level" not in claims
+
+
+def test_decode_invalid_token_raises() -> None:
+    with pytest.raises(ValueError):
+        decode_token("not-a-real-jwt")

tests/conftest.py

@@ -3,6 +3,10 @@
 Uses SQLite for speed (see ARCHITECTURE.md §4.8). Tests exercise the framework
 against an in-memory Postgres-shaped SQLAlchemy Base — no Alembic migrations
 in the test path; models create their tables from metadata.
+
+Environment defaults are set at *module import* time (not via fixture) so
+they are in place before pytest collects any test module that transitively
+imports `ypms.main` (which instantiates the FastAPI app at import time).
 """
 
 from __future__ import annotations
@@ -10,22 +14,18 @@ from __future__ import annotations
 import os
 from collections.abc import Iterator
 
-import pytest
-from fastapi.testclient import TestClient
-from sqlalchemy import create_engine
-from sqlalchemy.orm import Session, sessionmaker
-
-
-@pytest.fixture(scope="session", autouse=True)
-def _test_env() -> Iterator[None]:
-    """Set minimum env vars required for `Settings()` to construct in tests."""
 os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
 os.environ.setdefault("JWT_SECRET", "test-secret")
 os.environ.setdefault(
     "COLUMN_ENCRYPTION_KEY", "test-col-enc-key-32bytes-______________"
 )
 os.environ.setdefault("ENV", "test")
-    yield
+
+import pytest  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+from sqlalchemy import create_engine  # noqa: E402
+from sqlalchemy.orm import Session, sessionmaker  # noqa: E402
+from sqlalchemy.pool import StaticPool  # noqa: E402
 
 
 @pytest.fixture
@@ -41,7 +41,17 @@ def db() -> Iterator[Session]:
     import ypms.users.models  # noqa: F401
     from ypms.db import Base
 
-    engine = create_engine("sqlite:///:memory:", future=True)
+    # StaticPool + check_same_thread=False share the same in-memory DB across
+    # threads. TestClient runs handlers in a worker thread, so without this
+    # the fixture-scoped session and the request-scoped session see different
+    # empty databases (SADeprecationWarning: SQLite objects created in a
+    # thread can only be used in that same thread).
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+        future=True,
+    )
     Base.metadata.create_all(engine)
     SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
     session = SessionLocal()