Python实战：从零编写Linux SSH暴力破解IP分析脚本-洪萨配资

1. 项目概述：为什么我们需要自己动手分析安全日志？

如果你刚接触Linux运维或者安全分析，可能会觉得“安全日志”这个词听起来既专业又遥远。服务器跑得好好的，为什么要去分析那些枯燥的日志文件？直到某天，你发现服务器CPU莫名飙高，或者SSH连接变得异常缓慢，甚至完全登录不上去，这时候再去翻看/var/log/secure或/var/log/auth.log，很可能已经为时已晚——你的服务器可能正在遭受持续的暴力破解攻击。

暴力破解，尤其是针对SSH服务的，是互联网上最常见、最“朴素”的攻击方式之一。攻击者通过自动化脚本，使用常见的用户名密码组合（如 root/123456, admin/admin等），以极高的频率尝试登录你的服务器。一旦成功，你的服务器就沦为了“肉鸡”。传统的防御依赖于像fail2ban或DenyHosts这样的工具，它们能自动封禁尝试失败的IP。但作为一个新手，或者当你需要更灵活的定制分析时，自己写一个Python脚本来“读懂”安全日志，不仅能让你深刻理解攻击模式，更能让你掌握主动防御的第一手信息。这就像给你的服务器装上了一双你自己的“眼睛”，而不是完全依赖黑盒工具。

这个项目，就是带你从零开始，用Python写一个轻量级但功能强大的脚本，自动解析Linux系统的安全日志，快速定位那些可疑的、正在进行暴力破解的IP地址，并生成直观的报告。整个过程，你不需要是安全专家，只需要有基础的Python知识和一点好奇心。我们将从日志格式讲起，一步步拆解分析逻辑，最后给你一份可以直接运行、甚至能加入定时任务的完整代码。

2. 核心思路与日志格式深度解析

在动手写代码之前，我们必须先搞清楚我们要分析的对象——Linux安全日志——到底长什么样。不同的Linux发行版，日志的位置和格式略有差异，但核心内容是一致的。

2.1 主流Linux系统的安全日志文件

RHEL/CentOS/Fedora/Rocky Linux/AlmaLinux等：主要日志文件是/var/log/secure。这个文件记录了所有与安全相关的认证信息，包括SSH登录、sudo提权、用户切换等。
Debian/Ubuntu等：主要日志文件是/var/log/auth.log。其作用和/var/log/secure完全相同。你可以通过一个简单的命令来确定你的系统使用哪个文件：
```
ls -la /var/log/secure /var/log/auth.log 2>/dev/null | head -5
```
通常，只会有一个文件存在。

2.2 一条典型的失败登录日志长什么样？

这是我们分析的基础。一条SSH登录失败的记录通常包含以下关键信息：

CentOS/RHEL系列 (/var/log/secure) 示例：

May 10 15:23:41 my-server sshd[12345]: Failed password for invalid user admin from 192.168.1.100 port 55678 ssh2 May 10 15:23:43 my-server sshd[12346]: Failed password for root from 203.0.113.50 port 44322 ssh2

时间戳：May 10 15:23:41
主机名：my-server
服务进程：sshd[12345]
关键动作：Failed password
用户信息：for invalid user admin或for root
来源IP：from 192.168.1.100
来源端口：port 55678
协议：ssh2

Debian/Ubuntu系列 (/var/log/auth.log) 示例：格式几乎一样，只是服务进程的表示可能略有不同，但核心字段Failed password、from、IP是完全一致的。

2.3 我们的分析逻辑设计

基于日志格式，我们的脚本核心任务就清晰了：

定位文件：自动判断并使用正确的日志文件（secure或auth.log）。
读取与过滤：逐行读取日志文件，只筛选出包含“Failed password”关键词的行。这是暴力破解最直接的证据。
信息提取：从每一行失败记录中，准确提取出来源IP地址。这是我们追踪攻击者的唯一标识。
统计与聚合：对提取到的所有IP进行计数统计。同一个IP出现的失败次数越多，其进行暴力破解的嫌疑就越大。
阈值判断与输出：设定一个阈值（例如，1分钟内失败5次）。超过这个阈值的IP，我们将其判定为“疑似暴力破解IP”，并输出详细信息，包括IP、失败次数、首次及最后一次出现的时间。

注意：这里我们只分析“失败”日志。成功的登录日志（Accepted password）同样重要，但对于初级的暴力破解分析，失败日志的密集程度是更直接、更敏感的指标。一个IP在短时间内产生大量失败记录，其恶意意图非常明显。

3. 实战：一步步构建Python分析脚本

现在，让我们把思路转化为代码。我将把脚本拆解成几个函数，并逐一讲解。你可以先通读，然后复制完整的代码去运行。

3.1 环境准备与依赖

你只需要一个标准的Python 3环境（建议3.6以上）。不需要安装任何第三方库，我们完全使用Python标准库，确保脚本在任何Linux服务器上都能开箱即用。检查你的Python版本：

python3 --version

3.2 脚本代码逐行解析

以下是完整的脚本代码，我将其保存为analyze_ssh_bruteforce.py。

