通过分析Nginx日志实现阿里云ESA自动添加黑名单
因为使用了阿里云的ESA作为网站的CDN，所有请求都是通过ESA转发到源站的，这就导致了直接在源站封禁恶意IP没用了，所以现在防护都放在了阿里云ESA的Web应用防火墙中。
虽然阿里云ESA防火墙中没有提供黑名单功能，但是我们可以使用其自定义规则与全局分组功能，实现IP黑名单功能，然后再通过阿里云提供的OpenAPI接口实现本地分析Nginx，发现异常访问实时调用阿里云API将其自动添加到黑名单的功能。
下面记录一下具体的配置步骤供参考。
一、阿里云ESA配置
1.新建全局分组
首先需要建立一个分组blacklist，将黑名单IP都保存在分组内。
新建分组访问路径在阿里云ESA控制台首页，全局配置->分组，点击新建分组按钮：
2.新建防火墙规则
然后需要在防火墙中新建一个自定义规则，规则内引用我们新建的分组，设置客户端IP在分组内的直接拦截请求。
防火墙访问路径在安全防护->WAF页面内，点击自定义规则->新建规则：
规则设置为如果请求的客户端IP在blacklist分组中则执行拦截操作：
3.通过OpenAPI平台获取接口信息
阿里云OpenAPI平台地址：https://api.aliyun.com/
我们需要从OpenAPI平台找到ESA分组相关的接口名称和参数，在OpenAPI首页搜索ESA相关API：
服务区域列表页面可以阿里云ESA的API服务地址（esa.cn-hangzhou.aliyuncs.com）：
API文档中可以找到所有ESA支持的接口，我们主要使用的是下面这3个API接口：
接口信息如下：
UpdateList：更新自定义列表
GetList：获取单个自定义列表详情
ListLists：列举自定义列表
4.获取分组ID
分组ID可以通过OpenAPI平台的测试页面获取，直接在API详情页点击去调用：
直接填写我们创建的列表名称后点击调用就能拿到分组的ID等信息了：
5.获取阿里云AK和授权
因为阿里云OpenAPI是通过AccessKey作为访问凭据的，所以需要在RAM访问控制台新建用户并授权接口权限才可以使用。
RAM访问控制台地址：https://ram.console.aliyun.com/
在RAM控制台中新建黑名单用户，创建的时候勾选使用AccessKey访问：
创建完成可以拿到AccessKey ID和AccessKey Secret：
然后在权限策略里新增ESA分组列表相关权限：
在授权页面将新增的权限策略授权给新建的用户：
这里阿里云ESA的操作部分就完成了，下面的数据我们应该都获取到了：
a.阿里云ESA的API服务地址：esa.cn-hangzhou.aliyuncs.com
b.阿里云用户AccessKey ID和AccessKey Secret
c.需要使用的接口信息
d.全局分组的ID
二、Nginx日志格式调整
为了方便获取Nginx访问日志的状态码和客户端IP，需要调整Nginx日志格式：
a.将status字段调整到末位；
b.将http_x_forwarded_for字段调整到正数第二位，并使用双引号括起来。
我的Nginx日志格式配置如下：
log_format main '$remote_addr "$http_x_forwarded_for" $remote_user [$time_local] "$request" '
'$body_bytes_sent "$http_referer" "$http_user_agent" '
'"$request_time" "$upstream_response_time" "$scheme" "$host" $status'; 三、使用Agent编写代码
拿到了阿里云API参数和Nginx日志格式后，接下来就是核心代码的编写了。这里我们可以借助AI编程助手Agent来辅助完成。
Agent工具这里我们选择了腾讯的WorkBuddy，其内置的Hy3大模型限时免费，暂时可以免费使用到2026年8月31日，不知道后续会不会延期，不过当前是够用了。
1.向Agent描述任务需求
我将与AI交互的需求内容整理一下，总结如下：
我需要做一个实时分析Nginx日志自动添加黑名单到阿里云ESA的功能，帮我用Python实现以下功能：
1.实时监听nginx日志，根据日志中请求状态码判断，状态码是404、400、403、500的所属ip为黑名单；
2.如果请求IP为黑名单，通过api更新阿里云esa分组中的黑名单自定义列表，更新之前需要先获取原列表内容在后面追加；
3.需要记录日志，记录下规则触发，接口调用等关键信息；
4.nginx日志中状态码在最后一个字段，客户端IP在第二个字段；
5.因为阿里云SDK太占用内存了，需要使用原生HTTP调用阿里云ESA； 注：可以把需要的参考链接一块发给Agent处理。
2.Agent生成的完整代码
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ESA 黑名单监控（原生 V3 签名，无 SDK 依赖）。
监听 nginx 日志，对状态码 400/403/404/500 的 IP 加入 ESA 黑名单（先取原数据再追加，不覆盖）。
脚本启动即开始监控，日志路径仅从环境变量 NGINX_LOG_PATH 读取。
环境变量: ESA_BLACKLIST_LIST_ID(必填) ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET ESA_ENDPOINT
BLOCK_STATUS_CODES(默认400,403,404,500) NGINX_LOG_PATH
"""
import os, sys, json, time, uuid, ssl, ipaddress, hashlib, hmac, logging
import urllib.request, urllib.error, urllib.parse
from datetime import datetime, timezone
from logging.handlers import RotatingFileHandler
# ---------- 日志 ----------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
log = logging.getLogger('blacklist'); log.setLevel(logging.INFO)
_h = RotatingFileHandler(os.path.join(BASE_DIR, 'blacklist.log'),
maxBytes=10 * 1024 * 1024, backupCount=5, encoding='utf-8')
_h.setFormatter(logging.Formatter('%(asctime)s | %(message)s'))
log.addHandler(_h); log.addHandler(logging.StreamHandler())
# ---------- 配置 ----------
ESA_ENDPOINT = os.environ.get('ESA_ENDPOINT', 'esa.cn-hangzhou.aliyuncs.com')
LIST_ID = int(os.environ.get('ESA_BLACKLIST_LIST_ID', '0'))
LIST_NAME = os.environ.get('ESA_BLACKLIST_LIST_NAME', 'blacklist')
LIST_DESC = os.environ.get('ESA_BLACKLIST_LIST_DESC', '黑名单')
BLOCK_CODES = {c.strip() for c in os.environ.get('BLOCK_STATUS_CODES', '400,403,404,500').split(',') if c.strip()}
POLL = float(os.environ.get('BLACKLIST_POLL_INTERVAL', '0.3')) # 日志轮询间隔（秒），越小越实时
# ---------- V3 签名 + 请求（ACS3-HMAC-SHA256，零 SDK）----------
def _quote(s):
return urllib.parse.quote(str(s), safe='~')
def _sign_v3(method, query, headers, ak, sk):
cq = '&'.join(f'{_quote(k)}={_quote(v)}' for k, v in sorted(query.items())) if query else ''
keys = sorted(headers)
ch = ''.join(f'{k}:{str(headers[k]).strip()}\n' for k in keys)
sh = ';'.join(keys)
cr = '\n'.join([method.upper(), '/', cq, ch, sh, headers['x-acs-content-sha256']])
sts = 'ACS3-HMAC-SHA256\n' + hashlib.sha256(cr.encode()).hexdigest()
sig = hmac.new(sk.encode(), sts.encode(), hashlib.sha256).hexdigest()
return f'ACS3-HMAC-SHA256 Credential={ak},SignedHeaders={sh},Signature={sig}'
def _ssl_ctx():
ctx = ssl.create_default_context()
b = os.environ.get('ALIYUN_CA_BUNDLE')
if b and os.path.exists(b):
ctx.load_verify_locations(cafile=b)
return ctx
def _request(action, method='GET', query=None, body=None):
ak = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '').strip()
sk = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '').strip()
if not ak or not sk:
raise RuntimeError('未配置 AK/SK')
body_bytes = urllib.parse.urlencode(body).encode() if body else None
h = {
'host': ESA_ENDPOINT,
'x-acs-action': action,
'x-acs-version': '2024-09-10',
'x-acs-date': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
'x-acs-signature-nonce': uuid.uuid4().hex,
}
if body_bytes:
h['content-type'] = 'application/x-www-form-urlencoded'
h['x-acs-content-sha256'] = hashlib.sha256(body_bytes or b'').hexdigest()
qs = '&'.join(f'{_quote(k)}={_quote(v)}' for k, v in sorted((query or {}).items()))
url = f'https://{ESA_ENDPOINT}/' + (f'?{qs}' if qs else '')
h['Authorization'] = _sign_v3(method, query or {}, h, ak, sk)
req = urllib.request.Request(url, data=body_bytes, method=method)
for k, v in h.items():
req.add_header(k, v)
t0 = time.time()
try:
with urllib.request.urlopen(req, timeout=10, context=_ssl_ctx()) as r:
data = json.loads(r.read().decode('utf-8', 'replace'))
log.info(f'阿里云调用 | Action={action} | 耗时={ (time.time()-t0)*1000:.0f}ms | '
f'RequestId={data.get("RequestId", "")}')
return data
except urllib.error.HTTPError as e:
body = e.read().decode('utf-8', 'replace')
log.error(f'阿里云调用失败 | Action={action} | HTTP={e.code} | 响应={body[:500]}')
raise RuntimeError(f'HTTP {e.code}: {body}')
except Exception as e:
log.error(f'阿里云调用异常 | Action={action} | 错误={e}')
raise RuntimeError(f'请求失败: {e}')
# ---------- 黑名单读写 ----------
def _get():
if not LIST_ID:
raise RuntimeError('未配置 ESA_BLACKLIST_LIST_ID')
r = _request('GetList', query={'Id': str(LIST_ID)})
return ([str(i) for i in (r.get('Items') or [])],
r.get('Name') or LIST_NAME, r.get('Description') or LIST_DESC)
def _update(items, name, desc):
"""写回整个名单（全量 Items 字符串）。"""
return _request('UpdateList', method='POST', body={
'Id': str(LIST_ID), 'Name': name, 'Description': desc,
'Items': json.dumps(list(items), separators=(',', ':')),
})
# ---------- 实时监听 ----------
def parse(line):
p = line.split()
if len(p) < 2:
return None
ip = p[1].strip('"')
try:
ipaddress.ip_address(ip)
except Exception:
return None
for t in reversed(p):
t = t.strip('"')
if t in BLOCK_CODES:
return ip, t
return None
def monitor():
path = os.environ.get('NGINX_LOG_PATH', '').strip()
if not path:
log.error('未设置环境变量 NGINX_LOG_PATH，无法启动监控')
sys.exit(1)
if not os.path.exists(path):
log.error(f'日志不存在: {path}')
sys.exit(1)
log.info(f'监控启动 | 日志={path} | 列表ID={LIST_ID} | 监控状态码={sorted(BLOCK_CODES)} | '
f'轮询间隔={POLL}s')
f = open(path, 'r', encoding='utf-8', errors='replace')
f.seek(0, 2)
inode = os.stat(path).st_ino
try:
while True:
try:
st = os.stat(path)
except FileNotFoundError:
f = None
else:
if f is None or st.st_ino != inode or st.st_size < f.tell():
if f:
f.close()
f = open(path, 'r', encoding='utf-8', errors='replace')
inode = st.st_ino
log.info(f'日志文件重开（切割/截断）| 新inode={inode}')
if f:
for line in f:
r = parse(line)
if not r:
continue
ip, code = r
log.info(f'命中异常请求 | 真实IP={ip} | 状态码={code}')
try:
cur, name, desc = _get()
if ip in cur:
log.info(f'命中异常请求（黑名单中已存在，跳过）| 真实IP={ip} | 状态码={code}')
continue
cur.append(ip)
resp = _update(cur, name, desc)
rid = resp.get('RequestId', '')
log.info(f'写入黑名单成功 | 列表ID={LIST_ID} | 新增=1 | 原条数={len(cur)-1} | 新条数={len(cur)} | RequestId={rid} | IP={ip}')
except Exception as e:
log.error(f'写入黑名单失败 | 列表ID={LIST_ID} | 待写入={ip} | 错误={e}')
time.sleep(POLL)
except KeyboardInterrupt:
log.info('收到中断信号，已退出')
# ---------- 入口 ----------
if __name__ == '__main__':
monitor()
3.AI生成的服务文件
[Unit]
Description=ESA Blacklist Monitor (nginx log -> ESA)
After=network.target
[Service]
User=root
WorkingDirectory=/data/www/ip
ExecStart=/data/python3/bin/python3 /data/www/ip/blacklist.py
Restart=on-failure
RestartSec=5s
# ===== 阿里云 ESA 黑名单配置（仅保留必填项） =====
# 黑名单自定义列表 ID
Environment="ESA_BLACKLIST_LIST_ID=0"
# nginx 访问日志路径
Environment="NGINX_LOG_PATH=/var/log/nginx/access.log"
# 阿里云访问凭据（建议使用 RAM 子账号，仅授予 ESA 相关权限）
Environment="ALIBABA_CLOUD_ACCESS_KEY_ID=你的AccessKeyId"
Environment="ALIBABA_CLOUD_ACCESS_KEY_SECRET=你的AccessKeySecret"
# ===== 可选覆盖（均有默认值，需要时取消注释） =====
# Environment="ESA_ENDPOINT=esa.cn-hangzhou.aliyuncs.com"
# Environment="ESA_BLACKLIST_LIST_NAME=blacklist"
# Environment="ESA_BLACKLIST_LIST_DESC=黑名单"
# Environment="BLOCK_STATUS_CODES=400,403,404,500"
# Environment="BLACKLIST_POLL_INTERVAL=0.3"
[Install]
WantedBy=multi-user.target
四、部署代码到服务器
1.python环境部署
2.将代码上传到服务器
将AI生成的代码保存为blacklist.py文件，上传的服务器对应目录中。
3.将服务文件上传到服务器中
将服务文件保存为blacklist.service文件，上传到服务器/usr/lib/systemd/system/目录下，要注意修改一下环境变量：
NGINX_LOG_PATH：nginx日志目录。
ESA_BLACKLIST_LIST_ID：全局分组ID。
ALIBABA_CLOUD_ACCESS_KEY_ID：用户AccessKey ID。
ALIBABA_CLOUD_ACCESS_KEY_SECRET：用户AccessKey Secret。
BLOCK_STATUS_CODES：需要封禁的状态码。
4.使用systemd管理启动服务
systemctl daemon-reload
systemctl enable blacklist
systemctl start blacklist 五、功能测试
服务启动后可以测试了，访问一个404页面：
查看黑名单详情：
查看系统日志：
已经成功加入了黑名单列表，这样阿里云ESA自动添加黑名单的功能就算成功上线了。
扩展说明：
1.需要注意下这个全局列表最大支持5000个IP。
2.参考文档1：https://help.aliyun.com/zh/sdk/product-overview/v3-request-structure-and-signature
3.参考文档2：https://help.aliyun.com/zh/sdk/product-overview/request-structure-and-signature/
正文到此结束
- 本文标签： Python Linux
- 本文链接： https://blog.eyyyye.com/article/152
- 版权声明： 本文由爱做梦的比特原创发布，转载请遵循《署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)》许可协议授权
热门推荐
相关文章
关于我
运维之路
专注于探索实际应用问题的解决方案
