ipset create bad_ips hash:net family inet maxelem 500000
ipset create bad_subnets hash:net family inet maxelem 5000
ipset create bad_ports hash:ipport family inet maxelem 100000

http {
    # --- 信号 1：User-Agent 熵值 ---
    # 正常 UA 熵 ~4.2，爬虫池 UA 熵 ~6.8（随机串）
    map $http_user_agent $ua_class {
        default                     0;
        "~^(curl|python-requests)" 1;   # 脚本客户端
        "~^[a-f0-9]{20,}$"          2;   # 随机 hex 串（伪造 UA）
        "~^Mozilla/5.0"            0;    # 正常浏览器
    }

    # --- 敏感路径：直接限频 ---
    location ~ ^/(admin|wp-admin|xmlrpc\.php|/api/token|/api/login)$ {
        limit_req zone=admin_uri burst=5 nodelay;
        return 444;  # 444 = 无响应码，直接断连
    }

    # --- 信号 2：全局限流 ---
    limit_req_zone $binary_remote_addr zone=global_rps:10m rate=20r/s;
    limit_conn_zone $binary_remote_addr zone=global_conn:10m;

    # --- 信号 3：TLS 指纹（JA3 简化版）---
    # 用 ssl_preread 阶段记录 cipher suite 组合，写 access_log
    log_format j7_combined
        '$remote_addr - $ua_class - $request_uri '
        '$request_length $bytes_sent $upstream_response_time';

    # --- 信号 4：连接间隔指纹 ---
    # 记录同 IP 相邻两次请求的时间差，写自定义日志字段
}

# 自定义字段：请求间隔（ms）+ 路径哈希前缀
map $binary_remote_addr $client_fingerprint {
    # 用 IP 作为指纹键，值在 L3 里计算后回写
}

log_format fingerprint '
    fp|$time_iso8601|fp|
    $remote_addr|fp|
    $request|fp|
    $request_uri|fp|
    $http_user_agent|fp|
    $ua_class|fp|
    $request_length|fp|
    $bytes_sent|fp|
    $upstream_response_time';

access_log /var/log/nginx/fingerprint.log fingerprint buffer=32k flush=5s;

limit_req_zone $binary_remote_addr:$request_uri zone=uri_rps:20m rate=8r/s;

# 特征向量（8 维）
@dataclass
class IpFingerprint:
    ip: str
    req_count: int          # 请求总数
    uri_entropy: float      # 路径信息熵（Shannon）
    ua_entropy: float       # UA 多样性熵
    req_interval_jitter: float  # 请求间隔标准差 / 均值
    avg_response_time: float    # 平均上游耗时
    error_ratio: float        # 4xx/5xx 占比
    body_size_variance: float  # 请求体大小方差

def shannon_entropy(values: dict) -> float:
    total = sum(values.values())
    if total == 0:
        return 0.0
    ent = 0.0
    for v in values.values():
        p = v / total
        ent -= p * math.log2(p)
    return ent

# 训练阶段（每周跑一次）
normal_stats = {
    "uri_entropy": (2.1, 0.4),   # (mean, std) 来自 7 天正常流量
    "req_interval_jitter": (2.8, 0.9),
    "error_ratio": (0.015, 0.008),
    "ua_entropy": (0.8, 0.3),
}

def z_scores(fp: IpFingerprint) -> dict:
    return {
        k: abs(v - m) / s
        for (k, v), (m, s) in zip(fp.as_dict().items(),
                                   normal_stats.items())
    }

# 判定：任意 2 个特征 z > 2.5，或单特征 z > 4.0
def is_anomalous(fp: IpFingerprint) -> bool:
    zs = z_scores(fp)
    high = [k for k, z in zs.items() if z > 2.5]
    very_high = [k for k, z in zs.items() if z > 4.0]
    return len(high) >= 2 or len(very_high) >= 1

#!/usr/bin/env python3
"""L3 Behavior Learner — runs every 10 min via systemd timer"""
import json, math, os, subprocess, time
from collections import defaultdict
from dataclasses import dataclass, asdict
from pathlib import Path

FINGERPRINT_LOG = Path("/var/log/nginx/fingerprint.log")
IPSET_CMD      = "ipset"
BAD_IPS_SET   = "bad_ips"
STATE_FILE    = Path("/var/lib/l7-learner/seen_ips.json")
ANOMALY_LOG   = Path("/var/log/l7-learner/anomalies.log")

@dataclass
class IpWindow:
    req_count: int = 0
    uri_counts: dict = None
    ua_counts: dict = None
    intervals: list = None
    error_count: int = 0
    total_count: int = 0
    body_sizes: list = None
    last_ts: float = 0.0

def shannon(vals: dict) -> float:
    total = sum(vals.values())
    if total == 0: return 0.0
    return -sum((c/total)*math.log2(c/total) for c in vals.values() if c > 0)

