MAF v1 — Production hardening (Python + .NET)

Why this chapter

JWT auth and RBAC, as covered in the original series, get you a working login flow. They don’t get you through a real ops cycle. Three concrete moments will eventually force the issue:

1. A user forgets their password.
2. A refresh token leaks (a stolen device, a logged URL, a backup snapshot).
3. You need to roll the JWT signing key (or the x-agent-secret shared between agents) without a deploy window.

Each of these has a textbook answer; each of those answers has a footgun. This chapter walks the patterns the capstone settled on, in Python first, with the .NET twin called out where the idioms actually diverge.

Prerequisites

• Familiarity with Part 7, Production Readiness: Auth, RBAC, and Deployment. This picks up where that left off.
• Python 3.12+ with fastapi, pyjwt, bcrypt, asyncpg (or the .NET equivalents).
• Postgres 16 (the capstone’s database).
• An SMTP relay or transactional-email provider for password reset. The code below assumes a send_email(to, subject, body) helper; wire whatever you use.

Part A: Password reset

Threat model first

A password-reset endpoint is a credential issuer. A leak in this flow is worse than a leak in login: the attacker writes the new credential. Three rules cover most of what can go wrong:

• The token in the email is a one-shot bearer credential. Treat it like a password. Short TTL. Single-use. Never logged.
• Store the hash, not the token. A database snapshot must not let an attacker complete pending resets.
• Don’t leak account existence. POST /auth/forgot-password returns the same shape whether the email is registered or not. Otherwise the endpoint becomes an account-enumeration oracle.

Schema

CREATE TABLE password_reset_tokens (
    token_hash      TEXT PRIMARY KEY,           -- sha256(plaintext token)
    user_id         UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    expires_at      TIMESTAMPTZ NOT NULL,
    consumed_at     TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    requesting_ip   INET,
    user_agent      TEXT
);

CREATE INDEX idx_prt_user_unconsumed
    ON password_reset_tokens (user_id)
    WHERE consumed_at IS NULL;

The partial index on unconsumed rows lets us quickly invalidate prior pending tokens for the same user when a new one is issued (one outstanding token per user; see below).

Issuing a token

import hashlib, secrets
from datetime import datetime, timedelta, timezone

RESET_TOKEN_TTL = timedelta(minutes=15)

def _hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()

async def request_password_reset(email: str, request_ip: str, user_agent: str) -> None:
    """Generate and email a reset link. Never raises for unknown emails."""
    user = await db.fetch_one("SELECT id FROM users WHERE email = $1", email)
    if user is None:
        # Constant-time burn — same wall-clock as the success path
        await asyncio.sleep(0.05)
        return

    # Invalidate any outstanding pending token for this user
    await db.execute(
        "UPDATE password_reset_tokens SET consumed_at = now() "
        "WHERE user_id = $1 AND consumed_at IS NULL",
        user["id"],
    )

    plaintext = secrets.token_urlsafe(32)               # 256 bits
    await db.execute(
        "INSERT INTO password_reset_tokens "
        "(token_hash, user_id, expires_at, requesting_ip, user_agent) "
        "VALUES ($1, $2, $3, $4, $5)",
        _hash_token(plaintext),
        user["id"],
        datetime.now(timezone.utc) + RESET_TOKEN_TTL,
        request_ip,
        user_agent,
    )

    reset_url = f"{settings.frontend_base_url}/reset-password?token={plaintext}"
    await send_email(
        to=email,
        subject="Reset your ECommerce Agents password",
        body=f"Click to reset (expires in 15 minutes): {reset_url}",
    )

Three details that matter:

• secrets.token_urlsafe(32) generates 256 bits of entropy, URL-safe. random.choice and uuid4 are both wrong here: random is not cryptographic, and uuid4 only carries 122 bits.
• Plaintext goes only to the email. The DB never sees it. If the DB leaks, the attacker has hashes, which they can’t use to construct the URL.
• Old pending tokens are invalidated when a new one is issued. Otherwise an attacker who triggered an earlier reset (and intercepted that email) can race the legitimate user.

Consuming the token

