首页
/ Crawl4AI 代理与安全深度指南:按请求代理配置、轮换策略与 SSL 证书分析

Crawl4AI 代理与安全深度指南:按请求代理配置、轮换策略与 SSL 证书分析

2026-09-04 09:18:12作者:曹令琨Iris

本文围绕 Crawl4AI 的代理与安全能力展开:如何在 CrawlerRunConfig 层面按请求配置 HTTP/HTTPS/SOCKS5 代理(含认证与凭据安全)、通过 RoundRobinProxyStrategy 实现多代理轮换与粘性会话、以及在抓取时抓取并导出目标站点的 SSL 证书信息。读完后,你将能够完整搭建一套"代理 + 轮换 + 证书审计"的组合式网络出口方案,并理解其底层实现链路。

设计原则:为什么代理要"按请求"配置

Crawl4AI 推荐通过 CrawlerRunConfig.proxy_config 为每个请求配置代理,而不是在浏览器启动时全局绑定。这样做的好处:

  • 自包含:每个请求的网络设置完整描述在自己所在的 run config 中,便于复用与测试;
  • 可轮换:换代理或换轮换策略只需构造一个新的 run config,无需重启浏览器;
  • 可批量arun_many 批量抓取时可以为同一批 URL 应用统一的轮换策略。

从源码看,BrowserConfig 上旧的 proxy 字符串参数已标记弃用。crawl4ai/async_configs.py 中,传入 proxy 会触发 UserWarning,且当 proxyproxy_config 同时提供时,以 proxy_config 为准。

基础代理配置

最简用法是在 run config 中传入 ProxyConfig(也支持字典或直接字符串三种等价写法):

import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, ProxyConfig

run_config = CrawlerRunConfig(proxy_config=ProxyConfig(server="http://proxy.example.com:8080"))
# run_config = CrawlerRunConfig(proxy_config={"server": "http://proxy.example.com:8080"})
# run_config = CrawlerRunConfig(proxy_config="http://proxy.example.com:8080")


async def main():
    browser_config = BrowserConfig()
    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url="https://example.com", config=run_config)
        print(f"Success: {result.success} -> {result.url}")


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

ProxyConfig 的核心字段(见 crawl4ai/async_configs.py):

字段 说明
server 代理服务器 URL,例如 "http://127.0.0.1:8080"
username 可选,代理认证用户名
password 可选,代理认证密码
ip 可选,期望出口 IP,用于验证;不传时自动从 server 中解析

CrawlerRunConfig.proxy_config 的 setter 会调用 _normalize_proxy_config 自动把字符串/字典归一化为 ProxyConfig 对象(crawl4ai/async_configs.py),所以上面三种写法最终殊途同归。

支持的代理格式

ProxyConfig.from_string() 是格式解析的入口。当前源码(crawl4ai/async_configs.py)支持的格式比早期文档更多,共 5 类:

from crawl4ai import ProxyConfig

# HTTP 代理带认证(URL 内嵌凭据)
proxy1 = ProxyConfig.from_string("http://user:pass@192.168.1.1:8080")

# HTTPS 代理
proxy2 = ProxyConfig.from_string("https://proxy.example.com:8080")

# SOCKS5 代理
proxy3 = ProxyConfig.from_string("socks5://proxy.example.com:1080")

# 简单 IP:port 格式(自动补 http:// 前缀)
proxy4 = ProxyConfig.from_string("192.168.1.1:8080")

# IP:port:user:pass 格式
proxy5 = ProxyConfig.from_string("192.168.1.1:8080:user:pass")

解析规则(对应源码分支):

  1. @ 且含 ://:按 protocol://user:pass@host:port 拆分,username/password 单独存放,server 保留协议与地址;
  2. :// 且不含 @:整体作为 server 保留(支持 http/https/socks5 等任意 scheme);
  3. 冒号分隔 4 段:ip:port:user:pass,自动组装为 http://ip:port
  4. 冒号分隔 2 段:ip:port,自动组装为 http://ip:port
  5. 其他形式抛出 ValueError

此外还有 from_dict()to_dict()clone(**kwargs) 用于序列化与派生新配置;ProxyConfig.DIRECT(值为字符串 "direct")是一个哨兵常量,从源码注释看,它用于在 proxy_config 列表中显式表示"该请求不走代理"(crawl4ai/async_configs.py)。

注意:crawl4ai/proxy_strategy.py 中保留了一份旧版 ProxyConfig 用于向后兼容,其 from_string 仅支持后两种冒号格式;代码注释明确建议改为 from crawl4ai import ProxyConfigcrawl4ai/proxy_strategy.py)。新代码应始终使用 crawl4ai 顶层导出的版本。

