Architettura di sicurezza¶

import httpx
import ssl


def build_mtls_client(cert: str, key: str, ca: str) -> httpx.Client:
    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca)
    ctx.load_cert_chain(certfile=cert, keyfile=key)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_3
    ctx.verify_mode = ssl.CERT_REQUIRED
    return httpx.Client(verify=ctx, timeout=10.0)

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    default_limits=["200/minute", "2000/hour"],
)


def create_app() -> FastAPI:
    app = FastAPI(docs_url=None, redoc_url=None)  # disabilita /docs in prod
    app.state.limiter = limiter
    app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
    app.add_middleware(SlowAPIMiddleware)
    return app


@router.post("/chat")
@limiter.limit("30/minute")
async def chat(request: Request, body: ChatRequest):
    ...

from pydantic import BaseModel, Field, field_validator
from typing import Annotated
import re

ALLOWED_DEVICES = frozenset({"desktop", "mobile", "watch", "glasses", "vr"})


class ChatRequest(BaseModel):
    model_config = {"strict": True, "extra": "forbid"}
    message: Annotated[str, Field(min_length=1, max_length=4096)]
    device_id: Annotated[str, Field(pattern=r"^[a-zA-Z0-9\-]{8,64}$")]
    device_type: str
    session_id: Annotated[str, Field(pattern=r"^[a-f0-9\-]{36}$")]

    @field_validator("message")
    @classmethod
    def sanitize(cls, v: str) -> str:
        cleaned = re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", v)
        if not cleaned:
            raise ValueError("Empty after sanitization")
        return cleaned

    @field_validator("device_type")
    @classmethod
    def valid_device(cls, v: str) -> str:
        if v not in ALLOWED_DEVICES:
            raise ValueError(f"device_type must be one of {ALLOWED_DEVICES}")
        return v

import hvac
from functools import lru_cache


@lru_cache(maxsize=1)
def get_vault_client() -> hvac.Client:
    client = hvac.Client(url="https://vault.internal:8200")
    client.auth.approle.login(
        role_id=os.environ["VAULT_ROLE_ID"],
        secret_id=os.environ["VAULT_SECRET_ID"],
    )
    return client


def get_llm_api_key(provider: str) -> str:
    client = get_vault_client()
    secret = client.secrets.kv.v2.read_secret_version(
        path=f"jarvis/llm/{provider}", mount_point="secret"
    )
    return secret["data"]["data"]["api_key"]

import hashlib
import secrets
import base64
from dataclasses import dataclass


@dataclass(frozen=True)
class PKCEChallenge:
    verifier: str
    challenge: str
    method: str = "S256"


def generate_pkce_challenge() -> PKCEChallenge:
    verifier = secrets.token_urlsafe(64)  # 86 chars ≈ 512 bit
    digest = hashlib.sha256(verifier.encode()).digest()
    return PKCEChallenge(
        verifier=verifier,
        challenge=base64.urlsafe_b64encode(digest).rstrip(b"=").decode(),
    )

from datetime import datetime, timedelta, timezone
from jose import jwt
import uuid

ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=30)


def create_access_token(user_id: str, device_id: str, private_key: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id, "did": device_id,
        "iat": now, "exp": now + ACCESS_TTL,
        "jti": str(uuid.uuid4()),
        "iss": "open-jarvis", "aud": "jarvis-api",
    }
    return jwt.encode(payload, private_key, algorithm="ES256")


def create_refresh_token(user_id: str, family_id: str, private_key: str) -> str:
    now = datetime.now(timezone.utc)
    return jwt.encode(
        {
            "sub": user_id, "family": family_id,
            "iat": now, "exp": now + REFRESH_TTL,
            "jti": str(uuid.uuid4()), "type": "refresh",
        },
        private_key,
        algorithm="ES256",
    )

import hashlib
import hmac
import time
from fastapi import HTTPException, Header, Request

WEBHOOK_MAX_AGE = 300  # 5 min


async def verify_webhook(
    request: Request,
    x_signature: str = Header(..., alias="X-Jarvis-Signature"),
    x_timestamp: str = Header(..., alias="X-Jarvis-Timestamp"),
    secret: str = "",
) -> bytes:
    ts = int(x_timestamp)
    if abs(time.time() - ts) > WEBHOOK_MAX_AGE:
        raise HTTPException(400, "Timestamp expired")
    body = await request.body()
    expected = hmac.new(
        secret.encode(), f"{x_timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(f"sha256={expected}", x_signature):
        raise HTTPException(401, "Invalid signature")
    return body

from fastapi import APIRouter, Request
import logging

router = APIRouter()
hp = logging.getLogger("jarvis.honeypot")


@router.get("/api/v0/admin")
@router.post("/api/v0/admin")
@router.get("/.env")
@router.get("/wp-admin")
async def honeypot_trap(request: Request):
    hp.warning("HONEYPOT_HIT", extra={
        "ip": request.client.host,
        "path": request.url.path,
        "ua": request.headers.get("user-agent", ""),
        "method": request.method,
    })
    return {"detail": "Not Found"}

@router.delete("/gdpr/users/{user_id}/data")
async def right_to_erasure(user_id: str, current=Depends(get_current_user)):
    if current.sub != user_id:
        raise HTTPException(403)
    deleted_vectors = await qdrant.delete_by_user(user_id)
    deleted_db = await db.execute(
        "DELETE FROM user_data WHERE user_id = $1 RETURNING count(*)", user_id
    )
    await token_store.revoke_all_for_user(user_id)
    audit_log.info("GDPR_ERASURE", user_id=user_id, vectors=deleted_vectors)
    return {
        "status": "erased",
        "deleted_vectors": deleted_vectors,
        "deleted_records": deleted_db,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }