CC攻击之所以难防,根本原因在于它模拟的是正常用户的访问行为。传统的基于连接数或请求频率的封禁策略,在面对分布在全球各地的代理IP池时,往往力不从心。你封掉一个IP,攻击者马上切换另一个IP,而你的服务器资源就在这种拉锯战中被耗尽。将服务器本地日志分析与Cloudflare的全球边缘网络联动,是目前成本最低、效果最显著的一种防御策略。核心逻辑很简单:让fail2ban在服务器上实时分析Nginx或Apache的访问日志,一旦发现某个IP在短时间内请求了过多的动态页面或触发了大量403、404错误,立即调用Cloudflare API,在边缘节点层面将该IP拉入黑名单。这样,恶意请求甚至还没到达你的服务器就被拦截了。

环境准备与依赖安装

在开始配置之前,需要确保服务器已经安装了fail2ban,并且具备Python环境来运行Cloudflare API交互脚本。大多数现代Linux发行版的官方仓库都包含fail2ban,可以直接通过包管理器安装。以Debian或Ubuntu系统为例,执行以下命令即可完成基础安装:

apt update
apt install fail2ban -y

对于CentOS或RHEL系列,需要先启用EPEL仓库再安装。安装完成后,建议立即启动fail2ban服务并设置开机自启。接下来需要处理的是Cloudflare API的调用工具。虽然可以直接用curl命令,但为了处理复杂的JSON响应和错误重试机制,使用Python脚本会更加可靠。需要确保系统安装了requests库,可以通过pip安装:

pip install requests

最关键的一步是获取Cloudflare API令牌。登录Cloudflare控制台,在“我的个人资料”中找到“API令牌”选项。不要使用全局API密钥,权限过大存在安全隐患。应该创建一个具有特定权限的API令牌,至少需要包含“区域-防火墙规则:编辑”和“区域-区域:读取”这两个权限。将生成的令牌字符串妥善保存,后续脚本中会用到。同时,还需要记录下你的域名所在的区域ID,这个ID可以在Cloudflare仪表板右侧边栏找到。

编写Cloudflare API交互脚本

这个脚本是整套系统的核心执行单元。它需要接收fail2ban传递过来的IP地址和动作指令,然后通过Cloudflare API v4接口,将该IP添加或移出账户级别的IP列表。之所以推荐使用账户级别的列表而非区域级别的防火墙规则,是因为列表可以跨多个域名复用,管理起来更加集中。脚本的设计应该支持ban、unban两种操作,并且具备完善的错误处理逻辑。以下是一个经过生产环境验证的脚本示例:

#!/usr/bin/env python3
import requests
import sys
import json
import os

# Cloudflare 配置信息
CF_API_TOKEN = "你的API令牌"
CF_ACCOUNT_ID = "你的账户ID"
CF_LIST_ID = "你的IP列表ID"
CF_API_BASE = "https://api.cloudflare.com/client/v4"

def add_ip_to_list(ip_address):
    """将IP地址添加到Cloudflare账户级IP列表"""
    url = f"{CF_API_BASE}/accounts/{CF_ACCOUNT_ID}/rules/lists/{CF_LIST_ID}/items"
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = [{"ip": ip_address, "comment": f"Blocked by fail2ban"}]
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=10)
        result = response.json()
        if response.status_code == 200 and result.get("success"):
            print(f"成功将 {ip_address} 添加到IP列表")
            return True
        else:
            errors = result.get("errors", [])
            for error in errors:
                # 如果IP已存在,不算失败
                if error.get("code") == 1004:
                    print(f"{ip_address} 已存在于列表中")
                    return True
            print(f"API错误: {json.dumps(errors)}")
            return False
    except Exception as e:
        print(f"请求异常: {str(e)}")
        return False

