daddy/modbison
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Implement /auth/refresh endpoint' (#37) from agent/82c3ebd03921 into main

Browse Source
This commit was merged in pull request #37.
This commit is contained in:
reviewer-bot
2026-08-19 22:39:42 +00:00
parent 7d89e860fd 3cb4d01073
commit 5c21e5051c
2 changed files with 280 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/routers/auth.py
+79 -2
View File
@@ -5,7 +5,7 @@ from datetime import date, datetime, timedelta, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
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
@@ -31,6 +31,10 @@ class TokenResponse(BaseModel):
access_token_expires_minutes: int
class RefreshRequest(BaseModel):
refresh_token: str
class RegisterRequest(BaseModel):
username: str
email: EmailStr
@@ -62,8 +66,15 @@ class UserResponse(BaseModel):
# --- Helpers ---
def _create_access_token(data: dict) -> str:
"""Create a signed access token using RS256."""
"""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
)
@@ -232,6 +243,72 @@ async def login(form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession =
)
@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)
async def get_current_user(current_user: CurrentUser):
"""Get current user profile from the JWT token."""
backend/tests/test_refresh_endpoint.py
+201
View File
@@ -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