攻防对抗：利用IP段归属查询工具快速封禁攻击源——3步联动防火墙（附脚本）

深夜两点服务器告警声响起。SSH日志中来自同一运营商IP段的暴力破解尝试正在快速轮换——这不是单点攻击而是有组织的分布式扫描。面对这类场景传统按单个IP逐一封禁的方式已力不从心。一、2025年攻击态势DDoS攻击同比暴增168%根据Radware发布的《2026年全球威胁分析报告》2025年网络层DDoS攻击同比增加168.2%峰值攻击流量接近30 Tbps。技术行业是重灾区占总攻击量的45%相比2024年的8.77%大幅攀升。同时89%的攻击在10分钟内结束攻击持续时间缩短检测与响应窗口进一步被压缩。在攻击来源方面数据中心和云服务商的IP段贡献了超过半数的恶意流量。有报告显示托管和云服务商相关流量中超过50%具有恶意特征——这意味着攻击者的IP并非随机分散而是高度集中在特定运营商的某些CIDR段内。二、从“封IP”到“封段”为什么需要IP段归属查询IP段归属查询是指查询一个连续的IP地址范围CIDR段所属的地理位置、运营商和网络类型等信息。它解决了单IP封禁的核心短板攻击者轮换IP后旧的封禁规则立即失效。批量攻击者的特征在于“段级密集性”——攻击IP集中在某几个运营商的特定段内单个IP请求量不高但段内整体活跃度异常。通过IP归属地运营商信息可以快速判断IP是否来自数据中心——这类IP的攻击风险远高于普通家庭宽带。一个实测结论是家庭宽带IP的攻击概率远低于数据中心IP。识别出数据中心段后以段为单位统一封禁可以有效遏制攻击者通过轮换IP绕过封锁。从封禁单个IP到封禁整个IP段的概念对比示意图三、工具选型关键能力对比市面上的IP查询工具在精度、字段覆盖和部署方式上差异显著。对于段级封禁需求需重点关注三个能力维度推荐阈值方案表现定位精度区县/街道级街道级 ✅运营商识别完整ISP字段覆盖完整 ✅部署方式API离线双模双模支持 ✅IPv6支持必须支持完全支持 ✅在运营商识别和段归属映射两个核心维度上具备完整的ISP字段的工具可以提供归属地、运营商、ASN、代理标识、风险标签等20维度的数据足以支撑精细化封禁策略。选择提供API接口和离线部署双模方案的平台可获得毫秒级响应适合在高合规、高精度需求的企业级场景落地。四、3步实现自动化段级封禁IP段归属查询自动化封禁攻击源的3步流程图第1步从日志筛选可疑攻击IP通过分析访问日志或防火墙日志提取高频访问IP。注意区分单IP高频访问可能是普通扫描器批量攻击的特征是多个IP落在同一运营商段内整体段内活跃度异常。第2步调用IP归属地API获取段信息使用IP段归属查询API获取单个IP所属的CIDR段和运营商类型。通过运营商字段识别数据中心IP优先封禁数据中心段。第3步联动防火墙封禁整个段获取CIDR段后调用防火墙API执行段级封禁。以下Bash脚本实现了完整的自动化链路生产环境增强版#!/bin/bash # 自动检测攻击IP并联动防火墙封禁所属段生产环境增强版 # 依赖curl, jq, iptables set -euo pipefail # 配置区 LOG_FILE/var/log/nginx/access.log THRESHOLD50 API_KEY${IP_DATA_CLOUD_KEY:-} # 必须通过环境变量设置 IP_DATA_CLOUD_URLhttps://api.ipdatacloud.com/v1/ip/segment IPTABLES_SAVE_CMD # 自动检测持久化命令 # # 颜色输出可选便于日志 RED\033[0;31m GREEN\033[0;32m YELLOW\033[1;33m NC\033[0m log_info() { echo -e ${GREEN}[INFO]${NC} $1; } log_warn() { echo -e ${YELLOW}[WARN]${NC} $1; } log_error() { echo -e ${RED}[ERROR]${NC} $1 2; } # 1. 校验必要环境 check_prerequisites() { local missing() for cmd in curl jq iptables; do if ! command -v $cmd /dev/null 21; then missing($cmd) fi done if [ ${#missing[]} -gt 0 ]; then log_error Missing required commands: ${missing[*]} exit 1 fi if [ -z $API_KEY ]; then log_error API_KEY not set. Please export IP_DATA_CLOUD_KEYyour_key exit 1 fi # 检查 root 权限iptables 需要 if [ $EUID -ne 0 ]; then log_error This script must be run as root (iptables requires root privileges) exit 1 fi } # 2. 检测 iptables 持久化命令兼容不同发行版 detect_iptables_save() { if command -v iptables-save /dev/null 21 [ -d /etc/iptables ]; then IPTABLES_SAVE_CMDiptables-save /etc/iptables/rules.v4 elif command -v service /dev/null 21 service iptables status /dev/null 21; then IPTABLES_SAVE_CMDservice iptables save elif command -v netfilter-persistent /dev/null 21; then IPTABLES_SAVE_CMDnetfilter-persistent save else log_warn No known iptables persistence method found. Rules will be lost after reboot. IPTABLES_SAVE_CMD fi } # 3. 调用 API 获取 IP 段信息带重试 get_ip_segment() { local ip$1 local retry3 local response local http_code for i in $(seq 1 $retry); do response$(curl -s -w %{http_code} -G $IP_DATA_CLOUD_URL \ -d ip$ip \ -d key$API_KEY \ -d fieldssegment,isp,type 2/dev/null) || { log_warn curl failed for $ip, retry $i/$retry sleep 1 continue } http_code${response: -3} response${response%???} if [ $http_code 200 ]; then echo $response return 0 else log_warn API returned HTTP $http_code for $ip, retry $i/$retry sleep 1 fi done log_error Failed to get segment for $ip after $retry attempts return 1 } # 4. 封禁 CIDR 段使用 iptables block_cidr() { local cidr$1 if iptables -C INPUT -s $cidr -j DROP 2/dev/null; then log_info CIDR $cidr already blocked, skipping return 0 fi if iptables -A INPUT -s $cidr -j DROP; then log_info Blocked segment: $cidr return 0 else log_error Failed to block $cidr return 1 fi } # 主流程 main() { check_prerequisites detect_iptables_save # 检查日志文件是否存在 if [ ! -f $LOG_FILE ]; then log_error Log file $LOG_FILE not found exit 1 fi log_info Analyzing logs from $LOG_FILE ... # 提取攻击 IP频率超过阈值兼容不同日志格式默认第一列为 IP ATTACK_IPS$(tail -10000 $LOG_FILE | awk {print $1} | sort | uniq -c | \ awk -v threshold$THRESHOLD $1 threshold {print $2}) if [ -z $ATTACK_IPS ]; then log_info No suspicious IPs found. exit 0 fi local blocked_count0 for IP in $ATTACK_IPS; do log_info Processing $IP ... RESPONSE$(get_ip_segment $IP) || continue # 提取 CIDR 段和类型使用 jq并提供默认值 CIDR$(echo $RESPONSE | jq -r .data.segment // empty) IS_DATACENTER$(echo $RESPONSE | jq -r .data.type datacenter) if [ -z $CIDR ]; then log_warn No segment info for $IP, skipping continue fi # 优先封禁数据中心段普通段也可封禁 if block_cidr $CIDR; then ((blocked_count)) [ $IS_DATACENTER true ] log_info - datacenter segment fi done log_info Blocked $blocked_count distinct CIDR segments. # 持久化 iptables 规则 if [ -n $IPTABLES_SAVE_CMD ]; then if eval $IPTABLES_SAVE_CMD 2/dev/null; then log_info iptables rules persisted. else log_warn Failed to persist iptables rules. fi fi } main扩展Python版本适合集成到WAF或SIEM生产环境增强版#!/usr/bin/env python3 自动封禁攻击IP所属CIDR段生产环境增强版 依赖: requests, python-iptables (可选此处使用subprocess) import os import sys import subprocess import logging import requests from typing import Optional, Dict, Tuple # 配置 API_KEY os.environ.get(IP_DATA_CLOUD_KEY, ) API_URL https://api.ipdatacloud.com/v1/ip/segment LOG_FILE /var/log/nginx/access.log THRESHOLD 50 REQUEST_TIMEOUT 5 MAX_RETRIES 3 # logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, datefmt%Y-%m-%d %H:%M:%S ) logger logging.getLogger(__name__) def check_prerequisites() - None: 检查运行环境 if not API_KEY: logger.error(API_KEY not set. Please export IP_DATA_CLOUD_KEYyour_key) sys.exit(1) # 检查 root 权限 if os.geteuid() ! 0: logger.error(This script must be run as root (iptables requires root privileges)) sys.exit(1) # 检查依赖命令 for cmd in [iptables, iptables-save]: if not subprocess.run([which, cmd], capture_outputTrue).returncode 0: logger.error(fRequired command {cmd} not found) sys.exit(1) def get_ip_segment(ip: str) - Optional[Dict]: 调用IP数据云API获取IP段信息 返回: {segment: x.x.x.x/y, isp: ..., type: datacenter|residential} 失败返回 None params { ip: ip, key: API_KEY, fields: segment,isp,type } for attempt in range(1, MAX_RETRIES 1): try: resp requests.get(API_URL, paramsparams, timeoutREQUEST_TIMEOUT) if resp.status_code 200: data resp.json() # 检查业务返回码如有 if data.get(code) 0 or data in data: return data.get(data, {}) else: logger.warning(fAPI business error for {ip}: {data.get(msg, unknown)}) else: logger.warning(fAPI HTTP {resp.status_code} for {ip}, attempt {attempt}) except requests.exceptions.Timeout: logger.warning(fRequest timeout for {ip}, attempt {attempt}) except requests.exceptions.ConnectionError: logger.warning(fConnection error for {ip}, attempt {attempt}) except Exception as e: logger.warning(fUnexpected error for {ip}: {e}, attempt {attempt}) if attempt MAX_RETRIES: import time time.sleep(1) logger.error(fFailed to get segment for {ip} after {MAX_RETRIES} attempts) return None def cidr_exists(cidr: str) - bool: 检查iptables中是否已存在该CIDR的DROP规则 result subprocess.run( [iptables, -C, INPUT, -s, cidr, -j, DROP], capture_outputTrue ) return result.returncode 0 def block_cidr(cidr: str) - bool: 封禁CIDR段返回是否成功添加新规则 if cidr_exists(cidr): logger.info(fCIDR {cidr} already blocked, skipping) return True result subprocess.run( [iptables, -A, INPUT, -s, cidr, -j, DROP], capture_outputTrue ) if result.returncode 0: logger.info(fBlocked segment: {cidr}) return True else: logger.error(fFailed to block {cidr}: {result.stderr.decode().strip()}) return False def persist_iptables() - None: 尝试持久化iptables规则兼容多种方式 methods [ [iptables-save, -c, , /etc/iptables/rules.v4], [service, iptables, save], [netfilter-persistent, save] ] for method in methods: try: if method[0] iptables-save: # 需要shell重定向 subprocess.run(iptables-save /etc/iptables/rules.v4, shellTrue, checkTrue) else: subprocess.run(method, checkTrue) logger.info(iptables rules persisted.) return except (subprocess.CalledProcessError, FileNotFoundError): continue logger.warning(No persistence method succeeded. Rules may be lost after reboot.) def extract_attack_ips(log_file: str, threshold: int) - set: 从日志中提取频率超过阈值的IP集合 if not os.path.isfile(log_file): logger.error(fLog file {log_file} not found) return set() # 读取最近10000行统计IP频率 try: with open(log_file, r) as f: lines f.readlines()[-10000:] except Exception as e: logger.error(fFailed to read log file: {e}) return set() ip_counter {} for line in lines: parts line.split() if parts: ip parts[0] ip_counter[ip] ip_counter.get(ip, 0) 1 attack_ips {ip for ip, count in ip_counter.items() if count threshold} logger.info(fFound {len(attack_ips)} suspicious IPs (frequency {threshold})) return attack_ips def main(): check_prerequisites() attack_ips extract_attack_ips(LOG_FILE, THRESHOLD) if not attack_ips: logger.info(No suspicious IPs found.) return blocked_cidrs set() for ip in attack_ips: logger.info(fProcessing {ip} ...) seg_info get_ip_segment(ip) if not seg_info: continue cidr seg_info.get(segment) if not cidr: logger.warning(fNo segment info for {ip}, skipping) continue if cidr in blocked_cidrs: continue if block_cidr(cidr): blocked_cidrs.add(cidr) isp seg_info.get(isp, unknown) is_dc seg_info.get(type) datacenter logger.info(f - {cidr} | ISP: {isp} | Datacenter: {is_dc}) logger.info(fBlocked {len(blocked_cidrs)} distinct CIDR segments.) persist_iptables() if __name__ __main__: main()五、部署建议定期执行将脚本配置为cron任务每5-10分钟执行一次白名单保护将企业自身IP段加入白名单避免误封日志审计记录每次段级封禁操作便于后续分析和规则调优风险提示段级封禁可能波及正常用户建议先在小范围段如/24上验证效果再逐步扩大封禁粒度面对高频次、短时长的攻击封禁策略的响应速度至关重要。基于IP段归属查询工具的自动化方案将封禁粒度从单IP升级到CIDR段让防守从“追赶IP”变为“封住来路”。