Python FastAPI 9: Xác thực và ủy quyền – kỹ năng cần thiết để bảo mật API của bạn

pip install "passlib[bcrypt]" "python-jose[cryptography]" python-multipart

# app/security.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash password trước khi lưu DB"""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Kiểm tra password khi login"""
    return pwd_context.verify(plain_password, hashed_password)

# Khi đăng ký
password = "super_secret_123"
hashed = hash_password(password)
print(hashed)
# $2b$12$KIXxP.8dG...rất-dài...

# Khi đăng nhập
if verify_password("super_secret_123", hashed):
    print("Password đúng")
else:
    print("Password sai")

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjMiLCJleHAiOjE3...MzI.v8YD...xP2Q

# app/security.py
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from app.config import settings

ALGORITHM = "HS256"

def create_access_token(
    subject: str,
    expires_delta: timedelta = None,
    extra_claims: dict = None,
) -> str:
    """Tạo JWT access token"""
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=settings.access_token_expire_minutes)
    )

    payload = {
        "sub": str(subject),       # subject - thường là user_id
        "exp": expire,             # expiration time
        "iat": datetime.now(timezone.utc),  # issued at
        "type": "access",          # token type
    }

    if extra_claims:
        payload.update(extra_claims)

    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)


def decode_token(token: str) -> dict:
    """Decode và verify JWT. Raise JWTError nếu token không hợp lệ."""
    return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
import secrets

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    app_name: str = "TaskAPI"
    # QUAN TRỌNG: secret_key phải dài, ngẫu nhiên, và BÍ MẬT
    secret_key: str = secrets.token_urlsafe(32)
    access_token_expire_minutes: int = 30
    refresh_token_expire_days: int = 7

settings = Settings()

python -c "import secrets; print(secrets.token_urlsafe(32))"

# app/models/user.py
from datetime import datetime
from enum import Enum
from typing import Optional, List
from pydantic import BaseModel, EmailStr, Field

class UserRole(str, Enum):
    user = "user"
    admin = "admin"

class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str = Field(..., min_length=8, max_length=72)
    # Chú ý 72: giới hạn của bcrypt

class UserResponse(UserBase):
    id: int
    role: UserRole
    is_active: bool
    created_at: datetime

class UserInDB(UserResponse):
    """User với hashed password - CHỈ dùng nội bộ, không expose ra API"""
    hashed_password: str


class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenPayload(BaseModel):
    sub: str  # user_id
    exp: int
    type: str

# app/services/user_service.py
from datetime import datetime, timezone
from typing import Optional
from app.models.user import UserCreate, UserInDB, UserRole
from app.security import hash_password, verify_password
from app.exceptions import ConflictError, UnauthorizedError

_users_db: dict[int, UserInDB] = {}
_next_id = 1

class UserService:
    @staticmethod
    def create_user(data: UserCreate) -> UserInDB:
        global _next_id

        # Check email và username đã tồn tại
        if any(u.email == data.email for u in _users_db.values()):
            raise ConflictError(detail="Email đã được đăng ký")
        if any(u.username == data.username for u in _users_db.values()):
            raise ConflictError(detail="Username đã tồn tại")

        user = UserInDB(
            id=_next_id,
            email=data.email,
            username=data.username,
            full_name=data.full_name,
            role=UserRole.user,
            is_active=True,
            hashed_password=hash_password(data.password),
            created_at=datetime.now(timezone.utc),
        )
        _users_db[_next_id] = user
        _next_id += 1
        return user

    @staticmethod
    def get_by_id(user_id: int) -> Optional[UserInDB]:
        return _users_db.get(user_id)

    @staticmethod
    def get_by_username(username: str) -> Optional[UserInDB]:
        return next(
            (u for u in _users_db.values() if u.username == username),
            None,
        )

    @staticmethod
    def authenticate(username: str, password: str) -> UserInDB:
        """Xác thực user với username + password"""
        user = UserService.get_by_username(username)
        if not user:
            # Đừng để lộ rằng user không tồn tại
            raise UnauthorizedError(detail="Username hoặc password không đúng")

        if not verify_password(password, user.hashed_password):
            raise UnauthorizedError(detail="Username hoặc password không đúng")

        if not user.is_active:
            raise UnauthorizedError(detail="Tài khoản đã bị vô hiệu hoá")

        return user

