如何正确的优选属于你自己地区的CF 1

很多人都不知道怎么优选自己宽带的CF区域。

你所知道的延迟最低的CF实际上并非最快的，还是得看入口。
在这个鸡鸡闹鸡瘟的时间段，我建议大家不要溢价收鸡鸡。例如刚刚出现的dmit清退，claw清退，狐帝云跑路(必然)。

本文只针对电信网络，例如电信我只推荐NRT和SJC。基本上能让电信除了顶尖的cn2gia，没有比他更好的选择了，三网9929和cmi感觉都不如CF.

其实过程分为三部，以windows来举例。

1. 首先得去对应的github项目下载对应的优选工具
   https://github.com/XIU2/CloudflareSpeedTest
   解压并获得cfst.exe
   在该exe的目录执行powershell
   
   “.\cfst.exe -cfcolo NRT -dd -f ip.txt -tl 120”
   -cfcolo这个的意思是指定是NRT日本区域
   -dd是禁止下载(因为测速意义不是很大)
   -f输出为ip.txt
   -tl是最大延迟不能超过120。
   其他更多的参数请参考github上的自行添加

   

2. 当你获得NRT(实际不一定是NRT)的区域的ip.txt后，扔到python进行处理
   执行python trace.py 就可以知道对应地区的入口IP以及延迟。
   

3. 根据结果可知道100+个IP中只有不到10个IP是NRT的。

注意事项

1. 即使你指定了NRT，你优选出来的并非是NRT,还有可能是SIN。
2. 除了SIN和NRT，其实你的服务器在美国的话，可以尝试SJC也是挺快的，但是没人喜欢SJC入口，我也不知道为什么。
3. SIN时不时会抽风发颠，建议亚洲NRT，美国SJC/NRT。

人性的理解
肯定有人会跳出来说，每天疯狂测速、扫 IP，这不就是妥协和滥用 CF 吗？实际上，这和大家看待 Hysteria2 的态度一模一样。有人觉得 hy2 是没素质的“推土机”，挤占了公用带宽；但在这个处处 QoS、晚高峰丢包率起飞的恶劣网络环境下，普通人不过是为了求稳而选择了丛林生存法则。在国际出口带宽极其内卷的今天，没有什么绝对的正义、邪恶或者正确、错误(毕竟宽带交了钱)。我们在一堆 IP 里大海捞针，去分辨它到底去了 NRT 还是偷偷绕路到了 SIN，为的只是追求那一点点可控的稳定性。CF 既然提供了 Anycast 并且包容了这些流量，我们也算是为它的边缘节点做了海量的真实路由测试。说到底，工具没有善恶，全看你当下有多渴望一个不抽风的网络，以及等待大妈黑五打折(今年真的会打折吗)。

脚本内容：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sys
import os
import time
import socket
import urllib.request
import urllib.error
from concurrent.futures import ThreadPoolExecutor, as_completed

# 默认测试的单IP地址
DEFAULT_IP = "1.1.1.1"
# 默认的Host头信息，用于绕过直接访问IP的403限制
DEFAULT_HOST = "cloudflare.com"
# 连接和读取的超时时间（秒）
DEFAULT_TIMEOUT = 3.0
# 并发线程数
MAX_WORKERS = 20

def test_tcp_ping(ip, port=80, timeout=DEFAULT_TIMEOUT):
    """
    使用 Socket 建立 TCP 连接以精确测试 1 RTT 的网络延迟。
    """
    start_time = time.perf_counter()
    try:
        # 创建 TCP socket
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect((ip, port))
        s.close()
        return (time.perf_counter() - start_time) * 1000  # 转换为毫秒
    except Exception:
        return None

