Python IP拦截案例如何拦截恶意IP访问

wen python案例 2026-06-29 233

本文目录导读：

基于访问频率的IP拦截（Flask示例）
基于Redis的分布式IP拦截
基于IP段的批量拦截
完整的防火墙脚本
配置文件和用法
最佳实践建议

我来分享一个Python实现IP拦截的完整案例,包含多种实现方式：

基于访问频率的IP拦截（Flask示例）

from flask import Flask, request, abort
from collections import defaultdict
import time
from datetime import datetime
import redis
import ipaddress
app = Flask(__name__)
class IPBlocker:
    def __init__(self):
        # 本地内存存储（单机使用）
        self.ip_access = defaultdict(list)
        # 白名单
        self.whitelist = ['192.168.1.1', '10.0.0.1']
        # 黑名单
        self.blacklist = set()
        # 配置
        self.max_requests = 100  # 最大请求数
        self.time_window = 60    # 时间窗口（秒）
        self.block_time = 3600   # 封禁时间（秒）
    def is_banned(self, ip):
        """检查IP是否被封禁"""
        if ip in self.blacklist:
            return True
        return False
    def check_rate_limit(self, ip):
        """检查访问频率"""
        now = time.time()
        # 清除过期记录
        self.ip_access[ip] = [
            t for t in self.ip_access[ip] 
            if now - t < self.time_window
        ]
        # 添加当前请求时间
        self.ip_access[ip].append(now)
        # 判断是否超过限制
        if len(self.ip_access[ip]) > self.max_requests:
            self.blacklist.add(ip)
            return False
        return True
    def validate_ip(self, ip):
        """验证IP地址合法性"""
        try:
            ipaddress.ip_address(ip)
            return True
        except ValueError:
            return False
# 初始化IP拦截器
ip_blocker = IPBlocker()
@app.before_request
def block_malicious_ips():
    """请求前拦截恶意IP"""
    ip = request.remote_addr
    # 检查白名单
    if ip in ip_blocker.whitelist:
        return None
    # 检查黑名单
    if ip_blocker.is_banned(ip):
        abort(403, description="IP已被封禁")
    # 检查访问频率
    if not ip_blocker.check_rate_limit(ip):
        abort(429, description="请求过于频繁")
    return None
@app.route('/')
def index():
    return "正常访问"
if __name__ == '__main__':
    app.run(debug=True)

基于Redis的分布式IP拦截

import redis
import time
from functools import wraps
from flask import Flask, request, abort
app = Flask(__name__)
class RedisIPBlocker:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis_client = redis.Redis(
            host=redis_host, 
            port=redis_port, 
            decode_responses=True
        )
        self.block_prefix = "blocked_ip:"
        self.count_prefix = "ip_count:"
        self.max_requests = 100
        self.time_window = 60
        self.block_time = 3600
    def is_blocked(self, ip):
        """检查IP是否被封锁"""
        return self.redis_client.exists(f"{self.block_prefix}{ip}")
    def increment_count(self, ip):
        """增加访问计数"""
        key = f"{self.count_prefix}{ip}"
        count = self.redis_client.incr(key)
        # 设置过期时间
        if count == 1:
            self.redis_client.expire(key, self.time_window)
        return count
    def block_ip(self, ip):
        """封禁IP"""
        self.redis_client.setex(
            f"{self.block_prefix}{ip}", 
            self.block_time, 
            "blocked"
        )
        # 清除计数
        self.redis_client.delete(f"{self.count_prefix}{ip}")
    def check_ip(self, ip):
        """检查IP是否需要拦截"""
        # 检查是否已被封禁
        if self.is_blocked(ip):
            return False
        # 增加计数并检查是否超限
        count = self.increment_count(ip)
        if count > self.max_requests:
            self.block_ip(ip)
            return False
        return True
# 初始化
redis_blocker = RedisIPBlocker()
@app.before_request
def rate_limit():
    ip = request.remote_addr
    # 白名单检查（可放在Redis中）
    whitelist = redis_blocker.redis_client.smembers("whitelist")
    if ip in whitelist:
        return None
    if not redis_blocker.check_ip(ip):
        abort(403, description="IP已被限制访问")
@app.route('/')
def index():
    return "访问成功"
if __name__ == '__main__':
    app.run()

基于IP段的批量拦截

import ipaddress
from flask import Flask, request, abort
app = Flask(__name__)
class IPRangeBlocker:
    def __init__(self):
        # 配置需要拦截的IP段（来自威胁情报）
        self.blocked_ranges = [
            ipaddress.ip_network('192.168.0.0/16'),
            ipaddress.ip_network('10.0.0.0/8'),
            ipaddress.ip_network('172.16.0.0/12'),
        ]
        # 恶意IP名单
        self.malicious_ips = set()
        # 从文件加载恶意IP
        self.load_malicious_ips()
    def load_malicious_ips(self):
        """从文件加载恶意IP列表"""
        try:
            with open('malicious_ips.txt', 'r') as f:
                for line in f:
                    ip = line.strip()
                    if ip:
                        self.malicious_ips.add(ip)
        except FileNotFoundError:
            print("恶意IP文件未找到，使用默认配置")
    def is_blocked_ip(self, ip):
        """检查IP是否在拦截范围内"""
        try:
            ip_obj = ipaddress.ip_address(ip)
            # 检查是否在拦截的IP段内
            for network in self.blocked_ranges:
                if ip_obj in network:
                    return True
            # 检查是否在恶意IP列表中
            if ip in self.malicious_ips:
                return True
            return False
        except ValueError:
            return True  # 非法IP直接拦截