带认证的代理

代理需要用户名/密码时,直接以构造参数或字典方式传入:

import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, ProxyConfig

run_config = CrawlerRunConfig(
    proxy_config=ProxyConfig(
        server="http://proxy.example.com:8080",
        username="your_username",
        password="your_password",
    )
)
# 或字典风格:
# run_config = CrawlerRunConfig(proxy_config={
#     "server": "http://proxy.example.com:8080",
#     "username": "your_username",
#     "password": "your_password",
# })


async def main():
    browser_config = BrowserConfig()
    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url="https://example.com", config=run_config)
        print(f"Success: {result.success} -> {result.url}")


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

仓库中提供了更贴近真实服务商的示例,例如 docs/examples/nst_proxy/basic_proxy_example.pydocs/examples/nst_proxy/auth_proxy_example.pydocs/examples/nst_proxy/api_proxy_example.py,可参考其对动态代理 API 的封装方式。

从环境变量加载代理列表

批量场景下,把代理清单放进环境变量 PROXIES 可以避免把凭据写死在代码里:

import os
from crawl4ai import ProxyConfig, CrawlerRunConfig

# 设置环境变量
os.environ["PROXIES"] = "ip1:port1:user1:pass1,ip2:port2:user2:pass2,ip3:port3"

# 加载全部代理
proxies = ProxyConfig.from_env()
print(f"Loaded {len(proxies)} proxies")

# 使用第一个代理
if proxies:
    run_config = CrawlerRunConfig(proxy_config=proxies[0])

from_env(env_var="PROXIES") 的实现(crawl4ai/async_configs.py)逻辑是:读取环境变量 → 按逗号拆分 → 逐条调用 from_string 解析;空项会被跳过,解析异常会打印错误后返回已解析的部分。因此格式必须是 ip:port:user:pass,ip:port:user:pass,... 的逗号分隔形式,user:pass 可以省略。

Shell 中推荐的导出方式:

# 敏感凭据放环境变量,避免硬编码
export PROXIES="ip1:port1:user1:pass1,ip2:port2:user2:pass2"

代理轮换:RoundRobin 策略与粘性会话

Crawl4AI 内置轮换取代策略接口 ProxyRotationStrategy 与实现 RoundRobinProxyStrategycrawl4ai/proxy_strategy.py)。轮换在每次请求层面生效,通过 CrawlerRunConfig.proxy_rotation_strategy 挂接。

完整轮换示例

下面的示例用 arun_many 请求 httpbin.org/ip,验证每个请求确实走了不同的出口代理:

import asyncio
import re
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode, ProxyConfig
from crawl4ai.proxy_strategy import RoundRobinProxyStrategy

async def main():
    # 从环境变量加载代理
    proxies = ProxyConfig.from_env()
    if not proxies:
        print("No proxies found! Set PROXIES environment variable.")
        return

    # 创建轮换策略
    proxy_strategy = RoundRobinProxyStrategy(proxies)

    # 按请求配置代理轮换
    browser_config = BrowserConfig(headless=True, verbose=False)
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        proxy_rotation_strategy=proxy_strategy,
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        urls = ["https://httpbin.org/ip"] * (len(proxies) * 2)  # 每个代理测两轮

        print(f"Testing {len(proxies)} proxies with rotation...")
        results = await crawler.arun_many(urls=urls, config=run_config)

        for i, result in enumerate(results):
            if result.success:
                # 从响应中提取出口 IP
                ip_match = re.search(r'(?:[0-9]{1,3}\.){3}[0-9]{1,3}', result.html)
                if ip_match:
                    detected_ip = ip_match.group(0)
                    proxy_index = i % len(proxies)
                    expected_ip = proxies[proxy_index].ip

                    print(f"Request {i+1}: Proxy {proxy_index+1} -> IP {detected_ip}")
                    if detected_ip == expected_ip:
                        print("   IP matches proxy configuration")
                    else:
                        print(f"   IP mismatch (expected {expected_ip})")
                else:
                    print(f"Request {i+1}: Could not extract IP from response")
            else:
                print(f"Request {i+1}: Failed - {result.error_message}")

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

注意其中 cache_mode=CacheMode.BYPASS:轮换验证必须绕过缓存,否则可能命中缓存结果而拿不到真实的出口 IP。仓库附带可运行的同款演示脚本 docs/examples/proxy_rotation_demo.py

底层调用链与粘性会话