def get_trace_info(ip, host=DEFAULT_HOST, timeout=DEFAULT_TIMEOUT):
    """
    获取指定 IP 的 Cloudflare trace 信息，并进行 4 次 TCP Ping 测试其网络延迟。
    """
    url = f"http://{ip}/cdn-cgi/trace"
    req = urllib.request.Request(url)
    req.add_header('Host', host)
    req.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')
    
    trace_dict = {}
    content = ""
    success = False
    error_msg = ""
    
    # 1. 仅获取一次 trace 报文，用于解析数据中心和地区
    try:
        with urllib.request.urlopen(req, timeout=timeout) as response:
            content = response.read().decode('utf-8')
            lines = content.strip().split('\n')
            for line in lines:
                if '=' in line:
                    k, v = line.split('=', 1)
                    trace_dict[k.strip()] = v.strip()
            success = True
    except Exception as e:
        error_msg = str(e)
        
    if not success:
        return {
            'ip_address': ip,
            'success': False,
            'latency': 9999.0,
            'error': error_msg,
            'trace': {},
            'raw': ''
        }
        
    # 2. 进行 4 次 TCP Ping (建立 Socket 连接) 测量真实网络延迟，并取最小值
    latencies = []
    for _ in range(4):
        t = test_tcp_ping(ip, port=80, timeout=timeout)
        if t is not None:
            latencies.append(t)
            
    if latencies:
        return {
            'ip_address': ip,
            'success': True,
            'latency': min(latencies),  # 取 4 次测试中的最小延迟
            'trace': trace_dict,
            'raw': content
        }
    else:
        return {
            'ip_address': ip,
            'success': False,
            'latency': 9999.0,
            'error': "TCP Ping 失败",
            'trace': {},
            'raw': ''
        }

def read_ips_from_file(filepath):
    """
    从文件中读取 IP 列表，自动忽略注释行和空行。
    """
    if not os.path.exists(filepath):
        return []
    ips = []
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith('#'):
                ips.append(line)
    return ips

def print_single_ip_info(ip):
    """
    查询并打印单个 IP 的详细 trace 信息。
    """
    print(f"[*] 正在请求 IP: {ip} 的 trace 信息...")
    res = get_trace_info(ip)
    if res['success']:
        print("\n[+] 原始响应内容:")
        print("-" * 50)
        print(res['raw'].strip())
        print("-" * 50)
        print(f"连接延迟: {res['latency']:.2f} ms")
        
        # 打印结构化解析结果
        trace = res['trace']
        print("\n[+] 解析信息汇总:")
        print(f"  - 客户端 IP:    {trace.get('ip', '无')}")
        print(f"  - 数据中心 (Colo):   {trace.get('colo', '无')} ({trace.get('loc', '无')})")
        print(f"  - HTTP 版本: {trace.get('http', '无')}")
        print(f"  - WARP 状态:  {trace.get('warp', '无')}")
        print(f"  - 密钥交换协议: {trace.get('kex', '无')}")
    else:
        print(f"\n[-] 获取 IP {ip} trace 失败: {res['error']}")
        print(f"失败前耗时: {res['latency']:.2f} ms")

