IP池纯净度测试：用httpbin + 自建检测服务质量

连通性测试只是及格线，不是质量评估

搭好代理池之后，最常见的"测试"是这样的：用每个 IP 访问一下 httpbin.org/ip，能返回 IP 地址就算通过，返回超时就标记失败。

这个测试只能回答一个问题：这个 IP 能不能发 HTTP 请求。它回答不了：

这个 IP 有没有被目标站的安全机制标记过？它的出口地理位置是不是你需要的？它是不是和其他大量 IP 共享同一个 /24 子网段（意味着如果一个被限制，同段的其他 IP 也可能受影响）？它走的是住宅网络还是数据中心网络？请求头有没有被代理服务注入了额外字段？

这些才是决定 IP 在你的业务场景里能不能用的关键因素。我把这些统称为"IP 纯净度"——不是说 IP 干不干净的道德判断，而是工程意义上的"这个 IP 的网络环境、历史标记、出口特征有多适合你的目标任务"。

第一步：用 httpbin 做基础检测（6 个端点）

httpbin.org 是一个开源的 HTTP 测试服务，可以自托管。先用它做六项基础检测：

检测 1：出口 IP 确认。 访问 httpbin.org/ip，确认返回的 IP 是代理 IP 而不是你的真实出口。如果返回的是你本机 IP，说明代理没生效。

import requests

def check_exit_ip(proxy_addr):
    try:
        resp = requests.get(
            "https://httpbin.org/ip",
            proxies={
   "https": f"http://{proxy_addr}"},
            timeout=10
        )
        data = resp.json()
        return data.get("origin", "")
    except Exception as e:
        return f"FAIL: {e}"

检测 2：请求头透传检查。 访问 httpbin.org/headers，看代理有没有在请求头里注入额外字段。常见的泄露字段包括 X-Forwarded-For、Via、X-Real-IP。如果这些字段出现了且包含你的真实 IP 或代理信息，说明代理的请求环境隔离性不够。

def check_header_leak(proxy_addr):
    resp = requests.get(
        "https://httpbin.org/headers",
        proxies={
   "https": f"http://{proxy_addr}"},
        timeout=10
    )
    headers = resp.json().get("headers", {
   })
    leak_fields = ["X-Forwarded-For", "Via", "X-Real-Ip", "Proxy-Connection"]
    leaks = {
   k: v for k, v in headers.items() if k in leak_fields}
    return leaks  # 空字典 = 没泄露

检测 3：延迟基线。 访问 httpbin.org/delay/0（零延迟端点），记录往返时间。这个时间减去 httpbin 本身的处理时间（接近 0），就是纯网络延迟。

def check_latency(proxy_addr, target_url="https://httpbin.org/delay/0"):
    start = time.time()
    try:
        resp = requests.get(
            target_url,
            proxies={
   "https": f"http://{proxy_addr}"},
            timeout=15
        )
        latency_ms = (time.time() - start) * 1000
        return round(latency_ms, 1)
    except Exception:
        return -1  # 超时或失败

检测 4：HTTPS 支持。 访问 https://httpbin.org/ip（注意是 HTTPS），确认代理支持 HTTPS CONNECT。部分低质量代理只支持 HTTP。

检测 5：大响应体能力。 访问 httpbin.org/bytes/102400（返回 100KB 随机数据），测试带宽和完整性。如果代理在大响应体下断连或截断，说明它的连接稳定性有问题。

检测 6：并发稳定性。 用 10 个并发线程同时通过同一个 IP 访问 httpbin，看是否都能成功返回。有些代理 IP 单线程没问题，并发一高就开始丢连接。

第二步：自建检测服务做深度测试

httpbin 能做的检测有限——它只能告诉你"请求到了 httpbin 那边是什么样子"。但你真正关心的是"这个 IP 在目标站眼里是什么样子"。

自建一个检测服务，部署在你的目标区域（比如你的目标站在美国，就把检测服务部署在美国的 VPS 上），可以做更多事情：

