config.py

import warnings
from dataclasses import dataclass


@dataclass(frozen=True)
class Config:
    host: str = "::"
    port: int = 9000
    # Server access key
    key: str = "peerjs"
    # Server location
    path: str = "/"
    # Timeouts in seconds
    expire_timeout: float = 5.0  # queued message TTL

    # HTTP allow discovery of clients on path/peers
    allow_discovery: bool = False

    # Max time between pings before a client is dropped
    alive_timeout: float = 90.0

    concurrent_limit: int = 5000

    # CORS origin for HTTP API responses; None disables the header.
    # TS uses a CorsOptions object but GET-only endpoints need only the origin string.
    cors_origin: str | None = None

    # Rate limiting for HTTP API endpoints (/id and /peers)
    rate_limit_window: float = 60.0  # seconds per window
    rate_limit_max_requests: int = 10  # max requests per IP per window

    cleanup_interval: float = 1.0  # how often expire_messages sweeps

    # TS queues are unbounded; cap added here as an explicit hardening measure
    queue_max_size: int = 1000  # max queued messages per offline peer

    # How frequently we should check for clients that haven't sent a ping
    connectivity_check_interval: float = 0.3

    # Trust X-Forwarded-For header for real client IP (e.g. behind AWS ALB / nginx).
    # Only honoured when the direct connection comes from a known proxy IP in trusted_proxies.
    proxied: bool = False
    trusted_proxies: frozenset[str] = frozenset()

    def __post_init__(self) -> None:
        if not 0 <= self.port <= 65535:
            raise ValueError(f"port must be between 0 and 65535, got {self.port}")
        if not self.key:
            raise ValueError("key must not be empty")
        if not self.path.startswith("/"):
            raise ValueError(f"path must start with '/', got {self.path!r}")
        for field_name in (
            "expire_timeout",
            "alive_timeout",
            "rate_limit_window",
            "cleanup_interval",
            "connectivity_check_interval",
            "concurrent_limit",
            "rate_limit_max_requests",
            "queue_max_size",
        ):
            value = getattr(self, field_name)
            if value <= 0:
                raise ValueError(f"{field_name} must be positive, got {value}")

        if self.proxied and not self.trusted_proxies:
            raise ValueError("proxied=True requires at least one entry in trusted_proxies")
        if not self.proxied and self.trusted_proxies:
            warnings.warn(
                "trusted_proxies is set but proxied=False; X-Forwarded-For will not be read "
                "and the proxy entries have no effect. Set proxied=True to enable proxy support.",
                UserWarning,
                stacklevel=2,
            )

    @property
    def ws_path(self) -> str:
        return self.path.rstrip("/") + "/peerjs"