轮换的实际生效点在 crawl4ai/async_webcrawler.py:执行抓取前,若 config.proxy_rotation_strategy 存在:

  • 若设置了 config.proxy_session_id,调用 strategy.get_proxy_for_session(session_id, ttl),为同一 session_id 固定分配一个代理(粘性会话);
  • 否则调用 strategy.get_next_proxy(),按 itertools.cycle 顺序轮转;
  • 抓取结束后会调用 release_session 释放粘性会话(crawl4ai/async_webcrawler.py)。

粘性会话的设计意图(见 crawl4ai/proxy_strategy.py 的类文档):在深度爬取(deep crawling)中让同一站点会话内的多个请求保持同一出口 IP,避免频繁换 IP 触发风控。相关 API:

方法 作用
get_next_proxy() 轮转取出下一个代理
get_proxy_for_session(session_id, ttl) 为会话取/绑定代理,ttl(秒)到期后自动换绑
release_session(session_id) 释放会话绑定的代理
get_session_proxy(session_id) 只查询不创建
get_active_sessions() 列出未过期的全部会话
cleanup_expired_sessions() 清理过期会话,返回清理数量

线程安全上,会话操作由 asyncio.Lock 保护。对应的测试覆盖位于 tests/proxy/test_sticky_sessions.pytests/proxy/test_proxy_config.py,可作为行为基线参考。

SSL 证书分析:抓取 + 审计一步完成

在代理出口场景下(尤其是经过做 SSL 拦截/改写的代理时),抓取目标站点的证书并落盘审计是常见的安全动作。Crawl4AI 支持按请求开启 fetch_ssl_certificate

import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig

run_config = CrawlerRunConfig(
    proxy_config={
        "server": "http://proxy.example.com:8080",
        "username": "user",
        "password": "pass",
    },
    fetch_ssl_certificate=True,  # 为本请求开启证书分析
)


async def main():
    browser_config = BrowserConfig()
    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url="https://example.com", config=run_config)

        if result.success:
            print(f"Crawled via proxy: {result.url}")

            # 分析 SSL 证书
            if result.ssl_certificate:
                cert = result.ssl_certificate
                print("SSL Certificate Info:")
                print(f"   Issuer: {cert.issuer}")
                print(f"   Subject: {cert.subject}")
                print(f"   Valid until: {cert.valid_until}")
                print(f"   Fingerprint: {cert.fingerprint}")

                # 导出证书
                cert.to_json("certificate.json")
                print("Certificate exported to certificate.json")
            else:
                print("No SSL certificate information available")


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

证书抓取与代理配置是相互独立的两条链路:

  • 代理作用于浏览器上下文(请求经代理出口);
  • 证书抓取发生在导航之前,由 crawl4ai/async_crawler_strategy.pyconfig.fetch_ssl_certificate 为真时调用 SSLCertificate.from_url(url) 完成,是一条独立的 socket 直连(443 端口)。

SSLCertificate 的实现细节(crawl4ai/ssl_certificate.py):

  • ssl.create_default_context() 建立默认验证上下文,socket.create_connection((hostname, 443)) 连接后 wrap_socket,取 getpeercert(binary_form=True) 的二进制证书;
  • 交由 pyOpenSSL(OpenSSL.crypto.load_certificate(FILETYPE_ASN1, ...))解析出 subjectissuerserial_number(hex)、not_before/not_afterfingerprint(SHA-256 十六进制)、extensions,并保留 raw_cert(Base64 的 DER 原文);
  • 类继承自 dict,因此实例本身即可 JSON 序列化;常用属性包括 issuersubjectvalid_fromvalid_untilfingerprint
  • 导出方法:to_json(filepath=None)(写文件或返回 JSON 字符串)、to_pem(...)to_der(...),均支持传 filepath 直接落盘;
  • 失败语义:SSLCertVerificationError、DNS 解析失败、超时或任何异常都会打印提示并返回 None,即 result.ssl_certificateNone 不代表抓取本身失败。

CrawlResult.ssl_certificate 字段定义在 crawl4ai/models.py,抓取结果中可直接访问。独立使用证书工具的示例可参考 docs/examples/ssl_example.py

一个实用推论:若证书指纹(fingerprint)与你预期值不符,或 valid_until 异常临近,说明出口链路上的代理可能在做 MITM 替换证书——这正是"代理 + 证书审计"组合的价值所在。

安全最佳实践

1. 用轮换避免 IP 封禁

from crawl4ai import CrawlerRunConfig, ProxyConfig
from crawl4ai.proxy_strategy import RoundRobinProxyStrategy

# 多代理轮换,降低单 IP 被封概率
proxies = ProxyConfig.from_env("PROXIES")
strategy = RoundRobinProxyStrategy(proxies)

