امنیت میکروسرویس

۱۳.۱ مقدمه

امنیت در میکروسرویس‌ها چالش‌برانگیزتر از monolith است. ترافیک بین سرویس‌ها بیشتر، attack surface بزرگ‌تر، و trust boundaries پیچیده‌تر هستند.

۱۳.۲ Zero Trust Architecture

«Never trust, always verify» — هیچ سرویسی به‌صورت پیش‌فرض trusted نیست، حتی داخل cluster.

اصول Zero Trust

• Identity verification. هر سرویس identity مشخص داشته باشد.
• Least privilege. فقط دسترسی لازم.
• Encryption everywhere. mTLS بین همه سرویس‌ها.
• Continuous monitoring. همه ترافیک log و audit شود.
• Network segmentation. سرویس‌ها فقط با ضرورت با هم ارتباط برقرار کنند.

۱۳.۳ Authentication و Authorization

JWT (JSON Web Token)

import jwt
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

JWT_SECRET = os.getenv("JWT_SECRET")
JWT_ALGORITHM = "RS256"  # RSA — بهتر از HS256 برای میکروسرویس
JWT_ACCESS_TTL = timedelta(minutes=15)
JWT_REFRESH_TTL = timedelta(days=30)

def create_tokens(user_id: int, role: str) -> dict:
    now = datetime.utcnow()
    
    access_payload = {
        "sub": str(user_id),
        "role": role,
        "type": "access",
        "iat": now,
        "exp": now + JWT_ACCESS_TTL,
        "iss": "auth-service",
    }
    
    refresh_payload = {
        "sub": str(user_id),
        "type": "refresh",
        "iat": now,
        "exp": now + JWT_REFRESH_TTL,
        "jti": str(uuid.uuid4()),  # برای revoke
    }
    
    # با private key امضا
    with open("private.pem") as f:
        private_key = f.read()
    
    return {
        "access_token": jwt.encode(access_payload, private_key, algorithm=JWT_ALGORITHM),
        "refresh_token": jwt.encode(refresh_payload, private_key, algorithm=JWT_ALGORITHM),
    }

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
    token = credentials.credentials
    try:
        # با public key verify (سرویس‌های دیگر فقط نیاز به public key دارند)
        with open("public.pem") as f:
            public_key = f.read()
        
        payload = jwt.decode(
            token, 
            public_key, 
            algorithms=[JWT_ALGORITHM],
            issuer="auth-service",
        )
        
        if payload["type"] != "access":
            raise HTTPException(401, "Wrong token type")
        
        # بررسی revoke list
        if await is_revoked(payload.get("jti")):
            raise HTTPException(401, "Token revoked")
        
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError as e:
        raise HTTPException(401, f"Invalid token: {e}")
    

OAuth 2.0 و OIDC

برای third-party integration و SSO. ابزارهای آماده:

• Keycloak: Open Source identity provider
• Auth0: SaaS
• Okta: Enterprise
• AWS Cognito: Cloud-native

RBAC (Role-Based Access Control)

from enum import Enum

class Permission(str, Enum):
    READ_PRODUCTS = "products:read"
    WRITE_PRODUCTS = "products:write"
    DELETE_PRODUCTS = "products:delete"
    MANAGE_USERS = "users:manage"
    VIEW_ANALYTICS = "analytics:view"

ROLE_PERMISSIONS = {
    "customer": [Permission.READ_PRODUCTS],
    "vendor": [Permission.READ_PRODUCTS, Permission.WRITE_PRODUCTS],
    "admin": list(Permission),  # همه
}

def require_permission(perm: Permission):
    async def checker(payload = Depends(verify_token)):
        role = payload.get("role")
        if perm not in ROLE_PERMISSIONS.get(role, []):
            raise HTTPException(403, f"Missing permission: {perm}")
        return payload
    return checker

@app.delete("/products/{id}")
async def delete_product(
    id: int,
    user = Depends(require_permission(Permission.DELETE_PRODUCTS))
):
    ...
    

۱۳.۴ mTLS — Mutual TLS

در TLS عادی، فقط server خود را authenticate می‌کند. در mTLS، هم server و هم client از certificate استفاده می‌کنند.

TLS عادی:
[Client] ──verify cert──► [Server]
Client حواسش هست server کیست، اما server به Client اعتماد می‌کند.

mTLS:
[Client] ──verify cert──► [Server]
[Client] ◄──verify cert── [Server]
هر دو طرف verify می‌شوند.
    

چرا mTLS در میکروسرویس؟

• هر سرویس identity مشخص دارد
• ترافیک encrypt می‌شود
• impersonation امکان‌پذیر نیست
• compliance (PCI-DSS، HIPAA)

پیاده‌سازی دستی پیچیده است.

بهترین راه: استفاده از Service Mesh.

۱۳.۵ Service Mesh — Istio

Service Mesh یک لایه infrastructure است که ارتباطات بین سرویس‌ها را مدیریت می‌کند. در هر pod یک sidecar (Envoy) قرار می‌گیرد که همه ترافیک از آن عبور می‌کند.

بدون Service Mesh:
[Pod A: App] ────────► [Pod B: App]
هر سرویس باید خودش mTLS، retry، tracing را پیاده کند.

با Service Mesh:
[Pod A: App ↔ Sidecar] ──mTLS──► [Sidecar ↔ Pod B: App]
Sidecar مسئول mTLS، retry، tracing، metrics است.
App ساده می‌ماند.
    

قابلیت‌های Istio

