Socks List Processor：免费的Windows批量代理检测工具

#!/usr/bin/env python3
"""slpcheck 3.4 - multi-threaded proxy checker for Linux (SLP companion script).

Python 3.8+, standard library only. Edit the CONFIG block (or save values
with --set, the configurator on advanced.name/slp builds this command), then run:

    python3 slpcheck.py --set PROXY_URL='YOUR_LIST_ID' THREADS=300
    python3 slpcheck.py

PROXY_URL is the list ID from your link in the dashboard (cp.advanced.name):
https://get.advanced.name/<ID> for packages (<ID>.csv for the CSV export) or
https://rlists.advanced.name/<ID> for residential proxies. The script picks the
host by the look of the ID; the whole link, any other URL or a local file work too.
A package list is served only to the IPs in your subscription whitelist: if this
server is not there, the script prints its IP and stops.

The script downloads the list (or reads a local file), checks every proxy in
THREADS parallel threads and writes the good ones to OUTPUT_FILE, one per line
as the list has them. The list is either CSV with a header, for example
ip;port;country;speed;uptime;real_ip (real_ip or exit_ip is the exit address),
or one proxy per line in any of these forms:

    ip:port                        IP authorization, no login
    host:port:user:pass            proxies that need a login
    user:pass@host:port
    http://user:pass@host:port     the scheme sets the protocol of that line,
    socks5://user:pass@host:port   PROXY_TYPE covers the lines without one

The password may contain ':' and '@', the user name '+', '-' and '='
(USER+country=us+session=a1). HTTP proxies get Proxy-Authorization: Basic,
SOCKS5 ones the username/password step of RFC 1929. A proxy that turns the
login down is counted as "login refused", apart from the dead ones.
Lines with a login and no scheme use the protocol of their gateway port. The
ports of the Advanced.name gateways gw1.advanced.name, gw2.advanced.name and so
on are known (Resi Core: 6969 HTTP, 9696 SOCKS5; Resi Prime: 9000 and
10000-10900 HTTP, 11000 and 12000-12010 SOCKS5; mobile: 8080 HTTP, 1080
SOCKS5), any other host:port is asked once before the run: a port that answers
the SOCKS5 greeting gets SOCKS5, the others HTTP. PROXY_TYPE
covers ip:port lines and the ports that do not take the connection.

CHECK_MODE:

    connect  open CHECK_URL through every proxy, keep the ones that answer;
             REPORT_FILE lists them with the response time, fastest first
    rbl      the same live check, then the RBL lookup of the exit IP of every
             live proxy: only live and clean proxies reach OUTPUT_FILE;
             REPORT_FILE lists them with the zones, the dirtiest first

With CHECK_URL = "" the rbl mode skips the live check: a fast RBL pass over a
CSV list that has real_ip (seconds instead of minutes), dead proxies included.

Where the exit IP comes from in the rbl mode, in this order: the CHECK_URL
answer (ifconfig.me/ip and api.ipify.org answer with a bare IP), a request to
IP_URL through the same proxy when that answer has none (a judge like
google.com), real_ip from the CSV. Without the live check real_ip goes first.
A proxy whose exit IP stays unknown counts as "no exit IP": the gateway address
of a backconnect is never checked instead. Only with IP_URL = "" the proxy
address itself is used. The totals show how many exit IPs came from each source.

A proxy passes the RBL check when no more than RBL_MAX zones list its IP.
FTP_UPLOAD sends the good list to one or more FTP servers.

Before the run every zone must answer its test entry 127.0.0.2 through
RBL_DNS, a zone that fails is skipped. Spamhaus zones need your own
resolver (unbound on the server, RBL_DNS = "127.0.0.1"): through public DNS
they answer "not listed" for most addresses.

To refresh the list on a timer, run the script from cron (crontab -e):

    */30 * * * * cd ~/slpcheck && flock -n .lock python3 slpcheck.py > last-run.log 2>&1

OUTPUT_FILE and REPORT_FILE are names in the folder the script runs in (with
cron, the folder after cd) or full paths such as /var/www/html/socks.txt;
~ works and missing folders are created.

Every run replaces OUTPUT_FILE and REPORT_FILE with its own results in one step,
so the files never mix two runs and readers never see half a list. With no
good proxies OUTPUT_FILE becomes empty, FTP is skipped and the exit code is 1.
If the list cannot be downloaded, the run stops and the files stay untouched.
Ctrl+C stops at once, saves what is checked and prints the totals.

Versions (python3 slpcheck.py --version):
    3.4  PROXY_URL takes residential list IDs (rlists.advanced.name/<ID>);
         lines with a login and no scheme get the protocol of their gateway
         port (known Advanced.name ports, others asked once); a list that is
         not found says so
    3.3  lines with a login: host:port:user:pass, user:pass@host:port, the same
         after http:// or socks5://; a refused login is counted apart from dead
    3.2  PROXY_URL takes <ID>.csv, the CSV export of the list; a CSV without
         a header row is read as ip;port;country;speed;uptime;real_ip
    3.1  OUTPUT_FILE and REPORT_FILE take full paths and ~, missing folders
         are created
    3.0  PROXY_URL takes the list ID (get.advanced.name/<ID>); when this IP is
         not in the subscription whitelist, the script names the IP to put
         there in cp.advanced.name; an inactive subscription is reported too
    2.9  one request per proxy when CHECK_URL returns the IP (ifconfig.me), the
         plan line says so; IP_URL is never asked when it equals CHECK_URL
    2.8  two modes: connect and rbl (live check + RBL, the old "both", which
         still works as a name); rbl with CHECK_URL = "" is the fast RBL pass
         without the live check; response time in the report and the totals
    2.7  rbl mode gets the exit IP by itself: a proxy without real_ip is asked
         through at IP_URL, one that does not answer counts as "no exit IP"
    2.6  rbl mode on an ip:port list of backconnect gateways: the warning is
         repeated in the totals, the result covers only the gateway IPs
    2.5  version in the header and in the output; every run replaces good.txt
         and report.csv with its own results (a connect run removes the old
         report.csv, an empty result gives an empty good.txt)
    2.4  the check mode is printed first
    2.3  Ctrl+C saves what is checked and prints the totals without waiting
    2.2  exit IP from CHECK_URL, then IP_URL, CSV real_ip, the proxy address;
         the totals show the source of every exit IP
    2.1  CSV input with real_ip, backconnect warning, percentages in the totals
    2.0  RBL check, FTP upload, cron, --set
    1.0  connect check with a progress bar
"""