def compute_fingerprint(ip: str, w: IpWindow) -> dict:
    now = time.time()
    intervals = [b - a for a, b in zip(w.intervals[:-1], w.intervals[1:])]
    jitter = 0.0
    if intervals:
        mean_i = sum(intervals) / len(intervals)
        if mean_i > 0:
            jitter = (sum((x-mean_i)**2 for x in intervals)/len(intervals))**0.5 / mean_i
    return {
        "uri_entropy": shannon(w.uri_counts),
        "ua_entropy": shannon(w.ua_counts),
        "req_interval_jitter": jitter,
        "error_ratio": w.error_count / max(w.total_count, 1),
        "req_count": w.req_count,
    }

NORMAL_STATS = {
    "uri_entropy": (2.1, 0.4),
    "ua_entropy": (0.8, 0.3),
    "req_interval_jitter": (2.8, 0.9),
    "error_ratio": (0.015, 0.008),
}

def z_check(fp: dict) -> tuple[bool, list]:
    flagged = []
    for k, (m, s) in NORMAL_STATS.items():
        z = abs(fp[k] - m) / s
        if z > 4.0: flagged.append((k, z))
    high = [(k,z) for k,z in [(k, abs(fp[k]-m)/s) for k,(m,s) in NORMAL_STATS.items()] if z > 2.5]
    if any(z > 4.0 for _, z in flagged) or len(high) >= 2:
        return True, high + flagged
    return False, []

def add_to_ipset(ip: str, timeout: int = 86400):
    subprocess.run([IPSET_CMD, "add", BAD_IPS_SET, ip, f"timeout={timeout}"],
                   capture_output=True)

def main():
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    seen: dict = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {}

    # 增量读取：从上次位置继续
    offset = seen.get("_offset", 0)
    new_data = []
    with open(FINGERPRINT_LOG, "r", errors="replace") as f:
        f.seek(offset)
        for line in f:
            parts = line.strip().split("|fp|")
            if len(parts) < 10: continue
            ts_str, ip, method, uri, ua, ua_class = parts[:6]
            req_len, resp_len, up_time = parts[6:9]
            is_err = up_time.startswith("-") or int(req_len.split()[-1] if req_len else "0") > 500_000
            new_data.append((time.time(), ip, uri, ua, is_err))

    # 按 IP 聚合（10 分钟滑动窗口）
    windows: dict = defaultdict(IpWindow)
    for ts, ip, uri, ua, is_err in new_data:
        w = windows[ip]
        if not w.uri_counts:
            w.uri_counts = defaultdict(int); w.ua_counts = defaultdict(int)
            w.intervals = []; w.body_sizes = []
        if w.intervals:
            w.intervals.append(ts)
        w.req_count += 1; w.total_count += 1
        w.uri_counts[uri.split("?")[0][:64]] += 1
        w.ua_counts[ua[:80]] += 1
        if is_err: w.error_count += 1

    # 判定 + 封禁
    new_bans = []
    for ip, w in windows.items():
        if w.req_count < 5: continue          # 样本不足，不判
        fp = compute_fingerprint(ip, w)
        is_bad, reasons = z_check(fp)
        if is_bad:
            add_to_ipset(ip)
            new_bans.append((ip, reasons, fp))

    if new_bans:
        ANOMALY_LOG.parent.mkdir(parents=True, exist_ok=True)
        with open(ANOMALY_LOG, "a") as f:
            for ip, reasons, fp in new_bans:
                f.write(json.dumps({"ts": time.time(), "ip": ip,
                                     "reasons": [r[0] for r in reasons], **fp}) + "\n")

    seen["_offset"] = FINGERPRINT_LOG.stat().st_size
    json.dump(seen, open(STATE_FILE, "w"))
    print(f"[l7-learner] scanned {len(windows)} IPs, banned {len(new_bans)} new")

if __name__ == "__main__":
    main()

# /etc/systemd/system/l7-learner.service
[Unit]
Description=L7 Behavior Learner
After=network.target nginx.service

[Service]
Type=oneshot
User=root
ExecStart=/usr/local/bin/l7_learner.py
StandardOutput=journal
StandardError=journal
Nice=10

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/l7-learner.timer
[Unit]
Description=Run L7 learner every 10 minutes

[Timer]
OnCalendar=*:0/10
Persistent=true
AccuracySec=1s

[Install]
WantedBy=timers.target

systemctl enable --now l7-learner.timer
systemctl status l7-learner.timer   # 看下次触发时间

# /usr/local/bin/l7_retrain.py
"""用过去 7 天的正常流量（非异常日志里的 IP）重算 NORMAL_STATS"""
import json, subprocess, time
from collections import defaultdict
from statistics import mean, stdev

NORMAL_LOG = "/var/log/nginx/access.log"
ANOMALY_IPS = set(json.load(open("/var/log/l7-learner/banned_ips.json")).get("ips", []))

