অথেন্টিকেশন ও অথরাইজেশন

mkdir -p ~/projects/auth-api
cd ~/projects/auth-api
uv init --bare -q
uv add -q "fastapi[standard]" pyjwt "pwdlib[argon2]"
uv add -q --dev pytest httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pyjwt "pwdlib[argon2]" pytest httpx2
mkdir -p app tests
touch app/__init__.py
cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF
export JWT_SECRET="tutorial-only-secret-change-me-0123456789"
export BASE=http://localhost:8000

from pwdlib import PasswordHash

passwords = PasswordHash.recommended()          # Argon2id with sensible parameters

hashed = passwords.hash("correct-horse-1")
print(hashed.split("$")[1:3])                   # algorithm and version, then salt and hash
print(hashed.startswith("$argon2id$v=19$"))
print(passwords.verify("correct-horse-1", hashed), passwords.verify("wrong-guess", hashed))
print(hashed == passwords.hash("correct-horse-1"))

['argon2id', 'v=19']
True
True False
False

# file: app/security.py
"""Passwords, tokens, API keys and the login limiter. No FastAPI in here."""
import hashlib
import math
import os
import secrets
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

ALGORITHM = "HS256"
ISSUER = "bitbyte-auth-demo"
ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=7)

passwords = PasswordHash.recommended()
DUMMY_HASH = passwords.hash("not-a-real-password")   # for unknown usernames, see login


def secret_key() -> str:
    return os.environ["JWT_SECRET"]


def make_token(sub: str, kind: str, *, family: str | None = None, now: datetime | None = None,
               ttl: timedelta | None = None, jti: str | None = None) -> str:
    """Sign a token. `now`, `ttl` and `jti` are parameters so examples and tests can fix them."""
    now = now or datetime.now(timezone.utc)
    ttl = ttl or (ACCESS_TTL if kind == "access" else REFRESH_TTL)
    claims = {
        "iss": ISSUER,                       # who issued it
        "sub": sub,                          # whose it is (the user id, as a string)
        "typ": kind,                         # "access" or "refresh": never accept one as the other
        "fam": family or uuid.uuid4().hex,   # the login session this token belongs to
        "jti": jti or uuid.uuid4().hex,      # unique id of this token
        "iat": now,                          # issued at
        "exp": now + ttl,                    # expires at
    }
    return jwt.encode(claims, secret_key(), algorithm=ALGORITHM)


def decode_token(token: str, kind: str) -> dict:
    """Verify signature, algorithm, issuer, expiry and token type. Raises jwt.InvalidTokenError."""
    claims = jwt.decode(token, secret_key(), algorithms=[ALGORITHM], issuer=ISSUER,
                        options={"require": ["exp", "iat", "sub", "jti"]})
    if claims.get("typ") != kind:
        raise jwt.InvalidTokenError(f"expected a {kind} token")
    return claims


def new_api_key() -> str:
    return "bb_" + secrets.token_urlsafe(24)


def hash_api_key(key: str) -> str:
    """API keys are long and random, so a fast hash is fine (unlike passwords)."""
    return hashlib.sha256(key.encode()).hexdigest()


class LoginLimiter:
    """Block a key after `max_failures` failed logins within `window` seconds."""

    def __init__(self, max_failures: int = 5, window: float = 60.0,
                 clock: Callable[[], float] = time.monotonic) -> None:
        self.max_failures, self.window, self.clock = max_failures, window, clock
        self._failures: dict[str, list[float]] = {}

    def retry_after(self, key: str) -> int:
        """Seconds the caller must wait, or 0 if it may try."""
        now = self.clock()
        recent = [t for t in self._failures.get(key, []) if now - t < self.window]
        self._failures[key] = recent
        if len(recent) >= self.max_failures:
            return math.ceil(self.window - (now - recent[0]))
        return 0

    def record_failure(self, key: str) -> None:
        self._failures.setdefault(key, []).append(self.clock())

    def reset(self, key: str) -> None:
        self._failures.pop(key, None)

import base64
import json
from datetime import datetime, timezone

from app.security import make_token


def part(segment: str) -> dict:
    return json.loads(base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)))


FIXED = datetime(2026, 1, 15, 4, 0, tzinfo=timezone.utc)
token = make_token("2", "access", family="demo-session", jti="demo-token", now=FIXED)
head, payload, signature = token.split(".")
print(part(head))
print(part(payload))
print(len(signature), "characters of signature")

