WIP checkpoint: Write auth endpoint tests (pytest)

- auth.py: _create_access_token had no jti claim, so two refreshes
  within the same wall-clock second produced byte-identical access
  tokens (same failure mode already fixed for refresh tokens earlier)
- test_refresh_endpoint.py: usernames (jwttoken, refreshuser, etc.)
  collided with test_auth_endpoints.py's own test usernames once run
  in the same pytest session against a shared DB with no per-test
  isolation -- exactly how run_checkpoint_tests.py invokes pytest.
  Namespaced this file's usernames with a refep_ prefix.

Full backend suite (63 tests) passes clean end-to-end against a real
Postgres with real migrations applied.

backend/.env.example

@@ -4,10 +4,13 @@ DATABASE_URL=postgresql+asyncpg://agp_user:agp_password@localhost:5432/agp_db
 # Redis
 REDIS_URL=redis://localhost:6379/0
 
-# JWT
-SECRET_KEY=change-me-to-a-secure-random-string
-ALGORITHM=HS256
-ACCESS_TOKEN_EXPIRE_MINUTES=30
+# JWT / Auth
+ALGORITHM=RS256
+ACCESS_TOKEN_EXPIRE_MINUTES=60
+REFRESH_TOKEN_EXPIRE_DAYS=30
+# Directory where the RSA keypair is loaded/generated at runtime.
+# Defaults to ~/.agp/keys (outside the repo). Keys are never committed to git.
+RSA_KEY_DIR=~/.agp/keys
 
 # App
 APP_NAME=AGP Backend

backend/alembic/versions/0003_add_refresh_tokens_table.py

