"""Rate limit and security configuration with environment variable loading.""" from __future__ import annotations import os from dataclasses import dataclass, field from typing import Self @dataclass(frozen=True) class EndpointLimit: """Rate limit for a single endpoint pattern.""" max_requests: int window_seconds: int @dataclass(frozen=True) class RateLimitConfig: """Configuration for API rate limiting, export caps, and metrics access. All values are configurable via environment variables with sensible defaults. """ # Per-endpoint rate limits endpoint_limits: dict[str, EndpointLimit] = field(default_factory=lambda: { "/api/listing": EndpointLimit(30, 60), "/api/listing_geojson": EndpointLimit(10, 60), "/api/listing_geojson/stream": EndpointLimit(10, 60), "/api/refresh_listings": EndpointLimit(3, 300), "/api/task_status": EndpointLimit(200, 60), "/api/tasks_for_user": EndpointLimit(60, 60), "/api/cancel_task": EndpointLimit(10, 60), "/api/clear_all_tasks": EndpointLimit(5, 60), "/api/get_districts": EndpointLimit(20, 60), "/api/passkey": EndpointLimit(10, 60), }) # Bulk export caps listing_limit_cap: int = 100 geojson_limit_cap: int = 5_000 geojson_stream_limit_cap: int = 10_000 geojson_stream_batch_size_cap: int = 200 # Redis DB for rate limit counters rate_limit_redis_db: int = 3 # Metrics endpoint IP allowlist (comma-separated CIDRs) metrics_allowed_ips: str = "127.0.0.1,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16,::1" # X-Forwarded-For trusted proxy depth trusted_proxy_depth: int = 1 @classmethod def from_env(cls) -> Self: """Load configuration from environment variables. Environment variables: RATE_LIMIT_LISTING: /api/listing limit (default: 30/60s) RATE_LIMIT_GEOJSON: /api/listing_geojson limit (default: 10/60s) RATE_LIMIT_GEOJSON_STREAM: /api/listing_geojson/stream limit (default: 10/60s) RATE_LIMIT_REFRESH: /api/refresh_listings limit (default: 3/300s) RATE_LIMIT_TASK_STATUS: /api/task_status limit (default: 60/60s) RATE_LIMIT_TASKS_FOR_USER: /api/tasks_for_user limit (default: 30/60s) RATE_LIMIT_CANCEL_TASK: /api/cancel_task limit (default: 10/60s) RATE_LIMIT_CLEAR_TASKS: /api/clear_all_tasks limit (default: 5/60s) RATE_LIMIT_DISTRICTS: /api/get_districts limit (default: 20/60s) RATE_LIMIT_PASSKEY: /api/passkey/* limit (default: 10/60s) EXPORT_LISTING_LIMIT_CAP: Max listings per request (default: 100) EXPORT_GEOJSON_LIMIT_CAP: Max GeoJSON features per request (default: 5000) EXPORT_GEOJSON_STREAM_LIMIT_CAP: Max streamed features (default: 10000) EXPORT_GEOJSON_STREAM_BATCH_CAP: Max stream batch size (default: 200) RATE_LIMIT_REDIS_DB: Redis DB number for counters (default: 3) METRICS_ALLOWED_IPS: Comma-separated CIDRs for /metrics (default: private ranges) """ def _parse_limit(env_var: str, default_requests: int, default_window: int) -> EndpointLimit: raw = os.environ.get(env_var) if raw: parts = raw.split("/") if len(parts) == 2: return EndpointLimit(int(parts[0]), int(parts[1])) return EndpointLimit(default_requests, default_window) return cls( endpoint_limits={ "/api/listing": _parse_limit("RATE_LIMIT_LISTING", 30, 60), "/api/listing_geojson": _parse_limit("RATE_LIMIT_GEOJSON", 10, 60), "/api/listing_geojson/stream": _parse_limit("RATE_LIMIT_GEOJSON_STREAM", 10, 60), "/api/refresh_listings": _parse_limit("RATE_LIMIT_REFRESH", 3, 300), "/api/task_status": _parse_limit("RATE_LIMIT_TASK_STATUS", 200, 60), "/api/tasks_for_user": _parse_limit("RATE_LIMIT_TASKS_FOR_USER", 60, 60), "/api/cancel_task": _parse_limit("RATE_LIMIT_CANCEL_TASK", 10, 60), "/api/clear_all_tasks": _parse_limit("RATE_LIMIT_CLEAR_TASKS", 5, 60), "/api/get_districts": _parse_limit("RATE_LIMIT_DISTRICTS", 20, 60), "/api/passkey": _parse_limit("RATE_LIMIT_PASSKEY", 10, 60), }, listing_limit_cap=int(os.environ.get("EXPORT_LISTING_LIMIT_CAP", "100")), geojson_limit_cap=int(os.environ.get("EXPORT_GEOJSON_LIMIT_CAP", "5000")), geojson_stream_limit_cap=int(os.environ.get("EXPORT_GEOJSON_STREAM_LIMIT_CAP", "10000")), geojson_stream_batch_size_cap=int(os.environ.get("EXPORT_GEOJSON_STREAM_BATCH_CAP", "200")), rate_limit_redis_db=int(os.environ.get("RATE_LIMIT_REDIS_DB", "3")), metrics_allowed_ips=os.environ.get( "METRICS_ALLOWED_IPS", "127.0.0.1,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16,::1", ), )