Disattiva temporaneamente form prenotazioni

backend/app/services/booking_security.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+from collections import defaultdict, deque
+from datetime import UTC, datetime
+import base64
+import hashlib
+import hmac
+import json
+from pathlib import Path
+import random
+import threading
+import time
+from typing import Any
+from uuid import uuid4
+
+from fastapi import Request
+
+from app.config import settings
+from app.schemas.booking import BookingCaptchaChallenge, BookingRequestCreate
+
+
+_audit_lock = threading.Lock()
+_rate_limit_lock = threading.Lock()
+_rate_limit_hits: dict[str, deque[float]] = defaultdict(deque)
+
+
+def new_request_id() -> str:
+    return uuid4().hex
+
+
+def get_client_ip(request: Request) -> str:
+    forwarded_for = request.headers.get("x-forwarded-for", "")
+    if forwarded_for:
+        return forwarded_for.split(",", 1)[0].strip()
+
+    real_ip = request.headers.get("x-real-ip", "")
+    if real_ip:
+        return real_ip.strip()
+
+    if request.client:
+        return request.client.host
+
+    return "unknown"
+
+
+def is_rate_limited(client_ip: str) -> bool:
+    now = time.time()
+    window = settings.booking_rate_limit_window_seconds
+    max_requests = settings.booking_rate_limit_max_requests
+
+    with _rate_limit_lock:
+        hits = _rate_limit_hits[client_ip]
+        while hits and now - hits[0] > window:
+            hits.popleft()
+
+        if len(hits) >= max_requests:
+            return True
+
+        hits.append(now)
+        return False
+
+
+def create_captcha_challenge() -> BookingCaptchaChallenge:
+    left = random.randint(2, 9)
+    right = random.randint(2, 9)
+    expires_at = int(time.time()) + settings.booking_captcha_ttl_seconds
+    nonce = uuid4().hex
+    answer = str(left + right)
+    payload = f"{answer}:{expires_at}:{nonce}"
+    signature = _sign(payload)
+    token = _encode(f"{payload}:{signature}")
+
+    return BookingCaptchaChallenge(
+        captcha_id=token,
+        question=f"Quanto fa {left} + {right}?",
+    )
+
+
+def verify_captcha(captcha_id: str, captcha_answer: str) -> bool:
+    try:
+        decoded = _decode(captcha_id)
+        answer, expires_at_value, nonce, signature = decoded.split(":", 3)
+        payload = f"{answer}:{expires_at_value}:{nonce}"
+        expected_signature = _sign(payload)
+        expires_at = int(expires_at_value)
+    except (ValueError, TypeError):
+        return False
+
+    if not hmac.compare_digest(signature, expected_signature):
+        return False
+
+    if expires_at < int(time.time()):
+        return False
+
+    return hmac.compare_digest(captcha_answer.strip(), answer)
+
+
+def write_booking_audit_event(
+    event: str,
+    *,
+    request_id: str,
+    request: Request,
+    payload: BookingRequestCreate | None = None,
+    error: Exception | str | None = None,
+) -> None:
+    client_ip = get_client_ip(request)
+    record: dict[str, Any] = {
+        "timestamp": datetime.now(UTC).isoformat(),
+        "event": event,
+        "request_id": request_id,
+        "client_ip": client_ip,
+        "user_agent": request.headers.get("user-agent", "")[:300],
+        "referer": request.headers.get("referer", "")[:300],
+    }
+
+    if payload is not None:
+        record.update(
+            {
+                "email_hash": _hash_value(payload.email.lower().strip()),
+                "email_domain": _email_domain(payload.email),
+                "phone_hash": _hash_value(payload.phone.strip()),
+                "name_length": len(payload.name),
+                "pet_type": payload.pet_type,
+                "doctor": payload.doctor,
+                "service": payload.service,
+                "date": payload.date,
+                "time": payload.time,
+                "notes_length": len(payload.notes or ""),
+                "honeypot_filled": bool(payload.website),
+            }
+        )
+
+    if error is not None:
+        record["error_type"] = error.__class__.__name__ if isinstance(error, Exception) else "Error"
+        record["error"] = str(error)[:500]
+
+    log_path = Path(settings.booking_audit_log_path)
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with _audit_lock:
+        with log_path.open("a", encoding="utf-8") as log_file:
+            log_file.write(json.dumps(record, ensure_ascii=False) + "\n")
+
+
+def _sign(payload: str) -> str:
+    return hmac.new(
+        settings.booking_captcha_secret.encode("utf-8"),
+        payload.encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+
+
+def _encode(value: str) -> str:
+    return base64.urlsafe_b64encode(value.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def _decode(value: str) -> str:
+    padding = "=" * (-len(value) % 4)
+    return base64.urlsafe_b64decode(f"{value}{padding}".encode("ascii")).decode("utf-8")
+
+
+def _hash_value(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _email_domain(email: str) -> str:
+    if "@" not in email:
+        return ""
+    return email.rsplit("@", 1)[1].lower()[:120]