@@ -0,0 +1,36 @@
+"""add refresh_tokens table
+
+Revision ID: 0003_add_refresh_tokens_table
+Revises: 0002_add_date_of_birth_to_users
+Create Date: 2025-07-25 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "0003_add_refresh_tokens_table"
+down_revision: Union[str, None] = "0002_add_date_of_birth_to_users"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create the refresh_tokens table."""
+    op.create_table(
+        "refresh_tokens",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("token", sa.String(512), unique=True, index=True, nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("is_active", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    """Drop the refresh_tokens table."""
+    op.drop_table("refresh_tokens")

backend/app/config.py

@@ -1,10 +1,56 @@
 """Application configuration via pydantic-settings."""
 
 from functools import lru_cache
+from pathlib import Path
 
+from cryptography.hazmat.primitives.asymmetric.rsa import (
+    RSAPrivateKey,
+    RSAPublicKey,
+    generate_private_key,
+)
+from cryptography.hazmat.primitives.serialization import (
+    Encoding,
+    NoEncryption,
+    PrivateFormat,
+    PublicFormat,
+    load_pem_private_key,
+    load_pem_public_key,
+)
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
+def _load_or_generate_rsa_keys(key_dir: Path) -> tuple[bytes, bytes]:
+    """Load existing RSA keys or generate a new pair.
+
+    Keys are stored as PEM files in *key_dir*.  The directory is created
+    if it does not exist.  Files are named ``private.pem`` and
+    ``public.pem`` and are **not** committed to git.
+    """
+    private_path = key_dir / "private.pem"
+    public_path = key_dir / "public.pem"
+
+    if private_path.exists() and public_path.exists():
+        private_pem = private_path.read_bytes()
+        public_pem = public_path.read_bytes()
+    else:
+        # Generate a new 2048-bit RSA key pair
+        private_key = generate_private_key(public_exponent=65537, key_size=2048)
+        private_pem = private_key.private_bytes(
+            Encoding.PEM,
+            PrivateFormat.PKCS8,
+            NoEncryption(),
+        )
+        public_pem = private_key.public_key().public_bytes(
+            Encoding.PEM,
+            PublicFormat.SubjectPublicKeyInfo,
+        )
+        key_dir.mkdir(parents=True, exist_ok=True)
+        private_path.write_bytes(private_pem)
+        public_path.write_bytes(public_pem)
+
+    return private_pem, public_pem
+
+
 class Settings(BaseSettings):
     # App
     APP_NAME: str = "AGP Backend"
@@ -18,11 +64,13 @@ class Settings(BaseSettings):
     REDIS_URL: str = "redis://localhost:6379/0"
 
     # JWT / Auth
-    # NOTE: production must override SECRET_KEY with a secure random value
-    # (e.g. `python -c "import secrets; print(secrets.token_urlsafe(32))"`)
-    SECRET_KEY: str = "change-me-to-a-secure-random-string"
-    ALGORITHM: str = "HS256"
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+    ALGORITHM: str = "RS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
+    REFRESH_TOKEN_EXPIRE_DAYS: int = 30
+
+    # RS256 key storage.  Defaults to a directory outside the repo so the
+    # generated keypair is never accidentally committed.
+    RSA_KEY_DIR: str = str(Path.home() / ".agp" / "keys")
 
     # CORS
     CORS_ORIGINS: str = "http://localhost:8080,http://localhost:3000"
@@ -42,6 +90,32 @@ class Settings(BaseSettings):
     def cors_origins_list(self) -> list[str]:
         return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]
 
+    @property
+    def rsa_key_dir(self) -> Path:
+        return Path(self.RSA_KEY_DIR)
+
+    @property
+    def private_pem(self) -> bytes:
+        """Return the private key PEM bytes (loads/generates on first call)."""
+        private_pem, _ = _load_or_generate_rsa_keys(self.rsa_key_dir)
+        return private_pem
+
+    @property
+    def public_pem(self) -> bytes:
+        """Return the public key PEM bytes (loads/generates on first call)."""
+        _, public_pem = _load_or_generate_rsa_keys(self.rsa_key_dir)
+        return public_pem
+
+    @property
+    def private_key(self) -> RSAPrivateKey:
+        """Return the deserialized RSA private key object."""
+        return load_pem_private_key(self.private_pem, password=None)
+
+    @property
+    def public_key(self) -> RSAPublicKey:
+        """Return the deserialized RSA public key object."""
+        return load_pem_public_key(self.public_pem)
+
 
 @lru_cache()
 def get_settings() -> Settings:

backend/app/dependencies.py

@@ -27,7 +27,7 @@ async def get_current_user(
     )
     try:
         payload = jwt.decode(
-            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
+            token, settings.public_pem, algorithms=[settings.ALGORITHM]
         )
         user_id: str | None = payload.get("sub")
         if user_id is None:

backend/app/models/session.py

@@ -3,6 +3,7 @@ import uuid
 from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Boolean
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
+
 from app.database import Base
 
 
@@ -18,3 +19,23 @@ class Session(Base):
     updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
 
     user = relationship("User", back_populates="sessions")
+
+
+class RefreshToken(Base):
+    __tablename__ = "refresh_tokens"
+
+    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    # RS256 JWTs are ~420+ chars; 512 is sufficient and matches the migration.
+    token = Column(String(512), unique=True, index=True, nullable=False)
+    expires_at = Column(DateTime(timezone=True), nullable=False)
+    is_active = Column(Boolean, nullable=False, default=True, server_default="true")
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    updated_at = Column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+    )
+
+    user = relationship("User", back_populates="refresh_tokens")

backend/app/models/user.py

@@ -21,6 +21,7 @@ class User(Base):
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
 
     sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
+    refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan")
     worlds = relationship("World", back_populates="owner")
     assets = relationship("Asset", back_populates="owner")
 

backend/app/routers/auth.py

@@ -1,20 +1,21 @@
 """Authentication router - login, register, token refresh."""
 
+import uuid
 from datetime import date, datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
-from pydantic import BaseModel, EmailStr, field_validator
-
-from jose import jwt
+from jose import JWTError, jwt
 from passlib.context import CryptContext
+from pydantic import BaseModel, EmailStr, field_validator
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.database import get_db
 from app.dependencies import CurrentUser
+from app.models.session import RefreshToken
 from app.models.user import User
-from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
@@ -25,7 +26,13 @@ router = APIRouter()
 
 class TokenResponse(BaseModel):
     access_token: str
+    refresh_token: str
     token_type: str = "bearer"
+    access_token_expires_minutes: int
+
+
+class RefreshRequest(BaseModel):
+    refresh_token: str
 
 
 class RegisterRequest(BaseModel):
@@ -59,12 +66,43 @@ class UserResponse(BaseModel):
 # --- Helpers ---
 
 def _create_access_token(data: dict) -> str:
+    """Create a signed access token using RS256.
+
+    Includes a unique ``jti`` claim so that concurrent/same-second calls
+    (e.g. back-to-back /refresh requests) still produce distinct tokens —
+    without it, two tokens minted for the same user within the same
+    second are byte-identical since ``sub``/``exp`` are the only claims.
+    """
     to_encode = data.copy()
+    to_encode["jti"] = str(uuid.uuid4())
     expire = datetime.now(timezone.utc) + timedelta(
         minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
     )
     to_encode.update({"exp": expire})
-    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return jwt.encode(
+        to_encode,
+        settings.private_pem,
+        algorithm=settings.ALGORITHM,
+    )
+
+
+def _create_refresh_token(data: dict) -> str:
+    """Create a signed refresh token using RS256.
+
+    Includes a unique ``jti`` claim so that concurrent calls produce
+    distinct tokens even when issued within the same second.
+    """
+    to_encode = data.copy()
+    to_encode["jti"] = str(uuid.uuid4())
+    expire = datetime.now(timezone.utc) + timedelta(
+        days=settings.REFRESH_TOKEN_EXPIRE_DAYS
+    )
+    to_encode.update({"exp": expire})
+    return jwt.encode(
+        to_encode,
+        settings.private_pem,
+        algorithm=settings.ALGORITHM,
+    )
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
@@ -97,11 +135,25 @@ def calculate_age(birth_date: date, today: date | None = None) -> int:
     return age
 
 
+def _issue_refresh_token(user_id: int, db: AsyncSession) -> str:
+    """Create and persist a refresh token for the given user."""
+    refresh_token_value = _create_refresh_token({"sub": str(user_id)})
+    refresh_token_record = RefreshToken(
+        id=str(uuid.uuid4()),
+        user_id=user_id,
+        token=refresh_token_value,
+        expires_at=datetime.now(timezone.utc)
+        + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS),
+    )
+    db.add(refresh_token_record)
+    return refresh_token_value
+
+
 # --- Endpoints ---
 
 @router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
 async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)):
