أفضل ممارسات أمان واجهة برمجة التطبيقات لعام 2026: أفضل 10 ممارسات OWASP وJWT وتحديد المعدل

from datetime import datetime, timedelta, timezone
from typing import Optional
import jwt
import secrets
import hashlib

SECRET_KEY = secrets.token_hex(32)  # Store in env var!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = timedelta(days=30)

def create_access_token(user_id: int, scopes: list[str] = []) -> str:
    payload = {
        "sub": str(user_id),
        "type": "access",
        "scopes": scopes,
        "iat": datetime.now(timezone.utc),
        "exp": datetime.now(timezone.utc) + ACCESS_TOKEN_EXPIRE,
        "jti": secrets.token_hex(16),  # JWT ID for revocation
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Check if token is revoked (requires Redis/DB lookup)
        if is_token_revoked(payload["jti"]):
            raise jwt.InvalidTokenError("Token revoked")
        return payload
    except jwt.ExpiredSignatureError:
        raise ValueError("Token expired")
    except jwt.InvalidTokenError as e:
        raise ValueError(f"Invalid token: {e}")

# NEVER store secrets in JWTs
# ALWAYS verify signature
# Use short expiry + refresh tokens
# Store sensitive data server-side, not in JWT

from fastapi import FastAPI, Request, HTTPException
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 = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/api/auth/login")
@limiter.limit("5/minute")  # 5 login attempts per minute per IP
async def login(request: Request, credentials: LoginCredentials):
    pass

@app.get("/api/users")
@limiter.limit("100/minute")  # 100 requests per minute
async def list_users(request: Request, current_user: User = Depends(get_current_user)):
    pass

# Also implement:
# - Per-user rate limits (not just per-IP)
# - Exponential backoff on repeated auth failures
# - Account lockout after N failed attempts

from pydantic import BaseModel, EmailStr, Field, validator, field_validator
import re

class CreateUserRequest(BaseModel):
    name: str = Field(min_length=2, max_length=50, pattern=r"^[a-zA-Z\s'-]+$")
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    age: int = Field(ge=13, le=120)

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not re.search(r"[A-Z]", v):
            raise ValueError("Password needs uppercase letter")
        if not re.search(r"[0-9]", v):
            raise ValueError("Password needs a digit")
        if not re.search(r"[!@#$%^&*]", v):
            raise ValueError("Password needs special character")
        return v

    @field_validator("name")
    @classmethod
    def sanitize_name(cls, v: str) -> str:
        # Strip dangerous characters even if pattern matches
        return v.strip()

# NEVER trust client input
# Validate all fields: type, format, length, range
# Reject unexpected fields (use strict mode)
# Sanitize before storage and display

# VULNERABLE: Trusts user-supplied ID
@app.get("/api/invoices/{invoice_id}")
async def get_invoice(invoice_id: int, current_user: User = Depends(get_current_user)):
    return db.invoices.get(invoice_id)  # ANY user can access ANY invoice!

# SECURE: Always verify ownership
@app.get("/api/invoices/{invoice_id}")
async def get_invoice(invoice_id: int, current_user: User = Depends(get_current_user)):
    invoice = db.invoices.get(invoice_id)
    if not invoice:
        raise HTTPException(404, "Not found")
    if invoice.user_id != current_user.id and current_user.role != "admin":
        raise HTTPException(403, "Forbidden")  # Don't reveal existence to unauthorized users
    return invoice

# BEST PRACTICE: Filter by user_id in query
@app.get("/api/invoices/{invoice_id}")
async def get_invoice(invoice_id: int, current_user: User = Depends(get_current_user)):
    invoice = db.invoices.get_for_user(invoice_id, user_id=current_user.id)
    if not invoice:
        raise HTTPException(404, "Not found")
    return invoice

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

# HTTPS redirect (production only)
app.add_middleware(HTTPSRedirectMiddleware)

# CORS: be specific, never use "*"
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "https://www.myapp.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["Authorization", "Content-Type"],
)

# Security headers middleware
@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    # Remove server header
    response.headers.pop("server", None)
    return response

import bcrypt
import secrets
from cryptography.fernet import Fernet

# Password hashing (NEVER store plaintext)
def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=12)).decode()

def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode(), hashed.encode())

# Encrypt sensitive data (credit cards, SSNs, etc.)
ENCRYPTION_KEY = Fernet.generate_key()  # Store in secrets manager!
cipher = Fernet(ENCRYPTION_KEY)

def encrypt_pii(data: str) -> str:
    return cipher.encrypt(data.encode()).decode()

def decrypt_pii(encrypted: str) -> str:
    return cipher.decrypt(encrypted.encode()).decode()

# API key generation
def generate_api_key() -> str:
    return "tp_" + secrets.token_urlsafe(32)

# Mask sensitive data in responses
def mask_credit_card(card: str) -> str:
    return "*" * 12 + card[-4:]

def mask_email(email: str) -> str:
    user, domain = email.split("@")
    return user[:2] + "***@" + domain