mkdir fastapi-auth-demo && cd fastapi-auth-demo

# Tạo môi trường ảo
python -m venv venv

# Kích hoạt nó
./venv/Scripts/activate # Windows
source venv/bin/activate # macOS/Linux

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
itsdangerous==2.2.0

pip install -r ./requirements.txt

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="Mô phỏng xác thực FastAPI", version="1.0.0")

# Thêm middleware CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

security = HTTPBasic()

fake_users_db = {
    "john": {"username": "john", "password": "secret123", "role": "user"},
    "admin": {"username": "admin", "password": "supersecret", "role": "admin"}
}

def authenticate_user(credentials: HTTPBasicCredentials = Depends(security)):
    user = fake_users_db.get(credentials.username)
    if not user or not secrets.compare_digest(credentials.password, user["password"]):
        raise HTTPException(status_code=401, detail="Thông tin xác thực không hợp lệ")
    return user

@app.get("/basic/profile")
def read_profile(user: dict = Depends(authenticate_user)):
    return {"message": f"Chào mừng {user['username']}!"}

@app.get("/basic/admin")
def admin_area(user: dict = Depends(authenticate_user)):
    if user["role"] != "admin":
       raise HTTPException(status_code=403, detail="Chỉ dành cho quản trị viên")
    return {"message": "Bạn đang ở khu vực quản trị."}

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

api_key_header = HTTPBearer()

fake_api_keys = {
    "sk_test_123": {"key_id": "1", "permissions": ["read", "write"]},
    "sk_live_456": {"key_id": "2", "permissions": ["read"]}
}

def validate_api_key(credentials: HTTPAuthorizationCredentials = Depends(api_key_header)):
    key = credentials.credentials
    if key not in fake_api_keys:
        raise HTTPException(status_code=401, detail="API Key không hợp lệ")
    return fake_api_keys[key]

@app.get("/apikey/data")
def read_data(api_key=Depends(validate_api_key)):
    return {"data": ["item1", "item2"]}

@app.post("/apikey/create")
def create_item(item: dict, api_key=Depends(validate_api_key)):
    if "write" not in api_key["permissions"]:
        raise HTTPException(status_code=403, detail="Cần quyền ghi")
    return {"message": "Đã tạo item", "item": item}

from fastapi import FastAPI, Request, Form, Depends, HTTPException, status
from fastapi.responses import JSONResponse
from starlette.middleware.sessions import SessionMiddleware
from pydantic import BaseModel

# Thêm middleware phiên
app.add_middleware(SessionMiddleware, secret_key="super-secret-session-key")

# Cơ sở dữ liệu người dùng mô phỏng
users_db = {
    "alice": "password123",
    "bob": "securepass"
}

class LoginSchema(BaseModel):
    username: str
    password: str

@app.post("/login")
async def login(request: Request, payload: LoginSchema):
    username = payload.username
    password = payload.password

    if username in users_db and users_db[username] == password:
        request.session["user"] = username
        return JSONResponse({"message": "Đăng nhập thành công"})
    raise HTTPException(status_code=401, detail="Thông tin xác thực không hợp lệ")

@app.get("/dashboard")
async def dashboard(request: Request):
    user = request.session.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Chưa xác thực")
    return {"message": f"Chào mừng đến bảng điều khiển của bạn, {user}!"}

@app.post("/logout")
async def logout(request: Request):
    request.session.clear()
    return {"message": "Đăng xuất thành công"}

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    """Bộ xử lý ngoại lệ HTTP tùy chỉnh"""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "message": exc.detail,
                "type": "authentication_error" if exc.status_code == 401 else "authorization_error",
                "status_code": exc.status_code
            }
        },
        headers=exc.headers
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """Xử lý lỗi xác thực"""
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "message": "Lỗi xác thực",
                "type": "validation_error",
                "details": exc.errors()
            }
        }
    )

from passlib.context import CryptContext

# Sử dụng mã hóa mật khẩu đúng cách
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

from collections import defaultdict
import time

# Giới hạn tốc độ đơn giản (sử dụng Redis trong sản xuất)
request_counts = defaultdict(list)

def rate_limit(max_requests: int = 100, window_minutes: int = 15):
    def decorator(func):
        def wrapper(*args, **kwargs):
            client_ip = "127.0.0.1"
            now = time.time()
            window_start = now - (window_minutes * 60)

            # Xóa các yêu cầu cũ
            request_counts[client_ip] = [
                req_time for req_time in request_counts[client_ip]
                if req_time > window_start
            ]

            if len(request_counts[client_ip]) >= max_requests:
                raise HTTPException(
                    status_code=429,
                    detail="Giới hạn tốc độ đã vượt quá"
                )

            request_counts[client_ip].append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    secret_key: str = "your-secret-key-here"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"

settings = Settings()