web-check.py

#!/usr/bin/env python3
"""
argo-ip-radar: High-precision network diagnostic and IP optimization suite.
Supports both VLESS WebSocket (WS) and SplitHTTP (XHTTP) configurations.
Tests latency, real download/upload speed, best-practice TLS fingerprints,
and relevant ALPN combinations against Cloudflare CDN edge IPs.
"""

import argparse
import ipaddress
import json
import os
import signal
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime

from ip_gen import generate_weighted_ips
from radar_core import (
    BEST_PRACTICE_FPS,
    ALPN_COMBINATIONS_WS,
    ALPN_COMBINATIONS_XHTTP,
    benchmark_alpn_matrix,
    benchmark_fp_matrix,
    benchmark_single_ip,
    build_vless_url,
    check_xray_available,
    ensure_xray_binary,
    parse_vless_url,
)

# --- Defaults & Paths ---
SUBNETS_FILE = "subnets.json"
DEFAULT_IPS_FILE = "ips.txt"
RESULTS_DIR = "results"
DEFAULT_WORKERS = 16
DEFAULT_IP_COUNT = 100
DEFAULT_TOP_BENCHMARK = 5
MIN_SUBNET_WEIGHT = 0.1
MAX_SUBNET_WEIGHT = 10.0
BASE_FAIL_PENALTY = 0.9

# --- Colors ---
GREEN = "\033[92m"
RED = "\033[91m"
CYAN = "\033[96m"
YELLOW = "\033[93m"
PINK = "\033[95m"
BOLD = "\033[1m"
DIM = "\033[2m"
RESET = "\033[0m"


def disable_colors():
    global GREEN, RED, CYAN, YELLOW, PINK, BOLD, DIM, RESET
    GREEN = RED = CYAN = YELLOW = PINK = BOLD = DIM = RESET = ""


# --- Global State ---
stop_requested = False
state_lock = threading.RLock()
subnet_stats = {}
parsed_subnets = []


def signal_handler(sig, frame):
    global stop_requested
    stop_requested = True
    print(f"\n{YELLOW}[!] Interrupt received. Stopping gracefully...{RESET}")


signal.signal(signal.SIGINT, signal_handler)


# --- Subnet Management ---

def clamp_subnet_weight(weight):
    return round(max(MIN_SUBNET_WEIGHT, min(MAX_SUBNET_WEIGHT, weight)), 3)


def normalize_subnet_stats(raw_data):
    normalized = {}
    for cidr, value in raw_data.items():
        if isinstance(value, dict):
            normalized[cidr] = {
                "weight": clamp_subnet_weight(float(value.get("weight", MIN_SUBNET_WEIGHT))),
                "success_count": int(value.get("success_count", 0)),
                "fail_count": int(value.get("fail_count", 0)),
                "avg_speed": round(float(value.get("avg_speed", 0.0)), 2),
            }
        else:
            normalized[cidr] = {
                "weight": clamp_subnet_weight(float(value)),
                "success_count": 0,
                "fail_count": 0,
                "avg_speed": 0.0,
            }
    return normalized


def load_subnets(subnets_file=SUBNETS_FILE):
    global subnet_stats, parsed_subnets
    if not os.path.exists(subnets_file):
        return
    with open(subnets_file, "r") as f:
        subnet_stats = normalize_subnet_stats(json.load(f))
    parsed_subnets = [(cidr, ipaddress.ip_network(cidr)) for cidr in subnet_stats]


def record_subnet_result(ip: str, success: bool, speed_kbps: float = 0.0):
    global subnet_stats, parsed_subnets
    with state_lock:
        try:
            ip_obj = ipaddress.ip_address(ip)
        except ValueError:
            return

        for cidr, net in parsed_subnets:
            if ip_obj in net:
                stats = subnet_stats[cidr]
                if success:
                    prev_count = stats["success_count"]
                    next_count = prev_count + 1
                    stats["success_count"] = next_count
                    if speed_kbps > 0:
                        stats["avg_speed"] = round(
                            ((stats["avg_speed"] * prev_count) + speed_kbps) / next_count, 2
                        )
                        stats["weight"] = clamp_subnet_weight(stats["weight"] + (speed_kbps / 1000.0))
                    else:
                        stats["weight"] = clamp_subnet_weight(stats["weight"] + 0.05)
                else:
                    stats["fail_count"] += 1
                    stats["weight"] = clamp_subnet_weight(stats["weight"] * BASE_FAIL_PENALTY)
                break


