UBUNTU宝塔AMP环境,检测网站访问日志并防火墙拉黑

本次更新概要（auto_ban_4xx_ufw.sh）20260910

一、拉黑规则调整

检测项	更新前	更新后
404 / 403	连续 10 次即封，无时间限制（跨几天累计也算）	连续（仅 200 成功请求打断）满 10 次，且第 1 次与第 10 次首末落在 30 分钟（1800 秒）内才封
301	同一 IP 同一秒内超 30 次封	同一 IP 同一秒内超 20 次（即第 21 次）封

配置区新增/调整（L5-L10）：

MAX_ERR=10（连续次数，第 10 次即触发）
ERR_WINDOW_SEC=1800（新增，30 分钟时间窗，含 1800 秒边界）
BURST_301=20（原 30）

二、检测算法重写（awk 扫描段）

滑动窗口精确判定：每个 IP 用固定 10 槽环形数组只保留最近 10 次错误，任一时段有 10 次挤在 30 分钟内即命中——慢速扫描（如每 201 秒一次）不再误伤，快速爆发不漏抓。
命中后固定窗口起点，窗内后续错误计入通知明细，超窗不再延展；被 200 打断后重新计数。

三、速度优化

301 计数桶内存膨胀修复：旧版每个 200 请求都建一个空桶，高流量站点 END 阶段要遍历百万级死键；改为仅在桶存在时 delete，内存与遍历量随真实 301 事件数走。
4xx 时间戳由每行解析两遍（shortts + ts2epoch）合并为 parse_ts 一遍。
200（最常见行）判断前置，减少规则匹配开销。

四、其他

拉黑备注文案同步更新为「30分钟内连续10次404/403或1秒超20次301(200重置)」。
awk 输出行格式、bash 解析、C/B/A 段聚合合并、清理、宝塔 DB+ipset 双写、通知模式等其余逻辑零改动。
验证：17 个边界用例（1800/1801 秒边界、9 次不封、200 打断与重启、命中后续窗、301 的 20/21 次边界、蜘蛛 UA、404/403 互不打断等）全部通过，bash -n 语法检查通过。

脚本文件/root/auto_ban_4xx_ufw.sh内容如下:

#!/bin/bash
# ===================== 自定义配置区 =====================
LOG_DIR="/www/wwwlogs"
LOG_FILE_PATTERN="*-access_log"
MAX_ERR=10                 # 连续404/403达"此数量"(>=)且首末落在ERR_WINDOW_SEC窗内才拉黑(第10次即触发)
ERR_WINDOW_SEC=1800        # 连续404/403时间窗(秒)=30分钟;200成功访问打断连续段重新计
FIND_MTIME=-2
MERGE_THRESHOLD=3          # 同一/16(B段)下/24(C段)数达到此值即合并为/16
A_MERGE_THRESHOLD=30       # 同一/8(A段)下/16(B段)数超过此值即合并为/8
BURST_301=20               # 同一IP同一秒301次数"超过"此值(即>=21)才拉黑
CRAWLER_UA="Googlebot|Baiduspider|Yandex|Sogou|Bingbot"
EXCLUDE_SEG=("38.190.0.0/16" "27.43.0.0/16")
LOG_FILE="/tmp/ban_4xx.log"
WECOM_WEBHOOK="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=写上自己的KEY"
SERVER_TAG="香港服务器"
# ---- 通知模式开关 NOTIFY_MODE ----
#   控制每次扫描结果(有变更/无变更)的日志落盘与企业微信推送去向：
#   1: 有变更→日志+微信;  无变更→静默(都不发)
#   2: 有变更→日志+微信;  无变更→仅日志
#   3: 有变更→日志+微信;  无变更→仅微信
#   5: 全部(无论有无变更)→仅日志
#   6: 全部(无论有无变更)→仅微信
#   7: 有变更→仅微信;    无变更→静默
#   8: 有变更→仅日志;    无变更→静默
#   异常退出(目录缺失等)按"变更级"处理:1/2/3/5/8 落日志,6/7 仅微信
NOTIFY_MODE=7
# ---- 日志轮转阈值:超过任一即裁剪旧行 ----
LOG_MAX_DAYS=30            # 保留最近天数
LOG_MAX_BYTES=1048576      # 保留上限字节(1MB)
# =======================================================

# 段位术语: C段=/24(a.b.c.0/24)  B段=/16(a.b.0.0/16)  A段=/8(a.0.0.0/8)
# 合并链路: C段聚合→B段(MERGE_THRESHOLD) → B段聚合→A段(A_MERGE_THRESHOLD)

# 【内存化】异常IP用内存数组暂存，各阶段处理完即 unset，不再落 /tmp 临时文件
declare -a ips_4xx ips_301

