pip install passlib[bcrypt]

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

# 密碼雜湊設定
pwd_context = CryptContext(
    schemes=["bcrypt"],
    deprecated="auto"
)


def hash_password(password: str) -> str:
    """將密碼雜湊"""
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """驗證密碼"""
    return pwd_context.verify(plain_password, hashed_password)

from app.core.security import hash_password, verify_password

# 註冊時雜湊密碼
hashed = hash_password("my_password")
print(hashed)
# $2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW

# 登入時驗證密碼
is_valid = verify_password("my_password", hashed)
print(is_valid)  # True

is_valid = verify_password("wrong_password", hashed)
print(is_valid)  # False

from pydantic import BaseModel, field_validator
import re


class PasswordValidator:
    """密碼強度驗證器"""

    MIN_LENGTH = 8
    MAX_LENGTH = 128

    @classmethod
    def validate(cls, password: str) -> str:
        errors = []

        if len(password) < cls.MIN_LENGTH:
            errors.append(f"密碼至少需要 {cls.MIN_LENGTH} 個字元")

        if len(password) > cls.MAX_LENGTH:
            errors.append(f"密碼不能超過 {cls.MAX_LENGTH} 個字元")

        if not re.search(r"[A-Z]", password):
            errors.append("密碼需要包含至少一個大寫字母")

        if not re.search(r"[a-z]", password):
            errors.append("密碼需要包含至少一個小寫字母")

        if not re.search(r"\d", password):
            errors.append("密碼需要包含至少一個數字")

        if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", password):
            errors.append("密碼需要包含至少一個特殊字元")

        if errors:
            raise ValueError("; ".join(errors))

        return password


class UserCreate(BaseModel):
    username: str
    email: str
    password: str

    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        return PasswordValidator.validate(v)

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

app = FastAPI()
security = HTTPBasic()


def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)):
    """驗證 Basic Auth 憑證"""
    # 從資料庫取得使用者（這裡簡化為固定值）
    correct_username = "admin"
    correct_password = "secret"

    # 使用 secrets.compare_digest 防止 timing attack
    username_correct = secrets.compare_digest(
        credentials.username.encode("utf8"),
        correct_username.encode("utf8")
    )
    password_correct = secrets.compare_digest(
        credentials.password.encode("utf8"),
        correct_password.encode("utf8")
    )

    if not (username_correct and password_correct):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Basic"},
        )

    return credentials.username


@app.get("/protected")
def protected_route(username: str = Depends(verify_credentials)):
    return {"message": f"Hello, {username}!"}

pip install itsdangerous

from fastapi import FastAPI, Depends, HTTPException, Response, Request
from fastapi.security import APIKeyCookie
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired
from pydantic import BaseModel

app = FastAPI()

# 設定
SECRET_KEY = "your-super-secret-key"
SESSION_COOKIE_NAME = "session_id"
SESSION_MAX_AGE = 3600  # 1 小時

# Session 序列化器
serializer = URLSafeTimedSerializer(SECRET_KEY)

# Cookie 安全設定
cookie_scheme = APIKeyCookie(name=SESSION_COOKIE_NAME, auto_error=False)


class SessionData(BaseModel):
    user_id: int
    username: str


def create_session(data: SessionData) -> str:
    """建立 Session Token"""
    return serializer.dumps(data.model_dump())


def get_session_data(token: str) -> SessionData | None:
    """解析 Session Token"""
    try:
        data = serializer.loads(token, max_age=SESSION_MAX_AGE)
        return SessionData(**data)
    except (BadSignature, SignatureExpired):
        return None


async def get_current_user(
    session_token: str | None = Depends(cookie_scheme)
) -> SessionData:
    """取得當前使用者"""
    if not session_token:
        raise HTTPException(status_code=401, detail="Not authenticated")

    session_data = get_session_data(session_token)
    if not session_data:
        raise HTTPException(status_code=401, detail="Invalid or expired session")

    return session_data


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


@app.post("/login")
async def login(request: LoginRequest, response: Response):
    """登入"""
    # 驗證帳密（簡化版本）
    if request.username != "admin" or request.password != "password":
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # 建立 Session
    session_data = SessionData(user_id=1, username=request.username)
    session_token = create_session(session_data)

    # 設定 Cookie
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=session_token,
        max_age=SESSION_MAX_AGE,
        httponly=True,      # JavaScript 無法存取
        secure=True,        # 只在 HTTPS 傳送
        samesite="lax",     # 防止 CSRF
    )

    return {"message": "Login successful"}


