如何编写拦截恶意访问请求脚本

wen 实用脚本 2026-06-25 28

本文目录导读：

如何编写拦截恶意访问请求脚本

Nginx 层面的 IP 拦截脚本
Python 防火墙脚本
基于行为的恶意请求检测脚本
Fail2Ban 集成脚本
WAF（Web应用防火墙）规则脚本
使用建议：

Nginx 层面的 IP 拦截脚本

自动封禁异常 IP 的 Shell 脚本

#!/bin/bash
# 监控日志并封禁频繁请求的IP
LOG_FILE="/var/log/nginx/access.log"
BLACKLIST_FILE="/etc/nginx/blacklist.conf"
THRESHOLD=100  # 阈值：每分钟请求次数
# 分析日志并找出异常IP
tail -1000 $LOG_FILE | awk '{print $1}' | sort | uniq -c | sort -rn | \
while read count ip; do
    if [ $count -gt $THRESHOLD ]; then
        # 检查是否已封禁
        if ! grep -q $ip $BLACKLIST_FILE; then
            echo "deny $ip;" >> $BLACKLIST_FILE
            echo "Blocked: $ip (requests: $count)"
        fi
    fi
done
# 重载Nginx配置
nginx -s reload

Python 防火墙脚本

#!/usr/bin/env python3
import time
import subprocess
from collections import defaultdict
import re
class FirewallBlocker:
    def __init__(self, threshold=100, time_window=60):
        self.threshold = threshold
        self.time_window = time_window
        self.ip_requests = defaultdict(list)
        self.blocked_ips = set()
    def monitor_logs(self, log_file="/var/log/nginx/access.log"):
        """监控日志文件"""
        with open(log_file, 'r') as f:
            f.seek(0, 2)  # 移到文件末尾
            while True:
                line = f.readline()
                if line:
                    self.process_request(line)
                else:
                    time.sleep(0.1)
    def process_request(self, log_line):
        """处理单条请求日志"""
        # 解析IP地址（根据实际日志格式调整）
        ip_match = re.match(r'(\d+\.\d+\.\d+\.\d+)', log_line)
        if not ip_match:
            return
        ip = ip_match.group(1)
        current_time = time.time()
        # 如果IP已被封禁，跳过
        if ip in self.blocked_ips:
            return
        # 记录请求
        self.ip_requests[ip].append(current_time)
        # 清理过期记录
        self.ip_requests[ip] = [t for t in self.ip_requests[ip] 
                               if current_time - t < self.time_window]
        # 检查是否超过阈值
        if len(self.ip_requests[ip]) > self.threshold:
            self.block_ip(ip)
    def block_ip(self, ip):
        """封禁IP"""
        if ip in self.blocked_ips:
            return
        try:
            # iptables 封禁
            subprocess.run(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP'])
            self.blocked_ips.add(ip)
            print(f"封禁IP: {ip} - 时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
        except Exception as e:
            print(f"封禁失败 {ip}: {e}")
# 使用示例
if __name__ == "__main__":
    blocker = FirewallBlocker(threshold=50, time_window=30)  # 30秒内超过50次请求封禁
    blocker.monitor_logs()

基于行为的恶意请求检测脚本

import re
import time
from urllib.parse import urlparse
import requests
class MaliciousRequestDetector:
    def __init__(self):
        # 恶意特征规则
        self.suspicious_patterns = [
            r'(union.*select.*from)',
            r'<script.*?>.*?</script>',
            r'../../etc/passwd',
            r'(exec|system|passthru)\(.*\)',
            r'\x00',
            r'(delete|drop|truncate)\s+(from|table)',
            r'(exist|having|order)\s+by\s+\d+',
        ]
        # 非正常User-Agent列表
        self.suspicious_agents = [
            'sqlmap',
            'nikto',
            'nmap',
            'masscan',
            'zgrab',
            'python-requests',
            'curl',
            'wget'
        ]
    def check_request(self, request_data):
        """检测单个请求"""
        ip = request_data.get('ip', '')
        url = request_data.get('url', '')
        user_agent = request_data.get('user_agent', '')
        headers = request_data.get('headers', {})
        score = 0
        reasons = []
        # 1. 检测SQL注入
        for pattern in self.suspicious_patterns[:3]:  # 前几个是SQL注入模式
            if re.search(pattern, url, re.IGNORECASE):
                score += 50
                reasons.append(f"检测到SQL注入特征: {pattern}")
                break
        # 2. 检测XSS
        if re.search(r'<script>', url, re.IGNORECASE) or \
           re.search(r'on\w+=', url, re.IGNORECASE):
            score += 40
            reasons.append("检测到XSS攻击特征")
        # 3. 检测路径遍历
        if '../' in url or '..\\' in url:
            score += 30
            reasons.append("检测到路径遍历攻击")
        # 4. 检测恶意User-Agent
        ua_lower = user_agent.lower()
        for agent in self.suspicious_agents:
            if agent in ua_lower:
                score += 25
                reasons.append(f"检测到可疑User-Agent: {agent}")
                break
        # 5. 检测特殊头
        if 'X-Forwarded-For' in headers:
            score += 10
            reasons.append("使用代理访问")
        # 6. 检测异常请求频率
        if self._check_frequency(ip):
            score += 20
            reasons.append("请求频率异常")
        return {
            'is_malicious': score > 50,
            'score': score,
            'reasons': reasons,
            'ip': ip
        }
    def _check_frequency(self, ip):
        """简化的频率检测"""
        # 实际应用中应该维护一个请求计数器
        return False
# 集成到Flask应用示例
from flask import Flask, request, abort
app = Flask(__name__)
detector = MaliciousRequestDetector()
@app.before_request
def block_malicious():
    """在每个请求前检查恶意行为"""
    user_agent = request.headers.get('User-Agent', '')
    result = detector.check_request({
        'ip': request.remote_addr,
        'url': request.url,
        'user_agent': user_agent,
        'headers': dict(request.headers)
    })
    if result['is_malicious']:
        # 记录日志
        print(f"拦截恶意请求: {result}")
        # 可以在这里添加IP到黑名单
        abort(403)  # 返回403禁止访问
@app.route('/')
def home():
    return "Welcome!"
if __name__ == '__main__':
    app.run()

Fail2Ban 集成脚本

#!/bin/bash
# Fail2Ban 配置示例
cat > /etc/fail2ban/jail.local << 'EOF'
[nginx-badbots]
enabled = true
port = http,https
filter = nginx-badbots
logpath = /var/log/nginx/access.log
maxretry = 3
bantime = 86400
findtime = 300
action = iptables[name=nginx-badbots, port=http, protocol=tcp]
[nginx-noscript]
enabled = true
port = http,https
filter = nginx-noscript
logpath = /var/log/nginx/access.log
maxretry = 5
bantime = 3600
findtime = 600
EOF
# 创建过滤规则
cat > /etc/fail2ban/filter.d/nginx-badbots.conf << 'EOF'
[Definition]
failregex = ^<HOST> -.*"(GET|POST).*HTTP.*"(?:<badbots>)"$
badbots = sqlmap|nikto|nmap|masscan|zgrab
ignoreregex =
EOF
# 重启服务
systemctl restart fail2ban

WAF（Web应用防火墙）规则脚本

#!/bin/bash
# ModSecurity规则配置示例
cat > /etc/nginx/modsecurity/rules/block_malicious.conf << 'EOF'
# SQL注入防护
SecRule ARGS "@rx (union.*select.*from|drop\s+table|delete\s+from)" \
    "id:1001,phase:2,deny,status:403,msg:'SQL Injection detected'"
# XSS防护
SecRule ARGS "@rx <script[^>]*>.*</script>" \
    "id:1002,phase:2,deny,status:403,msg:'XSS attack detected'"
# 路径遍历防护
SecRule ARGS "@rx \.\./" \
    "id:1003,phase:2,deny,status:403,msg:'Path traversal detected'"
# 限制请求频率
SecAction \
    "id:1004,phase:1,initcol:ip=%{REMOTE_ADDR},setvar:ip.counter=+1,expirevar:ip.counter=60"
SecRule IP:COUNTER "@gt 100" \
    "id:1005,phase:2,deny,status:429,msg:'Too many requests'"
EOF
# 重新加载Nginx
nginx -s reload

使用建议：

部署步骤：
- 先在测试环境验证脚本
- 配置日志监控
- 设置合理的阈值（根据流量大小调整）
- 定期审查被拦截的IP
安全考虑：
- 白名单机制（允许合法爬虫）
- 不要拦截过多IP（可能误伤）
- 设置自动解封时间
- 记录所有封禁操作日志
性能优化：
- 使用缓存减少IO操作
- 异步处理日志分析
- 定期清理过期记录

这些脚本可以根据实际需求组合使用，建议先从温和的阈值开始,逐步调整到合适的参数。