Make SOCKS5 exit IP detection robust

app.py

@@ -373,35 +373,82 @@ async def socks5_connect(proxy: dict[str, Any], timeout: float) -> None:
     with contextlib.suppress(Exception): await writer.wait_closed()
 
 
-async def get_exit_ip_via_socks(proxy: dict[str, Any], timeout: float) -> str | None:
-    """Fetch the public egress IP through the SOCKS5 proxy without failing health checks."""
-    reader, writer = await socks5_open_connection(proxy, timeout, "api.ipify.org", 80)
+async def http_get_via_socks(proxy: dict[str, Any], timeout: float, host: str, path: str = "/") -> str:
+    reader, writer = await socks5_open_connection(proxy, timeout, host, 80)
     try:
-        req = b"GET /?format=text HTTP/1.1\r\nHost: api.ipify.org\r\nUser-Agent: socks5-monitor/1.0\r\nConnection: close\r\n\r\n"
+        req = (
+            f"GET {path} HTTP/1.1\r\n"
+            f"Host: {host}\r\n"
+            "User-Agent: socks5-monitor/1.0\r\n"
+            "Accept: text/plain, application/json;q=0.9, */*;q=0.1\r\n"
+            "Connection: close\r\n\r\n"
+        ).encode("ascii")
         writer.write(req); await writer.drain()
-        raw = await asyncio.wait_for(reader.read(8192), timeout=timeout)
-        text = raw.decode("utf-8", "replace")
-        body = text.split("\r\n\r\n", 1)[-1].strip()
-        m = re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", body)
-        if m:
-            return m.group(0)
-        m = re.search(r"\b[0-9a-fA-F:]{3,}\b", body)
-        return m.group(0) if m else None
+        chunks = []
+        deadline = asyncio.get_running_loop().time() + timeout
+        while True:
+            remaining = max(0.1, deadline - asyncio.get_running_loop().time())
+            if remaining <= 0.1 and chunks:
+                break
+            try:
+                chunk = await asyncio.wait_for(reader.read(4096), timeout=remaining)
+            except asyncio.TimeoutError:
+                break
+            if not chunk:
+                break
+            chunks.append(chunk)
+            if sum(len(x) for x in chunks) >= 65536:
+                break
+        raw = b"".join(chunks)
+        return raw.decode("utf-8", "replace")
     finally:
         writer.close()
         with contextlib.suppress(Exception): await writer.wait_closed()
 
 
+def extract_ip(text: str) -> str | None:
+    body = text.split("\r\n\r\n", 1)[-1].strip()
+    m = re.search(r"\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b", body)
+    if m:
+        return m.group(0)
+    # Very small IPv6 matcher, enough for plain text IP services.
+    m = re.search(r"\b(?:[0-9a-fA-F]{1,4}:){2,}[0-9a-fA-F]{0,4}\b", body)
+    return m.group(0) if m else None
+
+
+async def get_exit_ip_via_socks(proxy: dict[str, Any], timeout: float) -> str | None:
+    """Fetch the public egress IP through the SOCKS5 proxy without failing health checks."""
+    endpoints = [
+        ("api.ipify.org", "/?format=text"),
+        ("ipv4.icanhazip.com", "/"),
+        ("ifconfig.me", "/ip"),
+        ("ident.me", "/"),
+        ("checkip.amazonaws.com", "/"),
+    ]
+    errors = []
+    per_endpoint_timeout = max(2.0, min(float(timeout), 6.0))
+    for host, path in endpoints:
+        try:
+            text = await http_get_via_socks(proxy, per_endpoint_timeout, host, path)
+            ip = extract_ip(text)
+            if ip:
+                return ip
+            errors.append(f"{host}: no ip in response")
+        except Exception as e:
+            errors.append(f"{host}: {e}")
+    proxy["last_exit_ip_error"] = "; ".join(errors[-3:])
+    return None
+
+
 async def check_proxy(proxy: dict[str, Any]) -> tuple[bool, int | None, str | None, str | None]:
     timeout = float(proxy.get("timeout_seconds", 8))
     start = time.perf_counter()
     try:
         await socks5_connect(proxy, timeout)
         latency = int((time.perf_counter() - start) * 1000)
-        exit_ip = None
-        with contextlib.suppress(Exception):
         exit_ip = await get_exit_ip_via_socks(proxy, timeout)
-        return True, latency, None, exit_ip
+        err = None if exit_ip else proxy.get("last_exit_ip_error")
+        return True, latency, err, exit_ip
     except Exception as e:
         return False, int((time.perf_counter() - start) * 1000), str(e), None