VERSION = "3.4"

# ================================ CONFIG ====================================
PROXY_URL = "YOUR_LIST_ID"                         # list ID, a link or a local file, ip:port or CSV
THREADS = 200                                         # parallel checks
CHECK_MODE = "connect"                                # "connect" or "rbl" (live check + RBL)
CHECK_URL = "https://ifconfig.me/ip"                  # site to open through each proxy, "" = rbl without it
IP_URL = "https://api.ipify.org"                      # asked through the proxy for an unknown exit IP
PROXY_TYPE = "socks5"                                 # "socks5" or "http" for ip:port lines, login lines detect it
TIMEOUT = 10                                          # seconds per proxy
RBL_LISTS = "dnsbl.dronebl.org bl.blocklist.de psbl.surriel.com dnsbl-1.uceprotect.net"  # DNSBL zones
RBL_MAX = 0                                           # zones allowed to list a proxy, 0 = clean only
RBL_DNS = "1.1.1.1 8.8.8.8"                           # DNS servers for RBL lookups
OUTPUT_FILE = "good.txt"                              # good proxies go here
REPORT_FILE = "report.csv"                            # speed and RBL details, "" = off
FTP_UPLOAD = ""                                       # ftp://user:pass@host/dir/good.txt, "" = off
# ============================================================================

import base64
import ftplib
import io
import os
import random
import re
import shutil
import signal
import socket
import ssl
import struct
import sys
import threading
import time
import urllib.error
import urllib.request
from collections import Counter
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from urllib.parse import unquote, urlsplit

LEGACY_BOTH = CHECK_MODE == "both"
if LEGACY_BOTH:
    CHECK_MODE = "rbl"


def parse_target(url):
    """(url, https, host, port, path) of a site opened through the proxies."""
    u = urlsplit(url)
    https = u.scheme == "https"
    return url, https, u.hostname or "", u.port or (443 if https else 80), \
        (u.path or "/") + ("?" + u.query if u.query else "")


JUDGE = parse_target(CHECK_URL) if CHECK_URL else None
ECHO = parse_target(IP_URL) if IP_URL else None
# IP_URL is asked only when the CHECK_URL answer holds no IP, and never twice for the same address
SECOND = bool(ECHO and JUDGE and ECHO[0] != JUDGE[0])
ECHO_HOSTS = {"ifconfig.me", "api.ipify.org", "api64.ipify.org", "icanhazip.com", "ipv4.icanhazip.com",
              "checkip.amazonaws.com", "ifconfig.co", "ident.me", "ipecho.net", "ipinfo.io"}
TLS = ssl.create_default_context()
TLS.check_hostname = False
TLS.verify_mode = ssl.CERT_NONE
ZONES = RBL_LISTS.split()
DNS_SERVERS = RBL_DNS.split()
DNS_TIMEOUT = 3
PUBLIC_DNS = {"1.1.1.1", "1.0.0.1", "8.8.8.8", "8.8.4.4", "9.9.9.9", "149.112.112.112", "208.67.222.222"}
IP_RE = re.compile(r"\b\d{1,3}(?:\.\d{1,3}){3}\b")
META = {}
ACCOUNTS = {}
SCHEMES = {"http": "http", "socks5": "socks5", "socks5h": "socks5"}


class BadLogin(Exception):
    """The proxy turned down the user and password of the line."""


def is_ipv4(text):
    return bool(IP_RE.fullmatch(text)) and all(int(x) < 256 for x in text.split("."))


def hostlike(name):
    return is_ipv4(name) or name == "localhost" or ("." in name and "/" not in name)


def unpack_line(token):
    """(host, port, user, password, scheme) of a line with a login or a scheme, None for the rest.
    host:port:user:pass and user:pass@host:port, both also after http:// or socks5://;
    the password keeps every ':' and '@'. When both readings fit, the one whose host looks
    like an IP or a domain wins, then host:port:user:pass. scheme is "" without one."""
    head, sep, rest = token.partition("://")
    scheme = ""
    if sep:
        scheme = SCHEMES.get(head.lower(), "")
        if not scheme:
            return None
        token = rest.rstrip("/")
    readings = []
    parts = token.split(":", 3)
    if len(parts) == 4:
        readings.append(parts)
    who, at, where = token.rpartition("@")
    if at:
        user, _, password = who.partition(":")
        host, _, port = where.rpartition(":")
        readings.append([host, port, user, password])
    elif scheme:
        host, _, port = token.rpartition(":")
        readings.append([host, port, None, None])
    fit = [r for r in readings if r[0] and "@" not in r[0] and r[1].isdigit() and 0 < int(r[1]) < 65536]
    if not fit:
        return None
    host, port, user, password = next((r for r in fit if hostlike(r[0])), fit[0])
    return host, int(port), user, password, scheme


