feat: mihomo YAML subscription support + flow/fingerprint passthrough

- fetcher: parse mihomo/clash.meta YAML subscriptions (proxies: list),
  auto-detect format (mihomo YAML vs plain vless:// lines)
- config_builder: forward VLESS flow (xtls-rprx-vision) and per-node
  uTLS fingerprint to sing-box (required for Reality+Vision nodes)
- default DNS upstream: udp://10.35.99.172 (public UDP resolvers
  unreachable on this network; old-sub nodes are hostnames)
- Dockerfile: install PyYAML

Verified: 50-node reality+vision subscription works end-to-end
(sing-box check + live SOCKS tunnel); legacy vless:// subs unaffected.
This commit is contained in:
2026-09-09 10:56:11 +03:00
parent 1beb97e367
commit 56f28099c8
6 changed files with 143 additions and 25 deletions
+110 -1
View File
@@ -3,11 +3,12 @@
import base64
import logging
import re
import time
import urllib.error
import urllib.parse as up
import urllib.request
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from models import ProxyInfo
@@ -152,3 +153,111 @@ def url_to_proxy_info(url_str: str) -> ProxyInfo:
host=raw["host"],
flow=raw["flow"],
)
def _mihomo_entry_to_proxy(entry: Dict[str, Any]) -> Optional[ProxyInfo]:
"""Convert one mihomo YAML proxy entry to a ProxyInfo (VLESS only).
Args:
entry: Raw proxy dict from the mihomo `proxies` list.
Returns:
ProxyInfo instance, or None if the entry is not a usable VLESS node.
"""
if str(entry.get("type", "")).lower() != "vless":
return None
server = str(entry.get("server", "")).strip()
port = entry.get("port")
if not server or port is None:
return None
opts = entry.get("reality-opts") or {}
ws = entry.get("ws-opts") or {}
grpc = entry.get("grpc-opts") or {}
http = entry.get("http-opts") or {}
# Network name: "ws" and "h2" are both WebSocket transports
network = str(entry.get("network", "tcp") or "tcp").lower()
transport_type = "ws" if network == "h2" else network
# Security mode: explicit reality-opts -> reality; otherwise honour tls flag
security = "reality" if opts else ("tls" if entry.get("tls", False) else "none")
return ProxyInfo(
uuid=str(entry.get("uuid", "")),
server=server,
port=int(port),
security=security,
type=transport_type,
sni=str(entry.get("servername", "") or ""),
pbk=str(opts.get("public-key", "") or ""),
sid=str(opts.get("short-id", "") or ""),
path=str(ws.get("path", "") or http.get("path", "") or ""),
host=str(ws.get("host", "") or http.get("host", "") or ""),
serviceName=str(grpc.get("service-name", "") or ""),
flow=str(entry.get("flow", "") or ""),
fingerprint=str(entry.get("client-fingerprint", "") or "chrome"),
)
def extract_mihomo_nodes(text: str) -> List[ProxyInfo]:
"""Parse a mihomo/clash.meta YAML subscription into ProxyInfo nodes.
Args:
text: Decoded mihomo YAML subscription content.
Returns:
List of VLESS ProxyInfo instances (non-VLESS entries are skipped).
Raises:
yaml.YAMLError: If the content is not valid YAML.
"""
import yaml # PyYAML
try:
data = yaml.safe_load(text)
except yaml.YAMLError as exc:
logger.error("Mihomo YAML parse error: %s", exc)
raise
if not isinstance(data, dict):
return []
proxies = data.get("proxies") or []
result: List[ProxyInfo] = []
for entry in proxies:
if not isinstance(entry, dict):
continue
try:
proxy = _mihomo_entry_to_proxy(entry)
except Exception as exc:
logger.warning("Skipping unparseable mihomo entry %r: %s",
entry.get("name"), exc)
continue
if proxy is not None:
result.append(proxy)
return result
def extract_nodes(text: str) -> List[ProxyInfo]:
"""Extract all usable VLESS nodes from a subscription, any known format.
Detects mihomo/clash.meta YAML exports (top-level `proxies:` list) and
plain `vless://` URI subscriptions.
Args:
text: Raw (already decoded) subscription content.
Returns:
List of ProxyInfo instances.
"""
if re.search(r"(?m)^\s*proxies\s*:", text):
try:
nodes = extract_mihomo_nodes(text)
if nodes:
return nodes
logger.warning("Mihomo YAML detected but no VLESS nodes inside")
except Exception:
pass
# Fallback: plain vless:// lines (base64 or plain)
return [url_to_proxy_info(u) for u in extract_vless_lines(text)]