170 lines
4.7 KiB
Python
170 lines
4.7 KiB
Python
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]
|