refactor: split monolith into typed modules with retry logic
- models.py: ProxyInfo and DNSServerConfig dataclasses with type hints - fetcher.py: subscription fetcher with exponential backoff retries (3 attempts) plus VLESS URL parser - config_builder.py: sing-box config builder decomposed into focused helpers - main.py: orchestrator with setup_logging(), print_banner(), pick_node()
This commit is contained in:
@@ -0,0 +1,210 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Sing-box configuration builder."""
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from models import ProxyInfo
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Default SNI mapping for DNS-over-TLS with IP addresses
|
||||
SNI_MAP: Dict[str, str] = {
|
||||
"8.8.8.8": "dns.google",
|
||||
"8.8.4.4": "dns.google",
|
||||
"1.1.1.1": "cloudflare-dns.com",
|
||||
"1.0.0.1": "cloudflare-dns.com",
|
||||
}
|
||||
|
||||
|
||||
def make_dns_server(addr_str: str) -> Dict[str, Any]:
|
||||
"""Parse common DNS address strings into native sing-box v1.13+ server dict.
|
||||
|
||||
Supports: udp://host:port, tls://host(:port), https://host(:port), bare IP/hostname.
|
||||
|
||||
Args:
|
||||
addr_str: DNS server address string (e.g. "tls://8.8.8.8", "udp://1.1.1.1:53").
|
||||
|
||||
Returns:
|
||||
Dictionary compatible with sing-box DNS server schema.
|
||||
"""
|
||||
import urllib.parse as up
|
||||
|
||||
stripped = addr_str.strip()
|
||||
parts = up.urlparse(stripped)
|
||||
scheme = parts.scheme.lower() if parts.scheme else ""
|
||||
host = parts.hostname or stripped
|
||||
port = parts.port or 53
|
||||
|
||||
config: Dict[str, Any] = {
|
||||
"type": scheme or "udp",
|
||||
"server": host,
|
||||
"server_port": port,
|
||||
}
|
||||
|
||||
if scheme == "tls":
|
||||
config["tls"] = {"server_name": SNI_MAP.get(host, host)}
|
||||
elif scheme == "https":
|
||||
config["server_port"] = port or 443
|
||||
config["tls"] = {"server_name": host}
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def build_singbox_config(
|
||||
proxy: ProxyInfo,
|
||||
socks_port: int = 1080,
|
||||
socks_user: str = "",
|
||||
socks_pass: str = "",
|
||||
dns_server: str = "tls://8.8.8.8",
|
||||
log_level: str = "info",
|
||||
) -> Dict[str, Any]:
|
||||
"""Build a complete sing-box JSON configuration.
|
||||
|
||||
Args:
|
||||
proxy: Parsed VLESS proxy information.
|
||||
socks_port: Local SOCKS5 listen port.
|
||||
socks_user: Optional username for SOCKS auth (empty disables auth).
|
||||
socks_pass: Optional password for SOCKS auth.
|
||||
dns_server: Upstream DNS resolver address string.
|
||||
log_level: Sing-box log level.
|
||||
|
||||
Returns:
|
||||
Dictionary representing the full sing-box configuration.
|
||||
"""
|
||||
outbound = _build_proxy_outbound(proxy)
|
||||
inbound = _build_socks_inbound(
|
||||
port=socks_port,
|
||||
username=socks_user,
|
||||
password=socks_pass,
|
||||
)
|
||||
dns = _build_dns_config(dns_server)
|
||||
|
||||
config: Dict[str, Any] = {
|
||||
"log": {"level": log_level},
|
||||
"dns": dns,
|
||||
"inbounds": [inbound],
|
||||
"outbounds": [
|
||||
outbound,
|
||||
{"type": "direct", "tag": "direct"},
|
||||
],
|
||||
"route": {
|
||||
"default_domain_resolver": "remote",
|
||||
"rules": [{
|
||||
"clash_mode": "direct",
|
||||
"outbound": "direct",
|
||||
}],
|
||||
"final": "proxy",
|
||||
},
|
||||
}
|
||||
|
||||
logger.debug("Built sing-box config with outbound tag 'proxy'")
|
||||
return config
|
||||
|
||||
|
||||
def _build_proxy_outbound(proxy: ProxyInfo) -> Dict[str, Any]:
|
||||
"""Build the VLESS outbound proxy object."""
|
||||
outbound: Dict[str, Any] = {
|
||||
"type": "vless",
|
||||
"tag": "proxy",
|
||||
"server": proxy.server,
|
||||
"server_port": proxy.port,
|
||||
"uuid": proxy.uuid,
|
||||
"packet_encoding": "xudp",
|
||||
}
|
||||
|
||||
security = proxy.security or "tls"
|
||||
|
||||
if security == "reality":
|
||||
outbound["tls"] = {
|
||||
"enabled": True,
|
||||
"server_name": proxy.sni,
|
||||
"utls": {"enabled": True, "fingerprint": "chrome"},
|
||||
"reality": {
|
||||
"enabled": True,
|
||||
"public_key": proxy.pbk,
|
||||
"short_id": proxy.sid or proxy.spx,
|
||||
},
|
||||
}
|
||||
elif security == "tls":
|
||||
tls_config: Dict[str, Any] = {
|
||||
"enabled": True,
|
||||
"server_name": proxy.sni,
|
||||
"utls": {"enabled": True, "fingerprint": "chrome"},
|
||||
}
|
||||
if proxy.alpn:
|
||||
tls_config["alpn"] = [a.strip() for a in proxy.alpn.split(",") if a.strip()]
|
||||
outbound["tls"] = tls_config
|
||||
|
||||
transport = _build_transport(proxy)
|
||||
if transport:
|
||||
outbound["transport"] = transport
|
||||
|
||||
return outbound
|
||||
|
||||
|
||||
def _build_transport(proxy: ProxyInfo) -> Optional[Dict[str, Any]]:
|
||||
"""Build the transport configuration (only for non-TCP types)."""
|
||||
transport_type = proxy.type or "tcp"
|
||||
transport: Dict[str, Any] = {}
|
||||
|
||||
if transport_type == "ws":
|
||||
ws: Dict[str, Any] = {"type": "ws", "path": proxy.path or "/"}
|
||||
if proxy.host:
|
||||
ws["headers"] = {"Host": proxy.host}
|
||||
return ws
|
||||
|
||||
elif transport_type == "http":
|
||||
http_cfg: Dict[str, Any] = {"type": "http"}
|
||||
if proxy.host:
|
||||
http_cfg["host"] = [h.strip() for h in proxy.host.split(",") if h.strip()]
|
||||
http_cfg["path"] = proxy.path or "/"
|
||||
return http_cfg
|
||||
|
||||
elif transport_type == "grpc":
|
||||
grpc: Dict[str, Any] = {"type": "grpc"}
|
||||
svc_name = proxy.serviceName or proxy.path
|
||||
if svc_name:
|
||||
grpc["service_name"] = svc_name
|
||||
return grpc
|
||||
|
||||
elif transport_type == "httpupgrade":
|
||||
hu: Dict[str, Any] = {"type": "httpupgrade", "path": proxy.path or "/"}
|
||||
if proxy.host:
|
||||
hu["headers"] = {"Host": proxy.host}
|
||||
return hu
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _build_socks_inbound(
|
||||
port: int,
|
||||
username: str = "",
|
||||
password: str = "",
|
||||
) -> Dict[str, Any]:
|
||||
"""Build the SOCKS5 inbound object."""
|
||||
inbound: Dict[str, Any] = {
|
||||
"type": "socks",
|
||||
"listen": "::",
|
||||
"listen_port": port,
|
||||
}
|
||||
|
||||
if username:
|
||||
inbound["users"] = [
|
||||
{"username": username, "password": password},
|
||||
]
|
||||
|
||||
return inbound
|
||||
|
||||
|
||||
def _build_dns_config(dns_server_addr: str) -> Dict[str, Any]:
|
||||
"""Build the DNS section of the config."""
|
||||
dns_server_obj = make_dns_server(dns_server_addr)
|
||||
|
||||
return {
|
||||
"servers": [
|
||||
{**dns_server_obj, "tag": "remote"},
|
||||
{"type": "local", "tag": "local"},
|
||||
],
|
||||
"final": "remote",
|
||||
}
|
||||
Reference in New Issue
Block a user