Invalid hits still recorded by provider as false positive netscan / scansnarf-ng.

Proposed by our dear Claude:

import ipaddress
import requests
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
from typing import List, Dict, Set, Tuple

logger = logging.getLogger(__name__)

class TunnelProxyRoutability:
    def __init__(self, proxy1: str, proxy2: str):
        self.proxy1 = proxy1
        self.proxy2 = proxy2
        self.routable_addresses: Set[str] = set()
        self.invalid_addresses: Set[str] = set()
        self.cloudflare_addresses: Set[str] = set()

    @staticmethod
    def is_valid_ip(ip: str) -> bool:
        try:
            ipaddress.ip_address(ip)
            return True
        except ValueError:
            return False

    @staticmethod
    def is_private_ip(ip: str) -> bool:
        return ipaddress.ip_address(ip).is_private

    def make_tunneled_request(self, url: str, timeout: int = 10) -> Tuple[bool, requests.Response]:
        proxies = {
            'http': self.proxy1,
            'https': self.proxy1,
        }
        try:
            with requests.Session() as session:
                session.proxies = proxies
                response = session.get(
                    url,
                    headers={'Proxy-Authorization': f'Basic {self.proxy2}'},
                    timeout=timeout
                )
            return True, response
        except requests.RequestException as e:
            logger.error(f"Error making tunneled request to {url}: {str(e)}")
            return False, None

    def is_cloudflare(self, ip: str) -> bool:
        success, response = self.make_tunneled_request(f"http://{ip}")
        if success and response:
            return 'cloudflare' in response.headers.get('Server', '').lower()
        return False

    def is_routable(self, ip: str) -> bool:
        if not self.is_valid_ip(ip) or self.is_private_ip(ip):
            return False

        if self.is_cloudflare(ip):
            logger.debug(f"{ip} is associated with Cloudflare")
            self.cloudflare_addresses.add(ip)
            return False

        success, response = self.make_tunneled_request(f"http://{ip}")
        return success and response and response.status_code == 200

    def validate_ip_list(self, ip_list: List[str], max_workers: int = 50) -> Dict[str, bool]:
        results = {}
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(self.is_routable, ip): ip for ip in ip_list}

            with tqdm(total=len(ip_list), desc="Validating IPs") as pbar:
                for future in as_completed(futures):
                    ip = futures[future]
                    try:
                        is_routable = future.result()
                        results[ip] = is_routable
                        if is_routable:
                            self.routable_addresses.add(ip)
                        elif ip not in self.cloudflare_addresses:
                            self.invalid_addresses.add(ip)
                    except Exception as exc:
                        logger.error(f"{ip} generated an exception: {exc}")
                        results[ip] = False
                        self.invalid_addresses.add(ip)
                    finally:
                        pbar.update(1)

        return results

# Example usage
if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    proxy1 = "http://proxy1.example.com:8080"
    proxy2 = "user:pass@proxy2.example.com:8080"
    routability = TunnelProxyRoutability(proxy1, proxy2)
    test_ips = ["8.8.8.8", "1.1.1.1", "172.217.167.78"]
    results = routability.validate_ip_list(test_ips)
    print("Results:", results)
    print("Routable:", routability.routable_addresses)
    print("Invalid:", routability.invalid_addresses)
    print("Cloudflare:", routability.cloudflare_addresses)```
phileasme / ProxyFilter

Invalid hits still recorded by provider as false positive netscan / scansnarf-ng. #1
