#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""proxy-filter.py -- split a list of IP addresses into clean.txt/flagged.txt
by proxy/VPN/Tor/hosting signal, via the ip.xakki.pro API.

Reads a text file with one IP address per line, calls
GET /api/v1/ip?ip=<addr>&include=security for each one, and writes:
  - clean.txt   -- IPs with no proxy/VPN/Tor/hosting signal.
  - flagged.txt -- IPs with at least one signal, one per line, with the
                    reason(s) appended after a tab.

An IP is flagged when ANY of these hold:
  - privacy.isProxy   -- a known anonymizing-proxy network (IPFire libloc).
  - privacy.isAnonymous -- isProxy OR isVpn OR isTor (the API's own rollup).
  - network.classification.type is "hosting" or "datacenter" -- not itself
    proof of a proxy, but a strong correlate (real residential ISPs almost
    never show up here) worth surfacing as its own reason.

An IP whose "security" group comes back rate-limited (withheld -- see the
rate-limit note in bulk-enrich.py's docstring) is CONSERVATIVELY flagged
with reason "unverified: rate-limited" rather than assumed clean -- silently
letting an unverified IP through to clean.txt would defeat the point of this
script. Re-run later (a repeat lookup of the same IP the same UTC day is
free once the group's budget resets) to get a real verdict.

A non-routable input (private/loopback/CGNAT/bogon) gets NO privacy signal
at all -- the API returns "privacy": null for those (there is no meaningful
proxy/VPN classification for an address that was never publicly routable),
so this script treats it as clean by omission. That is intentional, not a
gap: a private-range IP has no business appearing in a real proxy-filter
input in the first place.

Zero dependencies -- stdlib only. Python 3.8+.

Usage:
    python3 proxy-filter.py ips.txt
    python3 proxy-filter.py ips.txt --delay 1.0 --base-url http://localhost:11120
"""

import argparse
import ipaddress
import json
import sys
import time
import urllib.error
import urllib.request

DEFAULT_BASE_URL = "https://ip.xakki.pro"
USER_AGENT = "myip-proxy-filter-recipe/1.0 (+https://ip.xakki.pro/tools/scripts/proxy-filter.html)"
HOSTING_TYPES = {"hosting", "datacenter"}


def fetch_security(base_url, ip_value, max_retries=5, base_backoff=0.5):
    url = "{}/api/v1/ip?ip={}&include=security".format(base_url, ip_value)
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})
    attempt = 0
    while True:
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            if not (e.code == 429 or e.code >= 500) or attempt >= max_retries:
                raise
            wait = float(e.headers.get("Retry-After", 0)) or (base_backoff * (2 ** attempt))
        except (urllib.error.URLError, TimeoutError, ConnectionError):
            if attempt >= max_retries:
                raise
            wait = base_backoff * (2 ** attempt)
        attempt += 1
        time.sleep(wait)


def classify(base_url, ip_value):
    """Returns (is_flagged, reasons_list) for one IP. Never raises -- a
    transport failure or an invalid input is itself a flagged reason."""
    try:
        ipaddress.ip_address(ip_value)
    except ValueError:
        return True, ["invalid IP, could not check"]

    try:
        data = fetch_security(base_url, ip_value)
    except Exception as e:  # noqa: BLE001
        return True, ["unverified: request failed ({})".format(e)]

    try:
        if ipaddress.ip_address(data.get("ip", "")) != ipaddress.ip_address(ip_value):
            return True, ["unverified: unexpected response (ip mismatch)"]
    except ValueError:
        return True, ["unverified: unexpected response"]

    if "security" in (data.get("limited") or []):
        return True, ["unverified: rate-limited"]

    reasons = []
    privacy = data.get("privacy") or {}
    if privacy.get("isProxy"):
        reasons.append("privacy.isProxy")
    if privacy.get("isAnonymous"):
        reasons.append("privacy.isAnonymous")
    net_type = ((data.get("network") or {}).get("classification") or {}).get("type")
    if net_type in HOSTING_TYPES:
        reasons.append("network type = {}".format(net_type))

    return (len(reasons) > 0), reasons


def main():
    parser = argparse.ArgumentParser(description="Split an IP list into clean/flagged by proxy signal.")
    parser.add_argument("input_file", help="Path to a text file, one IP per line.")
    parser.add_argument("--clean-out", default="clean.txt", help="Output file for clean IPs (default: clean.txt).")
    parser.add_argument("--flagged-out", default="flagged.txt", help="Output file for flagged IPs (default: flagged.txt).")
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="API base URL (default: production).")
    parser.add_argument("--delay", type=float, default=0.5, help="Seconds between lookups (default: 0.5).")
    args = parser.parse_args()

    with open(args.input_file, encoding="utf-8") as f:
        ip_lines = [line.strip() for line in f if line.strip()]

    clean, flagged = [], []
    for i, ip_value in enumerate(ip_lines):
        is_flagged, reasons = classify(args.base_url, ip_value)
        if is_flagged:
            flagged.append((ip_value, reasons))
        else:
            clean.append(ip_value)
        print("[{}/{}] {} -> {}".format(i + 1, len(ip_lines), ip_value,
                                         "FLAGGED: " + ", ".join(reasons) if is_flagged else "clean"),
              file=sys.stderr)
        if i < len(ip_lines) - 1:
            time.sleep(args.delay)

    with open(args.clean_out, "w", encoding="utf-8") as f:
        f.write("\n".join(clean) + ("\n" if clean else ""))
    with open(args.flagged_out, "w", encoding="utf-8") as f:
        for ip_value, reasons in flagged:
            f.write("{}\t{}\n".format(ip_value, "; ".join(reasons)))

    print("done: {} clean -> {}, {} flagged -> {}".format(
        len(clean), args.clean_out, len(flagged), args.flagged_out), file=sys.stderr)


if __name__ == "__main__":
    main()