#!/usr/bin/env python3 """ Linux SSH暴力破解IP分析脚本 作者：你的名字 功能：分析 /var/log/secure 或 /var/log/auth.log，找出短时间内多次登录失败的IP地址。 """ import re import sys from collections import defaultdict, Counter from datetime import datetime, timedelta import os import argparse def detect_auth_log_file(): """ 自动检测系统使用的认证日志文件。 优先检查 /var/log/secure，如果不存在则检查 /var/log/auth.log。 如果都找不到，则抛出异常。 """ possible_paths = ['/var/log/secure', '/var/log/auth.log'] for log_path in possible_paths: if os.path.exists(log_path): return log_path raise FileNotFoundError("未找到安全日志文件（尝试了 /var/log/secure 和 /var/log/auth.log）。请检查系统类型。") def parse_log_line(line): """ 解析单行日志，提取时间、IP地址、用户名。 返回一个字典，如果解析失败则返回None。 """ # 匹配时间戳（格式：Mmm DD HH:MM:SS） time_pattern = r'([A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})' # 匹配“Failed password”行并提取IP。这个正则表达式兼容两种常见格式。 # 格式1: Failed password for invalid user <username> from <IP> port ... # 格式2: Failed password for <username> from <IP> port ... fail_pattern = r'Failed\s+password\s+(?:for\s+(?:invalid\s+user\s+)?(\S+)\s+)?from\s+(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' time_match = re.search(time_pattern, line) fail_match = re.search(fail_pattern, line, re.IGNORECASE) if not (time_match and fail_match): return None # 不是我们关心的失败登录行 time_str = time_match.group(1) # 提取IP地址，fail_match.group(2)对应IP ip_address = fail_match.group(2) # 提取用户名，fail_match.group(1)可能为None（如果日志没记录用户） username = fail_match.group(1) if fail_match.group(1) else 'unknown' # 将时间字符串转换为datetime对象。这里假设年份是当前年（日志不记录年份）。 # 注意：跨年的日志处理需要额外逻辑，此处简化。 current_year = datetime.now().year try: # 构建完整的时间字符串，如 "May 10 15:23:41 2024" full_time_str = f"{time_str} {current_year}" log_time = datetime.strptime(full_time_str, '%b %d %H:%M:%S %Y') except ValueError as e: # 如果解析失败（比如2月30日这种非法日期），则跳过此行 print(f"警告：时间解析失败，跳过该行: {line.strip()}", file=sys.stderr) return None return { 'time': log_time, 'ip': ip_address, 'username': username, 'raw_line': line.strip() # 保留原始行用于调试 } def analyze_log_file(log_path, lookback_minutes=10, failure_threshold=5): """ 核心分析函数。 :param log_path: 日志文件路径 :param lookback_minutes: 分析最近多少分钟内的日志 :param failure_threshold: 判定为暴力破解的失败次数阈值 :return: 疑似攻击IP的列表，每个元素是(ip, 详情字典) """ # 计算分析的时间起点 time_threshold = datetime.now() - timedelta(minutes=lookback_minutes) # 使用defaultdict来按IP聚合数据 ip_data = defaultdict(lambda: {'count': 0, 'first_seen': None, 'last_seen': None, 'usernames': set()}) try: with open(log_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: parsed = parse_log_line(line) if not parsed: continue # 跳过无关行 log_time = parsed['time'] ip = parsed['ip'] username = parsed['username'] # 只分析指定时间窗口内的日志 if log_time < time_threshold: continue # 更新该IP的统计信息 ip_data[ip]['count'] += 1 ip_data[ip]['usernames'].add(username) if not ip_data[ip]['first_seen'] or log_time < ip_data[ip]['first_seen']: ip_data[ip]['first_seen'] = log_time if not ip_data[ip]['last_seen'] or log_time > ip_data[ip]['last_seen']: ip_data[ip]['last_seen'] = log_time except FileNotFoundError: print(f"错误：日志文件不存在 - {log_path}", file=sys.stderr) sys.exit(1) except PermissionError: print(f"错误：没有权限读取日志文件 - {log_path}。请使用sudo运行。", file=sys.stderr) sys.exit(1) # 筛选出超过阈值的IP suspicious_ips = [] for ip, data in ip_data.items(): if data['count'] >= failure_threshold: suspicious_ips.append((ip, data)) # 按失败次数降序排序 suspicious_ips.sort(key=lambda x: x[1]['count'], reverse=True) return suspicious_ips def generate_report(suspicious_ips, lookback_minutes, failure_threshold): """生成并打印分析报告。""" if not suspicious_ips: print(f"\n✅ 分析完成。在最近 {lookback_minutes} 分钟内，未发现失败次数超过 {failure_threshold} 次的IP地址。") return print(f"\n🚨 发现疑似暴力破解IP！分析时间窗口：最近 {lookback_minutes} 分钟，阈值：{failure_threshold} 次失败") print("=" * 80) print(f"{'IP地址':<20} {'失败次数':<12} {'尝试用户数':<12} {'首次出现时间':<25} {'最后出现时间':<25}") print("-" * 80) for ip, data in suspicious_ips: user_list = ', '.join(sorted(list(data['usernames']))[:3]) # 最多显示3个用户 if len(data['usernames']) > 3: user_list += f', ...(共{len(data['usernames'])}个)' first_time = data['first_seen'].strftime('%Y-%m-%d %H:%M:%S') last_time = data['last_seen'].strftime('%Y-%m-%d %H:%M:%S') print(f"{ip:<20} {data['count']:<12} {len(data['usernames']):<12} {first_time:<25} {last_time:<25}") # 可选：打印该IP尝试过的用户名 # print(f" 尝试过的用户: {user_list}") print("=" * 80) print(f"总计发现 {len(suspicious_ips)} 个可疑IP。") def main(): """主函数，处理命令行参数并执行分析。""" parser = argparse.ArgumentParser(description='分析Linux安全日志，找出暴力破解IP。') parser.add_argument('--log', '-l', help='指定日志文件路径（默认自动检测）') parser.add_argument('--minutes', '-m', type=int, default=10, help='分析最近多少分钟内的日志（默认：10分钟）') parser.add_argument('--threshold', '-t', type=int, default=5, help='判定为暴力破解的失败次数阈值（默认：5次）') parser.add_argument('--verbose', '-v', action='store_true', help='显示更详细的输出') args = parser.parse_args() # 确定日志文件路径 log_path = args.log if args.log else detect_auth_log_file() if args.verbose: print(f"🔍 正在分析日志文件: {log_path}") print(f"⏰ 时间窗口: 最近 {args.minutes} 分钟") print(f"📊 失败阈值: {args.threshold} 次") # 执行分析 suspicious_ips = analyze_log_file(log_path, args.minutes, args.threshold) # 生成报告 generate_report(suspicious_ips, args.minutes, args.threshold) if __name__ == '__main__': main()