def endpoint(proxy):
    """(host, port, user, password, protocol) of a list entry; user is None without a login."""
    if proxy in ACCOUNTS:
        host, port, user, password, scheme = ACCOUNTS[proxy]
        return host, port, user, password, scheme or DETECTED.get((host, port)) or PROXY_TYPE
    host, port = proxy.rsplit(":", 1)
    return host, int(port), None, None, PROXY_TYPE


def parse_list(text):
    """Unique proxies in their order: ip:port strings, lines with a login as written
    (ACCOUNTS[line] holds their parts). CSV rows (header with ip and port) also
    fill META[proxy] = (real_ip, country) from the real_ip/exit_ip and country columns."""
    seen, out, cols = set(), [], None
    for line in text.splitlines():
        line = line.strip().lstrip("\ufeff")
        if not line or line.startswith("#"):
            continue
        token = line.split()[0]
        spec = unpack_line(token)
        sep = "" if spec else ";" if ";" in line else "," if "," in line else ""
        real = country = ""
        if sep:
            fields = [f.strip().strip('"') for f in line.split(sep)]
            names = [f.lower().replace(" ", "_") for f in fields]
            if cols is None and "ip" in names and "port" in names:
                cols = {n: i for i, n in enumerate(names)}
                continue
            c = cols or ({"ip": 0, "port": 1, "country": 2, "real_ip": 5} if len(fields) >= 6 else {"ip": 0, "port": 1})
            get = lambda name: fields[c[name]] if name in c and c[name] < len(fields) else ""
            proxy = "%s:%s" % (get("ip"), get("port"))
            real = next((get(k) for k in ("real_ip", "exit_ip", "realip") if is_ipv4(get(k))), "")
            country = get("country")
        else:
            proxy = token
        host, _, port = proxy.rpartition(":")
        if not (spec or host and port.isdigit()) or proxy in seen:
            continue
        seen.add(proxy)
        out.append(proxy)
        if spec:
            ACCOUNTS[proxy] = spec
        if real or country:
            META[proxy] = (real, country)
    return out


LIST_HOST = "https://get.advanced.name/"
RESI_HOST = "https://rlists.advanced.name/"
RESI_ID = re.compile(r"^[0-9a-f]{40,64}(\.csv)?$", re.I)
PANEL = "https://cp.advanced.name"
DETECTED = {}


def placeholder(src):
    """The default value, or the one of versions before 3.0, is still in place."""
    return src in ("YOUR_LIST_ID", "YOUR_LIST_ID.csv") or src.endswith("/YOUR_TOKEN")


def list_location(src):
    """A link or an existing file stays as it is. A bare list ID becomes a link: 40 to 64 hex digits
    are a residential list on RESI_HOST (it has no CSV export), other IDs (and ID.csv) go to LIST_HOST."""
    src = src.strip()
    if src.startswith(("http://", "https://")) or os.path.exists(src):
        return src
    m = RESI_ID.match(src)
    if m:
        return RESI_HOST + src[:len(src) - len(m.group(1) or "")]
    if re.match(r"^[A-Za-z0-9][A-Za-z0-9-]{7,63}(\.csv)?$", src, re.I):
        return LIST_HOST + src
    return src


def refused(text):
    """Stop with a clear message when the list server turns this host away."""
    low = text.lower()
    if "not authorized" in low:
        m = re.search(r"not authorized\s*-\s*([0-9A-Fa-f:.]+[0-9A-Fa-f])", text)
        ip = m.group(1) if m else "the IP of this server"
        sys.exit("Your IP %s is not authorized for this proxy list.\n"
                 "Open %s and put %s into the IP whitelist (ips) of your subscription\n"
                 "instead of the old address, then run slpcheck again." % (ip, PANEL, ip))
    if "not active" in low and len(text) < 300:
        sys.exit("The subscription of this list is not active: %s\n"
                 "Renew or start it in %s, then run slpcheck again." % (text.strip(), PANEL))


def load_list(src):
    """Return unique proxies from a list ID, a URL or a file, keeping their order."""
    if placeholder(src):
        sys.exit("Set PROXY_URL to the list ID from your dashboard link %s<ID>:\n"
                 "    python3 slpcheck.py --set PROXY_URL='YOUR_LIST_ID'" % LIST_HOST)
    src = list_location(src)
    try:
        if src.startswith(("http://", "https://")):
            req = urllib.request.Request(src, headers={"User-Agent": "slpcheck/" + VERSION})
            with urllib.request.urlopen(req, timeout=60) as r:
                text = r.read().decode("utf-8", "replace")
        else:
            with open(src, encoding="utf-8", errors="replace") as f:
                text = f.read()
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "replace")
        refused(body)
        if e.code == 404:
            sys.exit("slpcheck: the list %s was not found (HTTP 404).\n"
                     "Copy the link or the ID of the list again from %s." % (src, PANEL))
        sys.exit("slpcheck: cannot load the list: HTTP %d %s" % (e.code, body.strip()[:200] or e.reason))
    except (OSError, ValueError) as e:
        sys.exit("slpcheck: cannot load the list: %s" % e)
    refused(text)
    return parse_list(text)


KNOWN_GATES = re.compile(r"^gw[0-9a-z-]*\.advanced\.name$", re.I)


def known_protocol(gate):
    """Protocol of a port of the gateways gw*.advanced.name, None for other hosts and ports."""
    host, port = gate
    if not KNOWN_GATES.search(host):
        return None
    if port in (6969, 9000, 8080) or 10000 <= port <= 10900:
        return "http"
    if port in (9696, 11000, 1080) or 12000 <= port <= 12010:
        return "socks5"
    return None


