SSH爆破行为IP自动封禁系统：基于日志监控的自动化蓝队防御方案 | ZapcoMan's world

本文仅供合法授权的安全研究与学习使用，请勿用于非法用途。读者应确保行为符合当地法律法规。

前言

SSH服务是Linux服务器远程管理的核心入口，同时也是攻击者最常尝试突破的目标之一。根据各类安全报告统计，SSH暴力破解（Brute Force）始终位列服务器遭受攻击类型的前三名。攻击者通过批量扫描互联网上的22端口，利用字典进行密码爆破，一旦成功即可获取服务器控制权，进而进行横向渗透、数据窃取或植入后门。

对于蓝队防守人员而言，面对每天成千上万次的SSH爆破尝试，仅靠人工查看日志显然不现实。虽然社区有Fail2Ban、DenyHosts等成熟工具，但理解其底层原理并能够自主实现一套轻量级的自动封禁系统，对于深度掌握安全防御技术、定制化防护策略至关重要。

本文基于真实项目源码，讲解如何使用Python监控SSH安全日志、识别爆破行为并自动封禁恶意IP，从基础版逐步演进到带有白名单和时间窗口的增强版。

技术背景

SSH安全日志机制

Linux系统（以CentOS/RHEL为例）的SSH认证日志记录在 /var/log/secure 文件中。当发生认证失败时，日志中会出现类似以下格式的记录：

Invalid user admin from 192.168.1.100 port 54322
Failed password for invalid user root from 10.0.0.5 port 38912

这些日志包含了攻击来源IP、尝试的用户名、端口号等关键信息，是我们识别爆破行为的数据源。

hosts.deny 访问控制

Linux的TCP Wrapper机制通过 /etc/hosts.deny 和 /etc/hosts.allow 两个文件实现基于IP的访问控制。当在 hosts.deny 中写入 sshd:1.2.3.4 时，IP为1.2.3.4的主机将被拒绝SSH连接。这是一种轻量级、无需配置防火墙规则即可生效的封禁方式。

关键技术点

• tail -f 实时日志流：通过持续读取日志新增行，实现近实时响应
• 正则表达式提取：从非结构化日志文本中精确提取IP地址
• 计数器与阈值机制：区分”偶发输错密码”与”恶意爆破行为”
• 时间窗口滑动：在限定时间范围内的计数才有意义，避免误封长期正常用户

实现思路

整体架构是一个日志监控→模式匹配→计数判定→自动封禁的闭环流程：

1. 加载已有黑名单：程序启动时读取 /etc/hosts.deny，将已封禁IP载入内存字典，避免重复操作
2. 实时监控日志流：通过 subprocess.Popen 执行 tail -f /var/log/secure，逐行读取新增日志
3. 正则匹配攻击特征：使用正则表达式匹配两类关键日志行——“Invalid user”和”Failed password”
4. 阈值判定与封禁：
   • 对于”Invalid user”（无效用户名），说明攻击者在猜用户名，直接封禁
   • 对于”Failed password”（密码错误），可能是正常用户输错密码，需累计超过阈值后封禁
5. 写入黑名单：将 sshd:<IP> 追加写入 /etc/hosts.deny 实现封禁

增强版在此基础上增加了白名单机制和时间窗口滑动，提升安全性。

核心代码解析

基础版：自动封禁核心逻辑

以下代码来自 autoMaticallyBlockIP.py，实现了基本的日志监控与自动封禁功能：

import re
import subprocess
import time

# 定义安全日志文件路径
securityLog = '/var/log/secure'
# 定义黑名单文件路径
hostDeny = '/etc/hosts.deny'
# 定义封禁阈值，即密码错误的次数达到多少次后触发封禁
BlockThreshold = 5


