R13: Fix X-Forwarded-For trust boundary — only trust from known proxies
- Add _is_trusted_proxy() to check client IP against trusted CIDRs - Only use X-Forwarded-For if connection came from trusted proxy - Add TRUSTED_PROXY_CIDRS config (default: 127.0.0.1/32, ::1/128) - Add parse_comma_separated_lists for env var parsing - Add 3 unit tests: trusted CIDR check, untrusted ignores forwarded, trusted uses forwarded
This commit is contained in:
@@ -33,6 +33,18 @@ class Settings(BaseSettings):
|
||||
rate_limit_secret: str = "change-rate-limit-secret"
|
||||
log_level: str = "INFO"
|
||||
cors_origins: list[str] = Field(default_factory=lambda: ["http://localhost:4321", "http://127.0.0.1:4321"])
|
||||
trusted_proxy_cidrs: list[str] = Field(default_factory=lambda: ["127.0.0.1/32", "::1/128"])
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def parse_comma_separated_lists(cls, data: dict) -> dict:
|
||||
"""Parse comma-separated string values into lists."""
|
||||
if isinstance(data, dict):
|
||||
for field_name in ["cors_origins", "trusted_proxy_cidrs"]:
|
||||
value = data.get(field_name)
|
||||
if isinstance(value, str) and value:
|
||||
data[field_name] = [item.strip() for item in value.split(",") if item.strip()]
|
||||
return data
|
||||
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
|
||||
|
||||
@model_validator(mode="after")
|
||||
|
||||
+19
-1
@@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from ipaddress import IPv4Address, IPv6Address, IPv4Network, IPv6Network
|
||||
import hashlib
|
||||
import hmac
|
||||
import logging
|
||||
@@ -538,13 +539,30 @@ def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_ad
|
||||
return Response(status_code=204)
|
||||
|
||||
|
||||
def _is_trusted_proxy(address: str, trusted_cidrs: list[str]) -> bool:
|
||||
"""Check if address is in trusted proxy CIDRs."""
|
||||
try:
|
||||
addr = IPv4Address(address) if ":" not in address else IPv6Address(address)
|
||||
except ValueError:
|
||||
return False
|
||||
for cidr in trusted_cidrs:
|
||||
try:
|
||||
network = IPv4Network(cidr) if ":" not in cidr else IPv6Network(cidr)
|
||||
if addr in network:
|
||||
return True
|
||||
except ValueError:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def _check_rate_limit(request: Request, db: Session) -> None:
|
||||
now = datetime.now(timezone.utc)
|
||||
cutoff = now - timedelta(minutes=10)
|
||||
# Extract real client IP from forwarded headers
|
||||
client = request.client.host if request.client else "unknown"
|
||||
forwarded = request.headers.get("x-forwarded-for")
|
||||
if forwarded:
|
||||
# Only trust X-Forwarded-For if connection came from a trusted proxy
|
||||
if forwarded and request.client and _is_trusted_proxy(request.client.host, settings.trusted_proxy_cidrs):
|
||||
client = forwarded.split(",")[0].strip()
|
||||
client_hash = hmac.new(settings.rate_limit_secret.encode(), client.encode(), hashlib.sha256).hexdigest()
|
||||
if db.get_bind().dialect.name == "postgresql":
|
||||
|
||||
Reference in New Issue
Block a user