def scan_multiple_ips(ip_list):
    """
    并发扫描 IP 列表，并在控制台输出摘要表格，最后按地区分类写入文件。
    """
    print(f"[*] 正在并发扫描 {len(ip_list)} 个 IP，并发线程数: {MAX_WORKERS}...")
    print(f"{'IP 地址':<18} | {'地区 (Colo)':<10} | {'位置 (Loc)':<10} | {'延迟':<10} | {'状态':<15}")
    print("-" * 70)
    
    results = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # 提交并发任务
        futures = {executor.submit(get_trace_info, ip): ip for ip in ip_list}
        
        for future in as_completed(futures):
            res = future.result()
            results.append(res)
            
            ip = res['ip_address']
            if res['success']:
                trace = res['trace']
                colo = trace.get('colo', 'N/A')
                loc = trace.get('loc', 'N/A')
                latency_str = f"{res['latency']:.1f} ms"
                status_str = "正常"
                print(f"{ip:<18} | {colo:<10} | {loc:<10} | {latency_str:<10} | {status_str:<15}")
            else:
                err_msg = str(res['error'])
                if len(err_msg) > 15:
                    # 截断过长的错误信息
                    err_msg = err_msg[:12] + "..."
                print(f"{ip:<18} | {'N/A':<10} | {'N/A':<10} | {'超时':<10} | {err_msg:<15}")
                
    # 筛选出请求成功的所有结果
    successful_results = [r for r in results if r['success']]
    
    # 1. 打印延迟最低的前5个IP
    successful_results.sort(key=lambda x: x['latency'])
    print("\n[+] 延迟最低的前 5 个 Cloudflare IP:")
    print("-" * 60)
    print(f"{'排名':<5} | {'IP 地址':<18} | {'地区 (Colo)':<10} | {'位置 (Loc)':<10} | {'延迟':<10}")
    print("-" * 60)
    for idx, res in enumerate(successful_results[:5], 1):
        trace = res['trace']
        print(f"{idx:<5} | {res['ip_address']:<18} | {trace.get('colo', 'N/A'):<10} | {trace.get('loc', 'N/A'):<10} | {res['latency']:.1f} ms")

    # 2. 按地区（colo）分类并写入文件
    print("\n[*] 正在按地区分类保存 IP 地址到根目录下...")
    colo_groups = {}
    for res in successful_results:
        trace = res['trace']
        colo = trace.get('colo', 'UNKNOWN').upper()
        if colo not in colo_groups:
            colo_groups[colo] = []
        colo_groups[colo].append(res)
        
    for colo, res_list in colo_groups.items():
        # 对同一地区的IP也按照延迟升序排列
        res_list.sort(key=lambda x: x['latency'])
        
        filename = f"ip_{colo}.txt"
        
        # 每次执行测完写入时，检查 ip_<COLO>.txt 是否存在
        # 如果存在，则删除或清空内容，重新写入，避免数据堆积
        if os.path.exists(filename):
            print(f"  - 检测到已存在 {filename}，正在清空其内容并重新写入最新数据...")
        else:
            print(f"  - 正在创建新文件: {filename}")
            
        try:
            # 使用 'w' 模式打开文件会清空该文件原有的所有内容
            with open(filename, 'w', encoding='utf-8') as f:
                for r in res_list:
                    trace = r['trace']
                    colo_val = trace.get('colo', 'N/A')
                    loc_val = trace.get('loc', 'N/A')
                    f.write(f"{r['ip_address']:<18} | {colo_val:<10} | {loc_val:<10} | {r['latency']:.1f} ms\n")
        except Exception as e:
            print(f"  [-] 写入文件 {filename} 失败: {e}")

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Cloudflare Trace 与测速工具")
    parser.add_argument("ip", nargs="?", default=None, help=f"单个要测试的 IP 地址 (默认: {DEFAULT_IP})")
    parser.add_argument("-f", "--file", default=None, help="包含 IP 列表的文件路径")
    parser.add_argument("-s", "--scan", action="store_true", help="强制扫描工作区中的 ip.txt 文件")
    
    args = parser.parse_args()
    
    # 决定是进行批量扫描还是检测单个IP
    if args.file:
        ips = read_ips_from_file(args.file)
        if not ips:
            print(f"[-] 在文件 {args.file} 中未找到有效的 IP 地址")
            sys.exit(1)
        scan_multiple_ips(ips)
    elif args.scan or (not args.ip and os.path.exists("ip.txt")):
        # 如果指定了 --scan 或者未提供参数但 ip.txt 存在，则读取 ip.txt 进行批量扫描
        target_file = "ip.txt"
        print(f"[*] 发现 {target_file} 文件，即将开始批量测试与分类...")
        ips = read_ips_from_file(target_file)
        if not ips:
            print(f"[-] 在 {target_file} 中未找到任何 IP 地址")
            sys.exit(1)
        scan_multiple_ips(ips)
    else:
        # 测试单个 IP
        ip_to_check = args.ip if args.ip else DEFAULT_IP
        print_single_ip_info(ip_to_check)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n[-] 用户取消了操作。")
        sys.exit(0)