"""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"