async def reset_password(token: str, new_password: str) -> bool:
    token_hash = _hash_token(token)
    async with db.transaction():
        row = await db.fetch_one(
            "SELECT user_id, expires_at, consumed_at "
            "FROM password_reset_tokens WHERE token_hash = $1 FOR UPDATE",
            token_hash,
        )
        if row is None:
            return False
        if row["consumed_at"] is not None:
            return False
        if row["expires_at"] \< datetime.now(timezone.utc):
            return False

        await db.execute(
            "UPDATE password_reset_tokens SET consumed_at = now() "
            "WHERE token_hash = $1",
            token_hash,
        )
        await db.execute(
            "UPDATE users SET password_hash = $1 WHERE id = $2",
            hash_password(new_password),
            row["user_id"],
        )
        # Critical: revoke all existing refresh tokens for this user
        await db.execute(
            "UPDATE refresh_tokens SET revoked_at = now() "
            "WHERE user_id = $1 AND revoked_at IS NULL",
            row["user_id"],
        )
    return True

The FOR UPDATE row lock on the reset-token row is what makes single-use atomic across concurrent requests. The transaction also covers the users update and the refresh-token revocation; if anything fails, nothing changes.

Endpoint

@router.post("/auth/forgot-password", status_code=204)
async def forgot_password(req: ForgotPasswordRequest, request: Request):
    await request_password_reset(
        email=req.email,
        request_ip=request.client.host,
        user_agent=request.headers.get("user-agent", ""),
    )
    return Response(status_code=204)


@router.post("/auth/reset-password")
async def reset_password_endpoint(req: ResetPasswordRequest):
    ok = await reset_password(req.token, req.new_password)
    if not ok:
        # Same response on expired/consumed/unknown — no oracle
        raise HTTPException(400, detail="Invalid or expired reset token")
    return {"ok": True}

forgot-password always returns 204. The endpoint is unconditionally rate-limited (10 per IP per hour and 3 per email per day are reasonable starting points; the capstone uses Redis for both counters). Rate limits exist to throttle the email send-budget, not to prove anything to an attacker.

.NET note

ASP.NET’s IDataProtector plus a small ResetTokenStore table covers the same ground. The DataProtectionTokenProvider<TUser> that ships with Microsoft.AspNetCore.Identity is useful only if you’re already using Identity; if not, the secrets.token_urlsafe(32) → SHA256 pattern translates one-for-one to RandomNumberGenerator.GetBytes(32) + Base64Url + SHA256.HashData. Don’t reach for Guid.NewGuid() here.

Part B: Refresh-token rotation with reuse detection

The original series ships short-lived access tokens (60 min) and long-lived refresh tokens (7 days), both as JWTs. That’s enough to log in once and stay logged in. It’s not enough to detect a stolen refresh.

The hardening: every refresh issues a new refresh token and invalidates the previous one. If a refresh token is presented after it has been used, the entire token family is revoked and the user is forced to re-authenticate.

This catches the canonical attack: an attacker steals a refresh token from a backup, the legitimate user later refreshes (rotating it), and then the attacker tries to refresh with the stale value. The replay attempt triggers a family-wide revocation: the legitimate user is signed out, but so is the attacker, and the security team gets a paged incident.

Token family lineage

Schema

CREATE TABLE refresh_tokens (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    family_id       UUID NOT NULL,                 -- shared across the lineage
    parent_id       UUID REFERENCES refresh_tokens(id),
    user_id         UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    token_hash      TEXT NOT NULL UNIQUE,          -- sha256 of the JWT
    issued_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    expires_at      TIMESTAMPTZ NOT NULL,
    rotated_at      TIMESTAMPTZ,                   -- set when used to mint a child
    revoked_at      TIMESTAMPTZ                    -- set when family is killed
);

CREATE INDEX idx_rt_family ON refresh_tokens (family_id);
CREATE INDEX idx_rt_user_active
    ON refresh_tokens (user_id)
    WHERE revoked_at IS NULL;

token_hash (not the raw JWT) is what we look up. JWTs are bearer credentials; we should not store them at rest in plaintext.

Refresh handler

