基于Python的SSH暴力破解自动防御脚本实现
服务器运维中，SSH暴力破解是最常见的攻击方式之一。通过监控系统日志（如/var/log/secure或/var/log/auth.log），可以及时发现异常登录尝试，并自动将发起攻击的IP地址加入防火墙黑名单。以下是一个高效、可定制的Python脚本实现方案。
核心设计思路
脚本维护一个异常计数字典，以IP地址为键，存储失败尝试次数和最后时间戳。当某个IP在指定时间窗口内的失败次数超过阈值（例如3次），且未被拉黑，则执行以下操作：
- 将该IP添加至防火墙规则（如iptables）永久拒绝
- 将IP记录到本地黑名单文件，避免重复处理
- 可选：发送报警通知或记录详细日志
关键代码解析
1. 加载已有黑名单
def load_ban_list(file_path):
try:
with open(file_path, 'r') as f:
return {line.strip() for line in f if line.strip()}
except FileNotFoundError:
return set()
该函数从文本文件中读取已封禁的IP列表，返回集合以便快速判断。文件不存在时返回空集合，保证脚本首次运行不会报错。
2. 解析日志并记录异常行为
def monitor_log(log_path, ban_path, alert_log_path, threshold=3, keywords=None):
if keywords is None:
keywords = ["Failed password", "Invalid user", "Bad protocol version"]
banned_ips = load_ban_list(ban_path)
suspect = {} # {ip: {'count': int, 'last_seen': float}}
with open(log_path, 'r') as log_file:
# 只读取最后N行以提高效率（此处为200行）
lines = log_file.readlines()[-200:]
for line in lines:
for kw in keywords:
if kw not in line:
continue
matches = re.findall(r'\b(?:\d{1,3}\.){3}\d{1,3}\b', line)
if not matches:
continue
ip = matches[0]
now = time.time()
if ip in suspect:
# 检查时间窗口（默认5分钟）
if now - suspect[ip]['last_seen'] <= 300:
suspect[ip]['count'] += 1
suspect[ip]['last_seen'] = now
else:
# 超过窗口则重置
suspect[ip] = {'count': 1, 'last_seen': now}
else:
suspect[ip] = {'count': 1, 'last_seen': now}
# 达到阈值且未被封禁则处理
if suspect[ip]['count'] >= threshold and ip not in banned_ips:
ban_ip(ip)
banned_ips.add(ip)
# 追加到黑名单文件
with open(ban_path, 'a') as ban_file:
ban_file.write(ip + '\n')
# 记录告警日志
with open(alert_log_path, 'a') as alert_file:
alert_file.write(f"{time.ctime()} - Blocked {ip} (attempts: {suspect[ip]['count']})\n")
# 可选：发送邮件报警
# send_alert(ip)
3. 执行防火墙封禁
import subprocess
def ban_ip(ip):
"""使用iptables添加DROP规则"""
try:
subprocess.run(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP'], check=True)
print(f" [+] 已封禁 {ip}")
except subprocess.CalledProcessError as e:
print(f" [!] 封禁失败 {ip}: {e}")
扩展功能示例
地理位置追踪
利用ip-api.com的免费API，可在封禁时记录攻击来源国家：
import requests
def get_country(ip):
try:
resp = requests.get(f'http://ip-api.com/json/{ip}', timeout=2)
data = resp.json()
return data.get('country', 'Unknown')
except:
return 'N/A'
邮件通知
通过SMTP发送告警邮件至管理员：
import smtplib
from email.mime.text import MIMEText
def send_alert(ip, attempts=3):
msg = MIMEText(f'SSH爆破攻击检测：IP {ip} 已被自动封禁，尝试次数 {attempts}')
msg['Subject'] = '服务器安全预警'
msg['From'] = 'monitor@example.com'
msg['To'] = 'admin@example.com'
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('monitor@example.com', 'password')
server.send_message(msg)
脚本部署建议
- 将脚本加入cron任务，每分钟执行一次，例如：
* * * * * python3 /opt/ssh_monitor.py - 确保脚本以root权限运行（才能执行iptables命令）
- 定期检查告警日志，必要时调整阈值或关键词
- 对于Ubuntu系统，可将日志路径改为
/var/log/auth.log
