tools/proxy_feasibility.py

import asyncio
import argparse
import socket
import sys
import json
import time
from urllib.parse import urlparse
from typing import List, Dict, Optional, Tuple

# Force UTF-8 output for Windows consoles
if sys.platform == "win32":
    sys.stdout.reconfigure(encoding='utf-8')

try:
    from playwright.async_api import async_playwright, ProxySettings
except ImportError:
    print("Error: playwright is not installed. Please run: pip install playwright")
    sys.exit(1)

# =============================================================================
# CONFIGURATION
# =============================================================================

DEFAULT_URL = "https://www.mavi.com/erkek/c/2"
IP_CHECK_SERVICE = "https://api64.ipify.org?format=json" # Dual stack, returns IP
TIMEOUT_MS = 30000

# Common block indicators in page title or content
BLOCK_INDICATORS = [
    "Attention Required",
    "Cloudflare",
    "Access Denied",
    "403 Forbidden",
    "Just a moment...",
    "Human Verification",
    "Challenge",
    "Security Check"
]

# =============================================================================
# CLASSES
# =============================================================================

class ProxyTester:
    def __init__(self, url: str, headless: bool = True):
        self.target_url = url
        self.headless = headless
        self.domain = urlparse(url).netloc
        self.results = []

    def resolve_dns(self) -> Dict[str, bool]:
        """Check if domain resolves to IPv4 and IPv6."""
        print(f"\n🔍 DNS Analysis for: {self.domain}")
        dns_info = {"ipv4": False, "ipv6": False, "details": []}
        
        try:
            # Check IPv4 (A record)
            ipv4s = socket.getaddrinfo(self.domain, None, socket.AF_INET)
            if ipv4s:
                dns_info["ipv4"] = True
                print(f"  ✅ IPv4 (A) Records found: {[x[4][0] for x in ipv4s]}")
            else:
                print(f"  ❌ No IPv4 (A) Records found.")
        except socket.gaierror:
             print(f"  ❌ IPv4 Resolution Failed.")

        try:
            # Check IPv6 (AAAA record)
            ipv6s = socket.getaddrinfo(self.domain, None, socket.AF_INET6)
            if ipv6s:
                dns_info["ipv6"] = True
                print(f"  ✅ IPv6 (AAAA) Records found: {[x[4][0] for x in ipv6s]}")
            else:
                print(f"  ❌ No IPv6 (AAAA) Records found. (IPv6-only proxies will fail unless they use NAT64)")
        except socket.gaierror:
            print(f"  ❌ IPv6 Resolution Failed.")
            
        return dns_info

    async def check_proxy(self, proxy_str: str, wait_on_page: int = 0) -> Dict:
        """Test a specific proxy against the target."""
        
        # Parse proxy string format: protocol://user:pass@ip:port or ip:port
        # Playwright expects a dictionary
        if "://" not in proxy_str and ":" in proxy_str:
            # Assume http if not specified but has port
            proxy_str = f"http://{proxy_str}"

        proxy_config = {"server": proxy_str}
        
        print(f"\nTesting Proxy: {proxy_str}")
        result = {
            "proxy": proxy_str,
            "ip_detected": None,
            "type": "Unknown",
            "target_access": False,
            "status": "Failed",
            "latency": 0,
            "block_reason": None
        }

        async with async_playwright() as p:
            # Launch browser with proxy
            try:
                browser = await p.chromium.launch(headless=self.headless, proxy=proxy_config)
                context = await browser.new_context(ignore_https_errors=True)
                page = await context.new_page()

                # Step 1: Check IP and Type (v4/v6) via Echo Service
                # Reduced timeout for IP check to fail fast
                try:
                    response = await page.goto(IP_CHECK_SERVICE, timeout=10000)
                    if response.ok:
                        content = await page.content()
                        # Simple extraction from json
                        import re
                        ip_match = re.search(r'ip":\s*"([^"]+)"', content)
                        if ip_match:
                            detected_ip = ip_match.group(1)
                            result["ip_detected"] = detected_ip
                            result["type"] = "IPv6" if ":" in detected_ip else "IPv4"
                            print(f"  ✅ Proxy Connection OK. Outbound IP: {detected_ip} ({result['type']})")
                    else:
                        print(f"  ⚠️  Could not reach IP check service (HTTP {response.status})")
                except Exception as e:
                    print(f"  ⚠️  Proxy connection check failed: {str(e)[:100]}")
                    # Continue to target test anyway, proxy might just ban the echo service

                # Step 2: Test Target URL
                print(f"  🚀  Navigating to target: {self.target_url}")
                start_target = time.time()
                try:
                    response = await page.goto(self.target_url, timeout=TIMEOUT_MS, wait_until="domcontentloaded")
                    latency = (time.time() - start_target) * 1000
                    result["latency"] = latency
                    
                    if wait_on_page > 0:
                        print(f"  ⏳ Waiting {wait_on_page}s on page (5-second rule)...")
                        await page.wait_for_timeout(wait_on_page * 1000)

                    title = await page.title()
                    content_text = await page.inner_text("body")
                    
                    # Check for blocks
                    is_blocked = False
                    block_reason = None
                    
                    if response.status in [403, 429]:
                        is_blocked = True
                        block_reason = f"HTTP {response.status}"
                    
                    if not is_blocked:
                        for indicator in BLOCK_INDICATORS:
                            if indicator.lower() in title.lower():
                                is_blocked = True
                                block_reason = f"Title: {indicator}"
                                break
                            if indicator in content_text[:1000]: # Check first 1000 chars
                                is_blocked = True
                                block_reason = f"Content: {indicator}"
                                break
                    
                    if is_blocked:
                        result["status"] = "Blocked"
                        result["block_reason"] = block_reason
                        print(f"  ⛔ BLOCKED by Target. Reason: {block_reason}")
                        # Take screenshot of block
                        await page.screenshot(path=f"block_{int(time.time())}.png")
                    else:
                        result["status"] = "Success"
                        result["target_access"] = True
                        print(f"  ✅ SUCCESS! Accessed Target. Title: {title[:30]}...")

                except Exception as e:
                    result["status"] = "Error"
                    result["block_reason"] = str(e)
                    print(f"  ❌ Error accessing target: {str(e)[:100]}")

                await browser.close()

            except Exception as e:
                result["status"] = "Connection Failed"
                print(f"  ❌ Proxy Connection/Launch Failed: {str(e)[:100]}")

        self.results.append(result)
        return result

    async def run_suite(self, proxy_list: List[str], loop: bool = False, delay: int = 5):
        """Run tests on all proxies"""
        # First, DNS check
        dns_info = self.resolve_dns()
        
        if not dns_info["ipv6"]:
            print("\n⚠️  WARNING: Target domain has no IPv6 (AAAA) records.")
            print("   pure IPv6 proxies will likely FAIL unless they support NAT64.")

        if loop:
            print(f"\n🔁 Starting Infinite Loop Mode (Interval: {delay}s)")
            print(f"   Proxies: {len(proxy_list)}")
            print("   Press Ctrl+C to stop...")
            
            idx = 0
            while True:
                proxy = proxy_list[idx % len(proxy_list)]
                print(f"\n--- Loop #{idx + 1} using {proxy} ---")
                
                # Use 5 second rule on page
                await self.check_proxy(proxy, wait_on_page=5)
                
                print(f"Waiting {delay}s before next test...")
                await asyncio.sleep(delay)
                idx += 1
        else:
            print(f"\nStarting Proxy Tests on {len(proxy_list)} proxies...")
            print("="*60)
            
            for proxy in proxy_list:
                await self.check_proxy(proxy, wait_on_page=5)
                
            self.print_summary()

    def print_summary(self):
        """Print final report"""
        print("\n" + "="*60)
        print("📊 PROXY FEASIBILITY REPORT")
        print("="*60)
        print(f"Target: {self.target_url}")
        print(f"Proxies Tested: {len(self.results)}")
        
        ipv4_success = 0
        ipv6_success = 0
        ipv4_total = 0
        ipv6_total = 0
        
        print("\nDetailed Results:")
        print(f"{'Proxy IP/Type':<25} | {'Status':<15} | {'Latency':<10} | {'Note'}")
        print("-" * 75)
        
        for r in self.results:
            p_type = r.get("type", "?")
            status = r.get("status", "Unknown")
            latency = f"{int(r.get('latency', 0))}ms"
            note = r.get("block_reason") or ""
            
            if p_type == "IPv4": ipv4_total += 1
            if p_type == "IPv6": ipv6_total += 1
            
            if status == "Success":
                if p_type == "IPv4": ipv4_success += 1
                if p_type == "IPv6": ipv6_success += 1
                
            display_ip = r.get("ip_detected") or "Unknown"
            print(f"{display_ip[:20]:<20} ({p_type}) | {status:<15} | {latency:<10} | {note}")

        print("-" * 75)
        
        # Summary Stats
        if ipv4_total > 0:
            v4_rate = (ipv4_success/ipv4_total)*100
            print(f"IPv4 Feasibility: {ipv4_success}/{ipv4_total} ({v4_rate:.1f}%)")
        else:
            print("IPv4 Feasibility: N/A (None tested)")

        if ipv6_total > 0:
            v6_rate = (ipv6_success/ipv6_total)*100
            print(f"IPv6 Feasibility: {ipv6_success}/{ipv6_total} ({v6_rate:.1f}%)")
        else:
            print("IPv6 Feasibility: N/A (None tested)")

