编写实时分析Nginx日志并自动调用nftables封禁异常访问IP的脚本

我来给你提供两个实用的脚本方案，刚好匹配你实时监控Nginx日志、自动封禁非合规访问IP的需求，一个是轻量的Bash脚本，另一个是更易扩展的Python脚本，你可以根据自己的偏好选择。

先确认前置条件 #

在运行脚本之前，你需要先确保nftables已经配置好对应的表和集合，否则脚本会执行失败。执行以下命令完成初始化：

# 创建inet类型的my_table表
nft add table inet my_table
# 创建存储恶意IP的badips集合（动态类型，支持自动添加）
nft add set inet my_table badips { type ipv4_addr; flags dynamic; }

方案一：Bash脚本（简单直接，快速部署） #

这个脚本用tail -F实时跟踪Nginx日志，配合awk提取关键信息，逻辑简洁，适合快速上线。

#!/bin/bash

# 定义需要忽略的请求路径（正则匹配）
IGNORED_PATHS="^/$|^/robots\.txt$|^/favicon\.ico$"
# 替换成你的Nginx访问日志路径，默认一般是/var/log/nginx/access.log
NGINX_LOG="/var/log/nginx/access.log"

# 实时监控日志，处理每一行内容
tail -F "$NGINX_LOG" | while read -r line; do
    # 提取客户端IP地址
    CLIENT_IP=$(echo "$line" | awk '{print $1}')
    # 提取请求路径（从日志的引号段中解析）
    REQUEST_PATH=$(echo "$line" | awk -F'"' '{print $2}' | awk '{print $2}')

    # 检查请求路径是否属于忽略列表
    if [[ ! "$REQUEST_PATH" =~ $IGNORED_PATHS ]]; then
        # 先检查IP是否已经在badips集合中，避免重复执行添加命令
        if ! nft list set inet my_table badips | grep -q "$CLIENT_IP"; then
            echo "$(date +'%Y-%m-%d %H:%M:%S') - 封禁异常IP: $CLIENT_IP，访问路径: $REQUEST_PATH"
            nft add element inet my_table badips { "$CLIENT_IP" }
        fi
    fi
done

脚本说明 #

1. tail -F会持续跟踪日志文件，即使日志滚动（比如logrotate切割日志）也能自动切换到新文件；
2. 用awk精准提取IP和请求路径，避免日志格式小变化导致的解析错误；
3. 增加了重复IP检查，减少不必要的nft命令调用，提升效率。

方案二：Python脚本（更灵活，易扩展） #

如果后续你需要添加更复杂的逻辑（比如记录封禁日志、定时清理过期IP、处理IPv6等），Python脚本会更适合。

import re
import subprocess
import shlex
from datetime import datetime

# 配置参数，根据你的实际情况修改
NGINX_LOG_PATH = "/var/log/nginx/access.log"
# 正则匹配需要忽略的请求路径
IGNORED_PATHS = re.compile(r'^/$|^/robots\.txt$|^/favicon\.ico$')
NFT_TABLE_SET = "inet my_table badips"

def is_ip_already_blocked(ip: str) -> bool:
    """检查IP是否已经在nftables的badips集合中"""
    cmd = f"nft list set {NFT_TABLE_SET}"
    result = subprocess.run(shlex.split(cmd), capture_output=True, text=True)
    return ip in result.stdout

def block_ip(ip: str) -> None:
    """将IP添加到nftables的封禁集合"""
    cmd = f"nft add element {NFT_TABLE_SET} {{ {ip} }}"
    try:
        subprocess.run(shlex.split(cmd), check=True, capture_output=True)
        print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 成功封禁IP: {ip}")
    except subprocess.CalledProcessError as e:
        print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 封禁IP {ip}失败: {e.stderr}")

def monitor_nginx_log():
    """实时监控Nginx访问日志"""
    cmd = ["tail", "-F", NGINX_LOG_PATH]
    with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as proc:
        for line in proc.stdout:
            line = line.strip()
            if not line:
                continue
            
            # 提取客户端IP
            ip_match = re.match(r'^(\d+\.\d+\.\d+\.\d+)', line)
            if not ip_match:
                continue
            client_ip = ip_match.group(1)
            
            # 提取请求路径（兼容GET/POST请求）
            path_match = re.search(r'"(GET|POST) (.*?) HTTP/1\.1"', line)
            if not path_match:
                continue
            request_path = path_match.group(2)
            
            # 检查是否是需要忽略的路径
            if IGNORED_PATHS.match(request_path):
                continue
            
            # 未被封禁则执行封禁操作
            if not is_ip_already_blocked(client_ip):
                block_ip(client_ip)

if __name__ == "__main__":
    try:
        print("开始实时监控Nginx日志...")
        monitor_nginx_log()
    except KeyboardInterrupt:
        print("\n监控已手动停止")

脚本说明 #

1. 用正则表达式更精准地解析日志内容，兼容GET和POST请求；
2. 完善的错误处理，封禁失败时会输出错误信息；
3. 自带时间戳日志，方便后续排查问题；
4. 代码模块化，后续可以轻松添加新功能（比如把封禁记录写入日志文件）。

额外优化建议 #

1. 后台运行脚本：可以把脚本配置成systemd服务，让它开机自动启动并后台运行。以Bash脚本为例，创建/etc/systemd/system/nginx-ip-blocker.service文件：

   [Unit]
   Description=Real-time Nginx Log Monitor for IP Blocking
   After=network.target nginx.service
   
   [Service]
   Type=simple
   ExecStart=/path/to/your/block-script.sh  # 替换成你的脚本实际路径
   Restart=always
   User=root
   
   [Install]
   WantedBy=multi-user.target
   

   然后执行以下命令启用服务：

   systemctl daemon-reload
   systemctl enable --now nginx-ip-blocker.service
   

2. 避免误封：如果你的网站有正常用户访问其他路径，建议先测试脚本，或者添加白名单逻辑（比如把信任的IP排除在封禁范围外）。

备注：内容来源于stack exchange，提问作者somenxavier