diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000000..3ffde9e6d0 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "manager"] + path = manager + url = https://github.com/wzdnzd/proxy-manager.git diff --git a/clash/clash-darwin-amd b/clash/clash-darwin-amd index 7a6122f093..d48f579cc5 100644 Binary files a/clash/clash-darwin-amd and b/clash/clash-darwin-amd differ diff --git a/clash/clash-darwin-arm b/clash/clash-darwin-arm index d7e9f7a1b5..b64be7f2f0 100644 Binary files a/clash/clash-darwin-arm and b/clash/clash-darwin-arm differ diff --git a/clash/clash-linux-amd b/clash/clash-linux-amd index 4e4e41bcd3..81322e8721 100644 Binary files a/clash/clash-linux-amd and b/clash/clash-linux-amd differ diff --git a/clash/clash-linux-arm b/clash/clash-linux-arm index d6a6c9ad43..96065b0239 100644 Binary files a/clash/clash-linux-arm and b/clash/clash-linux-arm differ diff --git a/clash/clash-windows-amd.exe b/clash/clash-windows-amd.exe index a21601b154..0cb4d7c2ce 100644 Binary files a/clash/clash-windows-amd.exe and b/clash/clash-windows-amd.exe differ diff --git a/manager b/manager new file mode 160000 index 0000000000..e3d93e51e7 --- /dev/null +++ b/manager @@ -0,0 +1 @@ +Subproject commit e3d93e51e762d269546e299490ff114031272e0b diff --git a/subconverter/subconverter-darwin-amd b/subconverter/subconverter-darwin-amd index b74b3a0091..e360517aad 100644 Binary files a/subconverter/subconverter-darwin-amd and b/subconverter/subconverter-darwin-amd differ diff --git a/subconverter/subconverter-darwin-arm b/subconverter/subconverter-darwin-arm index b74b3a0091..e360517aad 100644 Binary files a/subconverter/subconverter-darwin-arm and b/subconverter/subconverter-darwin-arm differ diff --git a/subconverter/subconverter-linux-amd b/subconverter/subconverter-linux-amd index 03dbba7f03..7749222c08 100644 Binary files a/subconverter/subconverter-linux-amd and b/subconverter/subconverter-linux-amd differ diff --git a/subconverter/subconverter-linux-arm b/subconverter/subconverter-linux-arm index fd48ed5683..9390376a51 100644 Binary files a/subconverter/subconverter-linux-arm and b/subconverter/subconverter-linux-arm differ diff --git a/subconverter/subconverter-windows-amd.exe b/subconverter/subconverter-windows-amd.exe index f2f9d6a3c3..b7dc348272 100644 Binary files a/subconverter/subconverter-windows-amd.exe and b/subconverter/subconverter-windows-amd.exe differ diff --git a/subscribe/airport.py b/subscribe/airport.py index b2b6d39b92..ea123925d5 100644 --- a/subscribe/airport.py +++ b/subscribe/airport.py @@ -533,14 +533,15 @@ def parse( with open(self.sub, "r", encoding="UTF8") as f: text = f.read() else: - headers = {"User-Agent": "Clash.Meta; Mihomo"} + client = f"{utils.USER_AGENT}; Clash.Meta; Mihomo; Shadowrocket;" + headers = {"User-Agent": client} trace = os.environ.get("TRACE_ENABLE", "false").lower() in ["true", "1"] text = utils.http_get( url=self.sub, headers=headers, retry=retry, - timeout=30, + timeout=120, trace=trace, interval=1, max_size=15 * 1024 * 1024, @@ -699,7 +700,7 @@ def parse( @staticmethod def check_protocol(link: str) -> bool: return re.match( - r"^(vmess|trojan|ss|ssr|vless|hysteria|hysteria2|tuic|snell|anytls)://[a-zA-Z0-9:.?+=@%&#_\-/]{10,}", + r"^(vmess|trojan|ss|ssr|vless|hysteria|hysteria2|tuic|snell|anytls|socks5|https?)://[a-zA-Z0-9:.?+=@%&#_\-/]{10,}", utils.trim(link).replace("\r", ""), flags=re.I, ) @@ -754,7 +755,7 @@ def clean_text(document: str) -> str: not is_b64encode and not is_json and not is_yaml - and all(AirPort.check_protocol(x) for x in text.split("\n") if x) + and all(AirPort.check_protocol(x) for x in text.split("\n") if x and not x.startswith("#")) ): text = base64.b64encode(text.encode(encoding="UTF8")).decode(encoding="UTF8") @@ -766,7 +767,7 @@ def clean_text(document: str) -> str: if os.path.exists(v2ray_file): os.remove(v2ray_file) - logger.error(f"save file fialed, artifact: {artifact}") + logger.error(f"save file failed, artifact: {artifact}") traceback.print_exc() generate_conf = os.path.join(PATH, "subconverter", "generate.ini") @@ -777,6 +778,7 @@ def clean_text(document: str) -> str: f"{artifact}.yaml", "clash", True, + True, ignore, ) if not success: diff --git a/subscribe/clash.py b/subscribe/clash.py index 193559f585..2733b89b88 100644 --- a/subscribe/clash.py +++ b/subscribe/clash.py @@ -192,7 +192,7 @@ def proxies_exists(proxy: dict, hosts: dict) -> bool: "xchacha20-ietf-poly1305", ] -# reference: https://github.com/MetaCubeX/sing-shadowsocks2/blob/dev/shadowaead_2022/method.go#L73-L86 +# reference: https://github.com/SagerNet/sing-shadowsocks2/blob/dev/shadowaead_2022/method.go#L72-L86 MIHOMO_SS_SUPPORTED_CIPHERS_SALT_LEN = { "2022-blake3-aes-128-gcm": 16, "2022-blake3-aes-256-gcm": 32, @@ -247,6 +247,31 @@ def proxies_exists(proxy: dict, hosts: dict) -> bool: SPECIAL_PROTOCOLS = set(["vless", "tuic", "hysteria", "hysteria2", "anytls"]) +VLESS_MLKEM_X25519_PLUS_PREFIX = "mlkem768x25519plus" +VLESS_MLKEM_X25519_PLUS_MODES = ("native", "xorpub", "random") +VLESS_MLKEM_X25519_PLUS_RTTS = ("1rtt", "0rtt") +VLESS_MLKEM_X25519_PLUS_PADDING_LIMIT = 20 +VLESS_MLKEM_X25519_PLUS_KEY_SIZES = (32, 1184) + +# mihomo ParseRange uses strconv.Atoi (64-bit int on our binaries) +# see: https://github.com/MetaCubeX/mihomo/blob/Alpha/transport/xhttp/config.go +XHTTP_RANGE_MAX = 2**63 - 1 +XHTTP_RANGE_FIELDS = ( + "sc-max-each-post-bytes", + "sc-min-posts-interval-ms", + "x-padding-bytes", + "uplink-chunk-size", + "session-length", +) +XHTTP_RANGE_POSITIVE_MAX = set(["sc-max-each-post-bytes", "sc-min-posts-interval-ms"]) +XHTTP_REUSE_RANGE_FIELDS = ( + "max-concurrency", + "max-connections", + "c-max-reuse-times", + "h-max-request-times", + "h-max-reusable-secs", +) + # xtls-rprx-direct and xtls-rprx-origin are deprecated and no longer supported # XTLS_FLOWS = set(["xtls-rprx-direct", "xtls-rprx-origin", "xtls-rprx-vision"]) @@ -290,10 +315,207 @@ def check_ports(port: str, ranges: str, protocol: str) -> bool: return True +def verify_vless_encryption(encryption: str) -> bool: + if not encryption or encryption == "none": + return True + + parts = encryption.split(".") + if ( + len(parts) < 4 + or parts[0] != VLESS_MLKEM_X25519_PLUS_PREFIX + or parts[1] not in VLESS_MLKEM_X25519_PLUS_MODES + or parts[2] not in VLESS_MLKEM_X25519_PLUS_RTTS + ): + return False + + for key in parts[3:]: + if len(key) < VLESS_MLKEM_X25519_PLUS_PADDING_LIMIT: + continue + if not re.fullmatch(r"[A-Za-z0-9_-]+", key): + return False + + try: + content = key + "=" * (-len(key) % 4) + decoded = base64.urlsafe_b64decode(content) + except: + return False + + if len(decoded) not in VLESS_MLKEM_X25519_PLUS_KEY_SIZES: + return False + + return True + + +def verify_ss_2022_password(cipher: str, password: str) -> bool: + # password is ":"-separated standard base64 PSKs; chacha20 rejects EIH (pskList > 1) + # see: https://github.com/SagerNet/sing-shadowsocks2/blob/dev/shadowaead_2022/method.go#L72-L86 + password = utils.trim(password) + if not password: + return False + + words = password.split(":") + if cipher == "2022-blake3-chacha20-poly1305" and len(words) > 1: + return False + + key_len = MIHOMO_SS_SUPPORTED_CIPHERS_SALT_LEN.get(cipher) + if not key_len: + return False + + for word in words: + # Go encoding/base64.StdEncoding: standard alphabet and padding required + if not word or not re.fullmatch(r"[A-Za-z0-9+/]+=*$", word) or len(word) % 4 != 0: + return False + try: + text = base64.b64decode(word, validate=True) + except: + return False + if len(text) != key_len: + return False + + return True + + +def verify_reality_public_key(public_key: str) -> bool: + # mihomo uses base64.RawURLEncoding (URL-safe, no padding) and requires 32 bytes + # see: https://github.com/MetaCubeX/mihomo/blob/Alpha/adapter/outbound/reality.go + public_key = utils.trim(public_key) + if not public_key or not re.fullmatch(r"[A-Za-z0-9_-]+", public_key): + return False + + try: + decoded = base64.urlsafe_b64decode(public_key + "=" * (-len(public_key) % 4)) + except: + return False + + if len(decoded) != 32: + return False + + # reject non-canonical encodings that Go RawURLEncoding.DecodeString rejects + canonical = base64.urlsafe_b64encode(decoded).decode("utf-8").rstrip("=") + return canonical == public_key + + +def parse_xhttp_range_bound(text: str): + # one ParseRange token: decimal int or scientific/float that is a whole number + text = utils.trim(text) + if not text: + return None + + if re.fullmatch(r"[0-9]+", text): + try: + value = int(text) + except: + return None + else: + try: + number = float(text) + except: + return None + if number != number or number < 0 or number == float("inf"): + return None + value = int(number) + if value != number: + return None + + if value < 0 or value > XHTTP_RANGE_MAX: + return None + return value + + +def normalize_xhttp_range(value): + # canonical "123" or "min-max" for mihomo ParseRange; None if invalid + if isinstance(value, bool) or value is None: + return None + if isinstance(value, int): + if value < 0 or value > XHTTP_RANGE_MAX: + return None + return str(value) + if isinstance(value, float): + if value != value or value < 0 or value == float("inf"): + return None + number = int(value) + if number != value or number > XHTTP_RANGE_MAX: + return None + return str(number) + + text = utils.trim(str(value)) + if not text: + return "" + + bound = parse_xhttp_range_bound(text) + if bound is not None: + return str(bound) + + # original value is already a range like "16-32" or "1e5-2e5", not a clamp + if text.count("-") != 1: + return None + left, right = text.split("-", 1) + min_val, max_val = parse_xhttp_range_bound(left), parse_xhttp_range_bound(right) + if min_val is None or max_val is None or max_val < min_val: + return None + if min_val == max_val: + return str(min_val) + return f"{min_val}-{max_val}" + + +def apply_xhttp_range_field(container: dict, key: str, min_positive: bool = False, max_positive: bool = False) -> bool: + if key not in container: + return True + + value = container[key] + if value is None or (isinstance(value, str) and not utils.trim(value)): + container.pop(key, None) + return True + + normalized = normalize_xhttp_range(value) + if normalized is None: + return False + if not normalized: + container.pop(key, None) + return True + + parts = normalized.split("-") + min_val, max_val = int(parts[0]), int(parts[-1]) + if min_positive and min_val <= 0: + return False + if max_positive and max_val <= 0: + return False + + container[key] = min_val if len(parts) == 1 else normalized + return True + + +def verify_xhttp_reuse_settings(settings: dict) -> bool: + if type(settings) != dict: + return False + + for key in XHTTP_REUSE_RANGE_FIELDS: + if not apply_xhttp_range_field(settings, key): + return False + + if "h-keep-alive-period" not in settings: + return True + + value = settings["h-keep-alive-period"] + if value is None or (isinstance(value, str) and not utils.trim(value)): + settings.pop("h-keep-alive-period", None) + return True + + normalized = normalize_xhttp_range(value) + if not normalized or "-" in normalized: + return False + + settings["h-keep-alive-period"] = int(normalized) + return True + + def verify(item: dict, mihomo: bool = True) -> bool: if not item or type(item) != dict or "type" not in item: return False + # remove dialer-proxy because target proxy maybe not exists + item.pop("dialer-proxy", None) + try: # name must be string name = str(item.get("name", "")).strip() @@ -337,17 +559,8 @@ def verify(item: dict, mihomo: bool = True) -> bool: return False if item["cipher"] in MIHOMO_SS_SUPPORTED_CIPHERS_SALT_LEN: - # will throw bad key length error - # see: https://github.com/MetaCubeX/sing-shadowsocks2/blob/dev/shadowaead_2022/method.go#L59-L108 - password = str(item.get(authentication, "")) - words = password.split(":") - for word in words: - try: - text = base64.b64decode(word) - if len(text) != MIHOMO_SS_SUPPORTED_CIPHERS_SALT_LEN.get(item["cipher"]): - return False - except: - return False + if not verify_ss_2022_password(item["cipher"], str(item.get(authentication, ""))): + return False plugin = item.get("plugin", "") @@ -521,10 +734,16 @@ def verify(item: dict, mihomo: bool = True) -> bool: elif item["type"] == "vless": authentication = "uuid" + + # see: https://github.com/MetaCubeX/mihomo/blob/Alpha/transport/vless/encryption/factory.go#L12 + encryption = utils.trim(item.get("encryption", "")) + if not verify_vless_encryption(encryption): + return False + network = utils.trim(item.get("network", "tcp")) # mihomo: https://wiki.metacubex.one/config/proxies/vless/#network - network_opts = ["ws", "tcp", "grpc", "http", "h2"] if mihomo else ["ws", "tcp", "grpc"] + network_opts = ["ws", "tcp", "grpc", "http", "h2", "xhttp"] if mihomo else ["ws", "tcp", "grpc"] if network not in network_opts: return False @@ -564,11 +783,11 @@ def verify(item: dict, mihomo: bool = True) -> bool: return False content = utils.trim(reality_opts["public-key"]) - content += "=" * (4 - len(content) % 4) - public_key = base64.urlsafe_b64decode(content) - if len(public_key) != 32: + if not verify_reality_public_key(content): return False + reality_opts["public-key"] = content + short_id = reality_opts["short-id"] if type(short_id) != str: if utils.is_number(short_id): @@ -576,10 +795,56 @@ def verify(item: dict, mihomo: bool = True) -> bool: else: return False - if len(short_id) != 8 or not is_hex(short_id) or re.match(r"\d+e\d+", short_id, flags=re.I): - return False + if short_id: + try: + sib = bytes.fromhex(short_id) + if len(sib) > 8: + return False + except ValueError: + return False reality_opts["short-id"] = QuotedStr(short_id) + if "xhttp-opts" in item: + if network != "xhttp": + return False + + xhttp_opts = item.get("xhttp-opts", {}) + if not xhttp_opts or type(xhttp_opts) != dict: + return False + if "path" in xhttp_opts and type(xhttp_opts["path"]) != str: + return False + if "host" in xhttp_opts and type(xhttp_opts["host"]) != str: + return False + + if "mode" in xhttp_opts: + xhttp_mode = utils.trim(xhttp_opts.get("mode", "")) + if xhttp_mode and xhttp_mode not in ["stream-one", "stream-up", "packet-up"]: + return False + if "headers" in xhttp_opts and type(xhttp_opts["headers"]) != dict: + return False + + # ParseRange fields must be decimal ints or min-max, not 1E+05 / 100000.0 + # see: https://github.com/MetaCubeX/mihomo/blob/Alpha/transport/xhttp/config.go + for key in XHTTP_RANGE_FIELDS: + min_positive = key == "session-length" + max_positive = key in XHTTP_RANGE_POSITIVE_MAX + if not apply_xhttp_range_field( + xhttp_opts, key, min_positive=min_positive, max_positive=max_positive + ): + return False + + if "reuse-settings" in xhttp_opts and not verify_xhttp_reuse_settings( + xhttp_opts.get("reuse-settings") + ): + return False + if "download-settings" in xhttp_opts: + download_settings = xhttp_opts.get("download-settings") + if type(download_settings) != dict: + return False + if "reuse-settings" in download_settings and not verify_xhttp_reuse_settings( + download_settings.get("reuse-settings") + ): + return False elif item["type"] == "tuic": # mihomo: https://wiki.metacubex.one/config/proxies/tuic token = wrap(item.get("token", "")) diff --git a/subscribe/crawl.py b/subscribe/crawl.py index 0c8cdc2bed..e999cedc6b 100644 --- a/subscribe/crawl.py +++ b/subscribe/crawl.py @@ -41,6 +41,56 @@ SEPARATOR = "-" +_SUSPICIOUS_URL_RE = re.compile( + r"(?i)(" + r"(?:^|//)(?:[^/]*\.)?(?:speedtest|librespeed|fast\.com|cachefly\.net|thinkbroadband\.com|speed\.cloudflare\.com)" + r"|/__down(?:\?|$)" + r"|\.(?:zip|iso|exe|mp4|mkv|avi|tar|tgz|gz|7z|rar|bin|img|dmg|apk|msi|pdf)(?:\?|$)" + r")" +) + +_REJECT_CONTENT_TYPE_PREFIXES = ( + "video/", + "audio/", + "image/", + "font/", + "application/zip", + "application/x-zip", + "application/gzip", + "application/x-gzip", + "application/x-tar", + "application/x-rar", + "application/pdf", + "application/x-msdownload", + "application/vnd.", +) + +_REJECT_DISPOSITION_EXT = ( + ".zip", + ".iso", + ".exe", + ".mp4", + ".mkv", + ".avi", + ".tar", + ".tgz", + ".gz", + ".7z", + ".rar", + ".bin", + ".img", + ".dmg", + ".apk", + ".msi", + ".pdf", + ".torrent", +) + +_VALIDATE_MAX_BYTES = 15 * 1024 * 1024 +_VALIDATE_UNIDENTIFIED_BYTES = 64 * 1024 +_VALIDATE_CHUNK_SIZE = 8 * 1024 +_VALIDATE_SNIFF_DEADLINE = 8 + @dataclass class ValidateResult(object): @@ -1271,15 +1321,33 @@ def check_status( if not url or retry <= 0: return False, connectable + # 垃圾 URL 提前拦截 + if _is_suspicious_url(url): + logger.debug(f"[Validate] skip suspicious url: {utils.mask(url)}") + return False, True + try: - headers = {"User-Agent": "clash.meta"} + headers = {"User-Agent": f"{utils.USER_AGENT}; Clash.Meta; Mihomo; Shadowrocket;"} request = urllib.request.Request(url=url, headers=headers) response = urllib.request.urlopen(request, timeout=10, context=utils.CTX) if response.getcode() != 200: return False, connectable - # in order to avoid the request to the speed test site causing constant data downloads, limit the maximum read to 15MB - content = str(response.read(15 * 1024 * 1024), encoding="utf8") + # 响应头检查 - 明显不是订阅 + if _should_reject_response(response): + response.close() + logger.debug(f"[Validate] reject by header: {utils.mask(url)}") + return False, True + + # 按需分块读取 + 嗅探，先确认像不像订阅，再决定读多少 + content = _read_subscription_body(response) + if content is None: + return False, True + + try: + content = content.decode("utf-8") + except UnicodeDecodeError: + return False, True # response text is too short, ignore if len(content) < 32: @@ -1310,7 +1378,7 @@ def check_status( return is_expired(header=subscription, remain=remain, spare_time=spare_time, tolerance=tolerance) except urllib.error.HTTPError as e: try: - message = str(e.read(), encoding="utf8") + message = str(e.read(4096), encoding="utf8") except: message = "" @@ -1326,7 +1394,8 @@ def check_status( ) return False, expired - except Exception as e: + except (socket.timeout, TimeoutError, ssl.SSLError, ConnectionError, OSError): + # 网络类错误继续重试 return check_status( url=url, retry=retry - 1, @@ -1335,9 +1404,15 @@ def check_status( tolerance=tolerance, connectable=connectable, ) + except Exception as e: + # 其他非网络错误不再重试 + logger.debug(f"[Validate] unexpected error for {utils.mask(url)}: {e}") + return False, connectable def _parse_yaml_proxies(content: str) -> list | None: + if not content or not re.search(r"^proxies\s*:", content, flags=re.M): + return None try: return yaml.load(content, Loader=yaml.SafeLoader).get("proxies", []) @@ -1385,6 +1460,151 @@ def is_expired(header: str, remain: float = 0, spare_time: float = 0, tolerance: return True, False +def _is_suspicious_url(url: str) -> bool: + """URL 前置垃圾拦截""" + if not url: + return False + + return bool(_SUSPICIOUS_URL_RE.search(url)) + + +def _should_reject_response(response) -> bool: + """响应头检查 - 明显不是订阅""" + ctype = (response.getheader("Content-Type") or "").split(";")[0].strip().lower() + if ctype: + for prefix in _REJECT_CONTENT_TYPE_PREFIXES: + if ctype.startswith(prefix): + return True + + disposition = response.getheader("Content-Disposition") or "" + if disposition: + m = re.search(r"filename\*?=(?:UTF-8''|\"|')?([^\";]+)", disposition, re.I) + if m: + name = urllib.parse.unquote(m.group(1).strip().strip("\"'")).lower() + if name.endswith(_REJECT_DISPOSITION_EXT): + return True + + return False + + +def _read_subscription_body(response) -> bytes | None: + """分块读取 + 嗅探 + 动态上限，大聚合订阅不受影响""" + start = time.monotonic() + buf = bytearray() + limit = _VALIDATE_UNIDENTIFIED_BYTES + verdict = "maybe" + + while len(buf) < limit: + # 嗅探阶段设置时限 + if verdict != "yes" and (time.monotonic() - start) > _VALIDATE_SNIFF_DEADLINE: + try: + response.close() + except Exception: + pass + return None + + to_read = min(_VALIDATE_CHUNK_SIZE, limit - len(buf)) + try: + chunk = response.read(to_read) + except (socket.timeout, TimeoutError, OSError): + try: + response.close() + except Exception: + pass + return None + + if not chunk: + break + + buf.extend(chunk) + + if verdict != "yes": + verdict = _looks_like_subscription(bytes(buf)) + if verdict == "no": + try: + response.close() + except Exception: + pass + return None + if verdict == "yes": + limit = _VALIDATE_MAX_BYTES + + # 读到 64KB 且 verdict 仍是 "maybe"，说明是大量未识别内容 + if verdict != "yes" and len(buf) >= _VALIDATE_UNIDENTIFIED_BYTES: + try: + extra = response.read(1) + except Exception: + extra = b"" + if extra: + try: + response.close() + except Exception: + pass + return None + + return bytes(buf) + + +def _looks_like_subscription(data: bytes) -> str: + """内容嗅探 - 返回 'yes' / 'no' / 'maybe'""" + if not data: + return "maybe" + + sample = data[:8192] + if b"\x00" in sample: + return "no" + + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + return "no" + + sample_text = text.lstrip(" \t\r\n") + if not sample_text: + return "maybe" + + head = sample_text[:64].lower() + if head.startswith(("= 32 and re.match(r"^[A-Za-z0-9+/=]+$", compact): + return "yes" + + # 纯注释开头，机场订阅常见前言 + lines = [ln.strip() for ln in sample_text.splitlines() if ln.strip()] + if lines and all(ln.startswith("#") or ln.startswith("//") for ln in lines): + return "maybe" + + return "maybe" + + def is_available(url: str, retry: int = 2, remain: float = 0, spare_time: float = 0) -> bool: available, _ = check_status(url=url, retry=retry, remain=remain, spare_time=spare_time) return available diff --git a/subscribe/iplibrary.py b/subscribe/iplibrary.py new file mode 100644 index 0000000000..832f1c7b0c --- /dev/null +++ b/subscribe/iplibrary.py @@ -0,0 +1,520 @@ +# -*- coding: utf-8 -*- + +# @Author : wzdnzd +# @Time : 2026-09-07 + +import html +import ipaddress +import json +import random +import re +import urllib.parse +from dataclasses import dataclass, field +from typing import Optional + +import utils +from logger import logger + +# HTTPS only. Fetched via CONNECT + TLS through the HTTP inbound. +EGRESS_IPV4_URLS = ( + "https://ipinfo.io/ip", + "https://api-ipv4.ip.sb/ip", + "https://ipv4.ip.sb/ip", + "https://ipv4.icanhazip.com", + "https://icanhazip.com", + "https://api4.ipify.org", + "https://api.ipify.org", + "https://v4.ident.me", + "https://ident.me", + "https://iplark.com/ipapi/public/ip", + "https://ipv4.wtfismyip.com/text", + "https://ifconfig.me/ip", + "https://ifconfig.co/ip", + "https://ifconfig.io/ip", + "https://ipecho.net/plain", + "https://wgetip.com", + "https://myexternalip.com/raw", + "https://ip.tyk.nu", + "https://eth0.me", + "https://ipapi.co/ip", + "https://ipv4.nsupdate.info/myip", + "https://myip.ipip.net", + "https://v4.ip.zxinc.org/getip", + "https://www.cloudflare.com/cdn-cgi/trace", + "https://1.1.1.1/cdn-cgi/trace", + "https://httpbin.org/ip", + "https://jsonip.com", + "https://ipv4.seeip.org", + "https://l2.io/ip", +) +_EGRESS_IPV4_TRIES = 8 +_IPV4_RE = re.compile(r"(?:\d{1,3}\.){3}\d{1,3}") +ECHO_HEADERS = { + "Accept": "text/plain, application/json, */*", + "Accept-Language": "en-US,en;q=0.8,zh-CN;q=0.6", + "Cache-Control": "no-cache", + "Origin": None, + "Referer": None, +} + + +def _valid_public_ipv4(text: str) -> str: + try: + addr = ipaddress.ip_address(utils.trim(text)) + except ValueError: + return "" + if addr.version != 4 or not addr.is_global: + return "" + return str(addr) + + +def _parse_egress_ipv4(content) -> str: + if content is None: + return "" + if isinstance(content, dict): + chunks = [] + for key in ("ip", "origin", "query", "address", "IPv4", "ipv4"): + value = content.get(key) + if value is not None: + chunks.append(str(value)) + content = "\n".join(chunks) if chunks else json.dumps(content, ensure_ascii=False) + elif not isinstance(content, str): + content = str(content) + + text = utils.trim(content) + if not text: + return "" + if text[0] in "{[": + try: + parsed = json.loads(text) + except Exception: + parsed = None + if isinstance(parsed, dict): + ip = _parse_egress_ipv4(parsed) + if ip: + return ip + + # Cloudflare trace lists h=1.1.1.1 before ip=; prefer the ip field + for match in re.finditer(r"(?im)^ip=([0-9.]+)\s*$", text): + ip = _valid_public_ipv4(match.group(1)) + if ip: + return ip + + for match in _IPV4_RE.finditer(text): + ip = _valid_public_ipv4(match.group(0)) + if ip: + return ip + return "" + + +def resolve_egress_ipv4(port: int, request, max_retries: int = 2, timeout: int = 15) -> str: + urls = list(EGRESS_IPV4_URLS) + random.shuffle(urls) + tries = min(len(urls), max(_EGRESS_IPV4_TRIES, max_retries + 1)) + for url in urls[:tries]: + success, content = request( + port=port, + url=url, + max_retries=1, + timeout=timeout, + headers=ECHO_HEADERS, + deserialize=False, + quiet=True, + ) + ip = _parse_egress_ipv4(content) if success else "" + if ip: + return ip + + return "" + + +@dataclass +class IPClassifyResult: + country_code: str = "" + company_type: str = "" + asn_type: str = "" + score: Optional[int] = None + raw: dict = field(default_factory=dict) + + +class IPLibrary: + name: str = "" + needs_egress_ip: bool = False + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + raise NotImplementedError + + def classify(self, data: dict) -> IPClassifyResult: + raise NotImplementedError + + def _resolve_egress_ip(self, port: int, request, max_retries: int = 2, timeout: int = 15, ip: str = "") -> str: + text = _valid_public_ipv4(ip) if ip else "" + if text: + return text + return resolve_egress_ipv4(port, request, max_retries=max_retries, timeout=timeout) + + def _get( + self, + request, + port: int, + url: str, + max_retries: int, + timeout: int, + headers: dict = None, + deserialize: bool = True, + ): + success, response = request( + port=port, + url=url, + max_retries=max_retries, + timeout=timeout, + headers=headers, + deserialize=deserialize, + ) + if not success: + return None + return response + + @staticmethod + def _parse_trust_score(value, *, invert: bool = False) -> Optional[int]: + if value is None: + return None + + try: + n = int(value) + except (TypeError, ValueError): + try: + n = int(float(value)) + except (TypeError, ValueError): + return None + + if invert: + n = 100 - n + + return max(0, min(100, n)) + + @staticmethod + def _nested(data, *keys) -> dict: + current = data + for key in keys: + if not isinstance(current, dict): + return {} + current = current.get(key) + return current if isinstance(current, dict) else {} + + +class IPNetCoffeeLibrary(IPLibrary): + name = "ipnetcoffee" + needs_egress_ip = True + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + ip = self._resolve_egress_ip(port, request, max_retries=max_retries, timeout=timeout, ip=ip) + if not ip: + return {} + + url = f"https://ip.net.coffee/api/ip/lookup/{urllib.parse.quote(ip, safe='')}" + response = self._get(request, port, url, max_retries, timeout) + return response if isinstance(response, dict) else {} + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + if data.get("isResidential") is True and utils.trim(data.get("company_type", "")) != "business": + company_type, asn_type = "isp", "isp" + else: + company_type, asn_type = "hosting", "hosting" + + return IPClassifyResult( + country_code=utils.trim(data.get("countryCode", "")).upper(), + company_type=company_type, + asn_type=asn_type, + score=self._parse_trust_score(data.get("trust_score")), + raw=data, + ) + + +class MeowVPSLibrary(IPLibrary): + name = "meowvps" + needs_egress_ip = True + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + ip = self._resolve_egress_ip(port, request, max_retries=max_retries, timeout=timeout, ip=ip) + if not ip: + return {} + + url = f"https://meowvps.com/api/ip-aggregator/{urllib.parse.quote(ip, safe='')}" + headers = { + "Accept": "*/*", + "Accept-Language": "zh-CN,zh;q=0.8", + "Origin": "https://meowvps.com", + "Referer": "https://meowvps.com/tools/ip-check/", + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/152.0.0.0 Safari/537.36" + ), + } + response = self._get(request, port, url, max_retries, timeout, headers=headers) + if not isinstance(response, dict): + return {} + if response.get("success") is False: + logger.warning(f"MeowVPS lookup failed, ip: {ip}") + return {} + + return response + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + core = self._nested(data, "core_data") + digital = self._nested(data, "api4", "digital") + traits = self._nested(data, "minfraud", "traits") + scores = self._nested(data, "risk_assessment", "ipdata", "scores") + + digital_type = "" if digital.get("type") is None else str(digital.get("type")).strip().lower() + user_type = str(traits.get("user_type") or "").strip().lower() + + if digital_type == "edu" or user_type in {"college", "education", "edu"}: + company_type, asn_type = "edu", "edu" + elif self._is_residential(digital_type, user_type, data): + company_type, asn_type = "isp", "isp" + else: + company_type, asn_type = "hosting", "hosting" + + country_code = utils.trim(core.get("country_code") or digital.get("country_code") or "").upper() + return IPClassifyResult( + country_code=country_code, + company_type=company_type, + asn_type=asn_type, + score=self._parse_trust_score(scores.get("trust_score")), + raw=data, + ) + + @classmethod + def _is_residential(cls, digital_type: str, user_type: str, data: dict) -> bool: + if digital_type in {"hosting", "edu"}: + return False + if user_type in {"hosting", "content_delivery_network", "college"}: + return False + if user_type in {"residential", "traveler", "cellular"}: + return True + if digital_type: + return False + + ipapi = cls._nested(data, "risk_assessment", "ipapi") + threat = cls._nested(data, "risk_assessment", "ipdata", "threat") + if ipapi.get("hosting") is True or threat.get("is_datacenter") is True: + return False + + return True + + +class IPPureLibrary(IPLibrary): + name = "ippure" + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + response = self._get(request, port, "https://my.ippure.com/v1/info", max_retries, timeout) + return response if isinstance(response, dict) else {} + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + if data.get("isResidential", False): + company_type, asn_type = "isp", "isp" + else: + company_type, asn_type = "hosting", "hosting" + + return IPClassifyResult( + country_code=utils.trim(data.get("countryCode", "")).upper(), + company_type=company_type, + asn_type=asn_type, + score=self._parse_trust_score(data.get("fraudScore"), invert=True), + raw=data, + ) + + +class IP2LocationLibrary(IPLibrary): + name = "ip2location" + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + headers = {"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"} + response = self._get( + request, + port, + "https://www.ip2location.com/demo", + max_retries, + timeout, + headers=headers, + deserialize=False, + ) + if not isinstance(response, str): + return {} + + data = self._extract_data(response) + if not data: + logger.warning("Failed to extract JSON payload from ip2location demo HTML") + return {} + return data + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + usage_type = utils.trim(data.get("usage_type", "")).lower() + as_info = data.get("as_info", {}) + if not isinstance(as_info, dict): + as_info = {} + as_usage_type = utils.trim(as_info.get("as_usage_type", "")).lower() + + check = lambda usage: usage.startswith("isp") or usage == "mob" + if check(usage_type) and check(as_usage_type): + company_type, asn_type = "isp", "isp" + else: + company_type, asn_type = "hosting", "hosting" + + return IPClassifyResult( + country_code=utils.trim(data.get("country_code", "")).upper(), + company_type=company_type, + asn_type=asn_type, + score=self._parse_trust_score(data.get("fraud_score"), invert=True), + raw=data, + ) + + @staticmethod + def _extract_data(content: str) -> dict: + if not content or not isinstance(content, str): + return {} + + pattern = r']*class=["\'][^"\']*\blanguage-json\b[^"\']*["\'][^>]*>(.*?)\s*' + groups = re.findall(pattern, content, flags=re.I | re.S) + if not groups: + return {} + + for group in groups: + payload = utils.trim(group) + if not payload: + continue + + payload = re.sub(r"<[^>]+>", "", payload, flags=re.I | re.S) + payload = html.unescape(payload) + + try: + data = json.loads(payload) + except Exception: + continue + + if isinstance(data, dict) and data: + return data + + return {} + + +class IPLarkLibrary(IPLibrary): + name = "iplark" + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + response = self._get(request, port, "https://iplark.com/ipapi/public/ipinfo", max_retries, timeout) + return response if isinstance(response, dict) else {} + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + node_type = utils.trim(data.get("type", "")).lower() + if node_type == "isp": + company_type, asn_type = "isp", "isp" + elif node_type == "business": + company_type, asn_type = "business", "business" + else: + company_type, asn_type = "hosting", "hosting" + + return IPClassifyResult( + country_code=utils.trim(data.get("country_code", "")).upper(), + company_type=company_type, + asn_type=asn_type, + raw=data, + ) + + +class IPInfoLibrary(IPLibrary): + name = "ipinfo" + needs_egress_ip = True + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + ip = self._resolve_egress_ip(port, request, max_retries=max_retries, timeout=timeout, ip=ip) + if not ip: + return {} + + url = f"https://ipinfo.io/widget/demo/{ip}" + response = self._get(request, port, url, max_retries, timeout) + if not isinstance(response, dict): + return {} + + data = response.get("data", response) + return data if isinstance(data, dict) else {} + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + company = data.get("company", {}) if isinstance(data.get("company"), dict) else {} + asn = data.get("asn", {}) if isinstance(data.get("asn"), dict) else {} + return IPClassifyResult( + country_code=utils.trim(data.get("country", "")).upper(), + company_type=utils.trim(company.get("type", "")).lower(), + asn_type=utils.trim(asn.get("type", "")).lower(), + raw=data, + ) + + +class IPApiLibrary(IPLibrary): + name = "ipapi" + + def __init__(self, api_key: str = ""): + self.api_key = utils.trim(api_key) + + def fetch(self, port: int, request, max_retries: int = 2, timeout: int = 12, ip: str = "") -> dict: + url = "https://api.ipapi.is" + if self.api_key: + url += f"?key={self.api_key}" + + response = self._get(request, port, url, max_retries, timeout) + return response if isinstance(response, dict) else {} + + def classify(self, data: dict) -> IPClassifyResult: + data = data if isinstance(data, dict) else {} + location = data.get("location", {}) if isinstance(data.get("location"), dict) else {} + company = data.get("company", {}) if isinstance(data.get("company"), dict) else {} + asn = data.get("asn", {}) if isinstance(data.get("asn"), dict) else {} + return IPClassifyResult( + country_code=utils.trim(location.get("country_code", "")).upper(), + company_type=utils.trim(company.get("type", "")).lower(), + asn_type=utils.trim(asn.get("type", "")).lower(), + raw=data, + ) + + +PROVIDERS_ORDER = ["ipnetcoffee", "meowvps", "ippure", "ip2location", "ipinfo"] + +LIBRARIES = { + "ipnetcoffee": IPNetCoffeeLibrary, + "meowvps": MeowVPSLibrary, + "ippure": IPPureLibrary, + "ip2location": IP2LocationLibrary, + "iplark": IPLarkLibrary, + "ipinfo": IPInfoLibrary, + "ipapi": IPApiLibrary, +} + + +def get_providers(preferred: str) -> list[str]: + library = utils.trim(preferred).lower() + if library not in PROVIDERS_ORDER: + if library != "": + logger.warning(f"IP library {library} is not be supported") + + library = "ipnetcoffee" + + return [library] + [item for item in PROVIDERS_ORDER if item != library] + + +def create_library(name: str, api_key: str = "") -> IPLibrary: + key = utils.trim(name).lower() + if key not in LIBRARIES: + key = "ipnetcoffee" + + cls = LIBRARIES[key] + if cls is IPApiLibrary: + return IPApiLibrary(api_key=api_key) + + return cls() diff --git a/subscribe/location.py b/subscribe/location.py index 1a00668397..e4a9d9940f 100644 --- a/subscribe/location.py +++ b/subscribe/location.py @@ -3,35 +3,49 @@ # @Author : wzdnzd # @Time : 2024-07-12 +import gzip +import http.client import json import math import os import random import re import socket +import ssl import subprocess -import sys import time -import urllib +import urllib.parse from collections import defaultdict from dataclasses import dataclass +from functools import partial +from typing import Optional import utils import yaml from executable import which_bin from geoip2 import database +from iplibrary import create_library, get_providers, resolve_egress_ipv4 from logger import logger from clash import is_mihomo @dataclass -class ProxyInfo: - """Proxy query result information""" +class GeoInfo: + """Country and CDN attributes of an IP""" - name: str = "" country: str = "" + is_cdn: bool = False + + +@dataclass +class ProxyInfo(GeoInfo): + """Proxy query result, including geo attributes""" + + name: str = "" ip_type: str = "" + score: Optional[int] = None + provider: str = "" @dataclass @@ -296,6 +310,34 @@ class ProxyQueryResult: } +# Pattern for CDN providers and Loyalsoldier custom ISO codes +CDN_PATTERN = r"cloudflare|cloudfront|fastly|google" +_CDN_NAME_RE = re.compile(CDN_PATTERN, flags=re.I) + + +def is_cdn_label(value: str) -> bool: + """Return True if a country name or ISO code refers to a CDN instead of a location""" + text = utils.trim(value) + return bool(text and _CDN_NAME_RE.search(text)) + + +def _mark_cdn(proxy: dict) -> None: + if isinstance(proxy, dict): + proxy["cdn"] = True + + +def _is_cdn_proxy(proxy: dict) -> bool: + return isinstance(proxy, dict) and (bool(proxy.get("cdn")) or is_cdn_label(str(proxy.get("name", "")))) + + +def _remove_temp_flags(proxies: list[dict]) -> list[dict]: + for proxy in proxies: + if isinstance(proxy, dict): + proxy.pop("cdn", None) + proxy.pop("renamed", None) + return proxies + + def download_mmdb(repo: str, target: str, filepath: str, retry: int = 3) -> bool: """ Download GeoLite2-City.mmdb from github release @@ -387,48 +429,51 @@ def load_mmdb( return database.Reader(filepath) -def query_ip_country(ip: str, reader: database.Reader) -> str: +def lookup_ip_geo(ip: str, reader: database.Reader) -> GeoInfo: """ Query country information for an IP address using mmdb database - Args: - ip: The IP address to query - reader: The mmdb database reader - - Returns: - The country name in Chinese + CDN ranges such as Cloudflare are reported via is_cdn and never as a country """ if not ip or not reader: - return "" + return GeoInfo() try: # fake ip if ip.startswith("198.18.0."): logger.warning("cannot get geolocation because IP address is faked") - return "" + return GeoInfo() response = reader.country(ip) + names = response.country.names or {} + iso_code = utils.trim(response.country.iso_code).upper() + country = utils.trim(names.get("zh-CN", "")) - # Try to get country name in Chinese - country = response.country.names.get("zh-CN", "") - - # If Chinese name is not available, try to convert ISO code to Chinese country name - if not country and response.country.iso_code: - iso_code = response.country.iso_code - # Try to get Chinese country name from ISO code mapping + if not country and iso_code: country = ISO_TO_CHINESE.get(iso_code, iso_code) - # Special handling for well-known IPs - if not country: - if ip == "1.1.1.1" or ip == "1.0.0.1": - country = "Cloudflare" - elif ip.startswith("8.8.8.") or ip.startswith("8.8.4."): - country = "Google" + well_known_cdn = ip in ("1.1.1.1", "1.0.0.1") or ip.startswith("8.8.8.") or ip.startswith("8.8.4.") + if well_known_cdn or is_cdn_label(iso_code) or is_cdn_label(country): + return GeoInfo(is_cdn=True) - return country + return GeoInfo(country=country) except Exception as e: logger.error(f"query ip country failed, ip: {ip}, error: {str(e)}") - return "" + return GeoInfo() + + +def query_ip_country(ip: str, reader: database.Reader) -> str: + """ + Query country information for an IP address using mmdb database + + Args: + ip: The IP address to query + reader: The mmdb database reader + + Returns: + The country name in Chinese + """ + return lookup_ip_geo(ip, reader).country def locate_by_geoip(proxy: dict, reader: database.Reader) -> dict: @@ -446,258 +491,82 @@ def locate_by_geoip(proxy: dict, reader: database.Reader) -> dict: return proxy ip = socket.gethostbyname(address) - country = query_ip_country(ip, reader) - - if country: - proxy["name"] = country + geo = lookup_ip_geo(ip, reader) + if geo.is_cdn: + _mark_cdn(proxy) + logger.debug(f"server IP belongs to CDN, skip as location, address: {address}") + elif geo.country: + proxy["name"] = geo.country proxy["renamed"] = True else: - logger.warning(f"cannot get geolocation and rename, address: {address}") + logger.warning(f"cannot get geolocation and name, address: {address}") except Exception as e: logger.error(f"query ip geolocation failed, address: {address}, error: {str(e)}") return proxy -# Cache for checked port statuses -_PORT_STATUS_CACHE = {} -_AVAILABLE_PORTS = set() - - -def get_listening_ports() -> set: - """Get the set of listening ports in the system, cross-platform compatible""" - listening_ports = set() - - try: - # Windows system - if os.name == "nt": - try: - # Use 'cp437' encoding to handle Windows command line output - output = subprocess.check_output("netstat -an", shell=True).decode("cp437", errors="replace") - for line in output.split("\n"): - if "LISTENING" in line: - parts = line.split() - if len(parts) >= 2: - addr_port = parts[1] - if ":" in addr_port: - try: - port = int(addr_port.split(":")[-1]) - listening_ports.add(port) - except ValueError: - pass - except Exception as e: - logger.warning(f"Windows netstat command failed: {str(e)}") - return listening_ports - - # macOS system - elif sys.platform == "darwin": - try: - output = subprocess.check_output("lsof -i -P -n | grep LISTEN", shell=True).decode( - "utf-8", errors="replace" - ) - for line in output.split("\n"): - if ":" in line: - try: - port_part = line.split(":")[-1].split(" ")[0] - port = int(port_part) - listening_ports.add(port) - except (ValueError, IndexError): - pass - except Exception as e: - logger.warning(f"macOS lsof command failed: {str(e)}") - return listening_ports - - # Linux and other systems - else: - # Try using ss command (newer Linux systems) - try: - output = subprocess.check_output("ss -tuln", shell=True).decode("utf-8", errors="replace") - for line in output.split("\n"): - if "LISTEN" in line: - parts = line.split() - for part in parts: - if ":" in part: - try: - port = int(part.split(":")[-1]) - listening_ports.add(port) - except ValueError: - pass - except Exception as e: - logger.warning(f"Linux ss command failed, trying netstat: {str(e)}") - # Fall back to netstat command (older Linux systems) - try: - output = subprocess.check_output("netstat -tuln", shell=True).decode("utf-8", errors="replace") - for line in output.split("\n"): - if "LISTEN" in line: - parts = line.split() - for part in parts: - if ":" in part: - try: - port = int(part.split(":")[-1]) - listening_ports.add(port) - except ValueError: - pass - except Exception as e: - logger.warning(f"Linux netstat command also failed: {str(e)}") - return listening_ports - except Exception as e: - logger.warning(f"Failed to get listening ports: {str(e)}") - - return listening_ports - - -def scan_ports_batch(start_port: int, count: int = 100) -> dict: - """Batch scan port statuses, return a dictionary of port statuses""" - global _PORT_STATUS_CACHE, _AVAILABLE_PORTS - - # Create a list of ports to scan (excluding ports with known status) - ports_to_scan = [p for p in range(start_port, start_port + count) if p not in _PORT_STATUS_CACHE] - - if not ports_to_scan: - # If all ports are already cached, return cached results directly - return {p: _PORT_STATUS_CACHE.get(p, True) for p in range(start_port, start_port + count)} - - # Use a more efficient way to check ports in batch - results = {} +class PortReservation: + """Reserve local TCP ports by binding them without listen/connect.""" - try: - # Get the ports that are currently listening in the system - listening_ports = get_listening_ports() - - # Update results - for port in ports_to_scan: - in_use = port in listening_ports - results[port] = in_use - _PORT_STATUS_CACHE[port] = in_use - if not in_use: - _AVAILABLE_PORTS.add(port) - except Exception as e: - logger.warning(f"Batch port scanning failed, falling back to individual port checks: {str(e)}") - # If batch checking fails, fall back to individual port checks - for port in ports_to_scan: - in_use = check_single_port(port) - results[port] = in_use - _PORT_STATUS_CACHE[port] = in_use - if not in_use: - _AVAILABLE_PORTS.add(port) - - # Merge cached and newly scanned results - return { - **{ - p: _PORT_STATUS_CACHE.get(p, True) for p in range(start_port, start_port + count) if p in _PORT_STATUS_CACHE - }, - **results, - } + def __init__(self): + self._sockets = [] + def reserve(self, n: int) -> list[int]: + if n <= 0: + return [] -def check_single_port(port: int) -> bool: - """Helper function for checking a single port, checks if the port is listening""" - try: - # Use socket to check TCP port - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(0.2) - result = sock.connect_ex(("127.0.0.1", port)) - sock.close() - if result == 0: - return True - - # Also check IPv6 + ports = [] try: - sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) - sock.settimeout(0.2) - result = sock.connect_ex(("::1", port)) - sock.close() - return result == 0 - except: - pass - - return False - except: - # Assume port is not in use when an error occurs - return False - - -def is_port_in_use(port: int) -> bool: - """Check if a port is in use (using cache)""" - global _PORT_STATUS_CACHE, _AVAILABLE_PORTS - - # If port is known to be available, return directly - if port in _AVAILABLE_PORTS: - return False - - # If port status is already cached, return directly - if port in _PORT_STATUS_CACHE: - return _PORT_STATUS_CACHE[port] - - # Otherwise check the port and cache the result - in_use = check_single_port(port) - _PORT_STATUS_CACHE[port] = in_use - if not in_use: - _AVAILABLE_PORTS.add(port) - return in_use + for _ in range(n): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + self._sockets.append(sock) + ports.append(sock.getsockname()[1]) + return ports + except Exception: + self.close() + raise + + def close(self) -> None: + sockets, self._sockets = self._sockets, [] + for sock in sockets: + try: + sock.close() + except Exception: + pass -def generate_mihomo_config(proxies: list[dict]) -> tuple[dict, dict]: +def generate_mihomo_config(proxies: list[dict], listener_ports: list[int], mixed_port: int) -> tuple[dict, dict]: """Generate mihomo configuration for the given proxies""" - # Base configuration config = { - "mixed-port": 7890, + "mixed-port": mixed_port, "allow-lan": True, "mode": "global", "log-level": "error", + "ipv6": False, + "tcp-concurrent": True, "proxies": proxies, "dns": { "enable": True, - "enhanced-mode": "fake-ip", - "fake-ip-range": "198.18.0.1/16", + "ipv6": False, + "enhanced-mode": "redir-host", "default-nameserver": ["114.114.114.114", "223.5.5.5", "8.8.8.8"], - "nameserver": ["https://doh.pub/dns-query"], + "nameserver": ["114.114.114.114", "8.8.8.8"], }, "listeners": [], } - # Record the port assigned to each proxy records = dict() - - # If there are no proxies, return directly if not proxies: return config, records - # Pre-scan ports in batch to improve efficiency - start_port = 32001 - - # Scan enough ports to ensure there are sufficient available ports - port_count = len(proxies) * 2 - port_status = scan_ports_batch(start_port, port_count) - - # Find all available ports - available_ports = [p for p, in_use in port_status.items() if not in_use] - - # If available ports are insufficient, scan more ports - if len(available_ports) < len(proxies): - additional_ports = scan_ports_batch(start_port + port_count, port_count * 2) - available_ports.extend([p for p, in_use in additional_ports.items() if not in_use]) - - # Assign an available port to each proxy for index, proxy in enumerate(proxies): - if index < len(available_ports): - port = available_ports[index] - else: - # If available ports are insufficient, use traditional method to find available ports - port = start_port + port_count + index - max_attempts = 1000 - attempts = 0 - - while is_port_in_use(port) and attempts < max_attempts: - port += 1 - attempts += 1 - - if attempts >= max_attempts: - logger.warning( - f"Could not find an available port for proxy {proxy['name']} after {max_attempts} attempts" - ) - continue + if index >= len(listener_ports): + logger.warning(f"No reserved port for proxy {proxy['name']}") + continue + port = listener_ports[index] listener = { "name": f"http-{index}", "type": "http", @@ -712,6 +581,223 @@ def generate_mihomo_config(proxies: list[dict]) -> tuple[dict, dict]: return config, records +def _idna_host(host: str) -> str: + host = utils.trim(host) + if not host: + return host + try: + return host.encode("idna").decode("ascii") + except Exception: + return host + + +def _origin_headers(url: str, extra: dict = None) -> dict: + parsed = urllib.parse.urlparse(url) + base = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme and parsed.netloc else "" + result = { + "User-Agent": utils.USER_AGENT, + "Accept": "application/json, text/plain, */*", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Accept-Encoding": "identity", + "Cache-Control": "no-cache", + "Pragma": "no-cache", + "Connection": "close", + "Referer": f"{base}/" if base else url, + "Origin": base if base else url, + } + if extra and isinstance(extra, dict): + for key, value in extra.items(): + name = utils.trim(str(key)) + if not name: + continue + if value is None: + result.pop(name, None) + else: + result[name] = value + return result + + +def _proxy_ssl_context() -> ssl.SSLContext: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.options |= ssl.OP_NO_TICKET + if hasattr(ssl, "OP_IGNORE_UNEXPECTED_EOF"): + ctx.options |= ssl.OP_IGNORE_UNEXPECTED_EOF + try: + ctx.set_alpn_protocols(["http/1.1"]) + except Exception: + pass + return ctx + + +def _recv_until(sock: socket.socket, marker: bytes, limit: int = 65536) -> bytes: + buf = bytearray() + while marker not in buf: + chunk = sock.recv(min(4096, max(1, limit - len(buf)))) + if not chunk: + raise OSError(f"tunnel: incomplete proxy response: {bytes(buf)!r}") + buf += chunk + if len(buf) > limit: + raise OSError("tunnel: proxy response too large") + index = buf.find(marker) + len(marker) + if index < len(buf): + raise OSError("tunnel: unexpected data after CONNECT") + return bytes(buf[:index]) + + +def _tls_server_hostname(host: str) -> Optional[str]: + text = _idna_host(host) + if not text: + return None + try: + socket.inet_pton(socket.AF_INET, text) + return None + except OSError: + pass + if ":" in text: + try: + socket.inet_pton(socket.AF_INET6, text.strip("[]")) + return None + except OSError: + pass + return text + + +def _open_http_tunnel(sock: socket.socket, host: str, port: int) -> None: + host = _idna_host(host) + target = f"{host}:{port}" + # Do not send Connection/Proxy-Connection: close. Mihomo hijacks after 200; + # a close flag makes Windows RST the socket (WinError 10054) at TLS. + sock.sendall(f"CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n".encode("ascii")) + head = _recv_until(sock, b"\r\n\r\n") + status_line = head.split(b"\r\n", 1)[0].decode("iso-8859-1", "replace") + parts = status_line.split(" ", 2) + if len(parts) < 2: + raise OSError(f"tunnel: invalid CONNECT response: {status_line}") + try: + code = int(parts[1]) + except ValueError as e: + raise OSError(f"tunnel: invalid CONNECT status: {status_line}") from e + if code != 200: + raise OSError(f"tunnel: CONNECT failed: {status_line}") + + +def _assert_tunnel_open(sock: socket.socket, wait: float = 0.2) -> None: + """Fail fast if the CONNECT already died; otherwise wait out mihomo's peek.""" + previous = sock.gettimeout() + sock.settimeout(max(wait, 0.01)) + try: + peeked = sock.recv(1, socket.MSG_PEEK) + if not peeked: + raise OSError("tunnel: closed before TLS") + except (TimeoutError, socket.timeout): + return + except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError) as e: + raise OSError(f"tunnel: closed before TLS ({e})") from e + finally: + sock.settimeout(previous) + + +def _http_get_on_sock(sock: socket.socket, host: str, path: str, headers: dict) -> tuple[int, str, bytes]: + host = _idna_host(host) + lines = [f"GET {path} HTTP/1.1", f"Host: {host}"] + sent = {"host"} + for key, value in (headers or {}).items(): + name = utils.trim(str(key)) + if not name or name.lower() in sent or name.lower().startswith("proxy-"): + continue + lines.append(f"{name}: {value}") + sent.add(name.lower()) + if "connection" not in sent: + lines.append("Connection: close") + + sock.sendall(("\r\n".join(lines) + "\r\n\r\n").encode("latin-1")) + response = http.client.HTTPResponse(sock, method="GET") + try: + response.begin() + return response.status, utils.trim(response.getheader("Location", "")), response.read() + finally: + try: + response.close() + except Exception: + pass + + +def _decode_body(body: bytes) -> str: + if not body: + return "" + if body[:2] == b"\x1f\x8b": + try: + body = gzip.decompress(body) + except Exception: + pass + for encoding in ("utf-8", "gbk"): + try: + return body.decode(encoding) + except Exception: + continue + return body.decode("utf-8", "replace") + + +def _request_through_proxy(port: int, url: str, headers: dict, timeout: int, redirects: int = 3) -> tuple[int, bytes]: + """ + Fetch URL via mihomo HTTP inbound. + + http:// → absolute-form GET (plain HTTP proxy). + https:// → CONNECT then TLS in Python. Do not send `GET https://...`: + mihomo handles that with client.Do + TLS on net.Pipe, which races the + outbound dial and comes back as HTTP 502. + """ + parsed = urllib.parse.urlparse(url) + scheme = utils.trim(parsed.scheme).lower() + host = parsed.hostname + if not host or scheme not in ("http", "https"): + raise OSError(f"http: unsupported url: {url}") + + dst_port = parsed.port or (443 if scheme == "https" else 80) + path = parsed.path or "/" + if parsed.query: + path += "?" + parsed.query + + sock = socket.create_connection(("127.0.0.1", port), timeout=timeout) + wrapped = None + try: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + sock.settimeout(timeout) + target = sock + request_path = url + if scheme == "https": + _open_http_tunnel(sock, host, dst_port) + _assert_tunnel_open(sock) + try: + wrapped = _proxy_ssl_context().wrap_socket( + sock, + server_hostname=_tls_server_hostname(host), + suppress_ragged_eofs=True, + ) + except OSError as e: + raise OSError(f"tls: {e}") from e + target = wrapped + request_path = path + status, location, body = _http_get_on_sock(target, host, request_path, headers) + finally: + for item in (wrapped, sock): + if item is None: + continue + try: + item.close() + except Exception: + pass + + if status in (301, 302, 303, 307, 308) and redirects > 0 and location: + text = urllib.parse.urljoin(url, location) + if text.lower().startswith("http"): + return _request_through_proxy(port, text, _origin_headers(text, headers), timeout, redirects - 1) + + return status, body + + def make_proxy_request( port: int, url: str, @@ -719,6 +805,7 @@ def make_proxy_request( timeout: int = 10, headers: dict = None, deserialize: bool = True, + quiet: bool = False, ) -> tuple[bool, dict]: """ Make an HTTP request through a proxy and return the response @@ -739,41 +826,26 @@ def make_proxy_request( logger.warning("No port provided for proxy") return False, {} - # Configure the proxy for the request - proxy_url = f"http://127.0.0.1:{port}" - proxies_config = {"http": proxy_url, "https": proxy_url} - - # Configure proxy handler - proxy_handler = urllib.request.ProxyHandler(proxies_config) - - # Build opener with proxy handler - opener = urllib.request.build_opener(proxy_handler) - if headers and isinstance(headers, dict): - opener.addheaders = [(k, v) for k, v in headers.items() if k] - else: - opener.addheaders = [ - ("User-Agent", utils.USER_AGENT), - ("Accept", "application/json"), - ("Connection", "close"), - ] - - # Try to get response with retry and backoff + default_headers = _origin_headers(url, headers) attempt, success, data = 0, False, None while not success and attempt < max(max_retries, 1): try: - # Random sleep to avoid being blocked by the API (increasing with each retry) if attempt > 0: wait_time = min(2**attempt * random.uniform(0.5, 1.5), 6) time.sleep(wait_time) - # Make request - response = opener.open(url, timeout=timeout) - if response.getcode() == 200: - content = response.read().decode("utf-8") - data = json.loads(content) if deserialize else content - success = True + status, body = _request_through_proxy(port, url, default_headers, timeout) + if status != 200: + raise OSError(f"http: status {status}") + content = _decode_body(body) + data = json.loads(content) if deserialize else content + success = True except Exception as e: - logger.warning(f"Attempt {attempt+1} failed to request {url} through proxy port {port}: {str(e)}") + message = f"Attempt {attempt+1} failed to request {url} through proxy port {port}: {str(e)}" + if quiet or attempt + 1 < max(max_retries, 1): + logger.debug(message) + else: + logger.warning(message) attempt += 1 @@ -781,44 +853,52 @@ def make_proxy_request( def get_ipv4(port: int, max_retries: int = 5) -> str: - """ - Get the IPv4 address by accessing https://api.ipify.org?format=json through a proxy - - Args: - port: The port of the proxy - max_retries: Maximum number of retry attempts - - Returns: - The IPv4 address or empty string if failed - """ + """Get the egress IPv4 address through a proxy listener.""" if not port: logger.warning("No port provided for proxy") return "" - success, data = make_proxy_request(port=port, url="https://api.ipify.org?format=json", max_retries=max_retries) - return data.get("ip", "") if success else "" + return resolve_egress_ipv4(port, make_proxy_request, max_retries=max_retries, timeout=10) + + +def _wait_listener(port: int, process: subprocess.Popen, timeout: int = 20) -> None: + deadline = time.time() + max(timeout, 1) + while time.time() < deadline: + if process.poll() is not None: + raise RuntimeError(f"mihomo exited before becoming ready, code={process.returncode}") + try: + sock = socket.create_connection(("127.0.0.1", port), timeout=1) + sock.close() + return + except OSError: + time.sleep(0.2) + raise RuntimeError(f"mihomo listener 127.0.0.1:{port} not ready after {timeout}s") # Online API services for IP location LOCATION_API_SERVICES = [ {"url": "https://ipinfo.io", "country_key": "country"}, - {"url": "https://ipapi.co/json/", "country_key": "country_code"}, + {"url": "https://api.ip2location.io", "country_key": "country_code"}, {"url": "https://ipwho.is", "country_key": "country_code"}, - {"url": "https://freeipapi.com/api/json", "country_key": "countryCode"}, - {"url": "https://api.country.is", "country_key": "country"}, + {"url": "https://free.freeipapi.com/api/json", "country_key": "countryCode"}, {"url": "https://api.ip.sb/geoip", "country_key": "country_code"}, ] -# Pattern for CDN providers -CDN_PATTERN = r"cloudflare|cloudfront|fastly|google" - def random_delay(min_delay: float = 0.01, max_delay: float = 0.5): """Random delay to avoid API rate limiting""" time.sleep(random.uniform(min_delay, max_delay)) -def check_residential(proxy: dict, port: int, api_key: str = "", use_ipinfo: bool = True) -> ProxyQueryResult: +def check_residential( + proxy: dict, + port: int, + api_key: str = "", + ip_library: str = "ipnetcoffee", + reader: database.Reader = None, + max_retries: int = 2, + timeout: int = 12, +) -> ProxyQueryResult: """ Check if a proxy is residential by making a request through it @@ -826,40 +906,14 @@ def check_residential(proxy: dict, port: int, api_key: str = "", use_ipinfo: boo proxy: The proxy information dict port: The port of the proxy api_key: Optional API key for ipapi.is. Uses free tier if not provided - use_ipinfo: Whether to use ipinfo.io instead of ipapi.is, defaults to True + ip_library: IP query provider, supported: ipnetcoffee/meowvps/ippure/ip2location/iplark/ipinfo/ipapi + reader: Optional mmdb reader used to detect CDN egress IPs + max_retries: Retry count for provider queries + timeout: Timeout in seconds for provider queries Returns: ProxyQueryResult: Complete proxy query result """ - - def _get_ipapi_url(key: str = "") -> str: - url, key = "https://api.ipapi.is", utils.trim(key) - if key: - url += f"?key={key}" - return url - - def _get_ipinfo_url(port: int, name: str) -> str: - # First, get the IP address - success, content = make_proxy_request( - port=port, - url="https://ipinfo.io/ip", - max_retries=2, - timeout=15, - deserialize=False, - ) - if not success or not content: - logger.warning(f"Failed to get IP from ipinfo.io for proxy {name}") - return "" - - # Extract IP from response - ip = utils.trim(content) - if not ip: - logger.warning(f"Invalid IP address from ipinfo.io for proxy {name}") - return "" - - # Now get detailed information using the IP - return f"https://ipinfo.io/widget/demo/{ip}" - name = proxy.get("name", "") result = ProxyInfo(name=name) @@ -867,49 +921,91 @@ def _get_ipinfo_url(port: int, name: str) -> str: logger.warning(f"No port found for proxy {name}") return ProxyQueryResult(proxy=proxy, result=result, success=False) - # Random delay to avoid being blocked by the API random_delay() try: - url = "" - if use_ipinfo: - url = _get_ipinfo_url(port=port, name=name) - - if not url: - url = _get_ipapi_url(key=api_key) - use_ipinfo = False - - # Call API for IP information through the proxy - success, response = make_proxy_request(port=port, url=url, max_retries=2, timeout=12) - - # Parse data from response - if success: - try: - data = response.get("data", {}) if use_ipinfo else response + providers = get_providers(ip_library) + classified, provider = None, "" + egress_ip = None + request = partial(make_proxy_request, quiet=True) + + def _cached_egress_ip() -> str: + nonlocal egress_ip + if egress_ip is None: + egress_ip = resolve_egress_ipv4( + port, + request, + max_retries=max_retries, + timeout=max(timeout, 15), + ) + if not egress_ip: + logger.debug(f"Failed to get egress IP for proxy {name}") + return egress_ip - # Extract country code from data - if use_ipinfo: - country_code = data.get("country", "") + if reader: + ip = _cached_egress_ip() + if ip and lookup_ip_geo(ip, reader).is_cdn: + result.is_cdn = True + _mark_cdn(proxy) + logger.debug(f"Egress IP for proxy {name} belongs to CDN, continue locating") + return ProxyQueryResult(proxy=proxy, result=result, success=False) + + for idx, item in enumerate(providers): + library = create_library(item, api_key=api_key) + ip = _cached_egress_ip() if library.needs_egress_ip else "" + if library.needs_egress_ip and not ip: + if idx < len(providers) - 1: + fallback = providers[idx + 1] + logger.debug(f"Skip {item} for proxy {name}, no egress IP, trying fallback: {fallback}") else: - country_code = data.get("location", {}).get("country_code", "") + logger.debug(f"Skip {item} for proxy {name}, no egress IP") + continue - result.country = ISO_TO_CHINESE.get(country_code, "") if country_code else "" + data = library.fetch(port, request, max_retries=max_retries, timeout=timeout, ip=ip) + if data: + provider = item + classified = library.classify(data) + logger.debug(f"IP info for proxy {name} successfully retrieved, provider: {provider}") + break - company_type = data.get("company", {}).get("type", "") - asn_type = data.get("asn", {}).get("type", "") + if idx < len(providers) - 1: + fallback = providers[idx + 1] + logger.debug(f"Failed to query {item} for proxy {name}, trying fallback: {fallback}") + else: + logger.debug(f"Failed to query {item} for proxy {name}") - # Check if it's residential (both company and asn type should be "isp") + if classified: + try: + result.provider = utils.trim(provider) + country_code = utils.trim(classified.country_code).upper() + if country_code: + result.country = ISO_TO_CHINESE.get(country_code, "") + + if not result.country: + raw = classified.raw or {} + result.country = utils.trim( + raw.get("country_zh", "") or raw.get("country", "") or raw.get("country_name", "") + ) + + company_type = utils.trim(classified.company_type).lower() + asn_type = utils.trim(classified.asn_type).lower() if company_type == "isp" and asn_type == "isp": result.ip_type = "isp" elif company_type in ["business", "isp"] and asn_type in ["business", "isp"]: result.ip_type = "business" + result.score = classified.score except Exception as e: - logger.error(f"Error parsing {url} response for proxy {name}: {str(e)}") + logger.error(f"Error parsing response for proxy {name}: {str(e)}") else: - logger.warning(f"Failed to query {url} for proxy {name}") + logger.warning(f"Failed to query residential info for proxy {name} with providers: {providers}") + + if is_cdn_label(result.country): + result.is_cdn = True + _mark_cdn(proxy) + logger.debug(f"Residential country for proxy {name} is CDN, continue locating") + return ProxyQueryResult(proxy=proxy, result=result, success=False) - # Determine if query was successful flag = result.country != "" or result.ip_type != "" return ProxyQueryResult(proxy=proxy, result=result, success=flag) @@ -920,88 +1016,65 @@ def _get_ipinfo_url(port: int, name: str) -> str: def locate_by_ipinfo(proxy: dict, port: int, reader: database.Reader = None) -> ProxyQueryResult: """Check the location of a single proxy by making a request through it""" + name = proxy.get("name", "") + + is_cdn = _is_cdn_proxy(proxy) - def _create_failed_result(reason: str = "") -> ProxyQueryResult: - """Helper to create failed query result""" - name = proxy.get("name", "") + def _failed(reason: str = "") -> ProxyQueryResult: if reason: logger.warning(f"Location query failed for proxy {name}: {reason}") - return ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=name), success=False) + if is_cdn: + _mark_cdn(proxy) + return ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=name, is_cdn=is_cdn), success=False) - def _create_success_result(country: str) -> ProxyQueryResult: - """Helper to create successful query result""" - info = ProxyInfo(name=proxy.get("name", "")) - info.country = country + def _success(country: str) -> ProxyQueryResult: + info = ProxyInfo(name=name, country=country) return ProxyQueryResult(proxy=proxy, result=info, success=True) - def _try_local_mmdb_lookup() -> str: - """Attempt to get country from local MMDB database""" - if not reader: - return "" + if not port: + return _failed("No port specified") - ip = get_ipv4(port=port, max_retries=2) - if ip: - return query_ip_country(ip, reader) or "" - return "" + random_delay() - def _try_online_api_services() -> str: - """Attempt to get country from online API services with retry logic""" - retries = 3 + try: + if reader: + ip = get_ipv4(port=port, max_retries=2) + geo = lookup_ip_geo(ip, reader) if ip else GeoInfo() + if geo.is_cdn: + is_cdn = True + _mark_cdn(proxy) + logger.debug(f"Egress IP for proxy {name} belongs to CDN, try online APIs") + elif geo.country: + logger.debug(f"Location found via MMDB for proxy {name}: {geo.country}") + return _success(geo.country) + retries = 3 for attempt in range(retries): - # Select a random service for this attempt service = random.choice(LOCATION_API_SERVICES) - - # Make the API request - success, data = make_proxy_request(port=port, url=service["url"], max_retries=1, timeout=12) - + success, data = make_proxy_request(port=port, url=service["url"], max_retries=1, timeout=12, quiet=True) if success and data: - # Parse country code from response - key = service["country_key"] - code = data.get(key, "") - + code = data.get(service["country_key"], "") if code: - # Convert to Chinese country name - return ISO_TO_CHINESE.get(code, code) + country = ISO_TO_CHINESE.get(code, code) + if is_cdn_label(code) or is_cdn_label(country): + is_cdn = True + _mark_cdn(proxy) + logger.debug(f"API country for proxy {name} is CDN, continue locating") + else: + logger.debug(f"Location found via API for proxy {name}: {country}") + return _success(country) - # Handle retry delay for failed attempts if attempt < retries - 1: delay = min(2**attempt * random.uniform(1, 2), 6) - logger.warning( - f"API attempt {attempt+1} failed for proxy {proxy.get('name', '')} " - f"using {service['url']}, retrying in {delay:.2f}s" + logger.debug( + f"API attempt {attempt+1} failed for proxy {name} using {service['url']}, retrying in {delay:.2f}s" ) time.sleep(delay) - return "" - - # Validate input parameters - if not port: - return _create_failed_result("No port specified") - - # Apply rate limiting - random_delay() - - # Main location detection logic - try: - # Strategy 1: Try local MMDB database first (faster and more reliable) - country = _try_local_mmdb_lookup() - if country: - logger.debug(f"Location found via MMDB for proxy {proxy.get('name', '')}: {country}") - return _create_success_result(country) - - # Strategy 2: Fall back to online API services - country = _try_online_api_services() - if country: - logger.debug(f"Location found via API for proxy {proxy.get('name', '')}: {country}") - return _create_success_result(country) - - # No location detected from any source - return _create_failed_result("Unable to determine location from any source") - + return _failed("Unable to determine location from any source") except Exception as e: - logger.error(f"Unexpected error during location query for {proxy.get('name', '')}: {str(e)}") - return _create_failed_result(f"Exception: {str(e)}") + logger.error(f"Unexpected error during location query for {name}: {str(e)}") + return _failed(f"Exception: {str(e)}") def batch_query( @@ -1011,8 +1084,6 @@ def batch_query( show_progress: bool = True, description: str = "Querying", digits: int = 2, - reader: database.Reader = None, - api_key: str = "", ) -> list[ProxyQueryResult]: """ Run mihomo to query proxies information using the specified function @@ -1024,84 +1095,87 @@ def batch_query( show_progress: Whether to show progress description: Description for progress display digits: Number of digits for proxy naming - reader: Optional mmdb reader for locate_by_ipinfo - api_key: Optional API key for check_residential function Returns: List of ProxyQueryResult with complete information """ - if not proxies or not is_mihomo(): + if not proxies: return [] - # Rename proxies for consistent naming - nodes = rename(proxies, digits, False) + if not is_mihomo(): + return [ + ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=proxy.get("name", "")), success=False) + for proxy in proxies + ] - logger.info(f"Generate clash listeners configuration for {len(nodes)} proxies") - # Generate mihomo configuration - config, records = generate_mihomo_config(nodes) + nodes = rename(proxies, digits, False) + failed = [ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=proxy["name"]), success=False) for proxy in nodes] - # Save the configuration to clash/config.yaml in the project directory workspace = os.path.join(os.path.abspath(os.path.dirname(os.path.dirname(__file__))), "clash") - config_path = os.path.join(workspace, "config.yaml") - with open(config_path, "w", encoding="utf-8") as f: - yaml.dump(config, f, allow_unicode=True) - - logger.info(f"Mihomo configuration saved to {config_path}") - - # Check if we can find the mihomo binary mihomo_bin = os.path.join(workspace, which_bin()[0]) if not os.path.exists(mihomo_bin) or not os.path.isfile(mihomo_bin): logger.error("Mihomo binary not found, skipping proxy check") - return [ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=proxy["name"]), success=False) for proxy in nodes] + return failed - # Make the binary executable utils.chmod(mihomo_bin) - # Start mihomo with the configuration - logger.info(f"Starting mihomo with configuration {config_path}") + logger.info(f"Generate clash listeners configuration for {len(nodes)} proxies") + reservation = PortReservation() process = None try: - # Run mihomo in background + ports = reservation.reserve(1 + len(nodes)) + mixed_port, listener_ports = ports[0], ports[1:] + config, records = generate_mihomo_config(nodes, listener_ports, mixed_port) + + config_path = os.path.join(workspace, "config.yaml") + with open(config_path, "w", encoding="utf-8") as f: + yaml.dump(config, f, allow_unicode=True) + + logger.info(f"Mihomo configuration saved to {config_path}") + reservation.close() + + logger.info(f"Starting mihomo with configuration {config_path}") process = subprocess.Popen( [mihomo_bin, "-d", workspace, "-f", config_path], stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, + stderr=subprocess.DEVNULL, ) - # Wait longer to ensure mihomo is fully started logger.info("Waiting for mihomo to start...") - time.sleep(8) + _wait_listener(mixed_port, process) + if listener_ports: + _wait_listener(listener_ports[0], process, timeout=10) - # Create proxy info mapping for task generation mappings = {proxy["name"]: proxy for proxy in nodes} - - # Generate tasks for each proxy - if reader is not None: - # For locate_by_ipinfo which needs reader parameter - tasks = [(mappings[name], port, reader) for name, port in records.items() if name in mappings] - elif api_key: - # For check_residential with API key - tasks = [(mappings[name], port, api_key) for name, port in records.items() if name in mappings] - else: - # For check_residential without API key - tasks = [(mappings[name], port) for name, port in records.items() if name in mappings] - - # Check proxies using the specified function - results = utils.multi_thread_run( - func=func, - tasks=tasks, - num_threads=num_threads, - show_progress=show_progress, - description=description, + tasks = [(mappings[name], port) for name, port in records.items() if name in mappings] + results = ( + utils.multi_thread_run( + func=func, + tasks=tasks, + num_threads=num_threads, + show_progress=show_progress, + description=description, + ) + or [] ) - return results + queried, normalized = set(), [] + for item in results: + if not item: + continue + normalized.append(item) + queried.add(item.proxy.get("name")) + + for proxy in nodes: + if proxy["name"] not in queried: + normalized.append(ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=proxy["name"]), success=False)) + return normalized except Exception as e: logger.error(f"Error during mihomo check: {str(e)}") - return [ProxyQueryResult(proxy=proxy, result=ProxyInfo(name=proxy["name"]), success=False) for proxy in nodes] + return failed finally: - # Always try to kill the mihomo process + reservation.close() if process: try: process.terminate() @@ -1110,13 +1184,16 @@ def batch_query( pass -def process_query_results(results: list[ProxyQueryResult], strategy: str) -> tuple[list[dict], list[dict]]: +def process_query_results( + results: list[ProxyQueryResult], strategy: str, score: bool = False +) -> tuple[list[dict], list[dict]]: """ Process proxy query results Args: results: List of query results strategy: Processing strategy ('residential' or 'location') + score: Whether to prefix node names with provider and trust score Returns: tuple: (list of successful proxies, list of failed proxies) @@ -1124,28 +1201,36 @@ def process_query_results(results: list[ProxyQueryResult], strategy: str) -> tup successes, fails = [], [] for item in results: - if item.success and item.result.country: + if not item: + continue + + country = utils.trim(item.result.country) if item.result else "" + is_cdn = bool(item.result and item.result.is_cdn) or is_cdn_label(country) + if is_cdn: + _mark_cdn(item.proxy) + + if item.success and country and not is_cdn: # Copy proxy info to avoid modifying original data proxy = item.proxy.copy() if strategy == "residential": # Residential IP check strategy - name = item.result.country + name = country if item.result.ip_type == "isp": name += "家宽" elif item.result.ip_type == "business": name += "商宽" - - proxy["name"] = name - successes.append(proxy) - elif strategy == "location": - # Location check strategy - proxy["name"] = item.result.country - successes.append(proxy) else: - # Unknown strategy, use query result directly - proxy["name"] = item.result.country - successes.append(proxy) + # Location check or unknown strategy + name = country + + if score and item.result.score is not None: + source = utils.trim(item.result.provider).upper() + if source: + name = f"[{source}|{str(item.result.score).zfill(3)}] {name}" + + proxy["name"] = name + successes.append(proxy) else: # Failed query proxies fails.append(item.proxy) @@ -1161,36 +1246,46 @@ def regularize( show_progress: bool = True, locate: bool = False, residential: bool = False, + ip_library: str = "", digits: int = 2, + score: bool = False, ) -> list[dict]: if not proxies or not isinstance(proxies, list): return proxies # Phase 1: Residential check if necessary successes, fails = [], [] + reader = None if residential: - logger.info(f"Starting residential check for {len(proxies)} proxies") - - # Enable locate if residential check is enabled locate = True + if residential or locate: + directory = utils.trim(directory) + if not directory: + directory = os.path.join(os.path.abspath(os.path.dirname(os.path.dirname(__file__))), "data") + reader = load_mmdb(directory=directory, repo="Loyalsoldier/geoip", filename="Country.mmdb", update=update) + if not reader: + logger.error("cannot load mmdb: Country.mmdb") + + if residential: + logger.info(f"Starting residential check for {len(proxies)} proxies") + # Get https://api.ipapi.is API key from environment variable api_key = utils.trim(os.environ.get("IPAPI_IS_API_KEY", "")) # Use mihomo to check for residential proxies results = batch_query( proxies=proxies, - func=check_residential, + func=partial(check_residential, api_key=api_key, ip_library=ip_library, reader=reader), num_threads=num_threads, show_progress=show_progress, description="Checking residential", digits=digits, - api_key=api_key, ) # Process residential check results - successes, fails = process_query_results(results, "residential") + successes, fails = process_query_results(results, "residential", score=score) logger.info(f"Residential check completed: {len(successes)} successful, {len(fails)} failed") else: fails = proxies @@ -1199,60 +1294,50 @@ def regularize( if locate and fails: logger.info(f"Starting location check for {len(fails)} proxies") - # Initialize reader for locate functionality and load mmdb database if available - directory = utils.trim(directory) - if not directory: - directory = os.path.join(os.path.abspath(os.path.dirname(os.path.dirname(__file__))), "data") - - repo, filename = "Loyalsoldier/geoip", "Country.mmdb" - reader = load_mmdb(directory=directory, repo=repo, filename=filename, update=update) - if not reader: - logger.error(f"Skipping location check due to cannot load mmdb: {filename}") - unconfirmed = list() if reader: # Try local mmdb lookup first - tasks = [[p, reader] for p in fails if p and isinstance(p, dict)] + sources = [p for p in fails if p and isinstance(p, dict)] + tasks = [[p, reader] for p in sources] mmdb_results = utils.multi_thread_run(locate_by_geoip, tasks, num_threads, show_progress, "") - # Separate confirmed and unconfirmed proxies by regex - regex = f"中国|{CDN_PATTERN}" - - for proxy in mmdb_results: - if proxy.pop("renamed", False) and not re.search(regex, proxy["name"], flags=re.I): - # Add to successes list if confirmed by mmdb lookup - successes.append(proxy) + for source, proxy in zip(sources, mmdb_results or []): + node = proxy if proxy and isinstance(proxy, dict) else source + name = str(node.get("name", "")) + cdn = bool(node.get("cdn")) or is_cdn_label(name) + if cdn: + _mark_cdn(node) + if node.pop("renamed", False) and "中国" not in name and not cdn: + successes.append(node) else: - # Add to unconfirmed list if not confirmed by mmdb lookup - unconfirmed.append(proxy) + unconfirmed.append(node) else: # No mmdb available, treat all as unconfirmed unconfirmed = fails - # For unconfirmed proxies, use online API services to get location info (fallback) + # For unconfirmed proxies, use online API services to get location info if unconfirmed: logger.info(f"Using online API services for {len(unconfirmed)} unconfirmed proxies") # Use mihomo to check IP locations query_results = batch_query( proxies=unconfirmed, - func=locate_by_ipinfo, + func=partial(locate_by_ipinfo, reader=reader), num_threads=num_threads, show_progress=show_progress, description="Querying location", digits=digits, - reader=reader, ) # Process location check results and handle CDN proxies - query_successes, query_fails = process_query_results(query_results, "location") + query_successes, query_fails = process_query_results(query_results, "location", score=score) # Add query successes to final results successes.extend(query_successes) - # Handle CDN proxies that failed location check + # CDN nodes without a real country fall back to US for proxy in query_fails: - if re.search(CDN_PATTERN, proxy["name"], flags=re.I): + if _is_cdn_proxy(proxy): logger.warning(f"Failed to get location for proxy {proxy['name']}, assume it's in US") proxy["name"] = "美国" @@ -1264,7 +1349,7 @@ def regularize( successes.extend(fails) # Return final results - return rename(proxies=successes, digits=digits, shuffle=True) + return rename(proxies=_remove_temp_flags(successes), digits=digits, shuffle=True) def rename(proxies: list[dict], digits: int = 2, shuffle: bool = False) -> list[dict]: diff --git a/subscribe/process.py b/subscribe/process.py index cc98014986..b767ae1e2a 100644 --- a/subscribe/process.py +++ b/subscribe/process.py @@ -62,6 +62,7 @@ def load_configs( only_check: bool = False, num_threads: int = 0, display: bool = True, + retry: int = 3, ) -> ProcessConfig: def parse_config(config: dict) -> None: tasks.extend(config.get("domains", [])) @@ -273,7 +274,7 @@ def verify(storage: dict, groups: dict) -> bool: url, ): headers = {"User-Agent": utils.USER_AGENT, "Referer": url} - content = utils.http_get(url=url, headers=headers) + content = utils.http_get(url=url, headers=headers, retry=max(retry, 1), timeout=120) if not content: logger.error(f"cannot fetch config from remote, url: {utils.hide(url=url)}") else: @@ -509,6 +510,7 @@ def aggregate(args: argparse.Namespace) -> None: clash_bin, subconverter_bin = executable.which_bin() display = not args.invisible + retry = min(max(1, args.retry), 10) # parse config server = utils.trim(args.server) or os.environ.get("SUBSCRIBE_CONF", "").strip() @@ -517,11 +519,11 @@ def aggregate(args: argparse.Namespace) -> None: only_check=args.check, num_threads=args.num, display=display, + retry=retry, ) storages = process_config.storage or {} pushtool = push.get_instance(config=push.PushConfig.from_dict(storages)) - retry = min(max(1, args.retry), 10) # generate tasks tasks, groups, sites = assign( @@ -572,7 +574,7 @@ def aggregate(args: argparse.Namespace) -> None: filename = "config.yaml" proxies = clash.generate_config(workspace, proxies, filename) - # filer + # filter skip = utils.trim(os.environ.get("SKIP_ALIVE_CHECK", "false")).lower() in ["true", "1"] nochecks, starttime = proxies, time.time() @@ -637,6 +639,8 @@ def aggregate(args: argparse.Namespace) -> None: if regularize and isinstance(regularize, dict) and regularize.get("enable", False): locate = regularize.get("locate", False) residential = regularize.get("residential", False) + ip_library = regularize.get("library", "") + score = regularize.get("score", False) try: bits = max(1, int(regularize.get("bits", 2))) except: @@ -648,7 +652,9 @@ def aggregate(args: argparse.Namespace) -> None: show_progress=display, locate=locate, residential=residential, + ip_library=ip_library, digits=bits, + score=score, ) source_file, data = "config.yaml", {"proxies": nochecks} diff --git a/subscribe/scripts/fofa.py b/subscribe/scripts/fofa.py index 30e8ef3e94..d1bbdd4695 100644 --- a/subscribe/scripts/fofa.py +++ b/subscribe/scripts/fofa.py @@ -67,7 +67,7 @@ def extract_one(url: str) -> list[str]: regex = r"(?:https?://)?(?:[a-zA-Z0-9\u4e00-\u9fa5\-]+\.)+[a-zA-Z0-9\u4e00-\u9fa5\-]+(?:(?:(?:/index.php)?/api/v1/client/subscribe\?token=[a-zA-Z0-9]{16,32})|(?:/link/[a-zA-Z0-9]+\?(?:sub|mu|clash)=\d)|(?:/(?:s|sub)/[a-zA-Z0-9]{32}))" - headers = {"User-Agent": "Clash.Meta; Mihomo"} + headers = {"User-Agent": f"{utils.USER_AGENT}; Clash.Meta; Mihomo; Shadowrocket;"} subscriptions, content = [], "" count, retry = 0, 2 diff --git a/subscribe/subconverter.py b/subscribe/subconverter.py index fa4b552ad0..d7fa3bc07a 100644 --- a/subscribe/subconverter.py +++ b/subscribe/subconverter.py @@ -103,7 +103,7 @@ def generate_conf( lines.extend(["emoji=false", "add_emoji=false"]) if ignore_exclude: - lines.append("exclude=流量|过期|剩余|时间|Expire|Traffic") + lines.append("exclude=[到过]期|Expire|Traffic|剩余流量|时间|官网|产品|联系") lines.append("\n") content = "\n".join(lines) diff --git a/subscribe/utils.py b/subscribe/utils.py index 957731a715..75f8962acb 100644 --- a/subscribe/utils.py +++ b/subscribe/utils.py @@ -35,7 +35,7 @@ CTX.verify_mode = ssl.CERT_NONE USER_AGENT = ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36" + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36" ) diff --git a/tools/clean.py b/tools/clean.py index dc7301cdb0..21f5c481ca 100644 --- a/tools/clean.py +++ b/tools/clean.py @@ -223,7 +223,7 @@ def main(args: argparse.Namespace) -> None: elif country: name = country else: - print("cannot get geolocation and rename because IP address is faked") + print("cannot get geolocation and name because IP address is faked") item["name"] = name except Exception: diff --git a/tools/proxy-gen.py b/tools/proxy-gen.py new file mode 100644 index 0000000000..26a275d610 --- /dev/null +++ b/tools/proxy-gen.py @@ -0,0 +1,335 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""Generate dynamic HTTP/SOCKS proxy nodes from CLI options""" + +from __future__ import annotations + +import argparse +import ipaddress +import json +import os +import re +import secrets +import string +import sys +from dataclasses import dataclass + +SCHEMES = ("socks5", "socks5h", "http", "https") +FORMATS = ("uri", "clash") +NAMING_MODES = ("sid", "index", "none") +SID_CHARS = string.ascii_letters + string.digits +KEYWORD_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$") +DOMAIN_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") +POSITIVE_INT = re.compile(r"^[1-9]\d*$") +ILLEGAL_CHARS = set(":@#/?%") +ISO_CODES = frozenset(""" + AD AE AF AG AI AL AM AO AQ AR AS AT AU AW AX AZ + BA BB BD BE BF BG BH BI BJ BL BM BN BO BQ BR BS BT BV BW BY BZ + CA CC CD CF CG CH CI CK CL CM CN CO CR CU CV CW CX CY CZ + DE DJ DK DM DO DZ + EC EE EG EH ER ES ET + FI FJ FK FM FO FR + GA GB GD GE GF GG GH GI GL GM GN GP GQ GR GS GT GU GW GY + HK HM HN HR HT HU + ID IE IL IM IN IO IQ IR IS IT + JE JM JO JP + KE KG KH KI KM KN KP KR KW KY KZ + LA LB LC LI LK LR LS LT LU LV LY + MA MC MD ME MF MG MH MK ML MM MN MO MP MQ MR MS MT MU MV MW MX MY MZ + NA NC NE NF NG NI NL NO NP NR NU NZ + OM + PA PE PF PG PH PK PL PM PN PR PS PT PW PY + QA + RE RO RS RU RW + SA SB SC SD SE SG SH SI SJ SK SL SM SN SO SR SS ST SV SX SY SZ + TC TD TF TG TH TJ TK TL TM TN TO TR TT TV TW TZ + UA UG UM US UY UZ + VA VC VE VG VI VN VU + WF WS + XK + YE YT + ZA ZM ZW + """.split()) + + +class ValidationError(ValueError): + pass + + +@dataclass +class Config: + scheme: str + username: str + password: str + host: str + port: int + count: int + region_key: str + country: str + state_key: str + state: str + session_key: str + sid_length: int + duration_key: str + duration: int + naming_mode: str + output_format: str + output_path: str + + +def trim(value: object | None) -> str: + return "" if value is None else str(value).strip() + + +def require_text(value: object | None, name: str) -> str: + text = trim(value) + if not text: + raise ValidationError(f"{name} is required") + if any(ch.isspace() for ch in text): + raise ValidationError(f"{name} must not contain whitespace") + if any(ch in ILLEGAL_CHARS for ch in text): + raise ValidationError(f"{name} contains illegal character") + return text + + +def require_keyword(value: object | None, name: str, required: bool = True) -> str: + text = trim(value) + if not text: + if required: + raise ValidationError(f"{name} is required") + return "" + if any(ch.isspace() for ch in text): + raise ValidationError(f"{name} must not contain whitespace") + if not KEYWORD_PATTERN.fullmatch(text): + raise ValidationError(f"{name} may contain only letters, digits, '-' and '_'") + return text + + +def require_integer(value: object | None, name: str, maximum: int | None = None) -> int: + text = trim(value) + if not POSITIVE_INT.fullmatch(text): + raise ValidationError(f"{name} must be an integer > 0") + number = int(text) + if maximum is not None and number > maximum: + raise ValidationError(f"{name} must be <= {maximum}") + return number + + +def require_host(value: object | None) -> str: + host = require_text(value, "host") + try: + ipaddress.IPv4Address(host) + return host + except ValueError: + pass + hostname = host[:-1] if host.endswith(".") else host + if not hostname or len(hostname) > 253: + raise ValidationError("host must be a domain or IPv4 address") + labels = hostname.split(".") + if not all(DOMAIN_LABEL.fullmatch(label) for label in labels): + raise ValidationError("host must be a domain or IPv4 address") + return host + + +def require_country(value: object | None) -> str: + text = trim(value) + if not text: + return "" + country = text.upper() + if country not in ISO_CODES: + raise ValidationError("country must be an ISO 3166-1 alpha-2 code") + return country + + +def require_choice(value: object | None, name: str, options: tuple[str, ...], default: str) -> str: + text = trim(value).lower() + if not text: + return default + if text not in options: + raise ValidationError(f"{name} must be one of {', '.join(options)}") + return text + + +def generate_sids(count: int, length: int) -> list[str]: + capacity = len(SID_CHARS) ** length + if count > capacity: + raise ValidationError(f"count {count} exceeds unique sid space {capacity} for sid-length {length}") + randomizer = secrets.SystemRandom() + if length == 1: + return randomizer.sample(list(SID_CHARS), count) + seen: set[str] = set() + sids: list[str] = [] + while len(sids) < count: + sid = "".join(randomizer.choice(SID_CHARS) for _ in range(length)) + if sid in seen: + continue + seen.add(sid) + sids.append(sid) + return sids + + +def build_username(config: Config, sid: str) -> str: + parts = [config.username] + if config.country: + parts.extend((config.region_key, config.country)) + if config.state: + parts.extend((config.state_key, config.state)) + parts.extend((config.session_key, sid, config.duration_key, str(config.duration))) + return "-".join(parts) + + +def build_node_name(config: Config, sid: str, index: int, index_width: int) -> str: + if config.naming_mode == "none": + return "" + suffix = sid if config.naming_mode == "sid" else f"{index:0{index_width}d}" + return f"{config.country}-{suffix}" if config.country else suffix + + +def format_uri(config: Config, username: str, node_name: str) -> str: + uri = f"{config.scheme}://{username}:{config.password}@{config.host}:{config.port}" + return f"{uri}#{node_name}" if node_name else uri + + +def format_clash(config: Config, username: str, node_name: str) -> str: + proxy_type = "http" if config.scheme in ("http", "https") else "socks5" + proxy = { + "name": node_name, + "type": proxy_type, + "server": config.host, + "port": config.port, + "username": username, + "password": config.password, + } + if config.scheme in ("socks5", "socks5h"): + proxy["udp"] = True + proxy["tls"] = True + proxy["skip-cert-verify"] = False + elif config.scheme == "https": + proxy["tls"] = True + proxy["skip-cert-verify"] = False + return json.dumps(proxy, ensure_ascii=False, separators=(",", ":")) + + +def parse_config(args: argparse.Namespace) -> Config: + scheme = require_choice(args.scheme, "scheme", SCHEMES, "socks5") + username = require_text(args.username, "username") + password = require_text(args.password, "password") + host = require_host(args.host) + port = require_integer(args.port, "port", maximum=65535) + count = require_integer(args.count, "count") + region_key = require_keyword(args.region_key, "region-key") + country = require_country(args.country) + state_key = require_keyword(args.state_key, "state-key", required=False) + state = require_keyword(args.state, "state", required=False) + session_key = require_keyword(args.session_key, "session-key") + sid_length = require_integer(args.sid_length, "sid-length", maximum=64) + duration_key = require_keyword(args.duration_key, "duration-key") + duration = require_integer(args.duration, "duration") + naming_mode = require_choice(args.naming, "naming", NAMING_MODES, "sid") + output_format = require_choice(args.format, "format", FORMATS, "uri") + output_path = trim(args.output) + + if bool(state_key) != bool(state): + raise ValidationError("state-key and state must be provided together") + if (state_key or state) and not country: + raise ValidationError("state requires country") + if naming_mode == "none" and output_format != "uri": + raise ValidationError("naming=none is only supported when format=uri") + + return Config( + scheme=scheme, + username=username, + password=password, + host=host, + port=port, + count=count, + region_key=region_key, + country=country, + state_key=state_key, + state=state, + session_key=session_key, + sid_length=sid_length, + duration_key=duration_key, + duration=duration, + naming_mode=naming_mode, + output_format=output_format, + output_path=output_path, + ) + + +def generate_nodes(config: Config) -> list[str]: + sids = generate_sids(config.count, config.sid_length) + index_width = len(str(config.count)) + nodes: list[str] = [] + for index, sid in enumerate(sids, start=1): + username = build_username(config, sid) + node_name = build_node_name(config, sid, index, index_width) + if config.output_format == "clash": + nodes.append(f" - {format_clash(config, username, node_name)}") + else: + nodes.append(format_uri(config, username, node_name)) + if config.output_format == "clash": + return ["proxies:", *nodes] + return nodes + + +def write_output(nodes: list[str], output_path: str) -> None: + content = "\n".join(nodes) + if content: + content += "\n" + if not output_path: + sys.stdout.write(content) + return + path = os.path.abspath(output_path) + directory = os.path.dirname(path) + if directory and not os.path.isdir(directory): + raise ValidationError(f"output directory not found: {directory}") + with open(path, "w", encoding="utf-8", newline="\n") as output_file: + output_file.write(content) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Generate dynamic HTTP/SOCKS proxy nodes", + formatter_class=argparse.RawTextHelpFormatter, + epilog=( + "example:\n" + " python proxies-gen.py --username alice --password secret --host gate.example.com --port 1080 --count 10 " + "--session-key sid --sid-length 8 --duration-key t --country US --state-key st --state california" + ), + allow_abbrev=False, + ) + parser.add_argument("--scheme", default="socks5", help="socks5, socks5h, http or https") + parser.add_argument("--username", required=True, help="base username") + parser.add_argument("--password", required=True, help="password") + parser.add_argument("--host", required=True, help="domain or IPv4 address") + parser.add_argument("--port", required=True, help="port, 1-65535") + parser.add_argument("--count", required=True, help="number of nodes") + parser.add_argument("--region-key", default="region", help="region keyword, default: region") + parser.add_argument("--country", default="", help="ISO 3166-1 alpha-2 country code") + parser.add_argument("--state-key", default="", help="state or city keyword, e.g. st") + parser.add_argument("--state", default="", help="state or city name") + parser.add_argument("--session-key", required=True, help="session keyword, e.g. sid") + parser.add_argument("--sid-length", required=True, help="random sid length") + parser.add_argument("--duration-key", required=True, help="session duration keyword, e.g. t") + parser.add_argument("--duration", default="5", help="session duration, default: 5") + parser.add_argument("--naming", default="sid", help="node name style: sid, index or none, none only for uri") + parser.add_argument("--format", default="uri", help="uri or clash") + parser.add_argument("--output", default="", help="output file, default: stdout") + return parser + + +def main() -> None: + parser = build_parser() + try: + config = parse_config(parser.parse_args()) + write_output(generate_nodes(config), config.output_path) + except ValidationError as error: + parser.error(str(error)) + except OSError as error: + parser.error(str(error)) + + +if __name__ == "__main__": + main() diff --git a/tools/socks-checker.py b/tools/socks-checker.py index c7bbb26388..7f8e4a8b37 100644 --- a/tools/socks-checker.py +++ b/tools/socks-checker.py @@ -11,13 +11,15 @@ import argparse import asyncio +import html import ipaddress +import json import re import sys from dataclasses import dataclass from datetime import datetime -from typing import Dict, List, Optional, Tuple -from urllib.parse import urlparse +from typing import Any, Callable, Dict, List, Optional, Tuple +from urllib.parse import quote, urlparse import aiohttp import yaml @@ -276,6 +278,97 @@ } +def country_flag_emoji(country_code: str) -> str: + if not country_code or len(country_code) != 2: + return "" + + code = country_code.upper() + if not code.isalpha(): + return "" + + return chr(0x1F1E6 + ord(code[0]) - ord("A")) + chr(0x1F1E6 + ord(code[1]) - ord("A")) + + +def country_name_zh(country_code: str) -> str: + if not country_code: + return "" + + return COUNTRY_NAME_ZH.get(country_code.upper(), "") + + +CHINA_PROVINCE_SUFFIXES = ( + "特别行政区", + "维吾尔自治区", + "壮族自治区", + "回族自治区", + "自治区", + "省", + "市", +) +CHINA_PROVINCE_ALIASES = { + "anhui": "安徽", + "beijing": "北京", + "chongqing": "重庆", + "fujian": "福建", + "gansu": "甘肃", + "guangdong": "广东", + "guangxi": "广西", + "guangxizhuang": "广西", + "guizhou": "贵州", + "hainan": "海南", + "hebei": "河北", + "heilongjiang": "黑龙江", + "henan": "河南", + "hubei": "湖北", + "hunan": "湖南", + "innermongolia": "内蒙古", + "jiangsu": "江苏", + "jiangxi": "江西", + "jilin": "吉林", + "liaoning": "辽宁", + "neimenggu": "内蒙古", + "neimongol": "内蒙古", + "ningxia": "宁夏", + "ningxiahuizu": "宁夏", + "qinghai": "青海", + "shaanxi": "陕西", + "shandong": "山东", + "shanghai": "上海", + "shanxi": "山西", + "sichuan": "四川", + "tianjin": "天津", + "tibet": "西藏", + "xinjiang": "新疆", + "xinjianguygur": "新疆", + "xinjianguyghur": "新疆", + "xizang": "西藏", + "yunnan": "云南", + "zhejiang": "浙江", +} +CHINA_MAINLAND_PROVINCES = frozenset(CHINA_PROVINCE_ALIASES.values()) +CHINA_PROVINCE_ALIAS_PREFIXES = tuple( + sorted(CHINA_PROVINCE_ALIASES.items(), key=lambda item: len(item[0]), reverse=True) +) +CHINA_PROVINCE_EN_SUFFIXES = ( + "autonomousregion", + "municipality", + "province", + "region", + "sheng", + "city", +) + +REGION_KEYS = set(["province", "province_name", "region", "region_name", "state", "state_name"]) + + +def short_company_name(value: str) -> str: + if not value: + return "UNKNOWN" + + parts = [part for part in re.split(r"[\s,\.\-_@;:]+", value.strip()) if part] + return parts[0].upper() if parts else "UNKNOWN" + + @dataclass class ProxyInfo: protocol: str @@ -309,6 +402,820 @@ def from_proxy(cls, proxy_info: ProxyInfo) -> "TestResult": ) +@dataclass +class IpLookupResult: + ip: Optional[str] + data: Optional[Dict] + error: Optional[str] = None + + +class IPLibrary: + name: str = "" + + def __init__(self): + self._caches: Dict[str, str] = {} + + async def lookup( + self, session: aiohttp.ClientSession, proxy_info: ProxyInfo, retries: int, timeout: int + ) -> IpLookupResult: + data, error = await self._fetch(session, proxy_info, retries, timeout) + if not data: + host = "" if not proxy_info else proxy_info.host + return IpLookupResult(None, None, error or f"Failed to get IP info from {self.name}, host: {host}") + + return self._verify(data, self.name) + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + raise NotImplementedError + + async def _fetch( + self, session: aiohttp.ClientSession, proxy_info: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + raise NotImplementedError + + @staticmethod + def _build_headers(url: str) -> Dict[str, str]: + result = urlparse(url) + base = f"{result.scheme}://{result.netloc}" if result.scheme and result.netloc else "" + + return { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36", + "Accept": "application/json, text/plain, */*", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Cache-Control": "no-cache", + "Pragma": "no-cache", + "Connection": "close", + "Referer": f"{base}/" if base else url, + "Origin": base if base else url, + } + + async def _make_request( + self, + session: aiohttp.ClientSession, + url: str, + retries: int, + timeout: int, + headers: Optional[Dict[str, str]] = None, + deserialize: bool = True, + parser: Optional[Callable[[str], Any]] = None, + ) -> Tuple[Optional[Any], Optional[str]]: + default_headers = self._build_headers(url) + if headers and isinstance(headers, dict): + default_headers.update({k: v for k, v in headers.items() if k and v is not None}) + + error = None + for attempt in range(1, retries + 1): + try: + async with session.get( + url, + headers=default_headers, + timeout=aiohttp.ClientTimeout(total=timeout), + ) as response: + if response.status == 200: + content = await response.text() + + if parser is not None: + data = parser(content) + if data: + return data, None + + error = "Invalid response payload" + elif deserialize: + try: + data = json.loads(content) + except Exception: + data = None + + if isinstance(data, dict): + return data, None + + error = "Invalid JSON response" + else: + return content, None + + else: + error = f"HTTP {response.status}" + except asyncio.TimeoutError: + error = "Timeout" + except Exception as e: + error = str(e)[:100] + + if attempt < retries: + await asyncio.sleep(attempt) + + return None, error + + @staticmethod + def _verify(data: Dict, source: str) -> IpLookupResult: + address = (data.get("ip") or "").strip() + if not address: + return IpLookupResult(None, None, f"Invalid IP from {source}") + + try: + ipaddress.ip_address(address) + except ValueError: + return IpLookupResult(None, None, f"Invalid IP from {source}, ip: {address}") + + return IpLookupResult(address, data, None) + + @staticmethod + def _format_remark( + country_code: str, + country: str, + label: str, + include_asn_name: bool, + company_name: str, + detail: str = "", + ) -> str: + flag = country_flag_emoji(country_code) + base = f"{flag} {country}{label}".strip() + + if include_asn_name and company_name: + if detail: + return f"{base} [{company_name}::{detail}]".strip() + + return f"{base} [{company_name}]".strip() + + return base + + async def _resolve_country( + self, + session: aiohttp.ClientSession, + ip: str, + country_code: str, + country: str, + retries: int, + timeout: int, + data: Optional[Dict] = None, + ) -> str: + country_code = (country_code or "").upper() + if country_code != "CN" or not ip: + return country + + if ip in self._caches: + return self._caches[ip] + + province = self._extract_province(data) + if not province: + return "中国" + + resolved = f"中国{province}" + self._caches[ip] = resolved + return resolved + + def _extract_province(self, data: Optional[Dict]) -> str: + if not isinstance(data, dict): + return "" + + candidates: List[Any] = [] + for key in REGION_KEYS: + candidates.append(data.get(key)) + + for parent_key in ("location", "geo"): + nested = data.get(parent_key) + if isinstance(nested, dict): + for key in REGION_KEYS: + candidates.append(nested.get(key)) + + for value in candidates: + province = self._normalize_province(value) + if province: + return province + + return "" + + @staticmethod + def _province_alias_key(province: str) -> str: + return re.sub(r"[^a-z0-9]+", "", province.lower()) + + @staticmethod + def _match_province_alias(key: str) -> str: + if not key or key in {"china", "cn", "mainlandchina", "unknown", "na", "null", "none"}: + return "" + + candidates = [key] + for suffix in CHINA_PROVINCE_EN_SUFFIXES: + if key.endswith(suffix): + stripped = key[: -len(suffix)] + if stripped: + candidates.append(stripped) + + for candidate in candidates: + province = CHINA_PROVINCE_ALIASES.get(candidate) + if province: + return province + + for candidate in candidates: + for alias, province in CHINA_PROVINCE_ALIAS_PREFIXES: + if candidate.startswith(alias): + return province + + return "" + + @classmethod + def _normalize_province(cls, province: Any) -> str: + if isinstance(province, dict): + for key in ("name", "name_en", "en", "value"): + value = cls._normalize_province(province.get(key)) + if value: + return value + + return "" + + if not isinstance(province, str): + return "" + + province = province.strip() + if not province: + return "" + + if province.lower() in {"-", "n/a", "na", "none", "null", "unknown"}: + return "" + + for suffix in CHINA_PROVINCE_SUFFIXES: + if province.endswith(suffix): + province = province[: -len(suffix)].strip() + break + + if re.search(r"[\u4e00-\u9fff]", province): + return province if province in CHINA_MAINLAND_PROVINCES else "" + + key = cls._province_alias_key(province) + return cls._match_province_alias(key) + + +class IPInfoLibrary(IPLibrary): + name = "ipinfo" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + country_code = (data.get("country") or "").upper() + flag = country_flag_emoji(country_code) + + asn_info = data.get("asn", {}) or {} + company_info = data.get("company", {}) or {} + asn_type = (asn_info.get("type") or "").lower() + company_type = (company_info.get("type") or "").lower() + + asn_name = (asn_info.get("domain") or "").strip() + if not asn_name or re.match(r"^as\d+\.", asn_name, flags=re.I): + asn_name = (asn_info.get("name") or "").strip() + + company_name = short_company_name(asn_name) + + if asn_type == "isp" and company_type == "isp": + label = "家宽" + elif asn_type == "isp" or company_type == "isp": + label = "商宽" + elif asn_type == "edu" or company_type == "edu": + label = "教育" + else: + label = "" + + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) or country_code or "未知", + retries=retries, + timeout=timeout, + data=data, + ) + base = f"{flag} {country}{label}".strip() + if include_asn_name and company_name: + return f"{base} [{company_name}]".strip() + + return base + + @staticmethod + def _is_ipv4(host: str) -> bool: + if not host: + return False + try: + return isinstance(ipaddress.ip_address(host), ipaddress.IPv4Address) + except ValueError: + return False + + async def _resolve_ip(self, session: aiohttp.ClientSession, host: str, retries: int, timeout: int) -> Optional[str]: + if self._is_ipv4(host): + return host + + url = "https://ipinfo.io/ip" + for attempt in range(1, retries + 1): + try: + async with session.get( + url, + headers=self._build_headers(url), + timeout=aiohttp.ClientTimeout(total=timeout), + ) as response: + if response.status == 200: + text = (await response.text()).strip() + try: + ipaddress.ip_address(text) + return text + except ValueError: + pass + except asyncio.TimeoutError: + pass + except Exception: + pass + + if attempt < retries: + await asyncio.sleep(attempt) + + return None + + async def _fetch( + self, session: aiohttp.ClientSession, proxy_info: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + host = proxy_info.host if proxy_info else "" + address = await self._resolve_ip(session, host, retries, timeout) + if not address: + return None, f"Failed to get IP from ipinfo.io/ip, host: {host}" + + url = f"https://ipinfo.io/widget/demo/{address}" + data, error = await self._make_request(session, url, retries, timeout) + if not data: + return None, error or f"Failed to get IP info from ipinfo.io, ip: {address}" + + return data.get("data", data), None + + +class IPPureLibrary(IPLibrary): + name = "ippure" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + residential = data.get("isResidential") + label = "家宽" if residential is True else "" + + country_code = (data.get("countryCode") or "").upper() + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) or (data.get("country") or "未知"), + retries=retries, + timeout=timeout, + data=data, + ) + + company_name = short_company_name(data.get("asOrganization") or "") + score = str(data.get("fraudScore")).zfill(3) if "fraudScore" in data else "NUL" + + # broadcast or native + categroy = "NUL" if "isBroadcast" not in data else "B" if data.get("isBroadcast") else "N" + + return self._format_remark( + country_code=country_code, + country=country, + label=label, + include_asn_name=include_asn_name, + company_name=company_name, + detail=f"{score}::{categroy}", + ) + + async def _fetch( + self, session: aiohttp.ClientSession, _: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + url = "https://my.ippure.com/v1/info" + return await self._make_request(session, url, retries, timeout) + + +class IP2LocationLibrary(IPLibrary): + name = "ip2location" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + as_info = data.get("as_info") or {} + + usage_type = (data.get("usage_type") or "").strip().lower() + as_usage_type = ((as_info.get("as_usage_type") if isinstance(as_info, dict) else "") or "").strip().lower() + + check = lambda usage: usage.startswith("isp") or usage == "mob" + label = "家宽" if check(usage_type) and check(as_usage_type) else "" + + country_code = (data.get("country_code") or "").upper() + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) + or data.get("country_name") + or data.get("country", {}).get("name", "") + or "未知", + retries=retries, + timeout=timeout, + data=data, + ) + + provider = (data.get("as", "") or data.get("isp", "") or "").strip() + if not provider and as_info and isinstance(as_info, dict): + provider = (as_info.get("as_name", "") or as_info.get("as_domain", "")).strip() + if not provider: + provider = data.get("domain", "").strip() or "" + + company_name = short_company_name(provider) + score = str(data.get("fraud_score")).zfill(3) if "fraud_score" in data else "NUL" + + return self._format_remark( + country_code=country_code, + country=country, + label=label, + include_asn_name=include_asn_name, + company_name=company_name, + detail=score, + ) + + @staticmethod + def _extract_data(content: str) -> Dict: + if not content or not isinstance(content, str): + return {} + + pattern = r']*class=["\'][^"\']*\blanguage-json\b[^"\']*["\'][^>]*>(.*?)\s*' + groups = re.findall(pattern, content, flags=re.I | re.S) + if not groups: + return {} + + for group in groups: + payload = group.strip() + if not payload: + continue + + payload = re.sub(r"<[^>]+>", "", payload, flags=re.I | re.S) + payload = html.unescape(payload) + + try: + data = json.loads(payload) + if isinstance(data, dict): + return data + except Exception: + continue + + return {} + + async def _fetch( + self, session: aiohttp.ClientSession, _: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + url = "https://www.ip2location.com/demo" + headers = {"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"} + data, error = await self._make_request( + session=session, + url=url, + retries=retries, + timeout=timeout, + headers=headers, + deserialize=False, + parser=self._extract_data, + ) + if not data: + return None, "Invalid HTML response" if error == "Invalid response payload" else error + + return data, None + + +class IPLarkLibrary(IPLibrary): + name = "iplark" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + node_type = (data.get("type") or "").strip().lower() + if node_type == "isp": + label = "家宽" + elif node_type == "business": + label = "商宽" + elif node_type == "education": + label = "教育" + else: + label = "" + + country_code = (data.get("country_code") or "").upper() + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) or (data.get("country_zh") or data.get("country") or "未知"), + retries=retries, + timeout=timeout, + data=data, + ) + + # native if registered country code equals country code else broadcast + categroy = "N" if (data.get("registered_country_code") or "").upper() == country_code else "B" + + asn = str(data.get("asn") or "").strip() + detail = f"{'AS'+asn if asn else 'NUL'}::{categroy}" + + company_name = short_company_name(data.get("organization") or "") + + return self._format_remark( + country_code=country_code, + country=country, + label=label, + include_asn_name=include_asn_name, + company_name=company_name, + detail=detail, + ) + + def _extract_province(self, data: Optional[Dict]) -> str: + return "" + + async def _fetch( + self, session: aiohttp.ClientSession, _: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + url = "https://iplark.com/ipapi/public/ipinfo" + return await self._make_request(session, url, retries, timeout) + + +class IPNetCoffeeLibrary(IPLibrary): + name = "ipnetcoffee" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + label = "家宽" if data.get("isResidential") is True and data.get("company_type", "") != "business" else "" + + country_code = (data.get("countryCode") or "").upper() + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) or (data.get("country") or "未知"), + retries=retries, + timeout=timeout, + data=data, + ) + + company_name = short_company_name( + data.get("asOrganization") or data.get("isp") or data.get("company_name") or "" + ) + score = str(data.get("trust_score")).zfill(3) if "trust_score" in data else "NUL" + + # native if registered country code equals country code else broadcast + category = "N" if (data.get("registered_country_code") or "").upper() == country_code else "B" + + return self._format_remark( + country_code=country_code, + country=country, + label=label, + include_asn_name=include_asn_name, + company_name=company_name, + detail=f"{score}::{category}", + ) + + def _extract_province(self, data: Optional[Dict]) -> str: + province = super()._extract_province(data) + if province: + return province + + if not isinstance(data, dict): + return "" + + for source in data.get("geo_sources") or []: + if not isinstance(source, dict): + continue + + province = self._normalize_province(source.get("region")) + if province: + return province + + return "" + + async def _resolve_ip(self, session: aiohttp.ClientSession, retries: int, timeout: int) -> Optional[str]: + url = "https://ipinfo.io/ip" + text, _ = await self._make_request(session, url, retries, timeout, deserialize=False) + if not isinstance(text, str): + return None + + address = text.strip() + try: + ipaddress.ip_address(address) + return address + except ValueError: + return None + + async def _fetch( + self, session: aiohttp.ClientSession, proxy_info: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + address = await self._resolve_ip(session, retries, timeout) + if not address: + host = "" if not proxy_info else proxy_info.host + return None, f"Failed to get egress IP, host: {host}" + + url = f"https://ip.net.coffee/api/ip/lookup/{quote(address, safe='')}" + data, error = await self._make_request(session, url, retries, timeout) + if not data: + return None, error or f"Failed to get IP info from ip.net.coffee, ip: {address}" + + return data, None + + +class MeowVPSLibrary(IPLibrary): + name = "meowvps" + + async def build_remark( + self, + session: aiohttp.ClientSession, + ip: str, + data: Dict, + include_asn_name: bool, + retries: int, + timeout: int, + ) -> str: + core = data.get("core_data") if isinstance(data.get("core_data"), dict) else {} + minfraud = data.get("minfraud") if isinstance(data.get("minfraud"), dict) else {} + traits = minfraud.get("traits") if isinstance(minfraud.get("traits"), dict) else {} + + country_code = (core.get("country_code") or "").upper() + country = await self._resolve_country( + session=session, + ip=ip, + country_code=country_code, + country=country_name_zh(country_code) or minfraud.get("country") or core.get("country") or "未知", + retries=retries, + timeout=timeout, + data=data, + ) + + company_name = short_company_name(core.get("as_name") or traits.get("isp") or core.get("as_domain") or "") + scores = self._nested(data, "risk_assessment", "ipdata", "scores") + score = str(scores.get("trust_score")).zfill(3) if "trust_score" in scores else "NUL" + + registered = (minfraud.get("registered_country") or "").strip() + current = (minfraud.get("country") or "").strip() + category = "N" if registered and current and registered == current else "B" + + return self._format_remark( + country_code=country_code, + country=country, + label=self._build_label(data), + include_asn_name=include_asn_name, + company_name=company_name, + detail=f"{score}::{category}", + ) + + @staticmethod + def _nested(data: Optional[Dict], *keys: str) -> Dict: + current: Any = data + for key in keys: + if not isinstance(current, dict): + return {} + current = current.get(key) + return current if isinstance(current, dict) else {} + + @classmethod + def _build_label(cls, data: Dict) -> str: + digital = cls._nested(data, "api4", "digital") + traits = cls._nested(data, "minfraud", "traits") + digital_type = "" if digital.get("type") is None else str(digital.get("type")).strip().lower() + user_type = str(traits.get("user_type") or "").strip().lower() + + if digital_type == "edu" or user_type in {"college", "education", "edu"}: + return "教育" + + if cls._is_residential(digital_type, user_type, data): + return "家宽" + + return "" + + @classmethod + def _is_residential(cls, digital_type: str, user_type: str, data: Dict) -> bool: + # Verified against representative IPs: empty api4.digital.type usually means ISP/residential, but 114.114.114.114 also has empty type while user_type/hosting/datacenter say DC + if digital_type in {"hosting", "edu"}: + return False + if user_type in {"hosting", "content_delivery_network", "college"}: + return False + if user_type in {"residential", "traveler", "cellular"}: + return True + if digital_type: + return False + + ipapi = cls._nested(data, "risk_assessment", "ipapi") + threat = cls._nested(data, "risk_assessment", "ipdata", "threat") + if ipapi.get("hosting") is True or threat.get("is_datacenter") is True: + return False + + return True + + def _extract_province(self, data: Optional[Dict]) -> str: + province = super()._extract_province(data) + if province: + return province + + minfraud = data.get("minfraud") if isinstance(data, dict) else None + if not isinstance(minfraud, dict): + return "" + + for item in minfraud.get("subdivisions") or []: + province = self._normalize_province(item) + if province: + return province + + return "" + + async def _resolve_ip(self, session: aiohttp.ClientSession, retries: int, timeout: int) -> Optional[str]: + url = "https://ipinfo.io/ip" + text, _ = await self._make_request(session, url, retries, timeout, deserialize=False) + if not isinstance(text, str): + return None + + address = text.strip() + try: + ipaddress.ip_address(address) + return address + except ValueError: + return None + + async def _fetch( + self, session: aiohttp.ClientSession, proxy_info: ProxyInfo, retries: int, timeout: int + ) -> Tuple[Optional[Dict], Optional[str]]: + address = await self._resolve_ip(session, retries, timeout) + if not address: + host = "" if not proxy_info else proxy_info.host + return None, f"Failed to get egress IP, host: {host}" + + url = f"https://meowvps.com/api/ip-aggregator/{quote(address, safe='')}" + headers = { + "Accept": "*/*", + "Accept-Language": "zh-CN,zh;q=0.8", + "Origin": "https://meowvps.com", + "Referer": "https://meowvps.com/tools/ip-check/", + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/152.0.0.0 Safari/537.36" + ), + } + data, error = await self._make_request(session, url, retries, timeout, headers=headers) + if not data: + return None, error or f"Failed to get IP info from meowvps, ip: {address}" + if data.get("success") is False: + return None, f"MeowVPS lookup failed, ip: {address}" + + return data, None + + +IP_LIBRARIES = { + "ip2location": IP2LocationLibrary, + "iplark": IPLarkLibrary, + "ipinfo": IPInfoLibrary, + "ipnetcoffee": IPNetCoffeeLibrary, + "ippure": IPPureLibrary, + "meowvps": MeowVPSLibrary, +} + + +def get_ip_library(name: str) -> IPLibrary: + key = (name or "ip2location").strip().lower() + library = IP_LIBRARIES.get(key) + if not library: + supported = ", ".join(sorted(IP_LIBRARIES.keys())) + raise ValueError(f"Unsupported ip library: {name}. Supported: {supported}") + + return library() + + class ProxyChecker: def __init__( self, @@ -316,6 +1223,7 @@ def __init__( format_pattern: Optional[str] = None, default_port: int = 1080, include_asn_name: bool = False, + ip_library: str = "ip2location", ): """ 初始化代理检测器 @@ -332,6 +1240,7 @@ def __init__( self.format_pattern = format_pattern self.default_port = default_port self.include_asn_name = include_asn_name + self.ip_library = get_ip_library(ip_library) self.results: List[TestResult] = [] self.summary: Optional[Dict[str, float]] = None @@ -340,7 +1249,7 @@ def parse_proxy(self, text: str, format_pattern: Optional[str] = None) -> Option 解析代理字符串，支持自定义格式 支持的格式占位符: - - {protocol}: 协议类型 (socks5/socks4/http等) + - {protocol}: 协议类型 (socks5/socks4/http/https等) - {username}: 用户名 - {password}: 密码 - {host}: 主机地址 @@ -382,10 +1291,7 @@ def parse_proxy(self, text: str, format_pattern: Optional[str] = None) -> Option prefix = f"socks5://{prefix}" result = urlparse(prefix) - protocol = result.scheme or "socks5" - if protocol == "https": - protocol = "http" return ProxyInfo( protocol=protocol, @@ -472,9 +1378,6 @@ def _parse_custom_format(self, text: str, format_pattern: str) -> Optional[Proxy elif placeholder == "host": host = value - if protocol == "https": - protocol = "http" - return ProxyInfo( protocol=protocol, username=username, @@ -490,33 +1393,35 @@ async def test_proxy(self, proxy_info: ProxyInfo, retries: int = 3) -> TestResul Test a single proxy with retries. """ result = TestResult.from_proxy(proxy_info) - - # Build proxy URL - if proxy_info.username and proxy_info.password: - proxy_url = ( - f"{proxy_info.protocol}://{proxy_info.username}:{proxy_info.password}" - f"@{proxy_info.host}:{proxy_info.port}" - ) - else: - proxy_url = f"{proxy_info.protocol}://{proxy_info.host}:{proxy_info.port}" - start_time = datetime.now() try: - connector = ProxyConnector.from_url(proxy_url) - async with aiohttp.ClientSession(connector=connector) as session: - ip_address = await self._resolve_ip_with_proxy(session, proxy_info.host, retries) - if not ip_address: - result.error = "Failed to get IP from ipinfo.io/ip" - return result - - ip_data, ip_error = await self._fetch_ipinfo(session, ip_address, retries) - if not ip_data: - result.error = ip_error or "Failed to get IP info from ipinfo.io" + protocol = (proxy_info.protocol or "").lower() + if protocol in ("http", "https"): + proxy_url = self._build_proxy_url(proxy_info, include_auth=False) + proxy_auth = self._build_proxy_auth(proxy_info) + connector = aiohttp.TCPConnector(ssl=False) + session = aiohttp.ClientSession(connector=connector, proxy=proxy_url, proxy_auth=proxy_auth) + else: + proxy_url = self._build_proxy_url(proxy_info, include_auth=True) + connector = ProxyConnector.from_url(proxy_url) + session = aiohttp.ClientSession(connector=connector) + + async with session: + lookup = await self.ip_library.lookup(session, proxy_info, retries, self.timeout) + if not lookup.ip or not lookup.data: + result.error = lookup.error or f"Failed to get IP info from {self.ip_library.name}" return result - remark = self._build_remark_from_ipinfo(ip_data) + remark = await self.ip_library.build_remark( + session=session, + ip=lookup.ip, + data=lookup.data, + include_asn_name=self.include_asn_name, + retries=retries, + timeout=self.timeout, + ) result.remark = remark - result.ip = ip_address + result.ip = lookup.ip result.status = "success" result.response_time = round((datetime.now() - start_time).total_seconds(), 2) result.error = None @@ -531,77 +1436,6 @@ async def test_proxy(self, proxy_info: ProxyInfo, retries: int = 3) -> TestResul result.error = str(e)[:100] return result - def _is_ipv4(self, host: str) -> bool: - if not host: - return False - try: - return isinstance(ipaddress.ip_address(host), ipaddress.IPv4Address) - except ValueError: - return False - - async def _resolve_ip_with_proxy(self, session: aiohttp.ClientSession, host: str, retries: int) -> Optional[str]: - if self._is_ipv4(host): - return host - - url = "https://ipinfo.io/ip" - for attempt in range(1, retries + 1): - try: - async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.timeout)) as response: - if response.status == 200: - text = (await response.text()).strip() - try: - ipaddress.ip_address(text) - return text - except ValueError: - pass - except asyncio.TimeoutError: - pass - except Exception: - pass - - if attempt < retries: - await asyncio.sleep(attempt) - - return None - - async def _fetch_ipinfo( - self, session: aiohttp.ClientSession, ip_address: str, retries: int - ) -> Tuple[Optional[Dict], Optional[str]]: - url = f"https://ipinfo.io/widget/demo/{ip_address}" - last_error = None - for attempt in range(1, retries + 1): - try: - async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.timeout)) as response: - if response.status == 200: - data = await response.json() - if isinstance(data, dict): - return data.get("data", data), None - last_error = "Invalid JSON response" - else: - last_error = f"HTTP {response.status}" - except asyncio.TimeoutError: - last_error = "Timeout" - except Exception as e: - last_error = str(e)[:100] - - if attempt < retries: - await asyncio.sleep(attempt) - - return None, last_error - - def _country_flag_emoji(self, country_code: str) -> str: - if not country_code or len(country_code) != 2: - return "" - code = country_code.upper() - if not code.isalpha(): - return "" - return chr(0x1F1E6 + ord(code[0]) - ord("A")) + chr(0x1F1E6 + ord(code[1]) - ord("A")) - - def _country_name_zh(self, country_code: str) -> str: - if not country_code: - return "" - return COUNTRY_NAME_ZH.get(country_code.upper(), "") - def _format_standard(self, proxy_info: ProxyInfo, remark: str) -> str: auth = "" if proxy_info.username or proxy_info.password: @@ -611,58 +1445,43 @@ def _format_standard(self, proxy_info: ProxyInfo, remark: str) -> str: return f"{base}#{remark}" return base + def _build_proxy_url(self, proxy_info: ProxyInfo, include_auth: bool) -> str: + auth = "" + if include_auth and (proxy_info.username or proxy_info.password): + username = quote(proxy_info.username or "", safe="") + password = quote(proxy_info.password or "", safe="") + auth = f"{username}:{password}@" + + return f"{proxy_info.protocol}://{auth}{proxy_info.host}:{proxy_info.port}" + + def _build_proxy_auth(self, proxy_info: ProxyInfo) -> Optional[aiohttp.BasicAuth]: + if not (proxy_info.username or proxy_info.password): + return None + + return aiohttp.BasicAuth(proxy_info.username or "", proxy_info.password or "") + def _yaml_quote(self, value: str) -> str: escaped = value.replace("\\", "\\\\").replace('"', '\\"') return f'"{escaped}"' def _format_yaml_line(self, result: TestResult) -> str: name = result.remark or result.host + protocol = (result.protocol or "").lower() + clash_type = "http" if protocol == "https" else protocol parts = [ f"name: {self._yaml_quote(name)}", f"server: {self._yaml_quote(result.host)}", f"port: {result.port}", - f"type: {self._yaml_quote(result.protocol)}", + f"type: {self._yaml_quote(clash_type)}", ] + if protocol == "https": + parts.append("tls: true") if result.username: parts.append(f"username: {self._yaml_quote(result.username)}") if result.password: parts.append(f"password: {self._yaml_quote(result.password)}") return " - {" + ", ".join(parts) + "}" - def _build_remark_from_ipinfo(self, ip_data: Dict) -> str: - country_code = (ip_data.get("country") or "").upper() - flag = self._country_flag_emoji(country_code) - country_display = self._country_name_zh(country_code) or "未知" - - asn_info = ip_data.get("asn", {}) or {} - company_info = ip_data.get("company", {}) or {} - asn_type = (asn_info.get("type") or "").lower() - company_type = (company_info.get("type") or "").lower() - - asn_name = (asn_info.get("domain") or "").strip() - if not asn_name: - asn_name = (asn_info.get("name") or "").strip() - - if asn_name: - parts = [p for p in re.split(r"[\s,\.]+", asn_name) if p] - company_name = parts[0].upper() if parts else "UNKNOWN" - else: - company_name = "UNKNOWN" - - if asn_type == "isp" and company_type == "isp": - label = "家宽" - elif asn_type == "isp" or company_type == "isp": - label = "商宽" - elif asn_type == "edu" or company_type == "edu": - label = "教育" - else: - label = "" - - base = f"{flag} {country_display}{label}".strip() - if self.include_asn_name and company_name: - return f"{base} [{company_name}]".strip() - return base - def _convert(self, input_file: str, output_file: str, output_format: str, digits: int = 2) -> None: proxies = read_proxies(input_file) if not proxies: @@ -765,7 +1584,7 @@ async def check_proxies( output_handle = None if not output_file: - output_file = f'{output_format}.{"txt" if output_format == "v2ray" else "yaml"}' + output_file = f'{output_format}-{self.ip_library}.{"txt" if output_format == "v2ray" else "yaml"}' output_handle = open(output_file, "w", encoding="utf-8") if output_format == "clash": @@ -797,12 +1616,22 @@ async def test_with_semaphore(proxy_info): # 实时输出结果 status_icon = "✓" if result.status == "success" else "✗" if result.status == "success": - print(f"{status_icon} {result.original[:60]}... | {result.response_time}s | IP: {result.ip}") + print( + f"{status_icon} {result.original[:60]}... | {result.response_time}s | Export IP: {result.ip}".encode( + "utf-8", errors="ignore" + ).decode( + "utf-8" + ) + ) if write_queue: line = self._format_yaml_line(result) if output_format == "clash" else result.proxy await write_queue.put(line + "\n") else: - print(f"{status_icon} {result.original[:60]}... | {result.error}") + print( + f"{status_icon} {result.original[:60]}... | {result.error}".encode( + "utf-8", errors="ignore" + ).decode("utf-8") + ) async with stats_lock: if result.status == "success": @@ -930,8 +1759,8 @@ def _build_proxy(entry: Dict) -> Optional[str]: return None protocol = str(entry.get("type") or "socks5").strip().lower() - if protocol == "https": - protocol = "http" + if protocol == "http" and entry.get("tls") is True: + protocol = "https" elif protocol == "socks": protocol = "socks5" @@ -982,7 +1811,8 @@ def _load_proxies(data) -> List[str]: def _parse_yaml(text: str) -> Tuple[Optional[List[str]], Optional[object]]: try: - data = yaml.safe_load(text) + content = text.lstrip("\ufeff").replace("\r\n", "\n").replace("\r", "\n").replace("\t", " ") + data = yaml.safe_load(content) except yaml.YAMLError: return None, None @@ -1055,7 +1885,7 @@ async def main(): %(prog)s -f proxies.txt --input-format "socks5://{host}:{port}:{username}:{password}" 支持的格式占位符: - {protocol} - 协议类型 (socks5/socks4/http等) + {protocol} - 协议类型 (socks5/socks4/http/https等) {username} - 用户名 {password} - 密码 {host} - 主机地址/IP @@ -1089,6 +1919,14 @@ async def main(): help="在备注中追加 ASN 名称 (默认不追加)", ) + parser.add_argument( + "--ip-library", + dest="ip_library", + choices=sorted(IP_LIBRARIES.keys()), + default="ip2location", + help="IP地址数据库服务商: ip2location、iplark、ipinfo、ipnetcoffee、ippure 或 meowvps (默认: ip2location)", + ) + args = parser.parse_args() # 获取代理列表 @@ -1116,6 +1954,7 @@ async def main(): format_pattern=args.format_pattern, default_port=args.default_port, include_asn_name=args.include_asn_name, + ip_library=args.ip_library, ) await checker.check_proxies(