-    """Register a new user and return an access token.
+    """Register a new user and return access + refresh tokens.
 
     Requires date_of_birth and rejects users under 18.
     """
@@ -138,16 +190,26 @@ async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)):
         date_of_birth=dob,
     )
     db.add(user)
-    await db.commit()
+    await db.flush()  # flush to get user.id before creating refresh token
     await db.refresh(user)
 
+    # Create access token
     access_token = _create_access_token({"sub": str(user.id)})
-    return TokenResponse(access_token=access_token)
+
+    # Create and persist refresh token
+    refresh_token_value = _issue_refresh_token(user.id, db)
+    await db.commit()
+
+    return TokenResponse(
+        access_token=access_token,
+        refresh_token=refresh_token_value,
+        access_token_expires_minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
+    )
 
 
 @router.post("/login", response_model=TokenResponse)
 async def login(form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
-    """Authenticate user and return an access token.
+    """Authenticate user and return access + refresh tokens.
 
     Accepts form-encoded credentials (username/password) as expected by
     OAuth2PasswordBearer, so the Swagger UI "Authorize" flow works correctly.
@@ -167,8 +229,84 @@ async def login(form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession =
             detail="User account is inactive",
         )
 
+    # Create access token
     access_token = _create_access_token({"sub": str(user.id)})
-    return TokenResponse(access_token=access_token)
+
+    # Create and persist refresh token
+    refresh_token_value = _issue_refresh_token(user.id, db)
+    await db.commit()
+
+    return TokenResponse(
+        access_token=access_token,
+        refresh_token=refresh_token_value,
+        access_token_expires_minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
+    )
+
+
+@router.post("/refresh", response_model=TokenResponse)
+async def refresh(req: RefreshRequest, db: AsyncSession = Depends(get_db)):
+    """Exchange a valid refresh token for a new access token.
+
+    Validates the refresh token against the database:
+    - Must exist in the refresh_tokens table
+    - Must not be expired (expires_at > now)
+    - Must be active (is_active == true)
+
+    On success, returns a new access token. The refresh token itself is
+    NOT rotated (the client keeps using the same refresh token).
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid or expired refresh token",
+    )
+
+    # Step 1: Decode and validate the JWT signature/claims
+    try:
+        payload = jwt.decode(
+            req.refresh_token,
+            settings.public_pem,
+            algorithms=[settings.ALGORITHM],
+        )
+    except JWTError:
+        raise credentials_exception
+
+    user_id_str: str | None = payload.get("sub")
+    if user_id_str is None:
+        raise credentials_exception
+
+    try:
+        user_id = int(user_id_str)
+    except (ValueError, TypeError):
+        raise credentials_exception
+
+    # Step 2: Look up the refresh token in the database
+    result = await db.execute(
+        select(RefreshToken).where(
+            RefreshToken.token == req.refresh_token,
+            RefreshToken.user_id == user_id,
+        )
+    )
+    stored_token = result.scalar_one_or_none()
+
+    if stored_token is None:
+        raise credentials_exception
+
+    # Step 3: Check if the token is expired
+    if datetime.now(timezone.utc) >= stored_token.expires_at:
+        raise credentials_exception
+
+    # Step 4: Check if the token is active
+    if not stored_token.is_active:
+        raise credentials_exception
+
+    # Step 5: Issue a new access token
+    access_token = _create_access_token({"sub": str(user_id)})
+
+    return TokenResponse(
+        access_token=access_token,
+        refresh_token=req.refresh_token,  # Return the same refresh token
+        access_token_expires_minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
+    )
 
 
 @router.get("/me", response_model=UserResponse)