# =============================================================================
# ENTRY POINT
# =============================================================================

async def main():
    parser = argparse.ArgumentParser(description="Proxy IPv4/IPv6 Feasibility Tester")
    parser.add_argument("--url", default=DEFAULT_URL, help="Target URL to test")
    parser.add_argument("--proxies", help="Comma separated list of queries OR path to file")
    parser.add_argument("--headless", action="store_true", default=True, help="Run headless")
    parser.add_argument("--show-browser", action="store_false", dest="headless", help="Show browser")
    parser.add_argument("--loop", action="store_true", help="Run in infinite loop")
    parser.add_argument("--delay", type=int, default=5, help="Delay between loops (seconds)")

    args = parser.parse_args()
    
    proxy_list = []
    if args.proxies:
        if args.proxies.endswith(".txt"):
            try:
                with open(args.proxies, 'r') as f:
                    proxy_list = [line.strip() for line in f if line.strip()]
            except FileNotFoundError:
                print(f"Proxy file not found: {args.proxies}")
                return
        else:
            proxy_list = args.proxies.split(",")
    else:
        # Default behavior: If no proxies, just do DNS check and maybe local connection check?
        # Or ask user to provide them
        print("No proxies provided via --proxies.")
        print("Running DNS feasibility check only.")
    
    tester = ProxyTester(args.url, headless=args.headless)
    
    if not proxy_list:
        tester.resolve_dns()
        print("\nTo test actual proxies, provide them with --proxies 'http://user:pass@ip:port,...'")
    else:
        await tester.run_suite(proxy_list, loop=args.loop, delay=args.delay)

if __name__ == "__main__":
    asyncio.run(main())