3.3 关键代码段详解与避坑指南

detect_auth_log_file()函数：
- 作用：提高脚本的通用性。让脚本能在CentOS和Ubuntu上不加修改地运行。
- 避坑：有些定制化系统或容器环境可能将日志放在其他路径。生产环境中，可以考虑通过os.system('which fail2ban-server > /dev/null 2>&1')等命令辅助判断，或者直接让用户通过--log参数指定。
parse_log_line()函数与正则表达式：
- 这是脚本的核心解析器，也是最容易出错的地方。
- fail_pattern这个正则表达式r'Failed\s+password\s+(?:for\s+(?:invalid\s+user\s+)?(\S+)\s+)?from\s+(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'是关键。
  - (?: ... )?是一个非捕获分组，表示“for username”这部分可能不存在。
  - (?:invalid\s+user\s+)?匹配可能出现的“invalid user”字样。
  - (\S+)捕获用户名。
  - from\s+(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})捕获IP地址。
- 实操心得：不同系统、不同SSH配置（如日志等级）产生的日志格式可能有细微差别。在正式使用前，强烈建议用你自己服务器的几行真实日志测试一下这个正则表达式。可以在Python交互环境中导入这个函数进行测试。
时间处理：
- 日志里的时间没有年份，我们用datetime.now().year补全当前年份。这会导致一个经典问题：如果脚本在12月31日运行，分析1月1日的日志（假设日志跨年），时间判断会出错。对于短期（如几分钟/几小时）分析，这个问题影响不大。如果需要分析长时间跨度，必须处理年份滚动，逻辑会复杂很多。
analyze_log_file()函数中的ip_data：
- 使用defaultdict非常方便，可以自动初始化每个IP的统计字典。
- 我们不仅统计次数，还记录了first_seen和last_seen，这能让我们知道攻击持续了多久。记录usernames集合有助于判断攻击者是针对特定用户还是“广撒网”。
命令行参数 (argparse)：
- 这是让脚本变得专业和易用的关键。用户可以通过-m 60 -t 10来分析最近1小时内失败超过10次的IP，灵活性大大增强。

4. 运行脚本与结果解读

4.1 如何运行脚本？

将上面的代码保存为analyze_ssh_bruteforce.py。
赋予脚本执行权限：chmod +x analyze_ssh_bruteforce.py
由于安全日志通常需要root权限读取，我们使用sudo运行：

基本用法（使用默认参数：分析最近10分钟，失败阈值5次）：

sudo python3 analyze_ssh_bruteforce.py

带参数的高级用法：

# 分析最近1小时，失败阈值设为3次 sudo python3 analyze_ssh_bruteforce.py --minutes 60 --threshold 3 # 指定自定义日志路径（例如分析一个归档的日志） sudo python3 analyze_ssh_bruteforce.py --log /var/log/secure-20240510.gz --minutes 1440 # 显示详细过程 sudo python3 analyze_ssh_bruteforce.py -v -m 30 -t 8

4.2 解读分析报告

运行后，你可能会看到类似下面的输出：

🚨 发现疑似暴力破解IP！分析时间窗口：最近 10 分钟，阈值：5 次失败 ================================================================================ IP地址 失败次数 尝试用户数 首次出现时间 最后出现时间 -------------------------------------------------------------------------------- 203.0.113.50 87 4 2024-05-10 15:20:01 2024-05-10 15:29:58 192.168.1.100 23 1 2024-05-10 15:25:12 2024-05-10 15:29:45 45.33.32.156 12 8 2024-05-10 15:22:33 2024-05-10 15:29:12 ================================================================================ 总计发现 3 个可疑IP。