{'alg': 'HS256', 'typ': 'JWT'}
{'iss': 'bitbyte-auth-demo', 'sub': '2', 'typ': 'access', 'fam': 'demo-session', 'jti': 'demo-token', 'iat': 1768449600, 'exp': 1768450500}
43 characters of signature

import jwt
from datetime import timedelta

from app.security import decode_token

forged_payload = base64.urlsafe_b64encode(json.dumps({**part(payload), "sub": "1"}).encode()).rstrip(b"=").decode()
now = datetime.now(timezone.utc)
cases = {
    "valid": make_token("2", "access"),
    "expired": make_token("2", "access", now=now - timedelta(hours=1)),
    "payload edited": f"{head}.{forged_payload}.{signature}",
    "alg none": jwt.encode({"sub": "1", "typ": "access", "iss": "bitbyte-auth-demo", "jti": "x",
                            "iat": now, "exp": now + timedelta(hours=1)}, key=None, algorithm="none"),
    "other secret": jwt.encode({"sub": "1"}, "an-attacker-secret-that-is-long-enough", algorithm="HS256"),
    "refresh as access": make_token("2", "refresh"),
}
for name, candidate in cases.items():
    try:
        decode_token(candidate, "access")
        print(f"{name:18} accepted")
    except jwt.InvalidTokenError as exc:
        print(f"{name:18} rejected: {type(exc).__name__}")

valid              accepted
expired            rejected: ExpiredSignatureError
payload edited     rejected: InvalidSignatureError
alg none           rejected: InvalidAlgorithmError
other secret       rejected: InvalidSignatureError
refresh as access  rejected: InvalidTokenError

# file: app/store.py
"""In-memory data for the demo. Swap for a database and Redis in a real service."""
from app.security import LoginLimiter, passwords


class Store:
    def __init__(self) -> None:
        self.users: dict[int, dict] = {}
        self.chats: dict[int, dict] = {}
        self.api_keys: dict[str, str] = {}        # sha256(key) -> service name
        self.used_refresh: set[str] = set()       # refresh jti values that were already exchanged
        self.revoked_families: set[str] = set()   # login sessions that have been ended
        self.limiter = LoginLimiter()

    def add_user(self, username: str, password: str, role: str = "user") -> dict:
        row = {"id": len(self.users) + 1, "username": username,
               "password_hash": passwords.hash(password), "role": role}
        self.users[row["id"]] = row
        return row

    def by_name(self, username: str) -> dict | None:
        return next((u for u in self.users.values() if u["username"] == username), None)

# file: app/main.py
import os
import uuid
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, ConfigDict, Field

from app.security import DUMMY_HASH, decode_token, hash_api_key, make_token, passwords
from app.store import Store

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


class RegisterIn(BaseModel):
    model_config = ConfigDict(extra="forbid")
    username: str = Field(min_length=3, max_length=30, pattern=r"^[a-z0-9_]+$")
    password: str = Field(min_length=8, max_length=128)


class UserOut(BaseModel):                      # no password_hash here, on purpose
    id: int
    username: str
    role: str