检测 A：地理位置验证。 记录请求到达检测服务时的源 IP，用 MaxMind GeoIP2 数据库（免费版 GeoLite2 足够）查这个 IP 的国家、城市、ASN。跟代理声称的地理位置做对比——如果代理声称是美国 IP 但实际出口在新加坡，说明地理信息不准确。

# 检测服务端（Flask 示例）
from flask import Flask, request, jsonify
import geoip2.database

app = Flask(__name__)
reader = geoip2.database.Reader('/path/to/GeoLite2-City.mmdb')
asn_reader = geoip2.database.Reader('/path/to/GeoLite2-ASN.mmdb')

@app.route('/check')
def check():
    ip = request.remote_addr
    try:
        geo = reader.city(ip)
        asn = asn_reader.asn(ip)
        return jsonify({
   
            "ip": ip,
            "country": geo.country.iso_code,
            "city": geo.city.name,
            "asn_number": asn.autonomous_system_number,
            "asn_org": asn.autonomous_system_organization,
            "is_datacenter": is_datacenter_asn(asn.autonomous_system_number),
        })
    except Exception as e:
        return jsonify({
   "ip": ip, "error": str(e)})

def is_datacenter_asn(asn_number):
    """粗略判断：常见数据中心ASN列表"""
    dc_asns = {
   
        14061,   # DigitalOcean
        16509,   # Amazon/AWS
        15169,   # Google
        8075,    # Microsoft/Azure
        13335,   # Cloudflare
        20473,   # Choopa/Vultr
        63949,   # Linode
        # ... 补充更多
    }
    return asn_number in dc_asns

检测 B：ASN 和子网段分析。 同一个 /24 子网段的 IP 通常共享命运——如果一个被限制，同段的其他 IP 被限制的概率更高。把池子里所有 IP 的 /24 前缀提取出来，看集中度。

def analyze_subnet_diversity(ip_list):
    """统计 /24 子网段分布"""
    from collections import Counter
    subnets = Counter()
    for ip in ip_list:
        parts = ip.split(".")
        subnet = ".".join(parts[:3])
        subnets[subnet] += 1

    total = len(ip_list)
    top10 = subnets.most_common(10)
    concentration = sum(c for _, c in top10) / total  # 前10个子网占总量的比例

    return {
   
        "total_ips": total,
        "unique_subnets": len(subnets),
        "top10_concentration": round(concentration, 3),
        "top10_subnets": top10,
    }

如果 1000 个 IP 只分布在 50 个 /24 子网段里（平均每个子网 20 个 IP），子网集中度就很高。理想情况下，/24 子网段数量应该接近 IP 数量（每个子网只有 1-2 个 IP）。

检测 C：数据中心 vs 住宅网络判断。 通过 ASN 可以粗略判断一个 IP 是来自数据中心还是住宅 ISP。不同场景对这个有不同要求：做广告验证通常需要住宅 IP（因为广告平台对数据中心 IP 可能展示不同内容），做公开数据采集两种都可以用，但数据中心 IP 被目标站限制的概率通常更高。

检测 D：DNS 泄露测试。 通过代理访问你的检测服务时，检测服务端记录 DNS 查询来源。如果 DNS 请求不是从代理 IP 发出的（而是从你的本机 DNS 服务器发出的），说明存在 DNS 泄露——目标站可以通过 DNS 查询来源发现你的真实网络环境。

实现方式：检测服务返回一个唯一子域名（如 {uuid}.dns.your-check-server.com），客户端通过代理访问这个子域名，检测服务的 DNS 服务器记录谁来查询了这个子域名。

第三步：把检测结果量化成"纯净度评分"

拿到上面的检测结果后，需要一个评分模型把它们汇总成一个可比较的数字。推荐的评分维度和权重：