IP地址：攻击来源。
失败次数：在分析时间窗口内，该IP登录失败的次数。像203.0.113.50在10分钟内失败87次，这是非常明显的自动化攻击。
尝试用户数：该IP尝试了多少个不同的用户名。45.33.32.156尝试了8个用户，属于“广撒网”型攻击；而192.168.1.100只尝试了1个用户，可能是针对性的密码猜测。
首次/最后出现时间：攻击的时间范围。这有助于你判断攻击是持续性的还是间歇性的。

4.3 发现攻击IP后，我该怎么办？

脚本帮你找到了“嫌疑人”，接下来就是采取措施：

立即封禁：对于非常可疑的IP（如失败次数极高），可以立即手动封禁。

# 使用iptables（传统） sudo iptables -A INPUT -s 203.0.113.50 -j DROP # 使用firewalld（CentOS/RHEL 7+） sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="203.0.113.50" reject' sudo firewall-cmd --reload

加入定时任务：让脚本定期（如每5分钟）运行，并自动封禁超过阈值的IP。这需要将脚本与iptables或firewalld命令结合，并注意避免重复封禁。这是一个进阶功能，操作需谨慎，错误的封禁规则可能把自己锁在外面。
分析攻击模式：观察攻击IP的来源（可以用whois或在线IP归属地查询），尝试的用户名列表。如果攻击者总是尝试admin,root,test,oracle等用户，那么你应该确保这些账户要么不存在，要么使用了极其强壮的密码。
加固SSH服务：这是治本之策。
- 禁用密码登录，改用密钥对：这是最有效的一招。在/etc/ssh/sshd_config中设置PasswordAuthentication no。
- 修改默认端口：将SSH端口从22改为一个高位端口（如2222），可以避开大部分自动化扫描。
- 使用Fail2ban：这正是我们脚本模仿的原理。直接安装配置fail2ban，它能自动完成“检测-封禁-解封”的全流程。
- 限制访问来源：如果服务器有固定IP访问，可以在防火墙只允许特定IP段连接SSH端口。

5. 脚本的进阶优化与扩展思路

基础的脚本已经能解决大部分问题，但如果你想让它更强大、更实用，可以考虑以下方向：

5.1 优化一：支持日志轮转与实时监控

当前的脚本每次运行都读取整个日志文件。对于繁忙的服务器，日志文件可能很大。我们可以优化为：

记录上次读取的位置：将上次读取到的文件偏移量（file.tell()）或时间戳存入一个状态文件，下次只读取新的内容。这非常适合放入crontab中每分钟执行一次。
使用pyinotify库进行实时监控：当/var/log/secure有新增内容时，立即触发分析，实现近乎实时的攻击检测和响应。

5.2 优化二：集成自动封禁功能

这是最直接的扩展。在generate_report函数后，添加一个auto_block_ips函数。

def auto_block_ips(suspicious_ips, block_command_template="sudo iptables -A INPUT -s {ip} -j DROP"): """自动封禁可疑IP（请谨慎使用，并确保有解封机制）""" blocked = [] for ip, data in suspicious_ips: # 可以添加更复杂的判断条件，比如失败次数超过50次才封禁 if data['count'] > 50: import subprocess try: command = block_command_template.format(ip=ip) subprocess.run(command, shell=True, check=True) blocked.append(ip) print(f"已封禁IP: {ip} (失败次数: {data['count']})") except subprocess.CalledProcessError as e: print(f"封禁IP {ip} 失败: {e}") return blocked

⚠️ 严重警告：自动封禁风险极高！务必确保：
设置一个非常高的阈值（如50或100次），避免误封。
最好将封禁规则设置为临时生效（如iptables -A INPUT -s IP -m recent --set --name SSH_BLOCK配合--seconds 3600），一小时后自动解除。
务必保留一个不会被封禁的后门访问方式（如通过本地控制台，或另一个白名单IP）。

5.3 优化三：生成可视化报告或告警