def socks_or_http(gate):
    """"socks5" when host:port answers the SOCKS5 greeting, "http" when it answers anything else or
    keeps silent (an HTTP proxy waits for a request line), None when it does not take the connection."""
    wait_s = min(TIMEOUT, 6)
    try:
        sock = socket.create_connection(gate, wait_s)
    except OSError:
        return None
    try:
        sock.settimeout(wait_s)
        sock.sendall(b"\x05\x02\x00\x02")
        return "socks5" if sock.recv(2)[:1] == b"\x05" else "http"
    except OSError:
        return "http"
    finally:
        sock.close()


def detect_protocols(proxies):
    """The protocol of every gateway port that carries lines with a login and no scheme, asked once."""
    gates = sorted({ACCOUNTS[p][:2] for p in proxies if p in ACCOUNTS and not ACCOUNTS[p][4]})
    if not gates:
        return
    ask = []
    for gate in gates:
        proto = known_protocol(gate)
        if proto:
            DETECTED[gate] = proto
        else:
            ask.append(gate)
    known = len(gates) - len(ask)
    if ask:
        with ThreadPoolExecutor(max_workers=min(THREADS, len(ask))) as pool:
            for gate, proto in zip(ask, pool.map(socks_or_http, ask)):
                if proto:
                    DETECTED[gate] = proto
    found = Counter(DETECTED.values())
    text = ", ".join("%s %s" % (format(n, ","), "SOCKS5" if p == "socks5" else "HTTP") for p, n in sorted(found.items()))
    lost = len(gates) - len(DETECTED)
    if lost:
        text += (", " if text else "") + "%s not reachable (%s assumed)" % (format(lost, ","), PROXY_TYPE)
    if known:
        text += "; %s by the Advanced.name port table, %s asked" % (format(known, ","), format(len(ask), ","))
    print("slpcheck: login lines without a scheme, protocol of their %s gateway port%s: %s"
          % (format(len(gates), ","), "" if len(gates) == 1 else "s", text))


def recv_exact(sock, n):
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("closed")
        buf += chunk
    return buf


def read_line(sock):
    buf = b""
    while not buf.endswith(b"\n") and len(buf) < 512:
        chunk = sock.recv(1)
        if not chunk:
            break
        buf += chunk
    return buf.decode("latin-1").strip()


def socks5_login(sock, login):
    """Greeting with the user/password method next to "no auth", then the RFC 1929 step."""
    user, password = (x.encode("utf-8") for x in login)
    if not user or len(user) > 255 or len(password) > 255:
        raise BadLogin("socks5 user or password longer than 255 bytes")
    sock.sendall(b"\x05\x02\x00\x02")
    ver, method = recv_exact(sock, 2)
    if ver != 5:
        raise ConnectionError("socks5 greeting")
    if method == 0xFF:
        raise BadLogin("socks5 refused the user/password method")
    if method == 2:
        sock.sendall(b"\x01" + bytes([len(user)]) + user + bytes([len(password)]) + password)
        if recv_exact(sock, 2)[1] != 0:
            raise BadLogin("socks5 login")
    elif method != 0:
        raise ConnectionError("socks5 auth %d" % method)


def tunnel_socks5(sock, host, port, login=None):
    """SOCKS5: greeting (with the login when the line has one), then CONNECT to host:port."""
    if login:
        socks5_login(sock, login)
    else:
        sock.sendall(b"\x05\x01\x00")
        if recv_exact(sock, 2) != b"\x05\x00":
            raise ConnectionError("socks5 auth")
    host = host.encode()
    sock.sendall(b"\x05\x01\x00\x03" + bytes([len(host)]) + host + port.to_bytes(2, "big"))
    head = recv_exact(sock, 4)
    if head[1] != 0:
        raise ConnectionError("socks5 connect %d" % head[1])
    if head[3] == 1:
        recv_exact(sock, 6)
    elif head[3] == 3:
        recv_exact(sock, recv_exact(sock, 1)[0] + 2)
    else:
        recv_exact(sock, 18)


def tunnel_http(sock, https, host, port, auth=""):
    """HTTP proxy: CONNECT for https targets, plain absolute-URI GET otherwise.
    auth is the Proxy-Authorization header line of a line with a login, else ''."""
    if https:
        sock.sendall(("CONNECT %s:%d HTTP/1.1\r\nHost: %s:%d\r\n%s\r\n" % (host, port, host, port, auth)).encode())
        status = read_line(sock)
        while read_line(sock):
            pass
        if auth and status.split()[1:2] == ["407"]:
            raise BadLogin(status)
        if " 200" not in status:
            raise ConnectionError(status)


def read_answer(sock):
    """Status line and the start of the body: enough for IP echo services."""
    data = b""
    while len(data) < 65536:
        head, sep, body = data.partition(b"\r\n\r\n")
        if sep:
            size = re.search(rb"(?im)^content-length:\s*(\d+)", head)
            if len(body) >= 512 or (size and len(body) >= int(size.group(1))):
                break
        try:
            chunk = sock.recv(4096)
        except socket.timeout:
            if sep:
                break
            raise
        if not chunk:
            break
        data += chunk
    head, _, body = data.partition(b"\r\n\r\n")
    return head.split(b"\r\n", 1)[0].decode("latin-1"), body


def body_ip(body):
    """Exit IP from a short answer like the one of ifconfig.me/ip, else None."""
    if len(body) >= 512:
        return None
    for ip in IP_RE.findall(body.decode("latin-1")):
        if all(int(x) < 256 for x in ip.split(".")):
            return ip
    return None