backend/requirements.txt

@@ -9,6 +9,14 @@ pydantic==2.10.3
 pydantic-settings==2.7.0
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
+# passlib 1.7.4 (last release, unmaintained since ~2020) detects its bcrypt
+# backend via bcrypt.__about__.__version__, which bcrypt>=4.1 removed --
+# hashing/verifying a password then raises "password cannot be longer than
+# 72 bytes" (a misleading downstream symptom of the real AttributeError:
+# module 'bcrypt' has no attribute '__about__') on every call. Without this
+# pin, `pip install -r requirements.txt` silently grabs whatever the latest
+# bcrypt is and breaks register/login outright. Pin below 4.1 explicitly.
+bcrypt==4.0.1
 python-multipart==0.0.20
 email-validator==2.2.0
 httpx==0.28.1

backend/tests/conftest.py

@@ -0,0 +1,38 @@
+"""Shared pytest fixtures.
+
+The auth integration tests hit the *real* Postgres database (per
+``DATABASE_URL``), so they persist data between runs.  Without cleanup a
+second run of the suite fails with 409 conflicts because the users it
+registered on the first run already exist.  The session-scoped fixture below
+truncates all application tables once before the first test, making the suite
+re-runnable.
+
+Truncation order doesn't strictly matter because ``CASCADE`` follows foreign
+keys, but the tables are listed explicitly for clarity.
+"""
+
+import asyncio
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from app.config import settings
+
+TEST_TABLES = ("assets", "worlds", "sessions", "refresh_tokens", "users")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _reset_database():
+    """Truncate all application tables once per test session."""
+
+    async def _truncate() -> None:
+        engine = create_async_engine(settings.DATABASE_URL)
+        async with engine.begin() as conn:
+            await conn.execute(
+                text(f"TRUNCATE TABLE {', '.join(TEST_TABLES)} CASCADE")
+            )
+        await engine.dispose()
+
+    asyncio.run(_truncate())
+    yield

backend/tests/test_auth_expired_token.py