输出HTML报告：使用Jinja2模板库，将分析结果生成一个漂亮的HTML页面，包含表格和简单的图表（通过Chart.js），方便每天查阅。
发送邮件或Slack告警：当发现高危IP时，通过smtplib库发送告警邮件，或通过requests库调用Slack Webhook，将信息推送到你的手机。
集成到监控系统：将脚本的输出格式化为JSON，方便被Zabbix, Prometheus等监控系统抓取，作为一个自定义监控项。

5.4 优化四：增强分析维度

地理信息查询：调用免费的IP地理信息API（如ip-api.com），在报告中显示攻击IP所在的国家和城市，这能帮你判断攻击来源。
攻击时间线分析：不只统计总数，还按分钟或小时聚合失败次数，绘制攻击频率图，看看攻击是均匀持续还是集中爆发。
用户名字典分析：统计所有被尝试的用户名，生成一个“最常被攻击用户名”排行榜，这能直观反映攻击者使用的字典。

6. 常见问题与排查技巧实录

在实际使用和教学过程中，我遇到过不少问题，这里总结一下：

Q1: 运行脚本提示“Permission denied”A1:安全日志(/var/log/secure或/var/log/auth.log)的默认权限是640，只有root和adm组用户可以读取。所以**必须使用sudo**来运行脚本。sudo python3 analyze_ssh_bruteforce.py

Q2: 脚本没有输出任何结果，连“未发现”的提示都没有。A2:按以下步骤排查：

检查日志路径：先用sudo python3 -c “print(open(‘/var/log/secure’).readline())”试试能否读出一行。确认脚本detect_auth_log_file函数找到的文件是对的。
检查时间窗口：默认只分析最近10分钟。如果你的攻击发生在更早之前，需要用-m参数扩大范围，例如-m 1440分析最近一天。
检查阈值：默认阈值是5次。如果某个IP只失败了3次，它不会被显示。可以尝试将阈值设为1：-t 1。
检查正则表达式：这是最常见的问题。用你的真实日志行测试一下parse_log_line函数。在脚本中临时添加几行调试代码，打印出它解析到的IP和时间。

Q3: 时间解析错误，提示“ValueError: unconverted data remains”A3:这通常是因为日志时间格式和脚本中strptime的格式字符串不匹配。有些系统日志可能在时间戳前有更长的前缀。你需要调整time_pattern正则表达式，或者先对日志行进行更通用的分割（比如按空格分割取前3个字段）。

Q4: 如何分析压缩过的历史日志？A4:脚本默认不能直接读取.gz文件。你可以：

先解压：gunzip -c /var/log/secure-20240510.gz | sudo python3 analyze_ssh_bruteforce.py --log -（注意--log -表示从标准输入读取）。
修改脚本，使用gzip.open代替open来智能处理。可以判断文件后缀名，如果是.gz就用gzip.open。

Q5: 误封了自己怎么办？A5:这是最可怕的运维事故。预防永远优于治疗：

本地测试：先在测试环境或虚拟机里充分测试封禁逻辑。
使用临时规则：使用iptables时，用-I（插入）而不是-A（追加）可能更安全，但最佳实践是为你的管理IP设置一个永久放行的规则，并放在链的最顶端。
```
sudo iptables -I INPUT 1 -s 你的公网IP -p tcp --dport 22 -j ACCEPT
```
准备后路：确保你有通过云服务商控制台（如AWS EC2的Instance Connect，阿里云的VNC）登录服务器的能力。这样即使网络层被完全封死，你还能从控制台进去修复规则。

写这个脚本的过程，本身就是一次绝佳的学习之旅。它强迫你去理解日志格式、正则表达式、时间处理、数据统计，甚至简单的命令行工具设计。当你看到脚本成功揪出那个疯狂尝试的IP时，那种“掌控感”是使用现成工具无法比拟的。安全是一个持续的对抗过程，这个脚本就是你亲手打造的第一件武器。