# 判断 big CIDR 是否包含 small CIDR(按网络前缀掩码比较)
# 【速度优化】纯bash算术+内置read解析,无$(ip2dec)子shell fork,清理阶段热路径零fork
cidr_contains() {
    local big="$1" small="$2"
    local b_ip=${big%/*} b_mask=${big#*/}
    local s_ip=${small%/*}
    [[ ! "$b_ip" =~ ^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$ ]] && return 1
    [[ ! "$s_ip" =~ ^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$ ]] && return 1
    local ba bb bc bd sa sb sc sd b_dec s_dec mask_dec
    IFS=. read -r ba bb bc bd <<< "$b_ip"
    IFS=. read -r sa sb sc sd <<< "$s_ip"
    b_dec=$(( (ba << 24) + (bb << 16) + (bc << 8) + bd ))
    s_dec=$(( (sa << 24) + (sb << 16) + (sc << 8) + sd ))
    mask_dec=$(( (0xFFFFFFFF << (32 - b_mask)) & 0xFFFFFFFF ))
    (( (b_dec & mask_dec) == (s_dec & mask_dec) ))
}

is_exclude() {
    local seg="$1"
    for e in "${EXCLUDE_SEG[@]}";do
        if cidr_contains "$e" "$seg" ;then
            return 0
        fi
    done
    return 1
}

rule_exists() {
    local seg="$1"
    # 真相源:面板DB firewall_ip 表(面板自动同步DB→ipset)
    # 优先用内存缓存(cache_ipset_rules 已填充),未缓存则回退单次DB查询
    if [[ -n "${CACHE_LOADED+x}" ]]; then
        [[ -n "${RULES_EXIST[$seg]+x}" ]] && return 0
        return 1
    fi
    # 回退:直接查DB
    command -v sqlite3 >/dev/null 2>&1 && [[ -f "$BT_FW_DB" ]] \
        && sqlite3 "$BT_FW_DB" \
            "SELECT 1 FROM firewall_ip WHERE address='${seg//\'/\'}'" 2>/dev/null \
        | grep -q 1
    return $?
}

covered_by_existing_16() {
    local seg="$1"
    # /24 只可能被同 a.b 的 /16 覆盖,先 O(1) 哈希查该/16是否存在
    local a b net16
    IFS=. read -r a b _ <<< "${seg%/*}"
    net16="${a}.${b}.0.0/16"
    [[ -n "${RULES_EXIST[$net16]+x}" ]] && return 0
    return 1
}

# /16 或 /24 是否已被现有 /8(A段) 覆盖:同 a 即可,先 O(1) 哈希查该/8是否存在
covered_by_existing_8() {
    local seg="$1"
    local a net8
    IFS=. read -r a _ <<< "${seg%/*}"
    net8="${a}.0.0.0/8"
    [[ -n "${RULES_EXIST[$net8]+x}" ]] && return 0
    return 1
}

# 【真相源:宝塔面板DB firewall_ip表 + ipset in_bt_user_drop_ipset 双写】
#   ⚠️ 关键发现: 面板DB→ipset自动同步只在UI操作时触发,
#      命令行 sqlite3 INSERT 不会自动同步到 ipset (已实测验证)
#   所以 bt_ban/bt_unban 必须双写: DB给面板显示 + ipset给即时拦截
#   策略: 只写 ipset 集合 (不碰 ufw/iptables 链)
#         → comModel.py 统计遍历时不会多计,面板数字=DB count
#   后续所有 rule_exists / covered_by_existing_* 都走 O(1) 内存查询,零fork
declare -A RULES_EXIST
declare -a ACTION_MSGS
CACHE_LOADED=""
cache_ipset_rules() {
    local cidr
    RULES_EXIST=()
    # 从面板DB firewall_ip 表读取所有 CIDR
    # sqlite3 不可用/DB不存在视为环境缺失,静默返回
    if command -v sqlite3 >/dev/null 2>&1 && [[ -f "$BT_FW_DB" ]]; then
        while IFS= read -r cidr; do
            [[ -z "$cidr" ]] && continue
            # 只接受合法 CIDR(带 /8 /16 /24 后缀)
            [[ "$cidr" =~ /(8|16|24)$ ]] && RULES_EXIST["$cidr"]=1
        done < <(sqlite3 "$BT_FW_DB" "SELECT DISTINCT address FROM firewall_ip;" 2>/dev/null)
    fi
    CACHE_LOADED=1
}

# 企业微信群机器人通知
# 入参: markdown文本(可含换行,用python3做JSON转义,避免shell转义地狱)
send_wecom_notify() {
    local content="$1"
    [[ -z "$WECOM_WEBHOOK" ]] && return 0
    [[ -z "$content" ]] && return 0
    # 企微markdown上限4096字节:明细加长后防超长被整条拒收(按字符截,避免切破多字节字符)
    if (( ${#content} > 1000 )); then
        content="${content:0:1000}
(内容过长已截断)"
    fi
    local payload
    payload=$(printf '%s' "$content" | python3 -c 'import json,sys;print(json.dumps({"msgtype":"markdown","markdown":{"content":sys.stdin.read()}}))' 2>/dev/null)
    [[ -z "$payload" ]] && return 0
    curl -s -o /dev/null --max-time 10 -X POST "$WECOM_WEBHOOK" -H 'Content-Type: application/json' -d "$payload" 2>/dev/null || true
}

# 日志轮转:超过 LOG_MAX_DAYS 天 或 LOG_MAX_BYTES 字节则删旧行
#   1) 按天裁剪:删除日期早于 cutoff 的带时间戳行(无时间戳行如分隔符保留)
#   2) 按字节裁剪:仍超上限则只保留尾部完整行
trim_log_file() {
    [[ ! -f "$LOG_FILE" ]] && return 0
    local cutoff tmp size
    cutoff=$(date -d "-${LOG_MAX_DAYS} days" '+%Y-%m-%d' 2>/dev/null) || return 0
    tmp="${LOG_FILE}.trim.$$"
    awk -v cut="$cutoff" '
        /^[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]/ && (substr($0,1,10) < cut) { next }
        { print }
    ' "$LOG_FILE" > "$tmp" 2>/dev/null && mv -f "$tmp" "$LOG_FILE"

    size=$(stat -c %s "$LOG_FILE" 2>/dev/null || echo 0)
    if (( size > LOG_MAX_BYTES )); then
        tail -c "$LOG_MAX_BYTES" "$LOG_FILE" 2>/dev/null | tail -n +2 > "$tmp" 2>/dev/null \
            && mv -f "$tmp" "$LOG_FILE"
    fi
    rm -f "$tmp" 2>/dev/null
    return 0
}

# 异常退出:按 NOTIFY_MODE 输出诊断(1/2/3/5/8 落日志;6/7 仅微信)
fail_exit() {
    local msg="$1" now_str
    now_str=$(date '+%Y-%m-%d %H:%M:%S')
    case "$NOTIFY_MODE" in
        6|7)
            send_wecom_notify "## ${SERVER_TAG}-自动拦截-异常
**时间**: $now_str
**错误**: $msg"
            ;;
        *)
            echo "====================> $now_str 扫描启动 <====================" >> "$LOG_FILE"
            echo "错误：$msg，本次终止" >> "$LOG_FILE"
            ;;
    esac
    exit 1
}

# 统一日志落盘+微信推送(按 NOTIFY_MODE 分发):
#   有变更:「ts 扫描N个文件共M行结束,用时X秒;<动作以--拼接>!」+ 微信变更明细
#   无变更:「ts 扫描N个文件共M行结束,用时X秒，无规则变更!」(+ 可选微信)
flush_log_buffer() {
    local now_epoch now_str dur has_changes do_log do_wecom log_line
    now_epoch=$(date +%s)
    dur=$(( now_epoch - START_EPOCH ))
    [[ $dur -lt 1 ]] && dur=1
    now_str=$(date '+%Y-%m-%d %H:%M:%S')
    has_changes=0
    [[ ${#ACTION_MSGS[@]} -gt 0 ]] && has_changes=1

    # 按 NOTIFY_MODE + 是否有变更 决定日志/微信开关
    do_log=0; do_wecom=0
    case "$NOTIFY_MODE" in
        1) [[ $has_changes -eq 1 ]] && { do_log=1; do_wecom=1; } ;;
        2) if [[ $has_changes -eq 1 ]]; then do_log=1; do_wecom=1; else do_log=1; fi ;;
        3) if [[ $has_changes -eq 1 ]]; then do_log=1; do_wecom=1; else do_wecom=1; fi ;;
        5) do_log=1 ;;
        6) do_wecom=1 ;;
        7) [[ $has_changes -eq 1 ]] && do_wecom=1 ;;
        8) [[ $has_changes -eq 1 ]] && do_log=1 ;;
        *) do_log=1; do_wecom=1 ;;
    esac

    if [[ $has_changes -eq 1 ]]; then
        # 单行日志摘要:动作以 -- 拼接
        local changes="" first=1 msg
        for msg in "${ACTION_MSGS[@]}"; do
            if (( first )); then changes="$msg"; first=0
            else changes="$changes--$msg"; fi
        done
        log_line="$now_str 扫描${LOG_COUNT}个文件共${LOG_LINES_TOTAL}行结束,用时${dur}秒;${changes}!"
        if [[ $do_log -eq 1 ]]; then
            echo "$log_line" >> "$LOG_FILE"
            trim_log_file
        fi
        if [[ $do_wecom -eq 1 ]]; then
            # 真实换行拼字符串,python转义
            local md="" first_md=1
            for msg in "${ACTION_MSGS[@]}"; do
                if (( first_md )); then md="- $msg"; first_md=0
                else md="$md
- $msg"; fi
            done
            send_wecom_notify "## ${SERVER_TAG}-自动拦截通知
**时间**: $now_str
**扫描**: $LOG_COUNT个文件/共$LOG_LINES_TOTAL行,用时$dur秒
**变更明细**:
$md"
        fi
    else
        log_line="$now_str 扫描${LOG_COUNT}个文件共${LOG_LINES_TOTAL}行结束,用时${dur}秒，无规则变更!"
        if [[ $do_log -eq 1 ]]; then
            echo "$log_line" >> "$LOG_FILE"
            trim_log_file
        fi
        if [[ $do_wecom -eq 1 ]]; then
            send_wecom_notify "## ${SERVER_TAG}-自动拦截通知
**时间**: $now_str
**扫描**: $LOG_COUNT个文件/共$LOG_LINES_TOTAL行,用时$dur秒
**状态**: 无规则变更"
        fi
    fi
}

# 【防并发】锁文件:同一时刻只允许一个实例运行
# flock 基于文件描述符,进程退出自动释放,无需清理
LOCK_FILE="/tmp/auto_ban_4xx_ufw.lock"
exec 9>"$LOCK_FILE"
if ! flock -n 9; then
    # 另一个实例正在跑,直接静默退出(不发通知,避免重复推送)
    exit 0
fi

# 预检查：LOG_DIR 是否存在
if [ ! -d "$LOG_DIR" ]; then
    fail_exit "日志目录 $LOG_DIR 不存在"
fi
# 统计匹配文件个数（行数由扫描阶段的awk一并输出，减少一次IO遍历）
LOG_COUNT=$(find "$LOG_DIR" -maxdepth 1 -type f -name "$LOG_FILE_PATTERN" 2>/dev/null | wc -l)
if [ "$LOG_COUNT" -eq 0 ]; then
    fail_exit "日志目录 $LOG_DIR 下未找到匹配 $LOG_FILE_PATTERN 的文件"
fi

START_EPOCH=$(date +%s)

# 【内存化】单遍awk扫描，结果带标签输出到stdout，bash捕获后分发到内存数组
# 不再写 TMP_4XX_IP/TMP_301_IP/TMP_LINES_FILE 三个临时文件，省3次写盘+2次cat读盘
scan_out=$(find "$LOG_DIR" -maxdepth 1 -type f -name "$LOG_FILE_PATTERN" -mtime "$FIND_MTIME" -print0 2>/dev/null \
  | xargs -0 cat 2>/dev/null \
  | awk \
      -v crawler="$CRAWLER_UA" \
      -v burst="$BURST_301" \
      -v max="$MAX_ERR" \
      -v win="$ERR_WINDOW_SEC" \
    '
    BEGIN {
        split("Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec", MN, " ")
        for (i = 1; i <= 12; i++) MON[MN[i]] = sprintf("%02d", i)
        split("0 31 59 90 120 151 181 212 243 273 304 334", MDAYS, " ")
    }
    # 时间戳缩写: [06/Sep/2026:10:23:45 +0800 → 09-06_10:23:45 (无空格,便于按空格切分字段)
    function shortts(t,   a, mo) {
        sub(/^\[/, "", t)
        if (split(t, a, /[\/:]/) < 6) return t
        mo = MON[a[2]]
        if (mo == "") return t
        return mo "-" a[1] "_" a[4] ":" a[5] ":" substr(a[6], 1, 2)
    }
    # 一次解析同时产出近似epoch(G_EP)+短时间戳(G_TS),避免4xx热路径对$4重复split
    # epoch忽略闰年,仅用于2天内时间窗比较,足够精确;解析失败返回0
    function parse_ts(t,   a, mo) {
        sub(/^\[/, "", t)
        if (split(t, a, /[\/:]/) < 6) return 0
        mo = MON[a[2]]
        if (mo == "") return 0
        G_EP = (a[3] * 365 + MDAYS[mo + 0] + a[1]) * 86400 + a[4] * 3600 + a[5] * 60 + substr(a[6], 1, 2)
        G_TS = mo "-" a[1] "_" a[4] ":" a[5] ":" substr(a[6], 1, 2)
        return 1
    }
    # UA为搜索引擎蜘蛛的行直接跳过（不参与任何计数）
    $0 ~ crawler { next }
    # 200最常见放最前:打断该IP当前连续404/403段;该秒301桶存在则删除(后续301从1重新累计)
    # 用 in/delete 而非赋0:避免给每个成功请求都建空桶,END无需遍历百万级死键
    $9 == "200" {
        c404[$1] = 0; c403[$1] = 0
        if ($1 in bad4) done4[$1] = 1
        if ($1 in bad3) done3[$1] = 1
        key = $1 "\t" $4
        if (key in cnt) delete cnt[key]
        next
    }
    # 404 → 连续计数(仅200打断)。环形槽只留最近max次事件,第c次与第c-max+1次首末
    # 落在win秒内即命中;命中后固定起点,同窗连续段后续事件计入明细(超窗不再延)
    $9 == "404" {
        if (!parse_ts($4)) next
        ip = $1; ep = G_EP; ts = G_TS
        c = c404[ip] + 1; c404[ip] = c
        r4e[ip, c % max] = ep
        r4t[ip, c % max] = ts
        if (!(ip in bad4)) {
            if (c >= max) {
                ss = (c - max + 1) % max
                if (ep - r4e[ip, ss] <= win) {
                    bad4[ip] = 1
                    q4se[ip] = r4e[ip, ss]; q4st[ip] = r4t[ip, ss]
                    q4ee[ip] = ep;            q4et[ip] = ts
                    q4n[ip] = max
                    out[ip] = 1
                }
            }
        } else if (!(ip in done4) && ep - q4se[ip] <= win) {
            q4ee[ip] = ep; q4et[ip] = ts; q4n[ip]++
        }
        next
    }
    # 403 → 同404的滑窗连续计数逻辑(独立计数,互不打断)
    $9 == "403" {
        if (!parse_ts($4)) next
        ip = $1; ep = G_EP; ts = G_TS
        c = c403[ip] + 1; c403[ip] = c
        r3e[ip, c % max] = ep
        r3t[ip, c % max] = ts
        if (!(ip in bad3)) {
            if (c >= max) {
                ss = (c - max + 1) % max
                if (ep - r3e[ip, ss] <= win) {
                    bad3[ip] = 1
                    q3se[ip] = r3e[ip, ss]; q3st[ip] = r3t[ip, ss]
                    q3ee[ip] = ep;            q3et[ip] = ts
                    q3n[ip] = max
                    out[ip] = 1
                }
            }
        } else if (!(ip in done3) && ep - q3se[ip] <= win) {
            q3ee[ip] = ep; q3et[ip] = ts; q3n[ip]++
        }
        next
    }
    # 301 → 按"IP + 秒级时间戳"累加(桶被200删过则从1重新计);END统一阈值过滤
    $9 == "301" { cnt[$1 "\t" $4]++ }
    END {
        # 4XX输出: ip cnt404 tf404 tl404 dur404 cnt403 tf403 tl403 dur403 (取命中时间窗)
        print "LINES " NR+0
        for (ip in out) {
            if (ip in bad4) {
                m4 = q4n[ip] + 0; f4 = q4st[ip]; l4 = q4et[ip]
                d4 = q4ee[ip] - q4se[ip]; if (d4 < 0) d4 = 0
            } else { m4 = 0; f4 = "-"; l4 = "-"; d4 = 0 }
            if (ip in bad3) {
                m3 = q3n[ip] + 0; f3 = q3st[ip]; l3 = q3et[ip]
                d3 = q3ee[ip] - q3se[ip]; if (d3 < 0) d3 = 0
            } else { m3 = 0; f3 = "-"; l3 = "-"; d3 = 0 }
            print "4XX " ip, m4, f4, l4, d4+0, m3, f3, l3, d3+0
        }
        for (k in cnt) if (cnt[k] > burst) {
            split(k, a, "\t")
            if (cnt[k] > bmax[a[1]]) { bmax[a[1]] = cnt[k]; bts[a[1]] = a[2] }
        }
        for (ip in bmax) print "301 " ip, bmax[ip], shortts(bts[ip])
    }')

# 按标签分发到内存变量（空格分隔，无ANSI-C quoting）
LOG_LINES_TOTAL=0
ips_4xx=()
ips_301=()
declare -A IP_REASON   # 每个异常IP的触发详情(用于通知明细: 次数+时间窗)
while IFS= read -r line; do
    [[ -z "$line" ]] && continue
    case "${line%% *}" in
        LINES) LOG_LINES_TOTAL="${line#LINES }";;
        4XX)   read -r ip e4 t1f t1l d1 e3 t2f t2l d2 <<< "${line#4XX }"
               ips_4xx+=("$ip")
               t1f="${t1f//_/ }"; t1l="${t1l//_/ }"
               t2f="${t2f//_/ }"; t2l="${t2l//_/ }"
               det=""
               (( e4 >= MAX_ERR )) && det="${ip}在${t1f}~${t1l}共${d1}秒之间发生${e4}次访问404"
               (( e3 >= MAX_ERR )) && det="${det:+${det}; }${ip}在${t2f}~${t2l}共${d2}秒之间发生${e3}次访问403"
               IP_REASON["$ip"]="$det";;
        301)   read -r ip n ts <<< "${line#301 }"
               ips_301+=("$ip")
               ts="${ts//_/ }"
               entry="${ip}在${ts} 1秒内发生${n}次访问301"
               prev="${IP_REASON[$ip]}"
               if [[ -n "$prev" ]]; then IP_REASON["$ip"]="${prev}; ${entry}"
               else IP_REASON["$ip"]="$entry"; fi;;
    esac
done <<< "$scan_out"
unset scan_out

# 快速路径：本轮无任何异常IP时，跳过全部规则查询/检测/清理，直接收尾
if [[ ${#ips_4xx[@]} -eq 0 && ${#ips_301[@]} -eq 0 ]]; then
    flush_log_buffer
    exit 0
fi

declare -A ab24_count
declare -A add_24list
declare -A SEG_IPS      # 每个/24网段内的异常IP列表(空格分隔,尾随空格)
declare -A ip_seen

# 合并4xx异常IP + 301 burst IP，直接喂入/24聚合（add_24list按/24去重）
for ip in "${ips_4xx[@]}" "${ips_301[@]}"; do
    [[ -z "$ip" ]] && continue
    [[ -n "${ip_seen[$ip]+x}" ]] && continue
    ip_seen["$ip"]=1
    # bash内置IFS+read一次解析IP四段，避免每个IP fork三次echo|cut
    IFS=. read -r a b c _ <<< "$ip"
    seg24="${a}.${b}.${c}.0/24"
    is_exclude "$seg24" && continue
    add_24list["$seg24"]=1
    SEG_IPS["$seg24"]+="${ip} "
done
unset ips_4xx ips_301 ip_seen   # 已聚合到/24，原IP列表清理释放内存

# ---------- 宝塔面板同步(可选,无宝塔/无sqlite3时静默跳过) ----------
# 原理: 脚本双写 DB firewall_ip表(面板显示) + ipset in_bt_user_drop_ipset(即时拦截)
#       ⚠️ 面板DB→ipset自动同步只在UI操作时触发,命令行INSERT不会同步
#       只写 ipset 集合,不碰 ufw/iptables 链 → 面板顶部数字=DB count(无重复)
BT_FW_DB="/www/server/panel/data/db/firewall.db"
BT_SYNC_FAIL=0   # SQL 执行失败置1(进警告);环境缺失不置1(静默)

# ipset 集合操作 (主流程直接调用,与DB双写保证即时拦截)
BT_IPSET="in_bt_user_drop_ipset"

# 【速度优化】批量落盘队列: bt_ban/bt_unban 只入队内存数组,由 bt_flush 统一执行
#   常规"新增N个网段"从 2N次fork(grep+tr+date) + N次sqlite3 + N次ipset 降为 1次sqlite3 + 1次ipset restore
declare -a PENDING_SQL PENDING_IPSET_ADD PENDING_IPSET_DEL
# ensure_ipset 结果记忆(必须保持未设=未探测; 0=就绪 1=ipset命令缺失 2=创建失败),记忆化避免每次fork ipset list

# 注入防破坏: 值里的单引号/分号清掉(纯bash零fork);网段做白名单字符校验后防拼接注入
_sql_lite() {
    local v="${1//\'/}"
    printf %s "${v//;/}"
}
bt_seg_clean() {
    local re='^([0-9]{1,3}\.){3}[0-9]{1,3}(/[0-9]{1,2})?$'
    [[ "$1" =~ $re ]] || return 1
    printf %s "$1"
}

# 确保ipset集合存在(不存在则自动创建; iptables链由宝塔面板管理,脚本不碰)
# 返回: 0=就绪 1=ipset命令缺失(静默) 2=创建失败(设置BT_SYNC_FAIL);结果记忆化,单次运行只探测一次
ensure_ipset() {
    local rc=0
    if [[ -n "${IPSET_READY+x}" ]]; then
        rc=$IPSET_READY
    elif ! command -v ipset >/dev/null 2>&1; then
        IPSET_READY=1; rc=1
    elif ipset list "$BT_IPSET" >/dev/null 2>&1; then
        IPSET_READY=0   # 集合已存在
    elif ipset create "$BT_IPSET" hash:net family inet hashsize 1024 maxelem 65536 2>/dev/null; then
        IPSET_READY=0   # 集合不存在 → 创建(hash:net支持IP/CIDR混合)
    else
        IPSET_READY=2; rc=2
    fi
    return $rc
}

# 面板ipset通道探测: 0=该段已被覆盖(含被更大段如/8覆盖) 1=未覆盖 2=环境缺失
bt_ipset_test() {
    ensure_ipset || return 2
    ipset test "$BT_IPSET" "${1%/*}" >/dev/null 2>&1
}

bt_fw_add() {  # $1=网段 $2=备注 (存在则更新,不存在则插入;SQL入队,由bt_flush统一执行)
    # 注意: types 必须用 'drop' 而不是 'deny'!
    # 因为 comModel.structure_ip_return_data() 从 ipset 解析 Strategy=drop/accept,
    # 匹配时直接 ip_rule['types'] == list_ip[j]['Strategy'] 没做 deny↔drop 映射(面板bug)
    # 用 deny 永远匹配不上 → brief/addtime 被置空 → 前端显示 "--"
    local seg brief
    seg=$(bt_seg_clean "$1") || return 0
    brief=$(_sql_lite "$2")
    PENDING_SQL+=("INSERT INTO firewall_ip (address,types,chain,brief,addtime)
SELECT '${seg}','drop','INPUT','${brief}','${now_full}' WHERE NOT EXISTS
(SELECT 1 FROM firewall_ip WHERE address='${seg}');
UPDATE firewall_ip SET brief='${brief}',addtime='${now_full}' WHERE address='${seg}';")
}

bt_fw_del() {  # $1=网段 (SQL入队)
    local seg
    seg=$(bt_seg_clean "$1") || return 0
    PENDING_SQL+=("DELETE FROM firewall_ip WHERE address='${seg}';")
}

# ipset通道: 掐断已建立连接 (入队,由bt_flush单次restore批量add)
bt_ipset_add() {  # $1=网段
    local seg
    seg=$(bt_seg_clean "$1") || return 0
    PENDING_IPSET_ADD+=("add ${BT_IPSET} ${seg}")
}

bt_ipset_del() {  # $1=网段 (入队;flush批量del,任一失败回退逐条确保不漏删)
    local seg
    seg=$(bt_seg_clean "$1") || return 0
    PENDING_IPSET_DEL+=("del ${BT_IPSET} ${seg}")
}

# 【速度优化】队列批量落盘: 全部网段合并为 1次sqlite3 + 1~2次ipset restore
#   ⚠️ 清理阶段的 cache_ipset_rules 前必须先 flush,覆盖检测才能看到新规则
#   环境缺失(无sqlite3/无DB/无ipset)静默跳过不告警;命令执行失败置BT_SYNC_FAIL
bt_flush() {
    local ret line
    (( ${#PENDING_SQL[@]} + ${#PENDING_IPSET_ADD[@]} + ${#PENDING_IPSET_DEL[@]} > 0 )) || return 0
    # DB通道: 单次sqlite3,SQL经stdin喂入(避免超长参数表)
    if [[ ${#PENDING_SQL[@]} -gt 0 ]] && command -v sqlite3 >/dev/null 2>&1 && [[ -f "$BT_FW_DB" ]]; then
        printf '%s\n' "${PENDING_SQL[@]}" \
            | sqlite3 -cmd '.timeout 3000' "$BT_FW_DB" 2>/dev/null || BT_SYNC_FAIL=1
    fi
    # ipset通道: add批量restore(-exist容忍重复); del批量restore,失败回退逐条(元素缺失报错不能漏删)
    if (( ${#PENDING_IPSET_ADD[@]} + ${#PENDING_IPSET_DEL[@]} > 0 )) && command -v ipset >/dev/null 2>&1; then
        ensure_ipset; ret=$?
        if [[ $ret -eq 2 ]]; then
            BT_SYNC_FAIL=1                       # 创建失败
        elif [[ $ret -eq 0 ]]; then
            if [[ ${#PENDING_IPSET_ADD[@]} -gt 0 ]]; then
                printf '%s\n' "${PENDING_IPSET_ADD[@]}" \
                    | ipset -exist restore >/dev/null 2>&1 || BT_SYNC_FAIL=1
            fi
            if [[ ${#PENDING_IPSET_DEL[@]} -gt 0 ]]; then
                if ! printf '%s\n' "${PENDING_IPSET_DEL[@]}" | ipset restore >/dev/null 2>&1; then
                    for line in "${PENDING_IPSET_DEL[@]}"; do
                        ipset ${line} >/dev/null 2>&1 || true
                    done
                fi
            fi
        fi
    fi
    PENDING_SQL=(); PENDING_IPSET_ADD=(); PENDING_IPSET_DEL=()
    return 0
}

bt_ban() {  # $1=网段 $2=备注 (双写:DB给面板显示+ipset给即时拦截)
    # 【重要】面板DB→ipset 的自动同步只在UI操作时触发,
    # 命令行sqlite3 INSERT 不会触发同步,所以必须同时写ipset
    # 只写ipset集合(不碰ufw/iptables链)→ comModel.py统计不会多计
    bt_fw_add "$1" "$2"
    bt_ipset_add "$1"
}

bt_unban() {  # $1=网段 (双删:DB+ipset同步删)
    bt_fw_del "$1"
    bt_ipset_del "$1"
}

# 收集某前缀下所有/24网段的异常IP列表(空格分隔)  $1=前缀("a.b"或"a")
collect_ips_prefix() {
    local pfx="$1" s
    for s in "${!add_24list[@]}"; do
        [[ "$s" == "$pfx".* ]] && printf '%s ' "${SEG_IPS[$s]}"
    done
}

# 组装通知明细: IP列表 → "ip详情; ip详情", 最多列3个, 超出附总数
seg_ip_detail() {
    local ips="$1" ip r out="" n=0 total=0
    for ip in $ips; do
        total=$((total+1))
        (( n >= 3 )) && continue
        r="${IP_REASON[$ip]}"
        if [[ -z "$out" ]]; then out="$r"; else out="${out}; ${r}"; fi
        n=$((n+1))
    done
    (( total > n )) && out="${out} 等共${total}个IP"
    [[ -n "$out" ]] && out="${out};"
    printf '%s' "$out"
}

# 【性能优化】提前一次加载面板DB规则到内存,后续rule_exists/covered判断都不再fork
cache_ipset_rules

# 把DB已有的/24规则读进add_24list(否则只记得本次新发现的/24,之前拉黑的/24不计入阈值)
for existing_seg in "${!RULES_EXIST[@]}"; do
    [[ -z "$existing_seg" ]] && continue
    [[ "$existing_seg" != */24 ]] && continue
    # 已被现有/16或/8覆盖的/24不重复计数(清理阶段会删掉)
    covered_by_existing_16 "$existing_seg" && continue
    covered_by_existing_8  "$existing_seg" && continue
    is_exclude "$existing_seg" && continue
    add_24list["$existing_seg"]=1