@@ -0,0 +1,171 @@
+"""Tests for expired-token rejection across auth endpoints.
+
+Covers two distinct expiry paths:
+
+1. An *access* JWT whose ``exp`` claim is in the past must be rejected
+   with 401 by ``GET /api/auth/me`` (jose's built-in ``exp`` validation).
+2. A *refresh* token whose persisted ``expires_at`` is in the past must be
+   rejected with 401 by ``POST /api/auth/refresh`` even though its JWT
+   signature is still valid.
+
+Requires migrations already applied (alembic upgrade head) before running.
+"""
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+from jose import jwt
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import settings
+from app.main import app
+from app.database import get_db
+from app.models.session import RefreshToken
+
+
+# ---------------------------------------------------------------------------
+# Test fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _override_db():
+    """Override get_db so the app uses a fresh session per test."""
+    test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
+    test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
+
+    async def _get_db():
+        async_session = test_session_factory()
+        try:
+            yield async_session
+            await async_session.commit()
+        except Exception:
+            await async_session.rollback()
+            raise
+        finally:
+            await async_session.close()
+
+    app.dependency_overrides[get_db] = _get_db
+    yield
+    app.dependency_overrides.clear()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _mint_access_token(exp: datetime) -> str:
+    """Mint an RS256 access token with an explicit ``exp`` claim."""
+    return jwt.encode(
+        {"sub": "1", "exp": exp},
+        settings.private_pem,
+        algorithm=settings.ALGORITHM,
+    )
+
+
+def _expire_refresh_token_in_db(token: str) -> None:
+    """Set the persisted refresh token's ``expires_at`` into the past."""
+
+    async def _do():
+        engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
+        session_factory = async_sessionmaker(engine, expire_on_commit=False)
+        async with session_factory() as session:
+            from sqlalchemy import update
+
+            await session.execute(
+                update(RefreshToken)
+                .where(RefreshToken.token == token)
+                .values(expires_at=datetime.now(timezone.utc) - timedelta(hours=1))
+            )
+            await session.commit()
+        await engine.dispose()
+
+    asyncio.run(_do())
+
+
+def _register_user(client: TestClient, username: str) -> str:
+    """Register a user and return its refresh token."""
+    resp = client.post(
+        "/api/auth/register",
+        json={
+            "username": username,
+            "email": f"{username}@example.com",
+            "password": "secret123",
+            "date_of_birth": "2000-01-01",
+        },
+    )
+    assert resp.status_code == 201
+    return resp.json()["refresh_token"]
+
+
+# ---------------------------------------------------------------------------
+# Expired access token (GET /api/auth/me)
+# ---------------------------------------------------------------------------
+
+class TestExpiredAccessToken:
+    def test_expired_access_token_is_rejected_by_me(self):
+        """An access JWT with a past exp claim returns 401 on /me."""
+        client = TestClient(app)
+        expired_token = _mint_access_token(
+            datetime.now(timezone.utc) - timedelta(minutes=5)
+        )
+        resp = client.get(
+            "/api/auth/me",
+            headers={"Authorization": f"Bearer {expired_token}"},
+        )
+        assert resp.status_code == 401
+        assert "Could not validate credentials" in resp.json()["detail"]
+
+    def test_garbage_access_token_is_rejected_by_me(self):
+        """A non-JWT bearer token returns 401 on /me."""
+        client = TestClient(app)
+        resp = client.get(
+            "/api/auth/me",
+            headers={"Authorization": "Bearer not-a-real-jwt"},
+        )
+        assert resp.status_code == 401
+
+    def test_missing_bearer_token_returns_401(self):
+        """No Authorization header at all returns 401 on /me."""
+        client = TestClient(app)
+        resp = client.get("/api/auth/me")
+        assert resp.status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# Expired refresh token (POST /api/auth/refresh)
+# ---------------------------------------------------------------------------
+
+class TestExpiredRefreshToken:
+    def test_expired_refresh_token_is_rejected(self):
+        """A refresh token whose DB expires_at is in the past returns 401."""
+        client = TestClient(app)
+        refresh_token = _register_user(client, "expired_rt_user")
+
+        # Force the persisted record to be expired while its JWT signature
+        # is still valid.
+        _expire_refresh_token_in_db(refresh_token)
+
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp.status_code == 401
+        assert "Invalid or expired refresh token" in resp.json()["detail"]
+
+    def test_expired_refresh_token_jwt_is_rejected(self):
+        """A refresh JWT with a past exp claim is rejected before DB lookup."""
+        client = TestClient(app)
+        expired_jwt = jwt.encode(
+            {"sub": "1", "exp": datetime.now(timezone.utc) - timedelta(minutes=5)},
+            settings.private_pem,
+            algorithm=settings.ALGORITHM,
+        )
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": expired_jwt},
+        )
+        assert resp.status_code == 401
+        assert "Invalid or expired refresh token" in resp.json()["detail"]

