import os from dataclasses import dataclass from urllib.parse import urlsplit from .adapters import Endpoint from .normalizer import normalize_hostname, normalize_ipv4 def parse_endpoint_list( plugin: str, value: str, default_port: int, ) -> tuple[Endpoint, ...]: if not 1 <= default_port <= 65535: raise RuntimeError( "Endpoint default port must be between 1 and 65535" ) raw_addresses = value.split(",") if any(not item.strip() for item in raw_addresses): raise RuntimeError( f"{plugin.upper()} endpoint list contains an empty address" ) endpoints: list[Endpoint] = [] identities: set[str] = set() for raw_address in raw_addresses: address = raw_address.strip().rstrip("/") if "://" in address: raise RuntimeError( f"{plugin.upper()} endpoint {raw_address!r} must not " "include a URL scheme" ) if any(character.isspace() for character in address): raise RuntimeError( f"{plugin.upper()} endpoint {raw_address!r} " "contains whitespace" ) parsed = urlsplit(f"//{address}") if ( not parsed.hostname or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment ): raise RuntimeError( f"Invalid {plugin.upper()} endpoint address: " f"{raw_address!r}" ) try: port = parsed.port or default_port except ValueError as exc: raise RuntimeError( f"Invalid {plugin.upper()} endpoint port: " f"{raw_address!r}" ) from exc host = parsed.hostname.lower() normalized_address = ( f"[{host}]:{port}" if ":" in host else f"{host}:{port}" ) endpoint = Endpoint(plugin, normalized_address) if endpoint.identity in identities: raise RuntimeError( f"Duplicate {plugin.upper()} endpoint: {endpoint.address}" ) identities.add(endpoint.identity) endpoints.append(endpoint) return tuple(endpoints) def env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() not in {"0", "false", "no", "off"} @dataclass(frozen=True) class AppConfig: target_ip: str domain_matching: bool domain: str | None include_disabled: bool = False @classmethod def from_env(cls, include_disabled: bool = False) -> "AppConfig": raw_target_ip = os.getenv("DNS_TARGET_IP") raw_proxy_ip = os.getenv("PROXY_IP") if raw_target_ip: target_ip = normalize_ipv4( raw_target_ip, "DNS_TARGET_IP", ) elif raw_proxy_ip: # Deprecated compatibility fallback. Core no longer treats # PROXY_IP as an NPM endpoint. target_ip = normalize_ipv4( raw_proxy_ip, "PROXY_IP", ) else: raise RuntimeError( "DNS_TARGET_IP is required " "(legacy PROXY_IP is also accepted)" ) domain_matching = env_bool("DOMAIN_MATCHING", False) raw_domain = os.getenv("DOMAIN") domain = None if raw_domain: domain, _ = normalize_hostname( raw_domain, allow_wildcard=False, ) if domain_matching and not domain: raise RuntimeError( "DOMAIN is required when DOMAIN_MATCHING=true" ) return cls( target_ip=target_ip, domain_matching=domain_matching, domain=domain, include_disabled=include_disabled, )