51. Project: an auth service with Argon2 and JWTs
Goal: a small FastAPI service with registration, login, and a protected endpoint. Passwords are stored with Argon2id; login returns a short-lived JWT signed with EdDSA; the protected route verifies it. Compare with the Go auth service and the Rust one.
# labs/authsvc.py (uv add fastapi uvicorn argon2-cffi "pyjwt[crypto]" pydantic)
import secrets
import time
from typing import Annotated
import jwt
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field
ISSUER, AUDIENCE, TTL = "authsvc.local", "orders-api", 15 * 60
signing_key = Ed25519PrivateKey.generate() # production: load from a KMS/secret store, rotate
public_key = signing_key.public_key()
ph = PasswordHasher()
USERS: dict[str, str] = {} # username -> argon2 hash (use a database)
DUMMY_HASH = ph.hash(secrets.token_urlsafe()) # for constant-ish time on unknown users
class Credentials(BaseModel):
username: Annotated[str, Field(pattern=r"^[a-z0-9_]{3,32}$")]
password: Annotated[str, Field(min_length=12, max_length=256)]
app = FastAPI()
bearer = HTTPBearer()
@app.post("/register", status_code=201)
def register(c: Credentials) -> dict[str, str]:
if c.username in USERS:
raise HTTPException(409, "username taken")
USERS[c.username] = ph.hash(c.password)
return {"username": c.username}
@app.post("/login")
def login(c: Credentials) -> dict[str, str | int]:
stored = USERS.get(c.username, DUMMY_HASH) # always run one hash: no user-enumeration by timing
try:
ph.verify(stored, c.password)
except VerifyMismatchError:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials")
if c.username not in USERS:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials")
now = int(time.time())
token = jwt.encode(
{"sub": c.username, "iss": ISSUER, "aud": AUDIENCE, "iat": now, "exp": now + TTL,
"jti": secrets.token_hex(8)},
signing_key, algorithm="EdDSA",
)
return {"access_token": token, "token_type": "bearer", "expires_in": TTL}
def current_user(cred: Annotated[HTTPAuthorizationCredentials, Depends(bearer)]) -> str:
try:
claims = jwt.decode(
cred.credentials, public_key,
algorithms=["EdDSA"], # pin the algorithm: blocks alg=none / alg confusion
audience=AUDIENCE, issuer=ISSUER,
options={"require": ["exp", "iat", "sub", "aud", "iss"]},
)
except jwt.PyJWTError as e:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"invalid token: {type(e).__name__}")
return claims["sub"]
@app.get("/me")
def me(user: Annotated[str, Depends(current_user)]) -> dict[str, str]:
return {"user": user}
# tests/test_authsvc.py (uv add --dev httpx)
from fastapi.testclient import TestClient
import jwt
from labs.authsvc import app
c = TestClient(app)
def test_flow():
assert c.post("/register", json={"username": "ada", "password": "correct horse battery"}).status_code == 201
assert c.post("/login", json={"username": "ada", "password": "wrong password!!"}).status_code == 401
tok = c.post("/login", json={"username": "ada", "password": "correct horse battery"}).json()["access_token"]
assert c.get("/me", headers={"Authorization": f"Bearer {tok}"}).json() == {"user": "ada"}
def test_rejects_unsigned_token():
forged = jwt.encode({"sub": "ada", "aud": "orders-api", "iss": "authsvc.local"}, None, algorithm="none")
assert c.get("/me", headers={"Authorization": f"Bearer {forged}"}).status_code == 401
What to notice, each tied to the authentication chapter:
algorithms=["EdDSA"]on decode is the most important line. JWT libraries that read the algorithm from the token's header have been tricked intoalg: noneand into verifying an RSA token with the public key as an HMAC secret.- Required claims:
exp(short-lived),aud(this token is for this API),iss. - The dummy hash makes unknown usernames cost the same time as wrong passwords, so login timing does not reveal which usernames exist.
- Missing on purpose, and needed in production: rate limiting on
/login, refresh tokens with rotation, revocation (ajtideny-list), key rotation with akidheader and a JWKS endpoint, and HTTPS.