def calc_purity_score(check_result):
    score = 100

    # 1. 连通性（基础分，不通直接0分）
    if not check_result["connectable"]:
        return 0

    # 2. 请求头泄露（扣20分）
    if check_result["header_leaks"]:
        score -= 20

    # 3. 地理位置不符（扣15分）
    if not check_result["geo_match"]:
        score -= 15

    # 4. 数据中心IP（扣10分，非绝对扣分，取决于场景）
    if check_result["is_datacenter"]:
        score -= 10

    # 5. 子网集中度过高（扣10分）
    if check_result["subnet_concentration"] > 0.3:
        score -= 10

    # 6. 延迟过高（>2000ms 扣10分，>5000ms 扣20分）
    latency = check_result["latency_ms"]
    if latency > 5000:
        score -= 20
    elif latency > 2000:
        score -= 10

    # 7. HTTPS 不支持（扣15分）
    if not check_result["https_supported"]:
        score -= 15

    # 8. 并发不稳定（10并发成功率<80% 扣10分）
    if check_result["concurrent_success_rate"] < 0.8:
        score -= 10

    # 9. DNS泄露（扣15分）
    if check_result["dns_leak"]:
        score -= 15

    return max(score, 0)

评分阈值建议：

80-100 分：高纯净度，适合对 IP 质量要求高的场景（广告验证、需要稳定访问的长期采集）。

60-79 分：中等纯净度，日常公开数据采集够用，但需要配合更积极的轮转和淘汰策略。

40-59 分：低纯净度，只适合对成功率要求不高、允许大量重试的场景。

40 分以下：建议直接从池子里移除。

批量检测的工程实现

池子里有几千个 IP 时，逐个检测太慢。需要并发检测，但要控制好节奏：

import asyncio
import aiohttp

async def batch_check(ip_list, check_url, max_concurrent=50):
    sem = asyncio.Semaphore(max_concurrent)
    results = []

    async def check_one(ip):
        async with sem:
            try:
                async with aiohttp.ClientSession() as session:
                    proxy_url = f"http://{ip}"
                    start = time.time()
                    async with session.get(
                        check_url,
                        proxy=proxy_url,
                        timeout=aiohttp.ClientTimeout(total=15)
                    ) as resp:
                        latency = (time.time() - start) * 1000
                        data = await resp.json()
                        results.append({
   
                            "proxy": ip,
                            "status": resp.status,
                            "latency_ms": round(latency, 1),
                            "exit_ip": data.get("origin", data.get("ip", "")),
                            "success": True,
                        })
            except Exception as e:
                results.append({
   
                    "proxy": ip,
                    "success": False,
                    "error": str(e),
                })

    tasks = [check_one(ip) for ip in ip_list]
    await asyncio.gather(*tasks)
    return results

几个注意事项：

并发数 50 是个安全起点。如果检测目标是你自建的服务，可以开到 100-200；如果是 httpbin.org 公开实例，50 以上容易被限速。

检测频率建议每天跑一次全量检测，日常运行时的健康检测（每 5 分钟一轮）只做连通性和延迟，不做完整的纯净度评估。

检测结果要持久化——存到数据库或文件里，方便追踪 IP 质量的变化趋势。一个 IP 上周纯净度 85 分、这周降到 60 分，说明它可能被标记了或者网络环境变了。

FAQ

Q：httpbin.org 挂了或被限速怎么办？

httpbin 是开源的（GitHub 上搜 postmanlabs/httpbin），可以自己部署一个。部署在你的目标区域的 VPS 上，既解决了可用性问题，又能做地理位置检测。自建 httpbin 的资源消耗很低，1 核 1G 的 VPS 就够用。

Q：纯净度评分的权重怎么调？

上面那组权重是通用起点。你需要根据自己的业务场景调整。比如做广告验证的场景，"数据中心 IP"这一项应该扣更多分（甚至直接淘汰）；做公开数据采集的场景，延迟的权重可以降低（因为你通常不需要毫秒级响应）。建议先用默认权重跑一个月，再根据"高分 IP 的业务成功率是不是真的比低分 IP 高"来校准。

Q：GeoIP 数据库多久更新一次？

MaxMind 的 GeoLite2 免费版每周更新。付费版 GeoIP2 更新更频繁。建议至少每月更新一次 GeoIP 数据库，否则新分配的 IP 段可能查不到准确的地理信息。