def fetch(ip, port, site, via=PROXY_TYPE, login=None):
    """(answered, ip_in_answer, ms): answered when the site gave 2xx/3xx through the proxy,
    ms from the connect to the proxy until the answer arrived. via is socks5 or http,
    login is (user, password) or None; a refused login raises BadLogin."""
    url, https, host, site_port, path = site
    auth = ""
    if login and via != "socks5":
        auth = "Proxy-Authorization: Basic %s\r\n" % base64.b64encode(("%s:%s" % login).encode("utf-8")).decode()
    started = time.time()
    try:
        sock = socket.create_connection((ip, port), TIMEOUT)
        sock.settimeout(TIMEOUT)
        if via == "socks5":
            tunnel_socks5(sock, host, site_port, login)
        else:
            tunnel_http(sock, https, host, site_port, auth)
        if https:
            sock = TLS.wrap_socket(sock, server_hostname=host)
        plain = not https and via != "socks5"
        target = url if plain else path
        sock.sendall(("GET %s HTTP/1.1\r\nHost: %s\r\nUser-Agent: slpcheck/%s\r\n%sConnection: close\r\n\r\n"
                      % (target, host, VERSION, auth if plain else "")).encode())
        status, body = read_answer(sock)
        sock.close()
        parts = status.split()
        if plain and auth and parts[1:2] == ["407"]:
            raise BadLogin(status)
        if len(parts) > 1 and parts[0].startswith("HTTP/") and parts[1][:1] in ("2", "3"):
            return True, body_ip(body), int((time.time() - started) * 1000)
    except (OSError, ValueError, IndexError):
        pass
    return False, None, None


def dns_lookup(names):
    """A records for several names at once over UDP: {name: [ip, ...]}, [] when
    the name does not exist, None when no server gave an answer."""
    result = dict.fromkeys(names)
    for server in DNS_SERVERS * 2:
        base = random.randrange(65536)
        pending = {(base + i) % 65536: n for i, n in enumerate(n for n in names if result[n] is None)}
        if not pending:
            break
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        deadline = time.time() + DNS_TIMEOUT
        try:
            for qid, name in pending.items():
                labels = b"".join(bytes([len(p)]) + p for p in name.encode("ascii").split(b"."))
                sock.sendto(struct.pack(">HHHHHH", qid, 0x0100, 1, 0, 0, 0) + labels + b"\x00\x00\x01\x00\x01",
                            (server, 53))
            while pending:
                sock.settimeout(max(0.05, deadline - time.time()))
                data = sock.recv(1024)
                name = pending.pop(struct.unpack(">H", data[:2])[0], None)
                rcode = data[3] & 15
                if name and rcode in (0, 3):
                    result[name] = answer_ips(data) if rcode == 0 else []
        except (OSError, IndexError, struct.error):
            pass
        finally:
            sock.close()
    return result


def skip_name(data, pos):
    while True:
        size = data[pos]
        if size == 0:
            return pos + 1
        if size >= 192:
            return pos + 2
        pos += size + 1


def answer_ips(data):
    count = struct.unpack(">H", data[6:8])[0]
    pos = skip_name(data, 12) + 4
    ips = []
    for _ in range(count):
        pos = skip_name(data, pos)
        rtype, _, _, size = struct.unpack(">HHIH", data[pos:pos + 10])
        pos += 10
        if rtype == 1 and size == 4:
            ips.append(socket.inet_ntoa(data[pos:pos + 4]))
        pos += size
    return ips


RBL_CACHE = {}
RBL_FAILED = Counter()
RBL_LOCK = threading.Lock()


def verdict(answer):
    """True for listed, False for clean, None for no answer or an error code like 127.255.255.254."""
    if answer is None or any(not a.startswith("127.") or a.startswith("127.255.255.") for a in answer):
        return None
    return bool(answer)


def rbl_zones(ip):
    """Zones that list this IPv4 address, all asked at once; a zone without a usable answer does not count."""
    with RBL_LOCK:
        if ip in RBL_CACHE:
            return RBL_CACHE[ip]
    rev = ".".join(reversed(ip.split(".")))
    answers = dns_lookup([rev + "." + zone for zone in ZONES])
    listed, failed = [], []
    for zone in ZONES:
        v = verdict(answers[rev + "." + zone])
        if v is None:
            failed.append(zone)
        elif v:
            listed.append(zone)
    with RBL_LOCK:
        if ip not in RBL_CACHE:
            RBL_CACHE[ip] = listed
            RBL_FAILED.update(failed)
        return RBL_CACHE[ip]


def zone_problem(zone):
    """Why a zone cannot be used through RBL_DNS, '' when it answers its RFC 5782 test entry 127.0.0.2."""
    if zone.endswith("spamhaus.org") and PUBLIC_DNS & set(DNS_SERVERS):
        return "Spamhaus refuses public resolvers, set RBL_DNS to your own resolver"
    name = "2.0.0.127." + zone
    if all(verdict(dns_lookup([name])[name]) for _ in range(3)):
        return ""
    return "no answer for its test entry through " + ", ".join(DNS_SERVERS)


def check(proxy):
    """(good, status, exit_ip, zones, source, ms) for one proxy.
    status: "dead" (no answer from CHECK_URL), "login" (the proxy refused the login
    of the line), "noip" (exit IP unknown) or "ok";
    zones is None in the connect mode; source says where the exit IP came from:
    judge, echo, csv or proxy; ms is the CHECK_URL response time."""
    host, port, user, password, via = endpoint(proxy)
    login = None if user is None else (user, password)
    real = META.get(proxy, ("", ""))[0]
    ip, source, ms = None, "", None
    try:
        if JUDGE:
            alive, ip, ms = fetch(host, port, JUDGE, via, login)
            if not alive:
                return False, "dead", None, None, "", None
            if CHECK_MODE == "connect":
                return True, "ok", ip, None, "", ms
            source = "judge" if ip else ""
        elif real:
            ip, source = real, "csv"
        if not ip and ECHO and (SECOND or not JUDGE):
            ip, source = fetch(host, port, ECHO, via, login)[1], "echo"
    except BadLogin:
        return False, "login", None, None, "", None
    if not ip and real:
        ip, source = real, "csv"
    if not ip and not ECHO:
        try:
            ip, source = socket.gethostbyname(host), "proxy"
        except OSError:
            pass
    if not ip:
        return False, "noip", None, None, "", ms
    zones = rbl_zones(ip)
    return len(zones) <= RBL_MAX, "ok", ip, zones, source, ms


