手把手教你用MaxMind GeoIP数据库分析fail2ban攻击日志（附Python代码）

实战指南利用MaxMind GeoIP深度解析fail2ban攻击日志最近在维护几台云服务器时发现fail2ban的拦截记录每天都在快速增长。作为运维人员我们不仅要拦截攻击更需要了解攻击来源。本文将分享如何通过MaxMind GeoIP数据库对fail2ban日志进行地理位置分析帮助您从海量日志中提取有价值的安全情报。1. 环境准备与工具选型在开始之前我们需要准备以下工具和环境MaxMind GeoLite2数据库提供IP地址与地理位置的映射关系Python 3.6环境用于编写分析脚本sqlite3命令行工具用于直接查看fail2ban数据库geoip2 Python库MaxMind官方提供的数据库查询接口提示虽然GeoLite2是免费版本但自2019年12月30日起MaxMind要求用户注册账户后才能下载。注册过程简单只需提供有效的电子邮箱地址。安装必要的Python依赖pip install geoip2 sqlite3 pandas matplotlib对于Ubuntu/Debian系统还需要安装sqlite3命令行工具sudo apt update sudo apt install sqlite32. 获取并理解fail2ban日志数据fail2ban默认将拦截记录存储在SQLite数据库中位置通常为/var/lib/fail2ban/fail2ban.sqlite3我们可以先用sqlite3命令行工具快速查看数据库结构sqlite3 /var/lib/fail2ban/fail2ban.sqlite3在sqlite交互界面中查看所有表SELECT name FROM sqlite_master WHERE typetable;典型输出可能包括bips存储所有被ban的IP及其相关信息bans记录每次ban操作的详细信息logsfail2ban的内部日志jails配置的jail信息对于攻击来源分析我们主要关注bips表它包含了被封锁IP的核心数据。可以通过以下查询了解表结构PRAGMA table_info(bips);3. 下载和配置MaxMind GeoLite2数据库访问MaxMind官网(https://www.maxmind.com)并登录后可以下载以下类型的免费数据库GeoLite2-ASNIP地址与自治系统号(ASN)的映射GeoLite2-City城市级别的IP地理位置数据GeoLite2-Country国家级别的IP地理位置数据对于安全分析推荐下载GeoLite2-City它提供了足够详细的位置信息。下载后会得到一个.mmdb格式的二进制文件。注意免费版的GeoLite2-City在某些地区(特别是东亚)的位置精度可能有限通常只能精确到国家或省级别。4. Python实战构建攻击来源分析工具下面是一个完整的Python脚本用于从fail2ban数据库提取IP并分析其地理位置import sqlite3 import geoip2.database from collections import Counter import pandas as pd import matplotlib.pyplot as plt # 配置路径 FAIL2BAN_DB /var/lib/fail2ban/fail2ban.sqlite3 GEOIP_DB /path/to/GeoLite2-City.mmdb def analyze_attacks(): # 连接fail2ban数据库 conn sqlite3.connect(FAIL2BAN_DB) cursor conn.cursor() # 获取所有被ban的IP cursor.execute(SELECT ip FROM bips) ips [row[0] for row in cursor.fetchall()] # 初始化GeoIP读取器 geo_reader geoip2.database.Reader(GEOIP_DB) # 分析每个IP的地理位置 countries [] cities [] for ip in ips: try: response geo_reader.city(ip) countries.append(response.country.name) cities.append(response.city.name if response.city.name else Unknown) except: countries.append(Unknown) cities.append(Unknown) # 统计结果 country_stats Counter(countries) city_stats Counter(cities) # 输出前10的攻击来源国家 print(\nTop 10攻击来源国家:) for country, count in country_stats.most_common(10): print(f{country}: {count}次) # 输出前10的攻击来源城市 print(\nTop 10攻击来源城市:) for city, count in city_stats.most_common(10): print(f{city}: {count}次) # 可视化展示 df pd.DataFrame({Country: countries, City: cities, IP: ips}) # 按国家统计图表 df[Country].value_counts().head(10).plot(kindbar, figsize(10,6)) plt.title(Top 10攻击来源国家) plt.ylabel(攻击次数) plt.tight_layout() plt.savefig(attack_by_country.png) # 关闭连接 geo_reader.close() conn.close() if __name__ __main__: analyze_attacks()这个脚本主要功能包括从fail2ban数据库提取所有被ban的IP地址使用MaxMind数据库查询每个IP的地理位置统计攻击来源国家和城市分布生成可视化图表5. 高级分析与实战技巧5.1 时间维度分析除了地理位置我们还可以分析攻击的时间模式# 在analyze_attacks函数中添加时间分析 cursor.execute(SELECT timeofban FROM bips ORDER BY timeofban) timestamps [row[0] for row in cursor.fetchall()] # 转换为datetime并分析 import datetime times [datetime.datetime.fromtimestamp(ts) for ts in timestamps] # 按小时统计攻击频率 hours [t.hour for t in times] pd.Series(hours).value_counts().sort_index().plot(kindbar, figsize(10,6)) plt.title(攻击时间分布(按小时)) plt.xlabel(小时) plt.ylabel(攻击次数) plt.savefig(attack_by_hour.png)5.2 自治系统(ASN)分析了解攻击IP所属的网络运营商也很有价值# 使用GeoLite2-ASN数据库分析 asn_reader geoip2.database.Reader(/path/to/GeoLite2-ASN.mmdb) asns [] for ip in ips: try: response asn_reader.asn(ip) asns.append(response.autonomous_system_organization) except: asns.append(Unknown) asn_stats Counter(asns) print(\nTop 10攻击来源ASN:) for asn, count in asn_stats.most_common(10): print(f{asn}: {count}次)5.3 数据持久化与历史分析为了长期跟踪攻击趋势可以将分析结果保存到数据库import sqlite3 from datetime import datetime def save_analysis_results(country_stats, city_stats): conn sqlite3.connect(attack_analysis.db) cursor conn.cursor() # 创建表(如果不存在) cursor.execute( CREATE TABLE IF NOT EXISTS attack_stats ( date TEXT PRIMARY KEY, total_attacks INTEGER, top_country TEXT, top_country_count INTEGER, top_city TEXT, top_city_count INTEGER ) ) # 插入当前分析结果 today datetime.now().strftime(%Y-%m-%d) top_country, top_country_count country_stats.most_common(1)[0] top_city, top_city_count city_stats.most_common(1)[0] cursor.execute( INSERT OR REPLACE INTO attack_stats VALUES (?, ?, ?, ?, ?, ?) , (today, len(ips), top_country, top_country_count, top_city, top_city_count)) conn.commit() conn.close()6. 可视化展示与报告生成将分析结果可视化能更直观地理解攻击模式。我们可以使用matplotlib和pandas生成专业图表def generate_visual_report(df): # 设置绘图风格 plt.style.use(seaborn) # 创建画布 fig, axes plt.subplots(2, 2, figsize(15, 12)) fig.suptitle(Fail2Ban攻击分析报告, fontsize16) # 国家分布 country_counts df[Country].value_counts().head(10) axes[0,0].bar(country_counts.index, country_counts.values) axes[0,0].set_title(Top 10攻击来源国家) axes[0,0].tick_params(axisx, rotation45) # 城市分布 city_counts df[df[City] ! Unknown][City].value_counts().head(10) axes[0,1].bar(city_counts.index, city_counts.values, colororange) axes[0,1].set_title(Top 10攻击来源城市) axes[0,1].tick_params(axisx, rotation45) # 时间分布 df[Hour] pd.to_datetime(df[Timestamp], units).dt.hour hour_counts df[Hour].value_counts().sort_index() axes[1,0].plot(hour_counts.index, hour_counts.values, markero) axes[1,0].set_title(攻击时间分布(24小时)) axes[1,0].set_xlabel(小时) axes[1,0].set_xticks(range(24)) # ASN分布 asn_counts df[ASN].value_counts().head(10) axes[1,1].barh(asn_counts.index, asn_counts.values, colorgreen) axes[1,1].set_title(Top 10攻击来源网络(ASN)) # 调整布局并保存 plt.tight_layout() plt.savefig(attack_analysis_report.png, dpi300, bbox_inchestight)7. 安全加固建议基于分析结果我们可以采取更有针对性的安全措施地理封锁如果发现大量攻击来自特定国家/地区可以考虑在防火墙层面屏蔽这些地区的IP段# 示例使用iptables屏蔽特定国家代码 iptables -A INPUT -m geoip --src-cc CN,RU -j DROP服务加固更改SSH默认端口禁用密码认证改用密钥认证设置fail2ban更严格的封禁规则监控与告警设置自动化脚本定期分析fail2ban日志当检测到异常攻击模式时发送告警网络架构优化将面向公网的服务放在跳板机后面使用VPN访问管理接口在实际项目中我发现将分析脚本设置为每日定时任务非常有用。通过长期收集数据可以识别出攻击模式的变化趋势为安全决策提供数据支持。