• mTLS خودکار
• Traffic management (canary، blue-green)
• Circuit breaking
• Distributed tracing خودکار
• Authorization policies
• Rate limiting

# Authorization Policy
apiVersion: security.istio.io/v1
kind: AuthorizationPolicy
metadata:
  name: allow-orders-from-frontend
  namespace: shop
spec:
  selector:
    matchLabels:
      app: order-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/shop/sa/api-gateway"]
    to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/orders/*"]

---
# mTLS اجباری
apiVersion: security.istio.io/v1
kind: PeerAuthentication
metadata:
  name: default
  namespace: shop
spec:
  mtls:
    mode: STRICT
    

Linkerd — جایگزین سبک‌تر

اگر Istio پیچیده است، Linkerd ساده‌تر و سریع‌تر است.

۱۳.۶ Secret Management

هیچ‌گاه secret ها را در:

• کد منبع
• Git repository
• Docker image
• Environment variables (در فایل اشتراکی)

قرار ندهید!

HashiCorp Vault

import hvac

client = hvac.Client(
    url="http://vault:8200",
    token=os.getenv("VAULT_TOKEN")
)

# Read
secret = client.secrets.kv.v2.read_secret_version(
    path="database/production"
)
db_password = secret["data"]["data"]["password"]

# Dynamic credentials (ایده‌آل!)
# Vault به DB connect می‌شود و هر بار username/password جدید می‌سازد
creds = client.secrets.database.generate_credentials(name="my-role")
# creds["data"]["username"], creds["data"]["password"]
# با TTL کوتاه (مثلاً ۱ ساعت) — حتی اگر leak شود، خطر کم
    

External Secrets Operator (Kubernetes)

secret ها را از Vault، AWS Secrets Manager یا Azure Key Vault به صورت Kubernetes Secret می‌آورد.

apiVersion: external-secrets.io/v1
kind: ExternalSecret
metadata:
  name: db-secret
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: vault-store
    kind: SecretStore
  target:
    name: db-credentials
  data:
  - secretKey: password
    remoteRef:
      key: database/production
      property: password
    

Sealed Secrets

secret ها را encrypt می‌کنیم و در Git قرار می‌دهیم. در cluster، Operator decrypt می‌کند.

۱۳.۷ API Security

OWASP API Top 10 (2023)

1. BOLA (Broken Object Level Auth): بررسی نکردن مالکیت در URL params
2. Broken Authentication: JWT ضعیف، session fixation
3. Broken Object Property Level Auth: خواندن/نوشتن field های غیرمجاز
4. Unrestricted Resource Consumption: نبود rate limit
5. Broken Function Level Auth: دسترسی به admin endpoint
6. SSRF: Server-Side Request Forgery
7. Security Misconfiguration: verbose errors، CORS غلط
8. Lack of Inventory: ندانستن چه API هایی exposed هستند
9. Insecure API Consumption: trust زدن به third-party APIs

BOLA Prevention

# ❌ غلط
@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    return await db.get_order(order_id)
# هر کاربر می‌تواند سفارش هر کس دیگری را ببیند!

# ✅ درست
@app.get("/orders/{order_id}")
async def get_order(order_id: int, user = Depends(verify_token)):
    order = await db.get_order(order_id)
    if order.user_id != user["user_id"] and user["role"] != "admin":
        raise HTTPException(403, "Forbidden")
    return order
    

Input Validation

from pydantic import BaseModel, Field, validator
import re

class CreateProduct(BaseModel):
    name: str = Field(..., min_length=3, max_length=200)
    price: float = Field(..., gt=0, le=1_000_000_000)
    sku: str = Field(..., regex=r"^[A-Z]{2}-[0-9]{6}$")
    description: str = Field("", max_length=5000)
    
    @validator("name")
    def name_no_html(cls, v):
        if re.search(r"<[^>]+>", v):
            raise ValueError("HTML not allowed")
        return v
    

۱۳.۸ Defense in Depth

چندین لایه امنیت — اگر یکی شکست، دیگری حمله را متوقف می‌کند.

[ ️🌐 Internet ]
       │
       ▼
[ 🛡️ WAF (Cloudflare) ] ← لایه ۱: SQL injection، XSS، DDoS
       │
       ▼
[ 🚪 API Gateway ] ← لایه ۲: rate limit، auth، CORS
       │
       ▼
[ 🔒 mTLS ] ← لایه ۳: encryption + identity
       │
       ▼
[ 🛂 Application ] ← لایه ۴: input validation، authorization
       │
       ▼
[ 🗄️ Database ] ← لایه ۵: prepared statements، least privilege
    

۱۳.۹ بهترین تجربیات

1. Never trust input. همیشه validate کنید.
2. Defense in depth. چندین لایه امنیت.
3. Least privilege. هر سرویس فقط دسترسی لازم.
4. HTTPS اجباری. هیچ HTTP در production.
5. mTLS بین سرویس‌ها. با service mesh.
6. JWT با RSA. کلید private فقط در auth-service.
7. Short token TTL. 15 دقیقه برای access.
8. Refresh tokens revocable.
9. Secret management proper. Vault، نه env file.
10. Image scanning. Trivy در CI/CD.
11. Dependency scanning. snyk، Dependabot.
12. Audit logging. همه actions حساس log شوند.
13. Rate limiting و DDoS protection.
14. Security headers. CSP، HSTS، X-Frame-Options.
15. Regular pentest.

۱۳.۱۰ خلاصه فصل