backend/tests/test_refresh_endpoint.py

@@ -0,0 +1,201 @@
+"""Tests for the POST /api/auth/refresh endpoint.
+
+These tests exercise the full stack: register → get refresh token →
+refresh → new access token.  Requires migrations already applied
+(alembic upgrade head) before running.
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import settings
+from app.main import app
+from app.database import get_db
+
+
+# ---------------------------------------------------------------------------
+# Test fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _override_db():
+    """Override get_db so the app uses a fresh session per test."""
+    test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
+    test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
+
+    async def _get_db():
+        async_session = test_session_factory()
+        try:
+            yield async_session
+            await async_session.commit()
+        except Exception:
+            await async_session.rollback()
+            raise
+        finally:
+            await async_session.close()
+
+    app.dependency_overrides[get_db] = _get_db
+    yield
+    app.dependency_overrides.clear()
+
+
+# ---------------------------------------------------------------------------
+# Refresh endpoint tests
+# ---------------------------------------------------------------------------
+
+class TestRefresh:
+    """Tests for POST /api/auth/refresh."""
+
+    def _register_user(self, username: str, password: str, email: str = None):
+        """Helper to register a user and return the refresh token."""
+        client = TestClient(app)
+        if email is None:
+            email = f"{username}@example.com"
+        resp = client.post(
+            "/api/auth/register",
+            json={
+                "username": username,
+                "email": email,
+                "password": password,
+                "date_of_birth": "2000-01-01",
+            },
+        )
+        assert resp.status_code == 201
+        return resp.json()["refresh_token"]
+
+    def test_refresh_success(self):
+        """A valid refresh token returns a new access token."""
+        refresh_token = self._register_user("refep_refreshuser", "secret123")
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "access_token" in data
+        assert data["token_type"] == "bearer"
+        # The refresh token in the response is the same one we sent
+        assert data["refresh_token"] == refresh_token
+
+    def test_refresh_returns_different_access_token(self):
+        """Refreshing produces a new access token (different from any previous)."""
+        refresh_token = self._register_user("refep_refreshtoken", "secret123")
+        client = TestClient(app)
+
+        # First refresh
+        resp1 = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp1.status_code == 200
+        access_token_1 = resp1.json()["access_token"]
+
+        # Second refresh
+        resp2 = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp2.status_code == 200
+        access_token_2 = resp2.json()["access_token"]
+
+        # Each refresh should produce a new access token
+        assert access_token_1 != access_token_2
+
+    def test_refresh_invalid_token(self):
+        """An invalid (garbage) refresh token returns 401."""
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": "not.a.jwt.token"},
+        )
+        assert resp.status_code == 401
+        assert "Invalid or expired refresh token" in resp.json()["detail"]
+
+    def test_refresh_empty_body(self):
+        """Missing refresh_token field returns 422 validation error."""
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            json={},
+        )
+        assert resp.status_code == 422
+
+    def test_refresh_missing_field(self):
+        """No body at all returns 422."""
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            content="{}",
+            headers={"Content-Type": "application/json"},
+        )
+        assert resp.status_code == 422
+
+    def test_refresh_with_login_token(self):
+        """A refresh token obtained from login also works for /refresh."""
+        client = TestClient(app)
+        # Register first
+        client.post(
+            "/api/auth/register",
+            json={
+                "username": "refep_loginrefuser",
+                "email": "loginref@example.com",
+                "password": "secret123",
+                "date_of_birth": "2000-01-01",
+            },
+        )
+        # Login to get a refresh token
+        login_resp = client.post(
+            "/api/auth/login",
+            data={"username": "refep_loginrefuser", "password": "secret123"},
+        )
+        assert login_resp.status_code == 200
+        refresh_token = login_resp.json()["refresh_token"]
+
+        # Use it to refresh
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp.status_code == 200
+        assert "access_token" in resp.json()
+
+    def test_refresh_access_token_is_valid_jwt(self):
+        """The access token returned by /refresh is a valid JWT."""
+        from jose import jwt
+
+        refresh_token = self._register_user("refep_jwttoken", "secret123")
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp.status_code == 200
+        access_token = resp.json()["access_token"]
+
+        # Verify it's a 3-part JWT
+        parts = access_token.split(".")
+        assert len(parts) == 3
+
+        # Verify it decodes with the public key
+        payload = jwt.decode(
+            access_token,
+            settings.public_pem,
+            algorithms=[settings.ALGORITHM],
+        )
+        assert "sub" in payload
+        assert "exp" in payload
+
+    def test_refresh_expires_minutes_in_response(self):
+        """The response includes access_token_expires_minutes."""
+        refresh_token = self._register_user("refep_expiretest", "secret123")
+        client = TestClient(app)
+        resp = client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+        assert resp.status_code == 200
+        assert "access_token_expires_minutes" in resp.json()
+        assert resp.json()["access_token_expires_minutes"] == settings.ACCESS_TOKEN_EXPIRE_MINUTES

backend/tests/test_refresh_token_jti.py

@@ -0,0 +1,79 @@
+"""Tests for refresh token uniqueness (jti claim) and model column length."""
+
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from jose import jwt
+
+from app.config import settings
+from app.routers.auth import _create_refresh_token
+from app.models.session import RefreshToken
+
+
+class TestRefreshTokenJtiUniqueness:
+    """Verify that _create_refresh_token produces unique tokens even when
+    called with identical data in rapid succession (issue #1 from review)."""
+
+    def test_two_calls_produce_different_tokens(self):
+        """Two calls with the same sub produce different JWTs because of jti."""
+        token_a = _create_refresh_token({"sub": "42"})
+        token_b = _create_refresh_token({"sub": "42"})
+        assert token_a != token_b
+
+    def test_jti_is_valid_uuid(self):
+        """The jti claim inside the refresh token is a valid UUID."""
+        token = _create_refresh_token({"sub": "1"})
+        payload = jwt.decode(
+            token,
+            settings.public_pem,
+            algorithms=[settings.ALGORITHM],
+        )
+        assert "jti" in payload
+        # Verify it parses as a UUID (raises ValueError if not)
+        uuid.UUID(payload["jti"])
+
+    def test_each_token_has_unique_jti(self):
+        """Generate 100 tokens for the same user — all jtis must differ."""
+        jtis = set()
+        for _ in range(100):
+            token = _create_refresh_token({"sub": "999"})
+            payload = jwt.decode(
+                token,
+                settings.public_pem,
+                algorithms=[settings.ALGORITHM],
+            )
+            jtis.add(payload["jti"])
+        assert len(jtis) == 100
+
+    def test_jti_survives_decode(self):
+        """The jti claim is present and correct after decode."""
+        token = _create_refresh_token({"sub": "7"})
+        payload = jwt.decode(
+            token,
+            settings.public_pem,
+            algorithms=[settings.ALGORITHM],
+        )
+        assert payload["sub"] == "7"
+        assert "jti" in payload
+        assert "exp" in payload
+
+
+class TestRefreshTokenModelColumnLength:
+    """Verify the model's token column length matches the migration (issue #2)."""
+
+    def test_token_column_has_length_512(self):
+        """RefreshToken.token column should be String(512), not unbounded."""
+        from sqlalchemy import inspect
+
+        # Get the column definition from the model's table
+        column = RefreshToken.__table__.columns["token"]
+        # The type should be String with length 512
+        assert column.type.length == 512
+
+    def test_token_column_type_is_string(self):
+        """The token column should be a String type."""
+        from sqlalchemy import String as SAString
+
+        column = RefreshToken.__table__.columns["token"]
+        assert isinstance(column.type, SAString)