def getDenies():
    """
    读取黑名单文件，将已经封禁的IP地址加载到内存中
    :return: 返回一个字典，键为已封禁的IP地址，值为标记'1'
    """
    deniedDict = {}
    blackIPList = open(hostDeny, 'r').readlines()
    for ip in blackIPList:
        # 使用正则从每行中提取IP地址
        group = re.search(r'(\d+\.\d+\.d+\.\d+)', ip)
        if group:
            deniedDict[group[1]] = '1'
    return deniedDict

getDenies() 函数在程序启动时加载已有黑名单，构建内存字典用于快速查询。使用字典的 O(1) 查找特性，在后续判断中能高效判断某个IP是否已被封禁。

核心监控函数 monitor() 是整个系统的心脏：

def monitor(securityLog):
    """
    监控安全日志，自动封禁可疑IP地址
    """
    # 初始化临时字典，用于记录每个IP的密码错误次数
    tempIp = {}
    # 初始化已封禁IP字典，加载已存在的封禁IP
    deniedDict = getDenies()
    # 使用subprocess实时读取安全日志的新增内容
    popen = subprocess.Popen('tail -f ' + securityLog, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, shell=True)
    while True:
        time.sleep(0.1)
        line = popen.stdout.readline().strip()
        if line:
            print(line)
            # 搜索无效用户登录尝试的IP地址
            group = re.search('Invalid user \w+ from (\d+\.\d+\.\d+\.\d+) port \d+.+user not known', str(line))
            # 如果找到无效用户且该IP未被封禁，则将其加入黑名单
            if group and not deniedDict.get(group[1]):
                subprocess.getoutput('echo \'sshd:{}\' >> {}'.format(group[1], hostDeny))
                deniedDict[group[1]] = '1'
                time_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
                print('{} >>>> add ip:{} to host.deny for invalid user 用户名无效'.format(time_str, group[1]))
                continue
            # 搜索有效用户密码错误的IP地址
            group = re.search('Failed password for invalid user \w+ from (\d+\.\d+\.\d+\.\d+) port \d+.+user not known', str(line))
            if group:
                ip = group[1]
                # 统计该IP的密码错误次数
                if not tempIp.get(ip):
                    tempIp[ip] = 1
                else:
                    tempIp[ip] += 1
                # 超过阈值且未被封禁，则封禁
                if tempIp[ip] > BlockThreshold and not deniedDict.get(ip):
                    del tempIp[ip]
                    subprocess.getoutput('echo \'sshd:{}\' >> {}'.format(ip, hostDeny))
                    deniedDict[ip] = '1'
                    print('{} >>>> add ip:{} to host.deny 密码无效次数超过阈值 自动封禁'.format(time_str, ip))

代码的关键设计在于区分两种攻击模式：

• “Invalid user”直接封禁：当攻击者尝试不存在的用户名登录时，这几乎是扫描器的标志性行为，正常用户不会频繁输入不存在的用户名，因此直接封禁
• “Failed password”阈值封禁：密码错误可能是正常用户的手误，因此通过 tempIp 字典累计计数，超过 BlockThreshold（默认5次）后才封禁

增强版：白名单 + 时间窗口

punchyAutoMaticallyBlockIP.py 在基础版上做了三个关键增强：

from collections import defaultdict

# 定义时间窗口（秒）
TimeWindow = 900  # 15分钟
# 定义白名单文件路径
whiteList = '/etc/hosts.allow'


def getWhites():
    """
    读取白名单文件，将可信的IP地址加载到内存中
    """
    whiteDict = {}
    with open(whiteList, 'r') as f:
        whiteIPList = f.readlines()
    for ip in whiteIPList:
        group = re.search(r'(\d+\.\d+\.\d+\.\d+)', ip)
        if group:
            whiteDict[group[1]] = '1'
    return whiteDict

白名单机制从 /etc/hosts.allow 加载可信IP，封禁前先检查白名单，避免误封运维人员的IP。

时间窗口滑动是增强版的核心改进，使用 defaultdict(list) 记录每次失败的时间戳：

