159 lines
4.7 KiB
Python
159 lines
4.7 KiB
Python
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:
|
|
source_name: str
|
|
provider_name: str
|
|
proxy_ip: str
|
|
proxy_api_port: int
|
|
proxy_api_scheme: str
|
|
target_ip: str
|
|
proxy_verify_tls: bool
|
|
domain_matching: bool
|
|
domain: str | None
|
|
include_disabled: bool = False
|
|
|
|
@property
|
|
def proxy_url(self) -> str:
|
|
return f"{self.proxy_api_scheme}://{self.proxy_ip}:{self.proxy_api_port}"
|
|
|
|
@classmethod
|
|
def from_env(cls, include_disabled: bool = False) -> "AppConfig":
|
|
source_name = os.getenv("PROXY", "npm").strip().lower()
|
|
provider_name = os.getenv("DNS_PROVIDER", "verizon").strip().lower()
|
|
|
|
raw_proxy_ip = os.getenv("PROXY_IP")
|
|
if not raw_proxy_ip:
|
|
raise RuntimeError("PROXY_IP is required")
|
|
proxy_ip = normalize_ipv4(raw_proxy_ip, "PROXY_IP")
|
|
|
|
try:
|
|
port = int(os.getenv("PROXY_API_PORT", "81"))
|
|
except ValueError as exc:
|
|
raise RuntimeError("PROXY_API_PORT must be an integer") from exc
|
|
if not 1 <= port <= 65535:
|
|
raise RuntimeError("PROXY_API_PORT must be between 1 and 65535")
|
|
|
|
scheme = os.getenv("PROXY_API_SCHEME", "http").strip().lower()
|
|
if scheme not in {"http", "https"}:
|
|
raise RuntimeError("PROXY_API_SCHEME must be http or https")
|
|
|
|
target_ip = normalize_ipv4(
|
|
os.getenv("DNS_TARGET_IP", proxy_ip),
|
|
"DNS_TARGET_IP",
|
|
)
|
|
|
|
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(
|
|
source_name=source_name,
|
|
provider_name=provider_name,
|
|
proxy_ip=proxy_ip,
|
|
proxy_api_port=port,
|
|
proxy_api_scheme=scheme,
|
|
target_ip=target_ip,
|
|
proxy_verify_tls=env_bool("PROXY_VERIFY_TLS", True),
|
|
domain_matching=domain_matching,
|
|
domain=domain,
|
|
include_disabled=include_disabled,
|
|
)
|