def remove_ip_from_list(ip_address):
    """从Cloudflare账户级IP列表中移除IP地址"""
    # 首先需要获取列表项的ID
    get_url = f"{CF_API_BASE}/accounts/{CF_ACCOUNT_ID}/rules/lists/{CF_LIST_ID}/items"
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    
    try:
        # 搜索该IP对应的列表项
        search_params = {"search": ip_address}
        get_response = requests.get(get_url, headers=headers, params=search_params, timeout=10)
        get_result = get_response.json()
        
        if not get_result.get("success") or not get_result.get("result"):
            print(f"未找到 {ip_address} 的列表项")
            return False
        
        items_to_remove = []
        for item in get_result["result"]:
            if item.get("ip") == ip_address:
                items_to_remove.append({"id": item["id"]})
        
        if not items_to_remove:
            print(f"未找到匹配的IP项")
            return False
        
        # 执行删除操作
        delete_url = f"{CF_API_BASE}/accounts/{CF_ACCOUNT_ID}/rules/lists/{CF_LIST_ID}/items"
        delete_payload = {"items": items_to_remove}
        delete_response = requests.delete(delete_url, headers=headers, json=delete_payload, timeout=10)
        delete_result = delete_response.json()
        
        if delete_response.status_code == 200 and delete_result.get("success"):
            print(f"成功将 {ip_address} 从IP列表中移除")
            return True
        else:
            print(f"删除失败: {json.dumps(delete_result.get('errors', []))}")
            return False
            
    except Exception as e:
        print(f"请求异常: {str(e)}")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: ban  或 unban ")
        sys.exit(1)
    
    action = sys.argv[1]
    ip = sys.argv[2] if len(sys.argv) > 2 else None
    
    if not ip:
        print("缺少IP地址参数")
        sys.exit(1)
    
    if action == "ban":
        success = add_ip_to_list(ip)
    elif action == "unban":
        success = remove_ip_from_list(ip)
    else:
        print("未知操作,仅支持 ban 或 unban")
        sys.exit(1)
    
    sys.exit(0 if success else 1)

将上述脚本保存到/usr/local/bin/cf-firewall.py,并赋予执行权限。务必先手动测试一下脚本,确保ban和unban操作都能正常工作。测试时可以先使用一个无害的IP地址,比如192.0.2.1这个文档专用地址,验证API调用链路是否通畅。如果测试过程中遇到权限错误,需要回头检查API令牌的权限设置是否包含了编辑防火墙规则的权限。

在Cloudflare中创建并配置IP列表

脚本中用到了一个IP列表ID,这个列表需要在Cloudflare后台预先创建。登录Cloudflare仪表板,进入任意一个域名的管理页面,在左侧菜单中找到“安全性”->“WAF”,然后切换到“工具”标签页。在这里可以看到“IP列表”选项。点击创建新列表,名称可以取为“fail2ban_blocklist”,类型选择“IP”。创建完成后,点击进入列表详情,浏览器地址栏中会显示类似accounts/xxxxx/rules/lists/xxxxx的路径,从中可以提取出账户ID和列表ID。将这两个ID以及API令牌填入脚本对应的变量位置。这个列表后续会自动被fail2ban填充和清理,无需手动维护。

配置fail2ban监狱规则

有了执行器,接下来需要定义fail2ban的监控逻辑。这部分配置决定了哪些行为会被判定为攻击。以Nginx为例,需要创建一个新的过滤器来匹配CC攻击的典型特征。在/etc/fail2ban/filter.d/目录下创建一个名为nginx-cc.conf的文件:

[Definition]
failregex = ^ .* "(GET|POST|HEAD).* HTTP/.*" (403|404|429|502) .*$
            ^ .* "(GET|POST|HEAD).* (xmlrpc|wp-login|wp-admin|install|setup|config|admin).* HTTP/.*" .*$
ignoreregex =

这个过滤器定义了两组匹配规则。第一组针对短时间内产生大量HTTP错误状态码的IP,包括403禁止访问、404未找到、429请求过多、502错误网关。这些状态码通常意味着扫描器或攻击工具在尝试探测漏洞。第二组规则专门针对请求敏感路径的行为,比如WordPress的登录页面、XMLRPC接口、后台管理地址等。这些路径是CC攻击和暴力破解的高频目标。可以根据自己的应用特点添加更多路径特征。

接下来在/etc/fail2ban/jail.local文件中定义具体的监狱配置。如果该文件不存在,可以从jail.conf复制一份。添加以下配置块:

[nginx-cc]
enabled = true
port = http,https
filter = nginx-cc
logpath = /var/log/nginx/access.log
maxretry = 30
findtime = 10
bantime = 3600
action = cf-ban
         iptables-multiport[name=nginx-cc, port="http,https", protocol=tcp]

