curo1305andClaude Sonnet 4.6
a826738e18
- D-11: replace get_client_ip body with trusted-proxy CIDR check (127/8, 172.16/12,
192.168/16, ::1/128); untrusted peers always return their own IP, preventing XFF
spoofing by external callers
- D-12: create services/rate_limiting.py with _account_key() keyed by user.id
(falls back to peer IP when no authenticated user in request.state); exports
account_limiter = Limiter(key_func=_account_key)
- Wire: auth.py switches from get_remote_address to get_client_ip; main.py imports
account_limiter; all 9 documents.py endpoints and 7 cloud.py endpoints decorated
@account_limiter.limit("100/minute") with request.state.current_user = current_user
as the first handler statement (A1 ordering invariant)
- Promote all 8 xfail stubs to real assertions; add autouse fixture in conftest.py
to reset MemoryStorage between tests preventing cross-test 429 contamination
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
"""Shared dependency utilities — request parsing helpers used across all API routers."""
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
_TRUSTED_PROXY_NETS = [
|
|
ipaddress.ip_network("127.0.0.0/8"),
|
|
ipaddress.ip_network("172.16.0.0/12"),
|
|
ipaddress.ip_network("192.168.0.0/16"),
|
|
ipaddress.ip_network("::1/128"),
|
|
]
|
|
|
|
|
|
def _is_trusted_proxy(host: str) -> bool:
|
|
try:
|
|
addr = ipaddress.ip_address(host)
|
|
return any(addr in net for net in _TRUSTED_PROXY_NETS)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def get_client_ip(request: Request) -> Optional[str]:
|
|
"""Extract best-effort client IP from request for audit logging (D-11 — trusted-proxy CIDR check).
|
|
|
|
Only honours X-Forwarded-For when the direct peer is a known trusted proxy
|
|
(RFC-1918 / loopback). Untrusted direct peers always return their own address,
|
|
preventing XFF spoofing by external callers.
|
|
|
|
TRUST BOUNDARY: this value is used for forensic audit logging only —
|
|
not for authentication or access control decisions.
|
|
"""
|
|
direct_peer = request.client.host if request.client else None
|
|
if direct_peer and _is_trusted_proxy(direct_peer):
|
|
xff = request.headers.get("X-Forwarded-For")
|
|
if xff:
|
|
return xff.split(",")[0].strip()
|
|
return direct_peer
|
|
|
|
|
|
def parse_uuid(value: str, detail: str = "Not found") -> uuid.UUID:
|
|
"""Parse *value* as a UUID, raising HTTP 404 with *detail* on failure.
|
|
|
|
Use at API boundaries to convert path/body string IDs to UUID objects.
|
|
Returns the parsed UUID so callers can use it directly without a try/except.
|
|
"""
|
|
try:
|
|
return uuid.UUID(value)
|
|
except ValueError:
|
|
raise HTTPException(status_code=404, detail=detail)
|