first commit

proxy_chain_manager/leak_detect.py

@@ -0,0 +1,64 @@
+"""VPN-aware chain leak detection."""
+from __future__ import annotations
+
+
+def is_same_subnet(ip_a: str | None, ip_b: str | None, prefix_len: int = 16) -> bool:
+    """True if two IPv4 addresses share the same /prefix_len subnet."""
+    if not ip_a or not ip_b:
+        return False
+    try:
+        a_parts = [int(x) for x in ip_a.split(".")]
+        b_parts = [int(x) for x in ip_b.split(".")]
+        if len(a_parts) != 4 or len(b_parts) != 4:
+            return False
+
+        def to_int(parts: list[int]) -> int:
+            return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]
+
+        mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF
+        return (to_int(a_parts) & mask) == (to_int(b_parts) & mask)
+    except Exception:
+        return False
+
+
+def is_chain_leak(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> bool:
+    """True when the chain is not forwarding (traffic still looks like direct/VPN exit).
+
+    With VPN: compare /16 — VPN IPs rotate but stay in-provider ranges.
+    Without VPN: exact IP match only (avoid false positives on same ISP /16).
+
+    **Fail-closed**: when ``real_ip`` is unknown we cannot prove the chain is
+    safe, so we conservatively return ``True``.  The service loop applies a
+    short warm-up grace period before this verdict kicks in so a transient
+    direct-IP lookup failure doesn't cause perpetual rotation.
+    """
+    if not exit_ip:
+        return True
+    if not real_ip:
+        # Fail-closed: README promises "fail closed" when trust dies.
+        return True
+    if exit_ip == real_ip:
+        return True
+    if vpn_active:
+        return is_same_subnet(exit_ip, real_ip)
+    return False
+
+
+def leak_reason(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> str:
+    if not exit_ip:
+        return "exit IP unreachable"
+    if not real_ip:
+        return (
+            "direct IP unknown — cannot prove the chain is forwarding "
+            "(fail-closed). Check internet connectivity and ip_check_url."
+        )
+    if exit_ip == real_ip:
+        if vpn_active:
+            return f"exit {exit_ip} equals VPN/direct IP (chain not forwarding)"
+        return f"exit {exit_ip} equals your real IP (no anonymization)"
+    if vpn_active and is_same_subnet(exit_ip, real_ip):
+        return (
+            f"exit {exit_ip} shares /16 with direct {real_ip} "
+            "(likely exiting via VPN tunnel, not proxy chain)"
+        )
+    return "ok"