david/dns-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Configure arbitrary NPM endpoint lists

Browse Source
This commit is contained in:
Lyncoln
2026-09-19 16:06:53 -04:00
parent b0af5349af
commit d4d200551e
2 changed files with 123 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dns_sync/config.py
+51 -10
View File
@@ -79,6 +79,15 @@ def parse_endpoint_list(
return tuple(endpoints)
def endpoint_host(endpoint: Endpoint) -> str:
parsed = urlsplit(f"//{endpoint.address}")
if not parsed.hostname:
raise RuntimeError(
f"Invalid endpoint address: {endpoint.address!r}"
)
return parsed.hostname
def env_bool(name: str, default: bool) -> bool:
value = os.getenv(name)
if value is None:
@@ -90,6 +99,7 @@ def env_bool(name: str, default: bool) -> bool:
class AppConfig:
source_name: str
provider_name: str
npm_endpoints: tuple[Endpoint, ...]
proxy_ip: str
proxy_api_port: int
proxy_api_scheme: str
@@ -101,18 +111,16 @@ class AppConfig:
@property
def proxy_url(self) -> str:
return f"{self.proxy_api_scheme}://{self.proxy_ip}:{self.proxy_api_port}"
return (
f"{self.proxy_api_scheme}://"
f"{self.npm_endpoints[0].address}"
)
@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:
@@ -124,10 +132,42 @@ class AppConfig:
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",
)
raw_proxy_ip = os.getenv("PROXY_IP")
raw_npm_endpoints = os.getenv("NPM_ENDPOINTS")
if raw_npm_endpoints is None:
if not raw_proxy_ip:
raise RuntimeError(
"NPM_ENDPOINTS or legacy PROXY_IP is required"
)
proxy_ip = normalize_ipv4(raw_proxy_ip, "PROXY_IP")
npm_endpoints = parse_endpoint_list("npm", proxy_ip, port)
else:
npm_endpoints = parse_endpoint_list(
"npm",
raw_npm_endpoints,
port,
)
if raw_proxy_ip:
proxy_ip = normalize_ipv4(raw_proxy_ip, "PROXY_IP")
else:
proxy_ip = endpoint_host(npm_endpoints[0])
raw_target_ip = os.getenv("DNS_TARGET_IP")
if raw_target_ip:
target_ip = normalize_ipv4(raw_target_ip, "DNS_TARGET_IP")
elif raw_proxy_ip:
target_ip = proxy_ip
elif len(npm_endpoints) == 1:
target_ip = normalize_ipv4(
endpoint_host(npm_endpoints[0]),
"DNS_TARGET_IP",
)
else:
raise RuntimeError(
"DNS_TARGET_IP is required when multiple NPM_ENDPOINTS "
"are configured without legacy PROXY_IP"
)
domain_matching = env_bool("DOMAIN_MATCHING", False)
raw_domain = os.getenv("DOMAIN")
@@ -147,6 +187,7 @@ class AppConfig:
return cls(
source_name=source_name,
provider_name=provider_name,
npm_endpoints=npm_endpoints,
proxy_ip=proxy_ip,
proxy_api_port=port,
proxy_api_scheme=scheme,
tests/test_config.py
+72 -1
View File
@@ -1,6 +1,8 @@
import os
import unittest
from unittest.mock import patch
from dns_sync.config import parse_endpoint_list
from dns_sync.config import AppConfig, parse_endpoint_list
class EndpointParsingTests(unittest.TestCase):
@@ -45,5 +47,74 @@ class EndpointParsingTests(unittest.TestCase):
)
class AppConfigEndpointTests(unittest.TestCase):
def test_legacy_proxy_ip_remains_supported(self):
with patch.dict(
os.environ,
{"PROXY_IP": "192.168.1.101"},
clear=True,
):
config = AppConfig.from_env()
self.assertEqual(
tuple(endpoint.identity for endpoint in config.npm_endpoints),
("npm://192.168.1.101:81",),
)
self.assertEqual(config.proxy_url, "http://192.168.1.101:81")
self.assertEqual(config.target_ip, "192.168.1.101")
def test_endpoint_list_controls_api_and_proxy_ip_remains_target(self):
with patch.dict(
os.environ,
{
"NPM_ENDPOINTS": (
"192.168.1.3,192.168.1.4:8081,192.168.1.5"
),
"PROXY_IP": "192.168.1.101",
},
clear=True,
):
config = AppConfig.from_env()
self.assertEqual(
tuple(endpoint.identity for endpoint in config.npm_endpoints),
(
"npm://192.168.1.3:81",
"npm://192.168.1.4:8081",
"npm://192.168.1.5:81",
),
)
self.assertEqual(config.proxy_url, "http://192.168.1.3:81")
self.assertEqual(config.target_ip, "192.168.1.101")
def test_explicit_dns_target_supports_multiple_endpoints(self):
with patch.dict(
os.environ,
{
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
"DNS_TARGET_IP": "192.168.1.101",
},
clear=True,
):
config = AppConfig.from_env()
self.assertEqual(config.proxy_url, "http://192.168.1.3:81")
self.assertEqual(config.target_ip, "192.168.1.101")
def test_multiple_endpoints_require_target_without_proxy_ip(self):
with patch.dict(
os.environ,
{
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
},
clear=True,
):
with self.assertRaisesRegex(
RuntimeError,
"DNS_TARGET_IP is required",
):
AppConfig.from_env()
if __name__ == "__main__":
unittest.main()