这里的参数需要根据实际业务流量仔细调整。maxretry设置为30,findtime设置为10秒,意味着同一个IP如果在10秒内触发了30次过滤器定义的模式,就会被封禁。这个阈值对于静态资源较多的网站来说可能偏低,建议先观察正常用户的访问日志,统计高峰期单个IP在10秒内的平均请求数,然后将阈值设置在该数值的3到5倍。bantime设置为3600秒即1小时,对于持续攻击的IP,这个时长可以适当延长到86400秒也就是24小时。action字段同时调用了cf-ban这个自定义动作和系统默认的iptables封禁,形成双重保险。

定义fail2ban的Cloudflare动作

fail2ban需要知道如何调用之前编写的Python脚本。在/etc/fail2ban/action.d/目录下创建cf-ban.conf文件:

[Definition]
actionstart = 
actionstop = 
actioncheck = 
actionban = /usr/local/bin/cf-firewall.py ban 
actionunban = /usr/local/bin/cf-firewall.py unban 

这个配置文件将fail2ban的标准动作接口映射到了自定义脚本上。actionban在IP被封禁时触发,actionunban在封禁到期时触发。需要注意的是,actionstart和actionstop留空,因为不需要在fail2ban服务启动或停止时执行特殊操作。如果希望fail2ban重启后能够同步封禁状态,可以在actionstart中添加从fail2ban数据库恢复封禁的逻辑,但这通常不是必需的,因为Cloudflare列表中的条目是持久化的。

日志路径匹配与权限问题排查

配置完成后,使用fail2ban-client reload重载配置,然后通过fail2ban-client status nginx-cc查看监狱状态。如果发现始终没有IP被封禁,最常见的原因是日志路径不匹配。很多Nginx安装会将访问日志放在/var/log/nginx/access.log,但如果使用了虚拟主机或多站点配置,日志路径可能不同。务必确保jail.local中logpath指向的是正确的日志文件。另一个常见问题是日志格式,fail2ban默认能够解析Nginx的combined格式,但如果自定义了日志格式,可能需要调整过滤器的正则表达式。

权限问题也经常导致封禁动作执行失败。fail2ban默认以root权限运行,但调用的Python脚本可能因为文件权限设置不当而无法执行。可以通过查看/var/log/fail2ban.log来诊断问题,日志中会详细记录每次封禁操作的执行结果和错误信息。如果看到类似“Permission denied”的错误,需要检查/usr/local/bin/cf-firewall.py是否具有可执行权限,以及脚本内部引用的API令牌是否有效。

精细化调优与误封处理

规则上线后,需要持续关注误封情况。搜索引擎爬虫、大型机构的NAT出口IP、CDN回源节点都可能因为请求频率较高而被误判。Cloudflare的IP列表支持添加备注,可以在封禁脚本中自动附加封禁原因和时间戳,方便后期排查。对于已知的善意爬虫,可以在fail2ban的ignoreip配置中添加白名单,支持CIDR格式的网段。例如:

[DEFAULT]
ignoreip = 127.0.0.1/8 10.0.0.0/8 172.16.0.0/12 192.168.0.0/16

如果网站使用了Cloudflare的CDN服务,获取真实客户端IP需要正确配置Nginx的real_ip模块。确保Nginx已经加载了ngx_http_realip_module,并在配置中设置了set_real_ip_from指令,将Cloudflare的所有IP段添加为可信代理。这样fail2ban分析日志时才能拿到真实的攻击者IP,而不是Cloudflare的边缘节点IP。

监控与效果评估

部署完成后,可以通过几个维度来评估防御效果。首先是服务器负载,在遭受攻击时,CPU和内存使用率应该保持在正常水平,因为大部分恶意请求被Cloudflare拦截在边缘。其次是带宽消耗,如果攻击流量以HTTP Flood为主,源服务器的入站流量应该显著下降。最后是Cloudflare仪表板中的防火墙事件日志,可以看到大量请求被标记为“由IP列表阻止”,这些就是fail2ban自动封禁的成果。建议定期检查封禁列表,清理长期未触发的条目,避免列表膨胀影响匹配性能。这套方案的优势在于完全自动化,无需人工干预,而且利用Cloudflare的全球网络进行分布式封禁,响应速度远超本地防火墙。