class TokenPair(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"


class RefreshIn(BaseModel):
    refresh_token: str


class ChatIn(BaseModel):
    title: str = Field(min_length=1, max_length=80)


class ChatOut(BaseModel):
    id: int
    title: str
    owner_id: int


def unauthorized(detail: str = "Could not validate credentials") -> HTTPException:
    return HTTPException(401, detail, headers={"WWW-Authenticate": "Bearer"})


def get_store(request: Request) -> Store:
    return request.app.state.store


StoreDep = Annotated[Store, Depends(get_store)]


def issue_pair(user: dict, family: str | None = None) -> TokenPair:
    family = family or uuid.uuid4().hex
    return TokenPair(access_token=make_token(str(user["id"]), "access", family=family),
                     refresh_token=make_token(str(user["id"]), "refresh", family=family))


def authenticate(form: OAuth2PasswordRequestForm, request: Request, store: Store) -> dict:
    """Check a username and password, with a limit on failed attempts."""
    key = f"{form.username}|{request.client.host if request.client else '?'}"
    wait = store.limiter.retry_after(key)
    if wait:
        raise HTTPException(429, "Too many failed logins, try again later", headers={"Retry-After": str(wait)})
    user = store.by_name(form.username)
    # Always run one hash check, even for an unknown user, so response time does not reveal which names exist.
    password_ok = passwords.verify(form.password, user["password_hash"] if user else DUMMY_HASH)
    if user is None or not password_ok:
        store.limiter.record_failure(key)
        raise unauthorized("Incorrect username or password")
    store.limiter.reset(key)
    return user


def current_claims(token: Annotated[str, Depends(oauth2_scheme)], store: StoreDep) -> dict:
    try:
        claims = decode_token(token, "access")
    except jwt.ExpiredSignatureError:
        raise unauthorized("Token expired") from None
    except jwt.InvalidTokenError:
        raise unauthorized() from None
    if claims["fam"] in store.revoked_families:
        raise unauthorized("Session ended")
    return claims


def current_user(claims: Annotated[dict, Depends(current_claims)], store: StoreDep) -> dict:
    user = store.users.get(int(claims["sub"]))
    if user is None:
        raise unauthorized()
    return user


CurrentUser = Annotated[dict, Depends(current_user)]


def require_role(*roles: str):
    def checker(user: CurrentUser) -> dict:
        if user["role"] not in roles:
            raise HTTPException(403, "You do not have permission to do this")
        return user
    return checker


def owned_chat(chat_id: int, user: CurrentUser, store: StoreDep) -> dict:
    """Resource-level check: the chat must exist AND belong to the caller (or the caller is an admin)."""
    chat = store.chats.get(chat_id)
    if chat is None or (chat["owner_id"] != user["id"] and user["role"] != "admin"):
        raise HTTPException(404, "Chat not found")      # same answer for "missing" and "not yours"
    return chat


def service_name(store: StoreDep, x_api_key: Annotated[str | None, Header()] = None) -> str:
    name = store.api_keys.get(hash_api_key(x_api_key)) if x_api_key else None
    if name is None:
        raise HTTPException(401, "Invalid or missing API key")
    return name


def create_app() -> FastAPI:
    app = FastAPI(title="BitByte Auth Demo")
    store = Store()
    store.add_user("root", "root-pass-1234", role="admin")      # demo only: create admins with a CLI command
    if key := os.environ.get("INGEST_API_KEY"):
        store.api_keys[hash_api_key(key)] = "ingest-service"
    app.state.store = store

    @app.post("/auth/register", status_code=201)
    def register(data: RegisterIn, store: StoreDep) -> UserOut:
        if store.by_name(data.username):
            raise HTTPException(409, "Username already taken")
        return store.add_user(data.username, data.password)

    @app.post("/auth/login")
    def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], request: Request, store: StoreDep) -> TokenPair:
        return issue_pair(authenticate(form, request, store))

    @app.post("/auth/refresh")
    def refresh(data: RefreshIn, store: StoreDep) -> TokenPair:
        try:
            claims = decode_token(data.refresh_token, "refresh")
        except jwt.InvalidTokenError:
            raise unauthorized("Invalid refresh token") from None
        if claims["fam"] in store.revoked_families:
            raise unauthorized("Session ended")
        if claims["jti"] in store.used_refresh:
            store.revoked_families.add(claims["fam"])      # replay: someone holds a stolen copy, end the session
            raise unauthorized("Refresh token reuse detected, log in again")
        user = store.users.get(int(claims["sub"]))
        if user is None:
            raise unauthorized("Invalid refresh token")
        store.used_refresh.add(claims["jti"])              # each refresh token works exactly once
        return issue_pair(user, family=claims["fam"])

    @app.post("/auth/logout", status_code=204)
    def logout(claims: Annotated[dict, Depends(current_claims)], store: StoreDep) -> None:
        store.revoked_families.add(claims["fam"])

    @app.get("/me")
    def me(user: CurrentUser) -> UserOut:
        return user

    @app.get("/admin/users", dependencies=[Depends(require_role("admin"))])
    def list_users(store: StoreDep) -> list[UserOut]:
        return list(store.users.values())

    @app.post("/chats", status_code=201)
    def create_chat(data: ChatIn, user: CurrentUser, store: StoreDep) -> ChatOut:
        chat = {"id": len(store.chats) + 1, "title": data.title, "owner_id": user["id"]}
        store.chats[chat["id"]] = chat
        return chat

    @app.get("/chats/{chat_id}")
    def get_chat(chat: Annotated[dict, Depends(owned_chat)]) -> ChatOut:
        return chat

    @app.post("/ingest")
    def ingest(service: Annotated[str, Depends(service_name)]) -> dict[str, str]:
        return {"accepted": "yes", "service": service}

    return app


