FastAPI Authentication & Authorization Complete authentication system with JWT tokens, OAuth2, and role-based access control. ## Usage ``bash # Install auth dependencies pip install python-jose[cryptography] passlib[bcrypt] python-multipart # Generate secret key openssl rand -hex 32 ` ## JWT Configuration `python # app/core/security.py from datetime import datetime, timedelta from typing import Optional, Union, Any from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import settings # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # JWT settings SECRET_KEY = settings.SECRET_KEY ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 REFRESH_TOKEN_EXPIRE_DAYS = 7 def create_access_token( subject: Union[str, Any], expires_delta: Optional[timedelta] = None ) -> str: """Create JWT access token.""" if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode = {"exp": expire, "sub": str(subject), "type": "access"} encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def create_refresh_token(subject: Union[str, Any]) -> str: """Create JWT refresh token.""" expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"} encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify password against hash.""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """Generate password hash.""" return pwd_context.hash(password) def decode_token(token: str) -> Optional[dict]: """Decode and verify JWT token.""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: return None ` ## Authentication Dependencies ``python # app/api/dependencies/auth.py from typing import Optional from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import decodetoken from app.db.database import getdb from app.models.user import User from app.repositories.user import UserRepository # OAuth2 scheme oauth2scheme = OAuth2PasswordBearer( tokenUrl="/api/v1/auth/login", schemename="JWT" ) # Bearer token scheme security = HTTPBearer() async def getcurrentuser( token: str = Depends(oauth2scheme), db: AsyncSession = Depends(getdb) ) -> User: """Get current authenticated user.""" credentialsexception = HTTPException( statuscode=status.HTTP401UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) payload = decodetoken(token) if payload is None: raise credentialsexception userid: str = payload.get("sub") tokentype: str = payload.get("type") if userid is None or tokentype != "access": raise credentialsexception userrepo = UserRepository(User, db) user = await userrepo.get(int(userid)) if user is None: raise credentialsexception if not user.isactive: raise HTTPException( statuscode=status.HTTP400BADREQUEST, detail="Inactive user" ) return user async def getcurrentactiveuser( currentuser: User = Depends(getcurrentuser) ) -> User: """Get current active user.""" if not currentuser.isactive: raise HTTPException( statuscode=status.HTTP400BADREQUEST, detail="Inactive user" ) return currentuser async def getcurrentsuperuser( currentuser: User = Depends(getcurrentuser) ) -> User: """Get current superuser.""" if not currentuser.issuperuser: raise HTTPException( statuscode=status.HTTP403FORBIDDEN, detail="Not enough permissions" ) return currentuser def requirepermissions(*permissions: str): """Decorator for permission-based access control.""" async def permissionchecker( currentuser: User = Depends(getcurrentactiveuser) ) -> User: # Check if user has required permissions userpermissions = set(currentuser.permissions or []) requiredpermissions = set(permissions) if not requiredpermissions.issubset(userpermissions) and not currentuser.issuperuser: raise HTTPException( statuscode=status.HTTP403FORBIDDEN, detail="Insufficient permissions" ) return curr