async def refresh(presented_token: str) -> tuple[str, str]:
    """Returns (new_access_token, new_refresh_token) or raises."""
    payload = jwt.decode(presented_token, settings.jwt_public_key, algorithms=["RS256"])
    if payload.get("type") != "refresh":
        raise InvalidTokenError("Wrong token type")

    token_hash = _hash_token(presented_token)

    async with db.transaction():
        row = await db.fetch_one(
            "SELECT id, family_id, user_id, rotated_at, revoked_at, expires_at "
            "FROM refresh_tokens WHERE token_hash = $1 FOR UPDATE",
            token_hash,
        )
        if row is None:
            raise InvalidTokenError("Unknown token")
        if row["revoked_at"] is not None:
            raise InvalidTokenError("Token revoked")
        if row["expires_at"] \< datetime.now(timezone.utc):
            raise InvalidTokenError("Token expired")

        # Reuse detection: this token has already been rotated
        if row["rotated_at"] is not None:
            await db.execute(
                "UPDATE refresh_tokens SET revoked_at = now() "
                "WHERE family_id = $1 AND revoked_at IS NULL",
                row["family_id"],
            )
            log.warning("refresh_reuse_detected", user_id=str(row["user_id"]),
                        family_id=str(row["family_id"]))
            raise InvalidTokenError("Token reuse detected — family revoked")

        # Mint child
        new_jwt = _issue_refresh_jwt(row["user_id"])
        new_id = uuid.uuid4()
        await db.execute(
            "INSERT INTO refresh_tokens "
            "(id, family_id, parent_id, user_id, token_hash, expires_at) "
            "VALUES ($1, $2, $3, $4, $5, $6)",
            new_id, row["family_id"], row["id"], row["user_id"],
            _hash_token(new_jwt),
            datetime.now(timezone.utc) + REFRESH_TTL,
        )
        await db.execute(
            "UPDATE refresh_tokens SET rotated_at = now() WHERE id = $1",
            row["id"],
        )

    return _issue_access_jwt(row["user_id"]), new_jwt

The whole sequence sits inside one transaction with a row lock on the presented token. That guarantees:

• Two concurrent refreshes with the same token can’t both succeed (one will block on FOR UPDATE, then see rotated_at set, then trigger family revocation, which is the right behaviour: simultaneous use of the same refresh is exactly the attack signature).
• The “mint child + mark parent rotated” pair is atomic. A crash mid-flight leaves the parent unmodified.

Logout

async def logout(refresh_jwt: str) -> None:
    token_hash = _hash_token(refresh_jwt)
    await db.execute(
        "UPDATE refresh_tokens SET revoked_at = now() "
        "WHERE family_id = (SELECT family_id FROM refresh_tokens WHERE token_hash = $1) "
        "AND revoked_at IS NULL",
        token_hash,
    )

Logout revokes the entire family, not just the presented token. This matters because it makes “Sign out everywhere” trivial: the same query, run for every active family belonging to the user.

What this doesn’t catch

Refresh-token rotation catches replay of stolen refresh tokens. It does not catch:

• Theft of the access token before it expires. Access tokens are stateless; the only mitigation is short TTL. 60 minutes is the upper bound; 15 is more honest.
• Compromise of the JWT signing key. That’s Part C.
• A logged-in attacker abusing the legitimate session. Audit your tool calls (Ch06 middleware), don’t try to fix this in auth.

.NET note

The pattern is identical with Microsoft.IdentityModel.Tokens for JWT validation and EF Core for the table. The one trap: JwtBearerOptions.SaveToken = true will stash the access JWT on HttpContext; that’s harmless but unnecessary if you’re already extracting it explicitly.

Part C: Graceful secret rotation

Two kinds of secrets need rotation in the capstone:

1. JWT signing keys. The asymmetric key pair used to sign access and refresh tokens. RS256 means there’s a public key (used by every agent for verification) and a private key (used only by the orchestrator’s auth service to sign).
2. Inter-agent shared secrets. The x-agent-secret header that a specialist accepts as proof the request came from an authorized orchestrator.

Rotation that requires a deploy window or simultaneous restart is the same as no rotation; it never happens. The pattern below makes both kinds rotatable on a live system.

The two-key window

The trick is the same in both cases: maintain a list of valid keys, sign with the newest, accept any. A new key is “promoted in” hours or days before the old one is “demoted out.” During the overlap, both work.

JWT_KEYS: dict[str, RSAKey] = {
    "2026-04-15": load_rsa_key("keys/jwt-2026-04-15.pem"),  # current signer
    "2026-03-01": load_rsa_key("keys/jwt-2026-03-01.pem"),  # accepted, retiring
}
JWT_CURRENT_KID = "2026-04-15"

Signing (always uses current)

