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来源IPfrom 192.168.1.100来源端口port 55678协议ssh2Debian/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 --version3.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 rFailed\spassword\s(?:for\s(?:invalid\suser\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()}, filesys.stderr) return None return { time: log_time, ip: ip_address, username: username, raw_line: line.strip() # 保留原始行用于调试 } def analyze_log_file(log_path, lookback_minutes10, failure_threshold5): 核心分析函数。 :param log_path: 日志文件路径 :param lookback_minutes: 分析最近多少分钟内的日志 :param failure_threshold: 判定为暴力破解的失败次数阈值 :return: 疑似攻击IP的列表每个元素是(ip, 详情字典) # 计算分析的时间起点 time_threshold datetime.now() - timedelta(minuteslookback_minutes) # 使用defaultdict来按IP聚合数据 ip_data defaultdict(lambda: {count: 0, first_seen: None, last_seen: None, usernames: set()}) try: with open(log_path, r, encodingutf-8, errorsignore) 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}, filesys.stderr) sys.exit(1) except PermissionError: print(f错误没有权限读取日志文件 - {log_path}。请使用sudo运行。, filesys.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(keylambda x: x[1][count], reverseTrue) 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, typeint, default10, help分析最近多少分钟内的日志默认10分钟) parser.add_argument(--threshold, -t, typeint, default5, help判定为暴力破解的失败次数阈值默认5次) parser.add_argument(--verbose, -v, actionstore_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 21)等命令辅助判断或者直接让用户通过--log参数指定。parse_log_line()函数与正则表达式这是脚本的核心解析器也是最容易出错的地方。fail_pattern这个正则表达式rFailed\spassword\s(?:for\s(?:invalid\suser\s)?(\S)\s)?from\s(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})是关键。(?: ... )?是一个非捕获分组表示“for username”这部分可能不存在。(?:invalid\suser\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 84.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 # 使用firewalldCentOS/RHEL 7 sudo firewall-cmd --permanent --add-rich-rulerule familyipv4 source address203.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_templatesudo 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(ipip) subprocess.run(command, shellTrue, checkTrue) 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.pyQ2: 脚本没有输出任何结果连“未发现”的提示都没有。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时那种“掌控感”是使用现成工具无法比拟的。安全是一个持续的对抗过程这个脚本就是你亲手打造的第一件武器。