done

# ============ C段(/24) → B段(/16) 聚合判定 ============
# 统计每个/16段下不同/24子网数量(决定是否触发合并,而非按IP数)
for seg in "${!add_24list[@]}";do
    IFS=. read -r a b _ <<< "$seg"
    ab="${a}.${b}"
    ((ab24_count["$ab"]++))
done

# 先算出本轮会触发合并(新增/16)的A/B集合
# 好处:/24添加循环能跳过"将被/16覆盖的/24",避免"先加、后删"的双倍操作
declare -A merge_ab
MERGED_16=0
for abkey in "${!ab24_count[@]}";do
    cnt=${ab24_count[$abkey]}
    if (( cnt >= MERGE_THRESHOLD ));then
        seg16="${abkey}.0.0/16"
        is_exclude "$seg16" && continue
        rule_exists "$seg16" && continue
        merge_ab["$abkey"]=1
        MERGED_16=1
    fi
done

# ============ B段(/16) → A段(/8) 聚合判定 ============
# 构建"活跃/16集合"add_16list:来自add_24list的/24所属/16 + 现有DB /16 + 本轮新增merge_ab /16
declare -A add_16list
for seg in "${!add_24list[@]}"; do
    IFS=. read -r a b _ <<< "$seg"
    add_16list["${a}.${b}.0.0/16"]=1