def ftp_upload(url, data):
    """Upload to ftp://user:pass@host:port/dir/name (ftps:// for TLS). Returns '' or the error."""
    u = urlsplit(url)
    folder, _, name = unquote(u.path[1:]).rpartition("/")
    name = name or os.path.basename(OUTPUT_FILE)
    ftp = ftplib.FTP_TLS() if u.scheme == "ftps" else ftplib.FTP()
    try:
        ftp.connect(u.hostname, u.port or 21, timeout=60)
        ftp.login(unquote(u.username or "anonymous"), unquote(u.password or ""))
        if u.scheme == "ftps":
            ftp.prot_p()
        if folder:
            ftp.cwd(folder)
        ftp.storbinary("STOR %s.tmp" % name, io.BytesIO(data))
        try:
            ftp.rename(name + ".tmp", name)
        except ftplib.error_perm:
            ftp.delete(name)
            ftp.rename(name + ".tmp", name)
        ftp.quit()
        return ""
    except ftplib.all_errors as e:
        return str(e) or e.__class__.__name__
    finally:
        ftp.close()


def write_file(path, text):
    """Write next to the target, then swap: a web server never serves half a list."""
    path = os.path.expanduser(path)
    folder = os.path.dirname(path)
    if folder:
        os.makedirs(folder, exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(text)
    os.replace(tmp, path)


def addr_key(proxy):
    parts = proxy.replace(":", ".").split(".")
    return (0, [int(x) for x in parts]) if all(x.isdigit() for x in parts) else (1, [proxy])


def fmt_time(sec):
    sec = int(sec)
    if sec >= 3600:
        return "%dh %02dm %02ds" % (sec // 3600, sec % 3600 // 60, sec % 60)
    if sec >= 60:
        return "%dm %02ds" % (sec // 60, sec % 60)
    return "%ds" % sec


class Progress:
    """wget-style bar on stderr: percent, bar, counters and ETA."""

    def __init__(self, total):
        self.total = total
        self.done = self.good = 0
        self.start = time.time()
        self.last = 0.0
        self.tty = sys.stderr.isatty()
        self.lock = threading.Lock()

    def step(self, good):
        with self.lock:
            self.done += 1
            self.good += good
            now = time.time()
            if self.done == self.total or now - self.last > 0.1:
                self.last = now
                self.draw()

    def draw(self):
        pct = self.done * 100 // self.total
        elapsed = time.time() - self.start
        eta = elapsed / self.done * (self.total - self.done) if self.done else 0
        w = len(format(self.total, ","))
        tail = " %*s/%s  good %*s  bad %*s  eta %-7s" % (
            w, format(self.done, ","), format(self.total, ","), w, format(self.good, ","),
            w, format(self.done - self.good, ","), fmt_time(eta))
        if not self.tty:
            if pct % 10 == 0 and pct != getattr(self, "_shown", -1):
                self._shown = pct
                sys.stderr.write("%3d%%%s\n" % (pct, tail))
            return
        width = shutil.get_terminal_size((80, 20)).columns
        bar_w = max(10, width - len(tail) - 8)
        filled = bar_w * self.done // self.total
        if filled >= bar_w:
            bar = "=" * bar_w
        elif filled == 0:
            bar = " " * bar_w
        else:
            bar = "=" * (filled - 1) + ">" + " " * (bar_w - filled)
        sys.stderr.write("\r%3d%% [%s]%s" % (pct, bar, tail))
        sys.stderr.flush()


def main():
    if CHECK_MODE not in ("connect", "rbl"):
        sys.exit('CHECK_MODE must be "connect" or "rbl".')
    if CHECK_MODE == "connect" and not JUDGE:
        sys.exit("CHECK_URL is empty: the connect mode needs a site to open through the proxies.")
    rbl = CHECK_MODE == "rbl"
    if rbl and not (ZONES and DNS_SERVERS):
        sys.exit("RBL_LISTS and RBL_DNS must not be empty in the rbl mode.")
    if not rbl:
        mode = "connect (live check only)"
    elif JUDGE:
        mode = "rbl (live check, then RBL)"
    else:
        mode = "rbl without the live check (CHECK_URL is empty)"
    print("slpcheck %s, mode %s%s" % (VERSION, mode, ', set as "both", the old name' if LEGACY_BOTH else ""))
    if not placeholder(PROXY_URL):
        print("slpcheck: loading %s" % list_location(PROXY_URL))
    proxies = load_list(PROXY_URL)
    if not proxies:
        sys.exit("The list is empty.")
    detect_protocols(proxies)
    if rbl:
        with ThreadPoolExecutor(max_workers=len(ZONES)) as pool:
            problems = dict(zip(ZONES, pool.map(zone_problem, ZONES)))
        for zone in ZONES:
            if problems[zone]:
                print("slpcheck: skipping %s: %s" % (zone, problems[zone]))
        ZONES[:] = [z for z in ZONES if not problems[z]]
        if not ZONES:
            sys.exit("No RBL zone answers through RBL_DNS: change RBL_DNS or RBL_LISTS.")
    with_real = sum(1 for p in proxies if META.get(p, ("",))[0])
    without = len(proxies) - with_real
    ends = [endpoint(p) for p in proxies]
    protocols = "/".join(sorted({e[4] for e in ends}))
    logins = sum(1 for e in ends if e[2] is not None)
    how = []
    if JUDGE:
        how.append("%s -> %s, timeout %ds" % (protocols, CHECK_URL, TIMEOUT))
    elif without and ECHO:
        how.append("%s -> %s for the exit IP, timeout %ds" % (protocols, IP_URL, TIMEOUT))
    if rbl:
        how.append("rbl: %d zones, max %d, dns %s" % (len(ZONES), RBL_MAX, ", ".join(DNS_SERVERS)))
    print("slpcheck: %s proxies%s, %d threads, %s" % (
        format(len(proxies), ","), " (%s with a login)" % format(logins, ",") if logins else "", THREADS,
        "; ".join(how)))
    if rbl and JUDGE:
        if not SECOND or JUDGE[2] in ECHO_HOSTS:
            text = "from the %s answer, one request per proxy" % JUDGE[2]
            if SECOND:
                text += "; %s is asked only if that answer holds no IP" % ECHO[2]
        else:
            text = ("from the %s answer if it holds the IP; if not (google.com never shows it), the live proxy"
                    " makes one more request, to %s" % (JUDGE[2], ECHO[2]))
        if with_real:
            text += "; real_ip from the CSV as the last resort"
        if not ECHO:
            text += "; the proxy address when the answer holds no IP"
        print("slpcheck: exit IP for RBL: %s" % text)
    elif rbl:
        plan = []
        if with_real:
            plan.append("real_ip from the CSV for %s proxies" % format(with_real, ","))
        if without and ECHO:
            plan.append("asked through the proxy at %s for %s proxies without real_ip, the ones that do not"
                        " answer get no exit IP" % (ECHO[2], format(without, ",")))
        elif without:
            plan.append("the proxy address for %s proxies without real_ip, because IP_URL is empty"
                        % format(without, ","))
        print("slpcheck: exit IP for RBL: %s" % "; ".join(plan))
    progress = Progress(len(proxies))
    stop = threading.Event()
    rows = {}

    def worker(proxy):
        if stop.is_set():
            return None
        result = check(proxy)
        if stop.is_set():
            return None
        progress.step(result[0])
        return proxy, result

    started = time.time()
    interrupted = False
    pool = ThreadPoolExecutor(max_workers=THREADS)
    futures = [pool.submit(worker, p) for p in proxies]
    pending = set(futures)
    try:
        while pending:
            finished, pending = wait(pending, timeout=0.5, return_when=FIRST_COMPLETED)
            for f in finished:
                done = f.result()
                if done:
                    rows[done[0]] = done[1]
    except KeyboardInterrupt:
        # Ctrl+C: keep what is checked, drop the queue and do not wait for proxies still in TIMEOUT
        interrupted = True
        stop.set()
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        for f in futures:
            f.cancel()
        sys.stderr.write("\nslpcheck: interrupted, %s checked so far, saving them\n" % format(len(rows), ","))
    else:
        pool.shutdown(wait=True)
    elapsed = time.time() - started
    if progress.tty:
        sys.stderr.write("\n")

    order = sorted(rows, key=addr_key)
    good = [p for p in order if rows[p][0]]
    live = [p for p in order if rows[p][1] in ("ok", "noip")]
    checked = [p for p in order if rows[p][1] == "ok"]
    refused = [p for p in order if rows[p][1] == "login"]
    total = len(rows)
    failed = False

    # the report: response time in both modes, the RBL zones in the rbl mode
    if REPORT_FILE:
        if rbl:
            items = sorted(checked, key=lambda p: (-len(rows[p][3]), rows[p][5] or 0))
        else:
            items = sorted(live, key=lambda p: rows[p][5] or 0)
        show_ip = rbl or any(rows[p][2] for p in items)
        show_country = any(META.get(p, ("", ""))[1] for p in items)
        cols = ["proxy"] + (["ms"] if JUDGE else []) + (["exit_ip"] if show_ip else []) \
            + (["country"] if show_country else []) + (["listed", "zones"] if rbl else [])
        lines = [";".join(cols)]
        for p in items:
            _, _, ip, zones, _, ms = rows[p]
            cells = ['"%s"' % p.replace('"', '""') if ";" in p or '"' in p else p]
            if JUDGE:
                cells.append(str(ms))
            if show_ip:
                cells.append(ip or "-")
            if show_country:
                cells.append(META.get(p, ("", ""))[1] or "-")
            if rbl:
                cells += [str(len(zones)), ",".join(zones) or "-"]
            lines.append(";".join(cells))
        write_file(REPORT_FILE, "\n".join(lines) + "\n")

    # every run replaces the output with its own result, an empty file when nothing passed
    text = "\n".join(good) + ("\n" if good else "")
    write_file(OUTPUT_FILE, text)
    ftp_errors = []
    for url in FTP_UPLOAD.split() if good and not interrupted else []:
        error = ftp_upload(url, text.encode())
        u = urlsplit(url)
        ftp_errors.append(("%s%s" % (u.hostname, unquote(u.path) or "/"), error))

    if interrupted:
        print("\nChecked %s of %s proxies in %s (interrupted, %s left unchecked)"
              % (format(total, ","), format(len(proxies), ","), fmt_time(elapsed), format(len(proxies) - total, ",")))
    else:
        print("\nChecked %s proxies in %s" % (format(total, ","), fmt_time(elapsed)))

    def share(n, of):
        return "%s (%.1f%%)" % (format(n, ","), 100.0 * n / max(of, 1))

    if good:
        print("  good: %s  -> %s" % (share(len(good), total), OUTPUT_FILE))
    else:
        print("  good: 0 (0.0%%)  -> %s is empty now" % OUTPUT_FILE)
        failed = True
    bad = total - len(good)
    if not rbl and not refused:
        print("  bad:  %s" % share(bad, total))
    elif not rbl:
        print("  bad:  %s: dead %s, login refused %s"
              % (share(bad, total), format(bad - len(refused), ","), format(len(refused), ",")))
    else:
        parts = ["dead %s" % format(total - len(live) - len(refused), ",")] if JUDGE else []
        if refused:
            parts.append("login refused %s" % format(len(refused), ","))
        parts.append("no exit IP %s" % format(len(live) - len(checked), ","))
        parts.append("over RBL_MAX %s" % format(len(checked) - len(good), ","))
        print("  bad:  %s: %s" % (share(bad, total), ", ".join(parts)))
    if refused and len(refused) == sum(1 for p in rows if endpoint(p)[2] is not None):
        print("  login: every proxy with a login refused it, check the user and password in the list")
    speeds = sorted(rows[p][5] for p in live if rows[p][5] is not None)
    if speeds:
        print("  speed: fastest %s ms, median %s ms, slowest %s ms"
              % (format(speeds[0], ","), format(speeds[len(speeds) // 2], ","), format(speeds[-1], ",")))
    if rbl:
        listed = Counter(min(len(rows[p][3]), 3) for p in checked)
        print("  rbl:  of %s checked: clean %s, in 1 zone %s, in 2 zones %s, in 3+ zones %s"
              % ((format(len(checked), ","),) + tuple(share(listed[n], len(checked)) for n in range(4))))
        per_zone = Counter(z for p in checked for z in rows[p][3])
        for zone in ZONES:
            note = "  (no answer for %s IPs)" % format(RBL_FAILED[zone], ",") if RBL_FAILED[zone] else ""
            print("        %-24s %s%s" % (zone, share(per_zone[zone], len(checked)), note))
        sources = Counter(rows[p][4] for p in checked)
        names = {"judge": JUDGE[2] if JUDGE else "", "echo": ECHO[2] if ECHO else "",
                 "csv": "CSV real_ip", "proxy": "proxy address"}
        if checked:
            print("  ip:   exit IP from " + ", ".join("%s %s" % (names[k], share(sources[k], len(checked)))
                                                   for k in ("judge", "echo", "csv", "proxy") if sources[k]))
        by_proxy = [p for p in checked if rows[p][4] == "proxy"]
        hosts = {endpoint(p)[0] for p in by_proxy}
        if len(by_proxy) >= 20 and len(hosts) * 5 <= len(by_proxy):
            print("  WARNING: %s proxies were looked up by %d shared addresses, most likely backconnect\n"
                  "           gateways, not exit IPs: IP_URL is empty. Set IP_URL = \"https://api.ipify.org\"."
                  % (format(len(by_proxy), ","), len(hosts)))
    if REPORT_FILE:
        print("  report: %s" % REPORT_FILE)
    for where, error in ftp_errors:
        print("  ftp:  %s %s" % (where, "FAILED: " + error if error else "uploaded"))
        failed = failed or bool(error)
    if interrupted and FTP_UPLOAD.split():
        print("  ftp:  skipped, the run was interrupted")
    elif not good and FTP_UPLOAD.split():
        print("  ftp:  skipped, no good proxies to upload")
    print("  time: %s, %.1f proxies/s" % (time.strftime("%Y-%m-%d %H:%M:%S"), total / max(elapsed, 0.001)))
    if interrupted:
        # threads still waiting for dead proxies would hold the exit for up to TIMEOUT: leave now
        sys.stdout.flush()
        sys.stderr.flush()
        os._exit(130)
    if failed:
        sys.exit(1)


def set_config(pairs):
    """--set KEY=VALUE ...: write the values into the CONFIG block of this file and print the block."""
    path = os.path.abspath(__file__)
    with open(path, encoding="utf-8") as f:
        lines = f.read().split("\n")
    first = next(i for i, line in enumerate(lines) if line.startswith("# =") and "CONFIG" in line)
    last = next(i for i in range(first + 1, len(lines)) if lines[i].startswith("# ="))
    found = {}
    for i in range(first + 1, last):
        m = re.match(r'([A-Z_]+) = ("(?:[^"\\]|\\.)*"|-?\d+)(\s+#.*)?$', lines[i])
        if m:
            found[m.group(1)] = (i, m)
    for pair in pairs:
        key, sep, value = pair.partition("=")
        if key not in found or not sep:
            sys.exit("Use KEY=VALUE, where KEY is one of: %s" % ", ".join(found))
        i, m = found[key]
        if m.group(2).startswith('"'):
            literal = '"%s"' % value.replace("\\", "\\\\").replace('"', '\\"')
        elif re.match(r"-?\d+$", value):
            literal = value
        else:
            sys.exit("%s takes a whole number, got %r" % (key, value))
        head = "%s = %s" % (key, literal)
        lines[i] = head + (" " * max(2, 54 - len(head)) + m.group(3).strip() if m.group(3) else "")
    with open(path, "w", encoding="utf-8") as conf:
        conf.write("\n".join(lines))
    print("\n".join(lines[first:last + 1]))


if __name__ == "__main__":
    if sys.argv[1:2] == ["--set"]:
        set_config(sys.argv[2:])
    elif sys.argv[1:] == ["--version"]:
        print("slpcheck %s" % VERSION)
    elif len(sys.argv) > 1:
        sys.exit("usage: python3 slpcheck.py              run the check\n"
                 "       python3 slpcheck.py --set KEY=VALUE ...   save settings into CONFIG\n"
                 "       python3 slpcheck.py --set          show the settings\n"
                 "       python3 slpcheck.py --version      show the version")
    else:
        try:
            main()
        except KeyboardInterrupt:
            sys.stderr.write("\nslpcheck: interrupted before the check started, nothing saved\n")
            sys.exit(130)