@app.post("/logout")
async def logout(response: Response):
    """登出"""
    response.delete_cookie(SESSION_COOKIE_NAME)
    return {"message": "Logout successful"}


@app.get("/me")
async def get_me(user: SessionData = Depends(get_current_user)):
    """取得當前使用者資訊"""
    return {"user_id": user.user_id, "username": user.username}

# 選項 1: Cookie（如上）
# - 無狀態
# - 不需要資料庫

# 選項 2: Redis
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379")

async def create_session_redis(user_id: int, data: dict) -> str:
    session_id = secrets.token_urlsafe(32)
    await redis_client.setex(
        f"session:{session_id}",
        SESSION_MAX_AGE,
        json.dumps(data)
    )
    return session_id

async def get_session_redis(session_id: str) -> dict | None:
    data = await redis_client.get(f"session:{session_id}")
    if data:
        return json.loads(data)
    return None

async def delete_session_redis(session_id: str):
    await redis_client.delete(f"session:{session_id}")

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader, APIKeyQuery

app = FastAPI()

# API Key 可以從 Header 或 Query 參數取得
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
api_key_query = APIKeyQuery(name="api_key", auto_error=False)

# 有效的 API Keys（實際應該從資料庫讀取）
VALID_API_KEYS = {
    "key_abc123": {"name": "Service A", "scopes": ["read", "write"]},
    "key_xyz789": {"name": "Service B", "scopes": ["read"]},
}


async def get_api_key(
    api_key_header: str | None = Security(api_key_header),
    api_key_query: str | None = Security(api_key_query),
) -> str:
    """驗證 API Key"""
    api_key = api_key_header or api_key_query

    if not api_key:
        raise HTTPException(
            status_code=401,
            detail="API Key required"
        )

    if api_key not in VALID_API_KEYS:
        raise HTTPException(
            status_code=403,
            detail="Invalid API Key"
        )

    return api_key


def require_scope(required_scope: str):
    """檢查 API Key 是否有特定權限"""
    async def check_scope(api_key: str = Depends(get_api_key)):
        key_info = VALID_API_KEYS[api_key]
        if required_scope not in key_info["scopes"]:
            raise HTTPException(
                status_code=403,
                detail=f"Scope '{required_scope}' required"
            )
        return key_info
    return check_scope


@app.get("/data")
async def read_data(key_info: dict = Depends(require_scope("read"))):
    """需要 read 權限"""
    return {"data": "some data", "accessed_by": key_info["name"]}


@app.post("/data")
async def write_data(key_info: dict = Depends(require_scope("write"))):
    """需要 write 權限"""
    return {"message": "Data written", "by": key_info["name"]}

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
import secrets


class APIKey(Base):
    """API Key Model"""
    __tablename__ = "api_keys"

    id = Column(Integer, primary_key=True)
    key = Column(String(64), unique=True, index=True)
    name = Column(String(100))
    scopes = Column(String(500))  # JSON 或逗號分隔
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    last_used_at = Column(DateTime, nullable=True)
    expires_at = Column(DateTime, nullable=True)


def generate_api_key() -> str:
    """產生 API Key"""
    return f"sk_{secrets.token_urlsafe(32)}"


async def create_api_key(
    db: AsyncSession,
    name: str,
    scopes: list[str],
    expires_in_days: int | None = None
) -> APIKey:
    """建立 API Key"""
    from datetime import timedelta

    key = APIKey(
        key=generate_api_key(),
        name=name,
        scopes=",".join(scopes),
        expires_at=(
            datetime.utcnow() + timedelta(days=expires_in_days)
            if expires_in_days else None
        )
    )
    db.add(key)
    await db.commit()
    return key


async def validate_api_key(db: AsyncSession, key: str) -> APIKey | None:
    """驗證 API Key"""
    from sqlalchemy import select

    stmt = select(APIKey).where(
        APIKey.key == key,
        APIKey.is_active == True
    )
    result = await db.execute(stmt)
    api_key = result.scalar_one_or_none()

    if not api_key:
        return None

    # 檢查是否過期
    if api_key.expires_at and api_key.expires_at < datetime.utcnow():
        return None

    # 更新最後使用時間
    api_key.last_used_at = datetime.utcnow()
    await db.commit()

    return api_key