done
for existing_seg in "${!RULES_EXIST[@]}"; do
    [[ "$existing_seg" != */16 ]] && continue
    is_exclude "$existing_seg" && continue
    covered_by_existing_8 "$existing_seg" && continue
    add_16list["$existing_seg"]=1
done
for abkey in "${!merge_ab[@]}"; do
    add_16list["${abkey}.0.0/16"]=1
done

# 统计每个/8(A段)下不同/16(B段)数量,超过阈值则合并为/8
declare -A a8_count
for seg16 in "${!add_16list[@]}"; do
    IFS=. read -r a _ <<< "$seg16"
    ((a8_count["$a"]++))
done

declare -A merge_a8
MERGED_8=0
for akey in "${!a8_count[@]}"; do
    cnt=${a8_count[$akey]}
    if (( cnt > A_MERGE_THRESHOLD )); then
        seg8="${akey}.0.0.0/8"
        is_exclude "$seg8" && continue
        rule_exists "$seg8" && continue
        merge_a8["$akey"]=1
        MERGED_8=1
    fi
done

# 统一取一次时间写入备注(宝塔UI"备注"列可见;面板"时间"列存于面板自有数据库)
now=$(date '+%Y-%m-%d %H:%M')
now_full=$(date '+%Y-%m-%d %H:%M:%S')   # 【速度优化】SQL的addtime统一用此值,避免每个网段fork一次date
RULES_CHANGED=0   # 本轮是否有新增/合并/清理(决定收尾是否刷新ipset缓存)