# 初始化
ip_range_blocker = IPRangeBlocker()
@app.before_request
def check_ip_range():
    ip = request.remote_addr
    if ip_range_blocker.is_blocked_ip(ip):
        abort(403, description="IP不在允许访问范围内")
@app.route('/')
def index():
    return "访问成功"
if __name__ == '__main__':
    app.run()

完整的防火墙脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess
import re
import sys
import time
from collections import defaultdict
import logging
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('ip_blocker.log'),
        logging.StreamHandler()
    ]
)
class IPFirewall:
    def __init__(self):
        self.iptables_cmd = '/sbin/iptables'
        self.ipset_cmd = '/usr/sbin/ipset'
        self.blacklist_file = '/etc/ip_blacklist.txt'
        self.whitelist_file = '/etc/ip_whitelist.txt'
        # 配置规则
        self.max_connections = 50
        self.check_interval = 10  # 检查间隔（秒）
    def check_root(self):
        """检查是否为root用户"""
        if os.geteuid() != 0:
            logging.error("需要root权限运行此脚本")
            sys.exit(1)
    def add_iptables_rule(self, ip):
        """添加iptables规则"""
        try:
            subprocess.run([
                self.iptables_cmd, '-A', 'INPUT',
                '-s', ip,
                '-j', 'DROP'
            ], check=True, capture_output=True)
            logging.info(f"已添加iptables规则：禁止 {ip}")
        except subprocess.CalledProcessError as e:
            logging.error(f"添加iptables规则失败：{e}")
    def remove_iptables_rule(self, ip):
        """移除iptables规则"""
        try:
            subprocess.run([
                self.iptables_cmd, '-D', 'INPUT',
                '-s', ip,
                '-j', 'DROP'
            ], check=True, capture_output=True)
            logging.info(f"已移除iptables规则：允许 {ip}")
        except subprocess.CalledProcessError as e:
            logging.error(f"移除iptables规则失败：{e}")
    def add_to_ipset(self, ip):
        """添加到ipset"""
        try:
            subprocess.run([
                self.ipset_cmd, 'add', 'blacklist', ip
            ], check=True, capture_output=True)
            logging.info(f"已将 {ip} 添加到ipset黑名单")
        except subprocess.CalledProcessError as e:
            logging.error(f"添加ipset失败：{e}")
    def block_ip_permanently(self, ip):
        """永久封禁IP"""
        # 添加到iptables
        self.add_iptables_rule(ip)
        # 写入黑名单文件
        with open(self.blacklist_file, 'a') as f:
            f.write(f"{ip}\n")
        logging.info(f"已永久封禁IP: {ip}")
    def unblock_ip(self, ip):
        """解封IP"""
        self.remove_iptables_rule(ip)
        logging.info(f"已解封IP: {ip}")
    def load_blacklist(self):
        """加载黑名单"""
        try:
            with open(self.blacklist_file, 'r') as f:
                for line in f:
                    ip = line.strip()
                    if ip and not ip.startswith('#'):
                        self.add_iptables_rule(ip)
        except FileNotFoundError:
            logging.warning("黑名单文件不存在")
    def analyze_logs(self, log_file='/var/log/auth.log'):
        """分析系统日志识别恶意IP"""
        ip_attempts = defaultdict(int)
        try:
            with open(log_file, 'r') as f:
                for line in f:
                    # 匹配失败登录尝试
                    if 'Failed password' in line:
                        match = re.search(r'from (\d+\.\d+\.\d+\.\d+)', line)
                        if match:
                            ip = match.group(1)
                            ip_attempts[ip] += 1
        except FileNotFoundError:
            logging.error(f"日志文件 {log_file} 不存在")
            return ip_attempts
        return ip_attempts
    def check_and_block(self):
        """检查并封禁恶意IP"""
        ip_attempts = self.analyze_logs()
        for ip, attempts in ip_attempts.items():
            if attempts >= self.max_connections:
                self.block_ip_permanently(ip)
                logging.warning(f"检测到暴力破解攻击：{ip} (尝试次数：{attempts})")
    def start_monitoring(self):
        """开始监控"""
        logging.info("IP防火墙监控已启动")
        while True:
            try:
                self.check_and_block()
                time.sleep(self.check_interval)
            except KeyboardInterrupt:
                logging.info("监控已停止")
                break
            except Exception as e:
                logging.error(f"监控出错：{e}")
                time.sleep(self.check_interval)
if __name__ == '__main__':
    import os
    firewall = IPFirewall()
    firewall.check_root()
    firewall.start_monitoring()

配置文件和用法

配置文件 config.py

# IP拦截配置
IP_BLOCK_CONFIG = {
    'redis': {
        'host': 'localhost',
        'port': 6379,
        'db': 0
    },
    'rate_limit': {
        'max_requests': 100,
        'time_window': 60,  # 60秒内
        'block_time': 3600  # 封禁1小时
    },
    'whitelist': [
        '127.0.0.1',
        '192.168.1.0/24',
        '10.0.0.0/8'
    ],
    'blacklist_file': '/etc/ip_blacklist.txt',
    'log_file': '/var/log/nginx/access.log',
    'monitor_interval': 10  # 监控间隔（秒）
}

Docker部署示例

FROM python:3.9-alpine
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# 添加iptables支持
RUN apk add --no-cache iptables ipset
EXPOSE 5000
CMD ["python", "app.py"]