app = create_app()

export INGEST_API_KEY=$(uv run python -c "from app.security import new_api_key; print(new_api_key())")
echo "${INGEST_API_KEY:0:3}... (${#INGEST_API_KEY} characters)"
uv run fastapi run --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 $BASE/docs -o /dev/null -w '%{http_code}\n'

bb_... (35 characters)
200

curl -s -w ' %{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
  -d '{"username": "nadia", "password": "correct-horse-1"}'
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
  -d '{"username": "nadia", "password": "another-pass-9"}'
curl -s -o /dev/null -w '%{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
  -d '{"username": "x", "password": "short", "is_admin": true}'

{"id":2,"username":"nadia","role":"user"} 201
{"detail":"Username already taken"} 409
422

curl -s -w ' %{http_code}\n' -X POST $BASE/auth/login -d 'username=nadia' -d 'password=wrong-password'
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/login -d 'username=nobody' -d 'password=wrong-password'
curl -s -X POST $BASE/auth/login -d 'username=nadia' -d 'password=correct-horse-1' > login.json
export ACCESS=$(python -c "import json; print(json.load(open('login.json'))['access_token'])")
export REFRESH=$(python -c "import json; print(json.load(open('login.json'))['refresh_token'])")
python -c "import json; d = json.load(open('login.json')); print(sorted(d), d['token_type'], d['access_token'].count('.'))"

{"detail":"Incorrect username or password"} 401
{"detail":"Incorrect username or password"} 401
['access_token', 'refresh_token', 'token_type'] bearer 2

curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer $ACCESS"
curl -s -i $BASE/me | tr -d '\r' | grep -iE '^(HTTP|www-authenticate)|detail'
curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer ${ACCESS}x"

{"id":2,"username":"nadia","role":"user"} 200
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
{"detail":"Not authenticated"}
{"detail":"Could not validate credentials"} 401

curl -s -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$REFRESH\"}" > rotated.json
export NEW_REFRESH=$(python -c "import json; print(json.load(open('rotated.json'))['refresh_token'])")
python -c "import os; print('new refresh token differs:', os.environ['NEW_REFRESH'] != os.environ['REFRESH'])"
echo "--- replay the OLD refresh token"
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$REFRESH\"}"
echo "--- the NEW one, and the access token, are dead too"
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$NEW_REFRESH\"}"
curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer $ACCESS"

new refresh token differs: True
--- replay the OLD refresh token
{"detail":"Refresh token reuse detected, log in again"} 401
--- the NEW one, and the access token, are dead too
{"detail":"Session ended"} 401
{"detail":"Session ended"} 401

# file: login.sh
# usage: sh login.sh USERNAME PASSWORD   prints an access token
curl -s -X POST "$BASE/auth/login" -d "username=$1" -d "password=$2" | python -c "import sys, json; print(json.load(sys.stdin)['access_token'])"

export NADIA=$(sh login.sh nadia correct-horse-1)
export ROOT=$(sh login.sh root root-pass-1234)
curl -s -o /dev/null -X POST $BASE/auth/register -H 'Content-Type: application/json' -d '{"username": "tanvir", "password": "tanvir-pass-77"}'
export TANVIR=$(sh login.sh tanvir tanvir-pass-77)

curl -s -w ' %{http_code}\n' -X POST $BASE/chats -H "Authorization: Bearer $NADIA" -H 'Content-Type: application/json' -d '{"title": "Refund policy questions"}'
for who in NADIA TANVIR ROOT; do
  echo -n "$who reads chat 1: "
  curl -s -w ' %{http_code}\n' $BASE/chats/1 -H "Authorization: Bearer ${!who}"
done
echo -n "TANVIR reads chat 99: "
curl -s -w ' %{http_code}\n' $BASE/chats/99 -H "Authorization: Bearer $TANVIR"
echo "--- roles"
curl -s -w ' %{http_code}\n' $BASE/admin/users -H "Authorization: Bearer $NADIA"
curl -s -w ' %{http_code}\n' $BASE/admin/users -H "Authorization: Bearer $ROOT"