# 添加/24规则:跳过本轮会合并为/16或/8的段(由大段统一覆盖,避免先加后删)
for seg in "${!add_24list[@]}";do
    if rule_exists "$seg";then continue; fi
    if covered_by_existing_16 "$seg";then continue; fi
    if covered_by_existing_8  "$seg";then continue; fi
    IFS=. read -r a b _ <<< "$seg"
    ab="${a}.${b}"
    [[ -n "${merge_ab[$ab]+x}" ]] && continue
    [[ -n "${merge_a8[$a]+x}"  ]] && continue
    bt_ban "$seg" "自动拦截,30分钟内连续${MAX_ERR}次404/403或1秒超${BURST_301}次301(200重置)"
    detail="$(seg_ip_detail "${SEG_IPS[$seg]}")"
    if [[ -n "$detail" ]]; then
        ACTION_MSGS+=("[新增] 拉黑网段 $seg 异常IP: $detail")
    else
        ACTION_MSGS+=("[新增] 拉黑网段 $seg")
    fi
    RULES_CHANGED=1
done

# 添加合并后的/16大网段:跳过本轮会合并为/8的A段(由/8统一覆盖)
for abkey in "${!merge_ab[@]}";do
    seg16="${abkey}.0.0/16"
    IFS=. read -r a _ <<< "$seg16"
    [[ -n "${merge_a8[$a]+x}" ]] && continue
    covered_by_existing_8 "$seg16" && continue
    bt_ban "$seg16" "自动合并:${now},同A/B多段聚合为/16"
    ACTION_MSGS+=("[合并] 新增大网段 $seg16 (${ab24_count[$abkey]}个C段达${MERGE_THRESHOLD}阈值, 异常IP: $(seg_ip_detail "$(collect_ips_prefix "$abkey")"))")
    RULES_CHANGED=1