# 解析 7 天 access_log，排除已知异常 IP
features = defaultdict(list)
with open(NORMAL_LOG, errors="replace") as f:
    for line in f:
        if line.startswith("|fp|"):
            parts = [p.strip() for p in line.strip().split("|fp|")]
            ip = parts[1]
            if ip in ANOMALY_IPS: continue
            uri_ent = float(parts[4]) if parts[4] != "-" else 0.0
            features["uri_entropy"].append(uri_ent)
            features["error_ratio"].append(float(parts[6]) if parts[6] not in ("-", "") else 0.0)

new_stats = {k: (round(mean(v), 3), round(max(stdev(v), 0.01), 4))
             for k, v in features.items() if len(v) > 100}

# 原子写
tmp = "/tmp/normal_stats.json.tmp"
json.dump(new_stats, open(tmp, "w"), indent=2)
os.replace(tmp, "/var/lib/l7-learner/normal_stats.json")
print(f"Retrained {len(new_stats)} features from {sum(len(v) for v in features.values())} normal requests")

# /usr/local/bin/l7_metrics.py  （cron 每分钟跑，写 textfile）
import subprocess, json, time, os

outdir = "/var/lib/prometheus-node-exporter/"
os.makedirs(outdir, exist_ok=True)

lines = []
for s in ["bad_ips", "bad_subnets", "bad_ports"]:
    out = subprocess.run(["ipset", "list", s], capture_output=True, text=True).stdout
    count = sum(1 for l in out.splitlines() if l and not l.startswith(("Name:", "Type:", "Parameters:")) )
    lines.append(f"l7_ipset_members_total {{set=\"{s}\"}} {count}")

# 最近 1h 封禁数
banned = json.load(open("/var/lib/l7-learner/banned_ips.json"))
recent = sum(1 for ts in banned.get("timestamps", []) if time.time() - ts < 3600)
lines.append(f"l7_learner_bans_last_hour {recent}")
lines.append(f"l7_scrape_timestamp {time.time()}")

path = f"{outdir}l7_defense.prom"
with open(path + ".tmp", "w") as f:
    f.write("\n".join(lines) + "\n")
os.replace(path + ".tmp", path)

# prometheus.yml 里加
scrape_configs:
  - job_name: l7_defense
    static_configs:
      - targets: ["localhost:9102"]
    # 或用 textfile collector

# 已知 NAT 出口段白名单（维护在 conf.d/whitelist.conf）
geo $is_whitelist {
    default 0;
    202.112.0.0/16 1;    # 某高校出口
    59.82.0.0/16   1;    # 某企业出口
}

# 白名单 IP 跳过 L2 频控
limit_req_zone $binary_remote_addr zone=global_rps:10m rate=20r/s;
# 白名单走更宽松的路径
map $is_whitelist $limit_zone {
    0 "global_rps";
    1 "wl_rps";   # 100r/s 宽松限制
}

ipset create known_crawlers hash:ip family inet
# 从 Google/Bing 官方 IP 列表导入
curl -s https://www.google.com/robots.txt | grep -oE '[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+' | \
  xargs -I{} ipset add known_crawlers {}

# nftables 里对 known_crawlers 放行
ip saddr { "known_crawlers" } counter accept

# ipset 持久化 cron
* * * * * ipset save /etc/ipset/bad_ips.dump 2>/dev/null
@reboot  ipset restore /etc/ipset/bad_ips.dump 2>/dev/null

#!/bin/bash
# deploy-l7-defense.sh — 在 VPS 上一键部署
set -euo pipefail

echo "=== [1/5] 安装依赖 ==="
apt-get install -y ipset nftables python3 libstatistic

echo "=== [2/5] 创建 ipset ==="
ipset list bad_ips &>/dev/null || ipset create bad_ips hash:ip family inet maxelem 500000
ipset list bad_subnets &>/dev/null || ipset create bad_subnets hash:net family inet maxelem 5000
mkdir -p /etc/ipset
ipset save > /etc/ipset/init.dump 2>/dev/null || true

echo "=== [3/5] nftables 规则 ==="
cat > /etc/nftables/l7-defense.nft << 'EOF'
table inet filter {
    chain INPUT {
        type filter input priority filter
        ct state established,related accept
        ct state invalid drop
        ip saddr { "bad_ips" } counter drop
        ip saddr { "bad_subnets" } counter drop
    }
}
EOF
nft -f /etc/nftables/l7-defense.nft
systemctl enable nftables

echo "=== [4/5] L3 学习器 ==="
cp /root/l7_learner.py /usr/local/bin/l7_learner.py
cp /root/l7-learner.service /etc/systemd/system/
cp /root/l7-learner.timer /etc/systemd/system/
systemctl daemon-reload
systemctl enable --now l7-learner.timer

echo "=== [5/5] Prometheus 指标 ==="
cp /root/l7_metrics.py /usr/local/bin/l7_metrics.py
chmod +x /usr/local/bin/l7_metrics.py
( echo "* * * * * /usr/local/bin/l7_metrics.py >/dev/null 2>&1" ) | crontab -

echo "部署完成。下一步：把 Nginx log_format 改为 fingerprint 格式并 reload。"