# 按请求配置轮换(推荐)
run_config = CrawlerRunConfig(proxy_rotation_strategy=strategy)

# 若所有请求固定走同一代理,直接复用同一个 run_config 实例即可
static_run_config = run_config

2. 始终开启证书校验/分析

from crawl4ai import CrawlerRunConfig

# 尽可能校验/分析 SSL 证书(按请求生效)
run_config = CrawlerRunConfig(fetch_ssl_certificate=True)

3. 环境变量存放凭据

# 敏感代理凭据放环境变量,避免硬编码用户名/密码
export PROXIES="ip1:port1:user1:pass1,ip2:port2:user2:pass2"

4. 优先 SOCKS5

from crawl4ai import CrawlerRunConfig

# SOCKS5 协议支持更完整,安全性更好
run_config = CrawlerRunConfig(proxy_config="socks5://proxy.example.com:1080")

5. 代理的安全日志

ProxyConfig 内直接存着明文凭据,写日志时不要 str(proxy) 整对象输出。文档给出的脱敏写法:

from crawl4ai import ProxyConfig

def safe_proxy_repr(proxy: ProxyConfig):
    if getattr(proxy, "username", None):
        return f"{proxy.server} (auth: ****)"
    return proxy.server

to_dict() 同样会包含 password 字段,落盘或上报前需自行过滤。)

从已弃用的 proxy 参数迁移

旧的 BrowserConfig(proxy=...) 全局代理参数已弃用,迁移方式:

# 旧(已弃用)
# browser_config = BrowserConfig(proxy="http://proxy.example.com:8080")

# 新(推荐)
from crawl4ai import CrawlerRunConfig
run_config = CrawlerRunConfig(proxy_config="http://proxy.example.com:8080")

行为对照(以 crawl4ai/async_configs.py 的兼容逻辑为准):

  • 仅传 proxy:触发弃用警告,并自动转成 proxy_config
  • 两者同传:proxy_config 优先,同时发出警告;
  • 迁移后每个请求的网络出口由 run config 显式控制,arun_many 等批量接口也能统一套用轮换策略。

故障排查

现象 排查方向
代理连接失败 确认代理服务器网络可达;核对认证凭据;确认协议前缀与代理实际类型一致(http / https / socks5
SSL 证书错误 部分代理会破坏 SSL 检查链(MITM 换证);反复失败可换代理,或临时关闭 fetch_ssl_certificate 以隔离问题
环境变量未生效 确认脚本运行前已设置 PROXIES(或自定义变量名);核对格式 ip:port:user:pass,ip:port:user:pass;用 len(ProxyConfig.from_env()) 验证是否真的解析出条目
轮换不生效 确认 ProxyConfig.from_env() 解析结果非空(len(proxies) > 0);确认 proxy_rotation_strategy 已挂到 CrawlerRunConfig 上;检查传入策略的代理定义本身是否有效

验证轮换是否真实生效的推荐方法,就是本文"代理轮换"一节中的 httpbin.org/ip 出口 IP 比对法(配合 CacheMode.BYPASS)。

延伸阅读

适用前提:以上配置项(CrawlerRunConfig.proxy_configproxy_rotation_strategyproxy_session_idfetch_ssl_certificate)均以当前仓库 develop 分支(0.9.x)源码为准;证书抓取依赖 pyOpenSSL,且仅对 443 端口的 HTTPS 站点生效。

登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库,实现网络在NPU上加速计算。
C++
1.12 K
2.72 K
kernelkernel
openEuler内核是openEuler操作系统的核心,既是系统性能与稳定性的基石,也是连接处理器、设备与服务的桥梁。
C
527
590
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库,实现网络在NPU上加速计算。
C++
904
1.82 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件,TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件,使 PyTorch 框架能够直接调用昇腾 NPU,为开发者提供昇腾 AI 处理器的超强算力。
Python
854
1.34 K
docsdocs
暂无描述
Markdown
889
5.78 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent,它能够将大语言模型的强大能力,通过你日常使用的各类通讯应用,直接延伸至你的指尖。
Python
3.52 K
1.01 K
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库,实现网络在NPU上加速计算。
C++
1.33 K
1.45 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓,支持在线互动运行、边学边练,提供教程、示例与优化方案,一站式助力昇腾开发者快速上手。
Jupyter Notebook
981
502
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR(Multi-Level Intermediate Representation)构建的,面向昇腾亲和算子编译时使用的中间表示,提供昇腾完备表达能力,通过编译优化提升昇腾AI处理器计算效率,支持通过生态框架使能昇腾AI处理器与深度调优
C++
540
384