done

# 添加合并后的/8大网段(B段超阈值聚合为A段)
for akey in "${!merge_a8[@]}"; do
    seg8="${akey}.0.0.0/8"
    bt_ban "$seg8" "自动合并:${now},同A段B段超${A_MERGE_THRESHOLD}阈值聚合为/8"
    ACTION_MSGS+=("[合并] 新增A段大网段 $seg8 (${a8_count[$akey]}个B段超${A_MERGE_THRESHOLD}阈值, 异常IP: $(seg_ip_detail "$(collect_ips_prefix "$akey")"))")
    RULES_CHANGED=1
done

# 【速度优化】批量落盘第1点: 新增的/24、/16、/8统一执行DB+ipset写入
# 必须在清理阶段 cache_ipset_rules 之前,覆盖检测才能看到本轮新增的大网段
bt_flush

# 【速度优化】仅当本轮新增了/16或/8大网段(发生合并)时,才清理被覆盖的冗余规则
# 清理目标:/24 被 /16 或 /8 覆盖 → 删;  /16 被 /8 覆盖 → 删
# 真相源:面板DB firewall_ip 表,清理直接遍历RULES_EXIST哈希(零fork)
CLEAN_DEL=0
if [[ $MERGED_16 -eq 1 || $MERGED_8 -eq 1 ]]; then
    # 刷新缓存:本轮 bt_ban 已新增了/16或/8大网段,需要把它们纳入覆盖检测
    cache_ipset_rules

    declare -A to_delete

    # 遍历所有/24:是否被同a.b的/16或同a的/8覆盖
    for cidr in "${!RULES_EXIST[@]}"; do
        [[ "$cidr" != */24 ]] && continue
        if covered_by_existing_16 "$cidr"; then to_delete["$cidr"]=1; continue; fi
        if covered_by_existing_8  "$cidr"; then to_delete["$cidr"]=1; continue; fi
    done

    # 遍历所有/16:是否被同a的/8覆盖
    for cidr in "${!RULES_EXIST[@]}"; do
        [[ "$cidr" != */16 ]] && continue
        covered_by_existing_8 "$cidr" && to_delete["$cidr"]=1
    done

    for cidr in "${!to_delete[@]}"; do
        bt_unban "$cidr"
        CLEAN_DEL=$((CLEAN_DEL+1))
        ACTION_MSGS+=("[清理] 已删除被覆盖的 $cidr")
        RULES_CHANGED=1
    done