def save_subnets_to_disk(subnets_file=SUBNETS_FILE):
    global subnet_stats
    if not subnet_stats:
        return
    with state_lock:
        try:
            with open(subnets_file, "w") as f:
                json.dump(subnet_stats, f, indent=2)
        except Exception as e:
            print(f"{RED}Error saving subnet stats: {e}{RESET}")


# --- CLI Parser ---

def parse_args():
    parser = argparse.ArgumentParser(
        description="argo-ip-radar: Test Cloudflare IPs for VLESS WS & XHTTP configs with ALPN, FP, and Speed benchmarking."
    )
    parser.add_argument(
        "config",
        nargs="?",
        help="vless:// URL configuration (xhttp or ws). Can also be provided via ARGO_IP_RADAR_CONFIG environment variable.",
    )
    parser.add_argument(
        "-w", "--workers",
        type=int,
        default=DEFAULT_WORKERS,
        help=f"Number of concurrent testing workers (default: {DEFAULT_WORKERS}).",
    )
    parser.add_argument(
        "-n", "--count",
        type=int,
        default=None,
        help=f"Number of Cloudflare IPs to randomly generate and test (default: {DEFAULT_IP_COUNT} if ips file not present).",
    )
    parser.add_argument(
        "--ips",
        default=DEFAULT_IPS_FILE,
        help=f"Path to input IPs file (default: {DEFAULT_IPS_FILE}).",
    )
    parser.add_argument(
        "--subnets",
        default=SUBNETS_FILE,
        help=f"Path to subnets.json file (default: {SUBNETS_FILE}).",
    )
    parser.add_argument(
        "--top",
        type=int,
        default=DEFAULT_TOP_BENCHMARK,
        help=f"Number of top responsive IPs to benchmark for Speed, ALPN, and Fingerprint (default: {DEFAULT_TOP_BENCHMARK}).",
    )
    parser.add_argument(
        "--skip-speed",
        action="store_true",
        help="Skip throughput (download/upload) speed testing and only test ping.",
    )
    parser.add_argument(
        "--skip-alpn",
        action="store_true",
        help="Skip ALPN combination benchmarking on top IPs.",
    )
    parser.add_argument(
        "--skip-fp",
        action="store_true",
        help="Skip fingerprint benchmarking on top IPs.",
    )
    parser.add_argument(
        "--dl-bytes",
        type=int,
        default=250000,
        help="Number of bytes to download during speed test (default: 250000 = ~250 KB).",
    )
    parser.add_argument(
        "--ul-bytes",
        type=int,
        default=100000,
        help="Number of bytes to upload during speed test (default: 100000 = ~100 KB).",
    )
    parser.add_argument(
        "--ping-timeout",
        type=float,
        default=2.5,
        help="Timeout in seconds for reachability ping (default: 2.5s).",
    )
    parser.add_argument(
        "--xray-path",
        default=None,
        help="Custom path to Xray binary (overrides automatic resolution and ARGO_XRAY_PATH).",
    )
    parser.add_argument(
        "--no-color",
        action="store_true",
        help="Disable ANSI terminal colors.",
    )
    return parser.parse_args()


# --- Main Application Workflow ---

def main():
    global stop_requested
    args = parse_args()

    if args.no_color:
        disable_colors()

    script_dir = os.path.dirname(os.path.abspath(__file__))
    last_print = [0.0]

    def download_progress(downloaded, total):
        now = time.time()
        if now - last_print[0] < 0.25 and downloaded != total:
            return
        last_print[0] = now
        if total:
            pct = (downloaded / total) * 100.0
            mb_down = downloaded / (1024 * 1024)
            mb_tot = total / (1024 * 1024)
            sys.stdout.write(f"\r{CYAN}[*] Downloading Xray core: {pct:5.1f}% ({mb_down:.1f}/{mb_tot:.1f} MB)...{RESET}")
        else:
            mb_down = downloaded / (1024 * 1024)
            sys.stdout.write(f"\r{CYAN}[*] Downloading Xray core: {mb_down:.1f} MB...{RESET}")
        sys.stdout.flush()

    try:
        xray_binary = ensure_xray_binary(
            target_dir=script_dir,
            allow_download=True,
            fallback_system=True,
            progress_callback=download_progress,
            custom_path=args.xray_path,
        )
    except Exception as e:
        sys.stdout.write("\n")
        print(f"{RED}Error: {e}{RESET}")
        sys.exit(1)

    raw_config = args.config or os.environ.get("ARGO_IP_RADAR_CONFIG")
    if not raw_config:
        print(f"{RED}Error: No VLESS configuration provided.{RESET}")
        print("Usage: python3 web-check.py 'vless://...' [options]")
        sys.exit(1)

    try:
        vless_cfg = parse_vless_url(raw_config)
    except Exception as e:
        print(f"{RED}Error parsing VLESS configuration: {e}{RESET}")
        sys.exit(1)

    # Load subnets
    load_subnets(args.subnets)

    # Determine IPs to test
    ip_list = []
    if args.count is not None or not os.path.exists(args.ips):
        count_to_gen = args.count if args.count is not None else DEFAULT_IP_COUNT
        print(f"{CYAN}[*] Generating {count_to_gen} Cloudflare IPs from {args.subnets}...{RESET}")
        ip_list = generate_weighted_ips(count=count_to_gen, output_file=args.ips, config_file=args.subnets)
    else:
        with open(args.ips, "r") as f:
            ip_list = [line.strip() for line in f if line.strip() and not line.startswith("#")]
        if args.count is not None and args.count < len(ip_list):
            ip_list = ip_list[:args.count]

    total_ips = len(ip_list)
    print(f"\n{BOLD}{CYAN}======================================================{RESET}")
    print(f"{BOLD}{CYAN}                 argo-ip-radar Scanner                {RESET}")
    print(f"{BOLD}{CYAN}======================================================{RESET}")
    print(f"Transport   : {YELLOW}{vless_cfg['type'].upper()}{RESET}")
    print(f"SNI / Host  : {vless_cfg['sni']} / {vless_cfg['host']}")
    print(f"Path        : {vless_cfg['path']}")
    print(f"ALPN        : {','.join(vless_cfg['alpn'])}")
    print(f"Fingerprint : {vless_cfg['fp']}")
    print(f"Workers     : {args.workers}")
    print(f"Total IPs   : {total_ips}")
    print(f"{BOLD}{CYAN}------------------------------------------------------{RESET}\n")

    # ----------------------------------------------------
    # PHASE 1: Rapid Concurrency Reachability Screening
    # ----------------------------------------------------
    print(f"{BOLD}[Phase 1] Screening {total_ips} IPs for reachability and latency...{RESET}")
    start_time = time.time()
    alive_results = []
    checked_count = 0

    def screen_worker(worker_id, ip):
        if stop_requested:
            return None
        res = benchmark_single_ip(
            cfg=vless_cfg,
            ip=ip,
            worker_id=worker_id,
            ping_timeout=args.ping_timeout,
            do_speed_test=False,
        )
        record_subnet_result(ip, res["success"], speed_kbps=0.0)
        return res

    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        futures = {
            executor.submit(screen_worker, idx % args.workers, ip): ip
            for idx, ip in enumerate(ip_list)
        }
        for future in as_completed(futures):
            if stop_requested:
                for f in futures:
                    f.cancel()
                break
            res = future.result()
            checked_count += 1
            if res and res["success"]:
                alive_results.append(res)
                status_color = GREEN
                status_text = f"ALIVE ({res['latency_ms']:.1f}ms)"
            else:
                status_color = RED
                err = res["error"] if res else "interrupted"
                status_text = f"FAIL ({err})"

            # Inline progress update
            pct = (checked_count / total_ips) * 100.0
            sys.stdout.write(
                f"\r{DIM}[{checked_count}/{total_ips} - {pct:4.1f}%]{RESET} "
                f"Alive: {GREEN}{len(alive_results)}{RESET} | "
                f"Latest: {futures[future]:<15} -> {status_color}{status_text:<20}{RESET}"
            )
            sys.stdout.flush()

    sys.stdout.write("\n")
    p1_duration = time.time() - start_time
    print(
        f"\n{BOLD}Phase 1 Completed in {p1_duration:.1f}s:{RESET} "
        f"{GREEN}{len(alive_results)}/{total_ips} IPs responsive.{RESET}\n"
    )

    if not alive_results:
        print(f"{RED}[!] No responsive Cloudflare IPs found. Check network or SNI / config.{RESET}")
        save_subnets_to_disk(args.subnets)
        sys.exit(0)

    # Sort alive candidates by initial latency
    alive_results.sort(key=lambda x: x["latency_ms"])
    top_candidates = alive_results[: min(len(alive_results), args.top)]

    # ----------------------------------------------------
    # PHASE 2: True Download & Upload Speed Benchmarking
    # ----------------------------------------------------
    speed_results = []
    if not args.skip_speed:
        print(f"{BOLD}[Phase 2] Measuring real Download & Upload throughput on top {len(top_candidates)} candidates...{RESET}")
        for idx, item in enumerate(top_candidates, 1):
            if stop_requested:
                break
            ip = item["ip"]
            print(f"  {CYAN}--> [{idx}/{len(top_candidates)}] Testing {ip} (Ping: {item['latency_ms']:.1f}ms)...{RESET}")
            res = benchmark_single_ip(
                cfg=vless_cfg,
                ip=ip,
                worker_id=0,
                ping_timeout=args.ping_timeout,
                do_speed_test=True,
                dl_bytes=args.dl_bytes,
                ul_bytes=args.ul_bytes,
            )
            if res["success"]:
                dl_mbps = res["download_kbps"] / 1024.0
                ul_mbps = res["upload_kbps"] / 1024.0
                print(
                    f"      {GREEN}PASS{RESET} - Download: {BOLD}{res['download_kbps']:.1f} kbps ({dl_mbps:.2f} Mbps){RESET} | "
                    f"Upload: {BOLD}{res['upload_kbps']:.1f} kbps ({ul_mbps:.2f} Mbps){RESET} | Score: {res['score']:.1f}"
                )
                record_subnet_result(ip, True, speed_kbps=res["download_kbps"])
                speed_results.append(res)
            else:
                print(f"      {RED}FAIL{RESET} - Speed test failed: {res['error']}")
        print("")
    else:
        speed_results = top_candidates

    if not speed_results:
        speed_results = top_candidates

    speed_results.sort(key=lambda x: -x["score"])
    best_candidate_ip = speed_results[0]["ip"]

    # ----------------------------------------------------
    # PHASE 3: ALPN Combination & Fingerprint Testing
    # ----------------------------------------------------
    recommended_alpn = vless_cfg["alpn"]
    recommended_fp = vless_cfg["fp"]

    if not args.skip_alpn and not stop_requested:
        print(f"{BOLD}[Phase 3a] Benchmarking all relevant ALPN combinations on candidate {best_candidate_ip}...{RESET}")
        alpn_res = benchmark_alpn_matrix(
            cfg=vless_cfg,
            ip=best_candidate_ip,
            worker_id=0,
            dl_bytes=args.dl_bytes,
            ul_bytes=args.ul_bytes,
        )
        if alpn_res:
            print(f"\n  {BOLD}{'ALPN Combination':<24} {'Latency':<10} {'Download':<14} {'Upload':<14} {'Score':<8}{RESET}")
            print(f"  {'-'*70}")
            for r in alpn_res:
                alpn_str = ",".join(r["alpn"])
                is_win = " *" if r == alpn_res[0] else ""
                print(
                    f"  {alpn_str:<24} {r['latency_ms']:>6.1f}ms   {r['download_kbps']:>8.1f}kbps  "
                    f"{r['upload_kbps']:>8.1f}kbps  {r['score']:>6.1f}{GREEN}{is_win}{RESET}"
                )
            recommended_alpn = alpn_res[0]["alpn"]
            print(f"\n  {GREEN}>> Optimal ALPN Combination: {BOLD}{','.join(recommended_alpn)}{RESET}\n")

    if not args.skip_fp and not stop_requested:
        print(f"{BOLD}[Phase 3b] Benchmarking best-practice TLS fingerprints on {best_candidate_ip}...{RESET}")
        fp_res = benchmark_fp_matrix(
            cfg=vless_cfg,
            ip=best_candidate_ip,
            worker_id=0,
            alpn=recommended_alpn,
            dl_bytes=args.dl_bytes,
            ul_bytes=args.ul_bytes,
        )
        if fp_res:
            print(f"\n  {BOLD}{'Fingerprint':<14} {'Latency':<10} {'Download':<14} {'Upload':<14} {'Score':<8}{RESET}")
            print(f"  {'-'*60}")
            for r in fp_res:
                is_win = " *" if r == fp_res[0] else ""
                print(
                    f"  {r['fp']:<14} {r['latency_ms']:>6.1f}ms   {r['download_kbps']:>8.1f}kbps  "
                    f"{r['upload_kbps']:>8.1f}kbps  {r['score']:>6.1f}{GREEN}{is_win}{RESET}"
                )
            recommended_fp = fp_res[0]["fp"]
            print(f"\n  {GREEN}>> Optimal Fingerprint: {BOLD}{recommended_fp}{RESET}\n")

    # ----------------------------------------------------
    # PHASE 4: Final Recommendation & URL Generation
    # ----------------------------------------------------
    best_entry = speed_results[0]
    final_vless_url = build_vless_url(
        cfg=vless_cfg,
        new_ip=best_entry["ip"],
        new_alpn=recommended_alpn,
        new_fp=recommended_fp,
        remark=f"Argo-{vless_cfg['type'].upper()}-Radar-{best_entry['ip']}",
    )

    print(f"{BOLD}{GREEN}======================================================================{RESET}")
    print(f"{BOLD}{GREEN}                        RADAR RECOMMENDATION                          {RESET}")
    print(f"{BOLD}{GREEN}======================================================================{RESET}")
    print(f"  {BOLD}Best IP Address      :{RESET} {CYAN}{best_entry['ip']}{RESET}")
    print(f"  {BOLD}Recommended ALPN     :{RESET} {YELLOW}{','.join(recommended_alpn)}{RESET}")
    print(f"  {BOLD}Optimal Fingerprint  :{RESET} {YELLOW}{recommended_fp}{RESET}")
    print(f"  {BOLD}Ping Latency         :{RESET} {best_entry['latency_ms']:.1f} ms")
    print(f"  {BOLD}Download Speed       :{RESET} {GREEN}{best_entry['download_kbps']:.1f} kbps ({best_entry['download_kbps']/1024.0:.2f} Mbps){RESET}")
    print(f"  {BOLD}Upload Speed         :{RESET} {GREEN}{best_entry['upload_kbps']:.1f} kbps ({best_entry['upload_kbps']/1024.0:.2f} Mbps){RESET}")
    print(f"  {BOLD}Performance Score    :{RESET} {best_entry['score']:.1f}")
    print(f"{BOLD}{GREEN}----------------------------------------------------------------------{RESET}")
    print(f"{BOLD}Ready-to-Use VLESS Link:{RESET}\n")
    print(f"{PINK}{final_vless_url}{RESET}\n")
    print(f"{BOLD}{GREEN}======================================================================{RESET}\n")

    # ----------------------------------------------------
    # Save Results & Summaries
    # ----------------------------------------------------
    os.makedirs(RESULTS_DIR, exist_ok=True)
    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M")
    results_txt_file = os.path.join(RESULTS_DIR, f"{timestamp}.txt")
    results_json_file = os.path.join(RESULTS_DIR, f"{timestamp}_summary.json")

    with open(results_txt_file, "w") as f:
        for r in speed_results:
            alpn_str = ",".join(r["alpn"])
            f.write(
                f"{r['ip']} | {r['latency_ms']:.1f}ms | DL: {r['download_kbps']:.1f}kbps | "
                f"UL: {r['upload_kbps']:.1f}kbps | ALPN: {alpn_str} | FP: {r['fp']} | Score: {r['score']:.1f}\n"
            )

    summary_data = {
        "timestamp": timestamp,
        "transport": vless_cfg["type"],
        "target_sni": vless_cfg["sni"],
        "target_host": vless_cfg["host"],
        "target_path": vless_cfg["path"],
        "workers": args.workers,
        "total_ips": total_ips,
        "alive_ips": len(alive_results),
        "best_result": {
            "ip": best_entry["ip"],
            "alpn": recommended_alpn,
            "fingerprint": recommended_fp,
            "latency_ms": best_entry["latency_ms"],
            "download_kbps": best_entry["download_kbps"],
            "upload_kbps": best_entry["upload_kbps"],
            "score": best_entry["score"],
            "vless_url": final_vless_url,
        },
        "ranked_candidates": speed_results,
    }

    with open(results_json_file, "w") as f:
        json.dump(summary_data, f, indent=2)

    save_subnets_to_disk(args.subnets)
    print(f"{DIM}Saved results to {results_txt_file} and {results_json_file}.{RESET}")
    print(f"{DIM}Updated subnet weights in {args.subnets}.{RESET}")


if __name__ == "__main__":
    main()