# app/routers/auth.py
from datetime import timedelta
from fastapi import APIRouter, Depends, status
from fastapi.security import OAuth2PasswordRequestForm
from app.models.user import Token, UserCreate, UserResponse
from app.services.user_service import UserService
from app.security import create_access_token
from app.config import settings

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post(
    "/register",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
def register(data: UserCreate):
    user = UserService.create_user(data)
    return user


@router.post("/login", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    Login với username + password.
    Response: access_token + refresh_token.
    """
    user = UserService.authenticate(form_data.username, form_data.password)

    access_token = create_access_token(
        subject=user.id,
        expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
        extra_claims={"role": user.role.value},
    )
    refresh_token = create_access_token(
        subject=user.id,
        expires_delta=timedelta(days=settings.refresh_token_expire_days),
        extra_claims={"type": "refresh"},
    )

    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
    )

# app/routers/auth.py
from pydantic import BaseModel
from jose import JWTError
from app.security import decode_token

class RefreshRequest(BaseModel):
    refresh_token: str

@router.post("/refresh", response_model=Token)
def refresh_access_token(data: RefreshRequest):
    try:
        payload = decode_token(data.refresh_token)
    except JWTError:
        raise UnauthorizedError(detail="Refresh token không hợp lệ")

    if payload.get("type") != "refresh":
        raise UnauthorizedError(detail="Token không phải refresh token")

    user_id = int(payload["sub"])
    user = UserService.get_by_id(user_id)
    if not user or not user.is_active:
        raise UnauthorizedError(detail="User không tồn tại hoặc đã bị vô hiệu hoá")

    # Tạo access token mới
    access_token = create_access_token(
        subject=user.id,
        extra_claims={"role": user.role.value},
    )
    # Có thể tạo luôn refresh token mới (token rotation - an toàn hơn)
    new_refresh_token = create_access_token(
        subject=user.id,
        expires_delta=timedelta(days=settings.refresh_token_expire_days),
        extra_claims={"type": "refresh"},
    )

    return Token(access_token=access_token, refresh_token=new_refresh_token)

# app/dependencies.py
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from app.security import decode_token
from app.services.user_service import UserService
from app.models.user import UserInDB
from app.exceptions import UnauthorizedError, ForbiddenError

# tokenUrl chỉ cho Swagger UI biết endpoint login ở đâu
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


def get_current_user(token: str = Depends(oauth2_scheme)) -> UserInDB:
    """
    Extract và verify JWT, return user.
    Dùng cho mọi endpoint cần authenticated user.
    """
    try:
        payload = decode_token(token)
    except JWTError:
        raise UnauthorizedError(detail="Token không hợp lệ hoặc đã hết hạn")

    if payload.get("type") != "access":
        raise UnauthorizedError(detail="Token type không hợp lệ")

    user_id_str = payload.get("sub")
    if not user_id_str:
        raise UnauthorizedError(detail="Token không chứa thông tin user")

    user = UserService.get_by_id(int(user_id_str))
    if not user:
        raise UnauthorizedError(detail="User không tồn tại")

    if not user.is_active:
        raise UnauthorizedError(detail="Tài khoản đã bị vô hiệu hoá")

    return user


def get_current_active_user(
    user: UserInDB = Depends(get_current_user),
) -> UserInDB:
    """Alias cho trường hợp muốn rõ ràng user phải active"""
    return user

# app/routers/users.py
from fastapi import APIRouter, Depends
from app.models.user import UserResponse, UserInDB
from app.dependencies import get_current_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=UserResponse)
def get_me(current_user: UserInDB = Depends(get_current_user)):
    """Lấy thông tin user đang đăng nhập"""
    return current_user

# app/dependencies.py
from app.models.user import UserRole

def require_role(*allowed_roles: UserRole):
    """
    Factory tạo dependency kiểm tra role.
    Cách dùng: Depends(require_role(UserRole.admin))
    """
    def dependency(user: UserInDB = Depends(get_current_user)) -> UserInDB:
        if user.role not in allowed_roles:
            raise ForbiddenError(
                detail=f"Yêu cầu role: {', '.join(r.value for r in allowed_roles)}"
            )
        return user
    return dependency


# Các shortcut thường dùng
require_admin = require_role(UserRole.admin)

from app.dependencies import require_admin, require_role

@router.delete("/users/{user_id}")
def delete_user(
    user_id: int,
    admin: UserInDB = Depends(require_admin),
):
    # Chỉ admin mới vào được đây
    ...

# Hoặc cho phép nhiều role
@router.get("/reports")
def get_reports(
    user: UserInDB = Depends(require_role(UserRole.admin, UserRole.manager)),
):
    ...

# app/dependencies.py
from app.services.task_service import TaskService
from app.exceptions import TaskNotFoundError, ForbiddenError

def get_task_with_permission(
    task_id: int,
    user: UserInDB = Depends(get_current_user),
) -> dict:
    """
    Lấy task và kiểm tra user có quyền truy cập.
    Rules:
    - Admin: truy cập mọi task.
    - User: chỉ task của mình (owner_id == user.id).
    """
    task = TaskService.get_task(task_id)  # raise TaskNotFoundError nếu không có

    # Admin bypass
    if user.role == UserRole.admin:
        return task

    # User thường: phải là owner
    if task["owner_id"] != user.id:
        # Quan trọng: trả 404 thay vì 403 để không leak sự tồn tại của task
        raise TaskNotFoundError(task_id)

    return task

@router.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task: dict = Depends(get_task_with_permission)):
    return task

@router.delete("/tasks/{task_id}", status_code=204)
def delete_task(
    task: dict = Depends(get_task_with_permission),
    user: UserInDB = Depends(get_current_user),
):
    # Đã kiểm tra permission trong dependency
    TaskService.delete(task["id"])
    return None

from fastapi import Security
from fastapi.security import SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/api/v1/auth/login",
    scopes={
        "tasks:read": "Đọc tasks",
        "tasks:write": "Tạo/sửa/xoá tasks",
        "users:read": "Đọc thông tin users",
        "admin": "Quyền admin",
    },
)

def get_current_user_with_scopes(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
) -> UserInDB:
    try:
        payload = decode_token(token)
    except JWTError:
        raise UnauthorizedError(detail="Token không hợp lệ")

    token_scopes = payload.get("scopes", [])
    user_id = payload.get("sub")

    # Check user tồn tại
    user = UserService.get_by_id(int(user_id))
    if not user:
        raise UnauthorizedError(detail="User không tồn tại")

    # Check scopes yêu cầu có trong token không
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise ForbiddenError(
                detail=f"Thiếu scope yêu cầu: {scope}",
            )

    return user


# Sử dụng
@router.get("/tasks")
def list_tasks(
    user: UserInDB = Security(
        get_current_user_with_scopes,
        scopes=["tasks:read"],
    ),
):
    ...

from pydantic import BaseModel, Field, field_validator

class UserCreate(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)
    password: str = Field(..., min_length=8, max_length=72)

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if v.lower() in {"password", "12345678", "qwerty", "abc12345"}:
            raise ValueError("Password quá yếu")
        if len(set(v)) < 4:
            raise ValueError("Password phải có ít nhất 4 ký tự khác nhau")
        return v

class ChangePasswordRequest(BaseModel):
    current_password: str
    new_password: str = Field(..., min_length=8, max_length=72)

@router.post("/me/change-password")
def change_password(
    data: ChangePasswordRequest,
    user: UserInDB = Depends(get_current_user),
):
    # QUAN TRỌNG: verify current password trước khi đổi
    # Nếu access token bị đánh cắp, kẻ tấn công không đổi được password
    # mà không biết current password
    if not verify_password(data.current_password, user.hashed_password):
        raise UnauthorizedError(detail="Password hiện tại không đúng")

    if data.current_password == data.new_password:
        raise BusinessRuleError(detail="Password mới phải khác password cũ")

    # Update hashed password
    user.hashed_password = hash_password(data.new_password)
    UserService.update(user)

    # Tốt nhất: revoke mọi access token và refresh token hiện tại
    # (cần token blacklist hoặc token versioning - xem phần 14)

    return {"message": "Đổi password thành công"}

import redis

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

def revoke_token(token_jti: str, expire_seconds: int):
    """Thêm token vào blacklist"""
    redis_client.setex(f"blacklist:{token_jti}", expire_seconds, "1")

def is_token_revoked(token_jti: str) -> bool:
    return redis_client.exists(f"blacklist:{token_jti}") > 0


# Thêm claim "jti" (JWT ID) khi tạo token
import uuid
def create_access_token(subject, ...):
    payload = {
        "sub": str(subject),
        "jti": str(uuid.uuid4()),  # unique ID cho mỗi token
        "exp": expire,
        ...
    }
    return jwt.encode(payload, ...)

# Check blacklist trong get_current_user
def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = decode_token(token)

    if is_token_revoked(payload["jti"]):
        raise UnauthorizedError(detail="Token đã bị thu hồi")
    ...

@app.middleware("http")
async def security_headers(request, call_next):
    response = await call_next(request)

    # Ngăn XSS và clickjacking
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"

    # HTTPS strict (chỉ bật khi thực sự dùng HTTPS)
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"

    # Chặn browser gửi Referer header chứa URL nhạy cảm
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

    return response

pip install slowapi

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@router.post("/login")
@limiter.limit("5/minute")
def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    ...

@router.post("/register")
@limiter.limit("3/hour")
def register(request: Request, data: UserCreate):
    ...

# app/main.py
from fastapi import FastAPI
from app.error_handlers import register_exception_handlers
from app.routers import auth, users, tasks

app = FastAPI(title="TaskAPI", version="0.2.0")
register_exception_handlers(app)

app.include_router(auth.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")
app.include_router(tasks.router, prefix="/api/v1")

# app/routers/tasks.py
from fastapi import APIRouter, Depends, status
from app.models.task import TaskCreate, TaskResponse
from app.models.user import UserInDB
from app.dependencies import (
    get_current_user,
    get_task_with_permission,
    require_admin,
)
from app.services.task_service import TaskService

router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.get("/", response_model=list[TaskResponse])
def list_tasks(user: UserInDB = Depends(get_current_user)):
    """User chỉ thấy task của mình, admin thấy tất cả"""
    if user.role == "admin":
        return TaskService.get_all()
    return TaskService.get_by_owner(user.id)

@router.post("/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
def create_task(
    data: TaskCreate,
    user: UserInDB = Depends(get_current_user),
):
    return TaskService.create(data, owner_id=user.id)

@router.get("/{task_id}", response_model=TaskResponse)
def get_task(task: dict = Depends(get_task_with_permission)):
    return task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(
    task: dict = Depends(get_task_with_permission),
):
    TaskService.delete(task["id"])

@router.delete("/admin/purge-completed", status_code=status.HTTP_204_NO_CONTENT)
def purge_completed_tasks(admin: UserInDB = Depends(require_admin)):
    """Chỉ admin: xoá mọi task đã hoàn thành"""
    TaskService.delete_all_completed()