def issue_access_jwt(user_id: UUID) -> str:
    return jwt.encode(
        {
            "sub": str(user_id),
            "type": "access",
            "exp": datetime.now(timezone.utc) + ACCESS_TTL,
        },
        settings.JWT_KEYS[settings.JWT_CURRENT_KID].private,
        algorithm="RS256",
        headers={"kid": settings.JWT_CURRENT_KID},
    )

The kid (key ID) header is what makes the verifier’s job trivial.

Verifying (looks up by kid, rejects unknown)

def decode_access_jwt(token: str) -> dict:
    header = jwt.get_unverified_header(token)
    kid = header.get("kid")
    if kid not in settings.JWT_KEYS:
        raise InvalidTokenError(f"Unknown key id: {kid}")
    return jwt.decode(
        token,
        settings.JWT_KEYS[kid].public,
        algorithms=["RS256"],
    )

Rotation lifecycle

The “Day 8” gap matters: it must be longer than the longest access-token TTL plus any refresh slack. With a 60-minute access TTL, 24 hours is plenty. With a 7-day refresh, you wait a week before pulling the old key (or you accept that a small number of users will get one forced re-login).

Operationalizing

The JWT_KEYS dict above is a static reload. In production you want it to come from a key vault or a config service that supports atomic updates:

• Azure Key Vault. Store keys as secrets keyed by kid. Cache them locally with a 60-second TTL; the cache miss is acceptable because rotation is a planned event.
• Filesystem with a watcher. inotify/FileSystemWatcher on the keys directory; the agent reloads JWT_KEYS when a .pem is added or removed. Cheaper than a vault for self-hosted deployments.
• A /admin/reload-keys endpoint. Gated behind admin RBAC, useful as a manual fallback when the watcher misses an event.

Whichever way, the application code stays the same: it reads from settings.JWT_KEYS and trusts that someone (orchestration, watcher, manual call) keeps the dict current.

x-agent-secret rotation

The agent-to-agent shared secret is even simpler. There’s no kid because the header value is the secret. Solution: accept a list of secrets.

AGENT_SECRETS: set[str] = {
    "current_value_set_at_2026_04_15",   # promoted today
    "previous_value_set_at_2026_03_01",  # being retired
}
AGENT_SECRET_CURRENT = "current_value_set_at_2026_04_15"


def verify_agent_secret(presented: str) -> bool:
    # Constant-time compare against any allowed value
    for allowed in settings.AGENT_SECRETS:
        if hmac.compare_digest(presented, allowed):
            return True
    return False


def outgoing_agent_secret() -> str:
    return settings.AGENT_SECRET_CURRENT

The orchestrator always sends the current secret; specialists accept either. The rotation script:

1. Generate new secret. Add it to AGENT_SECRETS on every specialist (rolling restart or live config reload). Don’t change AGENT_SECRET_CURRENT yet.
2. Once every specialist accepts the new secret, switch AGENT_SECRET_CURRENT on the orchestrator. The orchestrator now sends the new value.
3. After the access-token TTL, remove the old secret from AGENT_SECRETS on the specialists.

hmac.compare_digest instead of == matters: the latter short-circuits on the first different byte, which leaks length and content via timing.

.NET note

IConfiguration reload-on-change combined with IOptionsMonitor<JwtKeyringOptions> does the equivalent without a watcher. For Key Vault, the Azure.Extensions.AspNetCore.Configuration.Secrets package wires it up directly. The constant-time compare is CryptographicOperations.FixedTimeEquals, never string.Equals.

What changes for the capstone

The capstone today (agents/python/shared/auth.py) ships the original Part 7 design: single JWT signing key, simple refresh, no reset flow. Phase 8 of the refactor plan tracks the migration:

• agents/python/shared/auth/keyring.py: JWT_KEYS map, decode_access_jwt, outgoing_agent_secret.
• migrations/0042_password_reset_and_refresh_families.sql: the two new tables and the partial indexes.
• agents/python/orchestrator/auth_routes.py: /auth/forgot-password, /auth/reset-password, refactored /auth/refresh.

None of this changes the agent code. Specialists keep using verify_agent_secret; tools keep reading current_user_email. The hardening lives entirely in the auth boundary.

What’s next

That closes the production-hardening loop the original Part 7 left open. Ch21: Putting it all together is the final chapter: a guided tour of the live e-commerce stack with file:line citations for every concept covered across the series.