Merge pull request #63 from igorbenav/token-refresh

Token refresh

README.md

@@ -48,6 +48,7 @@
 - ⚡️ Fully async
 - 🚀 Pydantic V2 and SQLAlchemy 2.0
 - 🔐 User authentication with JWT
+- Cookie based refresh token
 - 🏬 Easy redis caching
 - 👜 Easy client-side caching
 - 🚦 ARQ integration for task queue
@@ -92,7 +93,8 @@
  9. [More Advanced Caching](#59-more-advanced-caching)
  10. [ARQ Job Queues](#510-arq-job-queues)
  11. [Rate Limiting](#511-rate-limiting)
- 12. [Running](#512-running)
+ 12. [JWT Authentication](#512-jwt-authentication)
+ 13. [Running](#512-running)
 6. [Running in Production](#6-running-in-production)
  1. [Uvicorn Workers with Gunicorn](#61-uvicorn-workers-with-gunicorn)
  2. [Running With NGINX](#62-running-with-nginx)
@@ -156,6 +158,7 @@ And then create in `.env`:
 SECRET_KEY= # result of openssl rand -hex 32
 ALGORITHM= # pick an algorithm, default HS256
 ACCESS_TOKEN_EXPIRE_MINUTES= # minutes until token expires, default 30
+REFRESH_TOKEN_EXPIRE_DAYS= # days until token expires, default 7
 ```
 
 Then for the first admin user:
@@ -1251,7 +1254,47 @@ Note that for flexibility (since this is a boilerplate), it's not necessary to p
 > [!WARNING]
 > If a user does not have a `tier` or the tier does not have a defined `rate limit` for the path and the token is still passed to the request, the default `limit` and `period` will be used, this will be saved in `app/logs`.
 
-### 5.12 Running
+### 5.12 JWT Authentication
+#### 5.12.1 Details
+The JWT in this boilerplate is created in the following way:
+1. **JWT Access Tokens:** how you actually access protected resources is passing this token in the request header.
+2. **Refresh Tokens:** you use this type of token to get an `access token`, which you'll use to access protected resources. 
+
+The `access token` is short lived (default 30 minutes) to reduce the damage of a potential leak. The `refresh token`, on the other hand, is long lived (default 7 days), and you use it to renew your `access token` without the need to provide username and password every time it expires.
+
+Since the `refresh token` lasts for a longer time, it's stored as a cookie in a secure way:
+
+```python
+# app/api/v1/login
+
+...
+response.set_cookie(
+ key="refresh_token",
+ value=refresh_token,
+ httponly=True, # Prevent access through JavaScript
+ secure=True, # Ensure cookie is sent over HTTPS only
+ samesite='Lax', # Default to Lax for reasonable balance between security and usability
+ max_age=<number_of_seconds> # Set a max age for the cookie
+)
+...
+```
+
+You may change it to suit your needs. The possible options for `samesite` are:
+- `Lax`: Cookies will be sent in top-level navigations (like clicking on a link to go to another site), but not in API requests or images loaded from other sites.
+- `Strict`: Cookies will be sent in top-level navigations (like clicking on a link to go to another site), but not in API requests or images loaded from other sites.
+- `None`: Cookies will be sent with both same-site and cross-site requests.
+
+#### 5.12.2 Usage
+What you should do with the client is:
+- `Login`: Send credentials to `/api/v1/login`. Store the returned access token in memory for subsequent requests.
+- `Accessing Protected Routes`: Include the access token in the Authorization header.
+- `Token Renewal`: On access token expiry, the front end should automatically call `/api/v1/refresh` for a new token.
+- `Login Again`: If refresh token is expired, credentials should be sent to `/api/v1/login` again, storing the new access token in memory.
+- `Logout`: Call /api/v1/logout to end the session securely.
+
+This authentication setup in the provides a robust, secure, and user-friendly way to handle user sessions in your API applications.
+
+### 5.13 Running
 If you are using docker compose, just running the following command should ensure everything is working:
 ```sh
 docker compose up

src/app/api/v1/login.py

@@ -1,20 +1,28 @@
 from typing import Annotated, Dict
-from datetime import timedelta
+from datetime import timedelta, datetime, timezone
 
-from fastapi import Depends
+from fastapi import Response, Request, Depends
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 import fastapi
 
+from app.core.config import settings
 from app.core.db.database import async_get_db
-from app.core.schemas import Token
-from app.core.security import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, authenticate_user
 from app.core.exceptions.http_exceptions import UnauthorizedException
+from app.core.schemas import Token
+from app.core.security import (
+ ACCESS_TOKEN_EXPIRE_MINUTES, 
+ create_access_token, 
+ authenticate_user, 
+ create_refresh_token,
+ verify_token
+)
 
 router = fastapi.APIRouter(tags=["login"])
 
 @router.post("/login", response_model=Token)
 async def login_for_access_token(
+ response: Response,
  form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
  db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> Dict[str, str]:
@@ -30,5 +38,37 @@ async def login_for_access_token(
  access_token = await create_access_token(
  data={"sub": user["username"]}, expires_delta=access_token_expires
  )
+
+ refresh_token = await create_refresh_token(data={"sub": user["username"]})
+ max_age = settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60
+
+ response.set_cookie(
+ key="refresh_token",
+ value=refresh_token,
+ httponly=True,
+ secure=True,
+ samesite='Lax',
+ max_age=max_age
+ )
 
- return {"access_token": access_token, "token_type": "bearer"}
+ return {
+ "access_token": access_token, 
+ "token_type": "bearer"
+ }
+
+
+@router.post("/refresh")
+async def refresh_access_token(
+ request: Request,
+ db: AsyncSession = Depends(async_get_db)
+) -> Dict[str, str]:
+ refresh_token = request.cookies.get("refresh_token")
+ if not refresh_token:
+ raise UnauthorizedException("Refresh token missing.")
+
+ user_data = await verify_token(refresh_token, db)
+ if not user_data:
+ raise UnauthorizedException("Invalid refresh token.")
+
+ new_access_token = await create_access_token(data={"sub": user_data.username_or_email})
+ return {"access_token": new_access_token, "token_type": "bearer"}

src/app/api/v1/logout.py

@@ -1,8 +1,6 @@
 from typing import Dict
 
-from datetime import datetime
-
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Response, Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 from jose import JWTError
 
@@ -14,11 +12,14 @@
 
 @router.post("/logout")
 async def logout(
- token: str = Depends(oauth2_scheme),
+ response: Response,
+ access_token: str = Depends(oauth2_scheme),
  db: AsyncSession = Depends(async_get_db)
 ) -> Dict[str, str]:
  try:
- await blacklist_token(token=token, db=db)
+ await blacklist_token(token=access_token, db=db)
+ response.delete_cookie(key="refresh_token")
+
  return {"message": "Logged out successfully"}
 
  except JWTError:

src/app/core/config.py

@@ -16,8 +16,9 @@ class AppSettings(BaseSettings):
 
 class CryptSettings(BaseSettings):
  SECRET_KEY: str = config("SECRET_KEY")
- ALGORITHM: str = config("ALGORITHM")
- ACCESS_TOKEN_EXPIRE_MINUTES: int = config("ACCESS_TOKEN_EXPIRE_MINUTES")
+ ALGORITHM: str = config("ALGORITHM", default="HS256")
+ ACCESS_TOKEN_EXPIRE_MINUTES: int = config("ACCESS_TOKEN_EXPIRE_MINUTES", default=30)
+ REFRESH_TOKEN_EXPIRE_DAYS: int = config("REFRESH_TOKEN_EXPIRE_DAYS", default=7)
 
 
 class DatabaseSettings(BaseSettings):

src/app/core/db/models.py

@@ -11,6 +11,7 @@ class TimestampMixin:
  created_at: datetime = Column(DateTime, default=datetime.utcnow, server_default=text("current_timestamp(0)"))
  updated_at: datetime = Column(DateTime, nullable=True, onupdate=datetime.utcnow, server_default=text("current_timestamp(0)"))
 
+
 class SoftDeleteMixin:
  deleted_at: datetime = Column(DateTime, nullable=True)
  is_deleted: bool = Column(Boolean, default=False)

src/app/core/security.py

@@ -14,6 +14,7 @@
 SECRET_KEY = settings.SECRET_KEY
 ALGORITHM = settings.ALGORITHM
 ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES
+REFRESH_TOKEN_EXPIRE_DAYS = settings.REFRESH_TOKEN_EXPIRE_DAYS
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/login")
 crypt_context = CryptContext(schemes=["sha256_crypt"])
@@ -50,6 +51,16 @@ async def create_access_token(data: dict[str, Any], expires_delta: timedelta | N
  encoded_jwt: str = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  return encoded_jwt
 
+async def create_refresh_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str:
+ to_encode = data.copy()
+ if expires_delta:
+ expire = datetime.utcnow() + expires_delta
+ else:
+ expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+ to_encode.update({"exp": expire})
+ encoded_jwt: str = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+ return encoded_jwt
+
 async def verify_token(token: str, db: AsyncSession) -> TokenData | None:
  """
  Verify a JWT token and return TokenData if valid.