tempIp = defaultdict(list)
# ...
current_time = time.time()
tempIp[ip].append(current_time)
# 移除时间窗口外的记录
tempIp[ip] = [t for t in tempIp[ip] if current_time - t <= TimeWindow]
if len(tempIp[ip]) > BlockThreshold and not deniedDict.get(ip):
    # 执行封禁

这段代码的精妙之处在于：它不是简单地累加计数，而是维护一个时间戳列表，每次新失败发生时，清除超过15分钟窗口的旧记录，只统计窗口内的失败次数。这样即使用户在过去一年里偶尔输错密码，也不会被封禁——只有短时间内高频失败才被视为爆破行为。

同时，增强版修复了 Popen 的命令注入风险，将 shell=True 改为列表传参：

popen = subprocess.Popen(['tail', '-f', securityLog], stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE)

部署与使用方法

环境要求

• 操作系统：Linux（CentOS/RHEL，日志在 /var/log/secure；Ubuntu需修改为 /var/log/auth.log）
• Python版本：3.6+
• 权限：需要root权限（读写 /etc/hosts.deny）

部署步骤

1. 配置白名单（推荐使用增强版）：将运维IP写入 /etc/hosts.allow

   sshd:192.168.1.100
   sshd:10.0.0.0/24

2. 运行脚本：

   # 基础版
   sudo python3 autoMaticallyBlockIP.py
   
   # 增强版（推荐）
   sudo python3 punchyAutoMaticallyBlockIP.py

3. 设为系统服务（推荐使用systemd实现开机自启）：

   [Unit]
   Description=SSH Auto Block IP Service
   After=network.target
   
   [Service]
   Type=simple
   ExecStart=/usr/bin/python3 /opt/security/punchyAutoMaticallyBlockIP.py
   Restart=always
   
   [Install]
   WantedBy=multi-user.target

参数调优建议

• BlockThreshold：建议设为3-5，过低会误封，过高则防护滞后
• TimeWindow：建议设为900秒（15分钟），平衡敏感度与误报率
• 白名单务必包含所有运维出口IP，避免”封禁自己”的事故

防御效果分析

有效性

该系统能有效应对以下攻击场景：

• SSH字典爆破：攻击者使用常见用户名密码组合批量尝试，几分钟内即被阈值捕获并封禁
• 用户名枚举扫描：尝试大量不存在的用户名（如admin、test、guest），触发”Invalid user”规则直接封禁
• 慢速爆破：增强版的时间窗口机制可识别低频持续攻击（在窗口内累计达到阈值）

局限性

• 仅依赖日志：若攻击者绕过SSH认证（如利用SSH漏洞），本系统无法检测
• IP可变：攻击者使用代理池轮换IP时，单个IP的失败次数可能不触发阈值
• 无解封机制：封禁后无法自动过期，需手动清理 hosts.deny
• hosts.deny 依赖：部分新版本Linux发行版默认不再编译支持TCP Wrapper的sshd，需确认sshd是否启用了libwrap支持

改进方向

对于生产环境，建议在上述方案基础上扩展：

• 集成iptables/nftables封禁，不依赖TCP Wrapper
• 增加封禁自动过期机制，避免黑名单无限增长
• 接入告警系统（如Webhook通知到企业微信/钉钉）
• 增加地理IP分析，对异常地区的连接提高敏感度

总结

本文基于真实Python源码，完整讲解了SSH爆破行为IP自动封禁系统的实现原理。从基础版的日志监控+正则匹配+阈值计数，到增强版的白名单保护+时间窗口滑动，展示了一个蓝队防御工具的完整演进过程。

这套方案的核心价值在于轻量、透明、可定制——无需安装复杂依赖，通过数百行Python代码即可实现与Fail2Ban类似的防护效果，且便于根据业务场景灵活调整规则。对于安全运维人员和蓝队工程师而言，理解并掌握这种基于日志分析的自建防御方案，不仅能提升对SSH攻击的响应能力，更能为后续构建更完善的安全自动化体系打下坚实基础。