{"id":1,"title":"Refund policy questions","owner_id":2} 201
NADIA reads chat 1: {"id":1,"title":"Refund policy questions","owner_id":2} 200
TANVIR reads chat 1: {"detail":"Chat not found"} 404
ROOT reads chat 1: {"id":1,"title":"Refund policy questions","owner_id":2} 200
TANVIR reads chat 99: {"detail":"Chat not found"} 404
--- roles
{"detail":"You do not have permission to do this"} 403
[{"id":1,"username":"root","role":"admin"},{"id":2,"username":"nadia","role":"user"},{"id":3,"username":"tanvir","role":"user"}] 200

curl -s -w ' %{http_code}\n' -X POST $BASE/ingest
curl -s -w ' %{http_code}\n' -X POST $BASE/ingest -H "X-API-Key: $INGEST_API_KEY"
curl -s -w ' %{http_code}\n' -X POST $BASE/ingest -H "X-API-Key: $NADIA"

{"detail":"Invalid or missing API key"} 401
{"accepted":"yes","service":"ingest-service"} 200
{"detail":"Invalid or missing API key"} 401

from fastapi import Response

response = Response()
response.set_cookie("access_token", "<token>", max_age=900, httponly=True, secure=True, samesite="lax", path="/")
print(response.headers["set-cookie"])

access_token="<token>"; HttpOnly; Max-Age=900; Path=/; SameSite=lax; Secure

import jwt
from jwt import PyJWKClient

jwks = PyJWKClient("https://accounts.example.com/.well-known/jwks.json")      # caches keys


def verify_id_token(id_token: str, *, nonce: str) -> dict:
    signing_key = jwks.get_signing_key_from_jwt(id_token)
    claims = jwt.decode(id_token, signing_key, algorithms=["RS256"],
                        audience="my-client-id", issuer="https://accounts.example.com")
    if claims.get("nonce") != nonce:
        raise jwt.InvalidTokenError("nonce mismatch")
    return claims

for i in 1 2 3 4 5 6 7; do
  curl -s -o /dev/null -w '%{http_code} ' -X POST $BASE/auth/login -d 'username=mallory' -d "password=guess$i"
done; echo
curl -s -i -X POST $BASE/auth/login -d 'username=mallory' -d 'password=guess8' | tr -d '\r' | grep -icE '^retry-after'
curl -s -o /dev/null -w 'nadia can still log in: %{http_code}\n' -X POST $BASE/auth/login -d 'username=nadia' -d 'password=correct-horse-1'

401 401 401 401 401 429 429 
1
nadia can still log in: 200

kill "$(cat server.pid)"

# file: tests/test_auth.py
import os
from datetime import datetime, timedelta, timezone

import jwt
import pytest
from fastapi.testclient import TestClient

os.environ.setdefault("JWT_SECRET", "tutorial-only-secret-change-me-0123456789")

from app.main import create_app  # noqa: E402
from app.security import make_token  # noqa: E402


@pytest.fixture
def client(monkeypatch):
    monkeypatch.setenv("INGEST_API_KEY", "bb_test-key")
    with TestClient(create_app()) as tc:
        yield tc


def login(client, username, password="correct-horse-1", register=True):
    if register:
        client.post("/auth/register", json={"username": username, "password": password})
    return client.post("/auth/login", data={"username": username, "password": password})