fi

if [[ $CLEAN_DEL -gt 0 ]]; then
    ACTION_MSGS+=("[清理汇总] 共删除${CLEAN_DEL}条被覆盖的冗余规则")
fi
# 【速度优化】批量落盘第2点: 清理阶段入队的删除在此统一执行(队列空则no-op)
bt_flush
# 拦截即时生效: 本轮内 bt_flush 已双写 DB + ipset 集合
# 刷新一次缓存供后续 rule_exists 查询保持一致
if [[ ${RULES_CHANGED:-0} -eq 1 ]]; then
    cache_ipset_rules
fi

# 宝塔同步失败告警(仅 SQL 执行失败才告警;环境缺失保持静默)
if [[ $BT_SYNC_FAIL -eq 1 ]];then
    ACTION_MSGS+=("[警告] 宝塔同步失败(DB写入):检查sqlite3、firewall_ip表结构")
fi

# 收尾:按 NOTIFY_MODE 分发日志/微信
flush_log_buffer

终端运行下面命令给文件权限

chmod +x /root/auto_ban_4xx_ufw.sh

设置定时任务（每分钟检测一次日志）宝塔 → 计划任务 → Shell 脚本,执行周期：每5分钟
脚本内容：/root/auto_ban_4xx.sh

下面是支持IPV6版本: