本文目录导读：
我来分享一个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"]
最佳实践建议
- 多层防御：结合多种拦截方法（频率、IP段、行为分析）
- 日志记录：记录所有拦截行为方便分析
- 定期更新：保持恶意IP库更新
- 性能优化：使用Redis、ipset等提高性能
- 误报处理：提供白名单机制
- 监控告警：集成告警系统
这些代码可以根据实际需求进行配置和扩展，建议先在测试环境运行,确认无误后再部署到生产环境。