def bearer(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def test_register_login_and_me(client):
    assert login(client, "nadia").status_code == 200
    token = login(client, "nadia", register=False).json()["access_token"]
    me = client.get("/me", headers=bearer(token))
    assert me.status_code == 200
    assert me.json() == {"id": 2, "username": "nadia", "role": "user"}      # no password_hash


def test_wrong_password_and_unknown_user_look_the_same(client):
    login(client, "nadia")
    wrong = login(client, "nadia", password="not-the-password", register=False)
    unknown = login(client, "nobody", register=False)
    assert wrong.status_code == unknown.status_code == 401
    assert wrong.json() == unknown.json()


def bad_token(kind: str) -> str:
    now = datetime.now(timezone.utc)
    claims = {"iss": "bitbyte-auth-demo", "sub": "1", "typ": "access", "fam": "f", "jti": "j",
              "iat": now, "exp": now + timedelta(hours=1)}
    return {
        "expired": make_token("1", "access", now=now - timedelta(hours=1)),
        "other_secret": jwt.encode(claims, "an-attacker-secret-that-is-long-enough", algorithm="HS256"),
        "alg_none": jwt.encode(claims, key=None, algorithm="none"),
        "refresh_as_access": make_token("1", "refresh"),
        "garbage": "not.a.token",
    }[kind]


@pytest.mark.parametrize("kind", ["expired", "other_secret", "alg_none", "refresh_as_access", "garbage"])
def test_protected_route_rejects_bad_tokens(client, kind):
    resp = client.get("/me", headers=bearer(bad_token(kind)))
    assert resp.status_code == 401
    assert resp.headers["WWW-Authenticate"] == "Bearer"


def test_refresh_rotation_and_reuse_detection(client):
    first = login(client, "nadia").json()
    second = client.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
    assert second.status_code == 200
    replay = client.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
    assert replay.status_code == 401
    # the replay ended the whole session: the newer tokens are dead as well
    assert client.post("/auth/refresh", json={"refresh_token": second.json()["refresh_token"]}).status_code == 401
    assert client.get("/me", headers=bearer(second.json()["access_token"])).status_code == 401


def test_roles_and_bola(client):
    nadia = login(client, "nadia").json()["access_token"]
    tanvir = login(client, "tanvir").json()["access_token"]
    root = login(client, "root", password="root-pass-1234", register=False).json()["access_token"]
    chat = client.post("/chats", json={"title": "Private"}, headers=bearer(nadia)).json()
    url = f"/chats/{chat['id']}"
    assert client.get(url, headers=bearer(nadia)).status_code == 200
    assert client.get(url, headers=bearer(tanvir)).status_code == 404       # BOLA regression test
    assert client.get(url, headers=bearer(root)).status_code == 200
    assert client.get("/admin/users", headers=bearer(nadia)).status_code == 403
    assert client.get("/admin/users", headers=bearer(root)).status_code == 200


def test_login_is_rate_limited(client):
    codes = [login(client, "mallory", password=f"guess-{i}-xx", register=False).status_code for i in range(7)]
    assert codes == [401] * 5 + [429] * 2


def test_api_key(client):
    assert client.post("/ingest").status_code == 401
    assert client.post("/ingest", headers={"X-API-Key": "bb_wrong"}).status_code == 401
    ok = client.post("/ingest", headers={"X-API-Key": "bb_test-key"})
    assert ok.status_code == 200 and ok.json()["service"] == "ingest-service"

uv run python -m pytest -q tests

...........                                                                                  [100%]
11 passed in 5.60s

# 1. Easy
import os
from datetime import datetime, timedelta, timezone

import jwt

from app.security import decode_token, make_token

fixed = datetime(2026, 1, 15, 4, 0, tzinfo=timezone.utc)
short = make_token("7", "access", now=fixed, ttl=timedelta(minutes=5))
claims = jwt.decode(short, os.environ["JWT_SECRET"], algorithms=["HS256"], options={"verify_exp": False})
print(claims["sub"], claims["exp"] - claims["iat"])
try:
    decode_token(make_token("7", "access"), "refresh")
except jwt.InvalidTokenError as exc:
    print("rejected:", exc)


# 2. Medium
from app.security import LoginLimiter

clock = [0.0]
limiter = LoginLimiter(max_failures=3, window=60, clock=lambda: clock[0])
for _ in range(3):
    limiter.record_failure("nadia|10.0.0.1")
print("blocked for", limiter.retry_after("nadia|10.0.0.1"), "seconds")
clock[0] = 61
print("after 61 s:", limiter.retry_after("nadia|10.0.0.1"))

7 300
rejected: expected a refresh token
blocked for 60 seconds
after 61 s: 0

# 3. Hard
from fastapi.testclient import TestClient

from app.main import create_app

with TestClient(create_app()) as tc:
    tc.post("/auth/register", json={"username": "nadia", "password": "correct-horse-1"})
    first = tc.post("/auth/login", data={"username": "nadia", "password": "correct-horse-1"}).json()
    second = tc.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
    replay = tc.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
    newest = tc.post("/auth/refresh", json={"refresh_token": second.json()["refresh_token"]})
    print(second.status_code, replay.status_code, newest.status_code)
    print(replay.json()["detail"], "|", newest.json()["detail"])

200 401 401
Refresh token reuse detected, log in again | Session ended