利用Python与GeoLite2数据库分析fail2ban攻击源地理位置

1. 项目概述与核心价值

最近在排查一台对外服务的Linux服务器时，发现fail2ban的日志里密密麻麻记录着大量被拦截的IP地址。单纯看IP，你只能知道“谁”在攻击，但如果你想知道“他从哪里来”、“他属于哪个组织”，甚至“他是不是惯犯”，那就需要给这些冷冰冰的IP地址赋予地理位置和网络归属信息。这正是MaxMind GeoLite2数据库的用武之地。这个项目，就是教你如何将fail2ban的“战果”与GeoLite2数据库结合，用Python写一个脚本，自动分析出攻击者的来源国家、城市甚至所属的ISP（互联网服务提供商），从而让你对安全态势有一个更立体、更直观的认知。

对于任何管理着暴露在公网服务器的运维、开发或安全爱好者来说，这都是一项极具实用价值的技能。它不仅能帮你快速识别攻击来源是否集中在某个特定地区（比如是否遭遇了有组织的区域性扫描），还能在发生安全事件时，为溯源和撰写报告提供关键的数据支撑。整个过程不复杂，但串联起了日志分析、数据库查询和数据处理多个环节，是一个很好的练手项目。

2. 核心工具与原理拆解

2.1 fail2ban：服务器的自动门卫

fail2ban本质上是一个日志分析工具。它实时监控系统日志（如/var/log/auth.log,/var/log/nginx/error.log等），寻找预设规则匹配的失败尝试（例如SSH密码错误、Web应用暴力登录）。一旦某个IP在指定时间窗口内失败次数超过阈值，fail2ban就会通过操作系统的防火墙（如iptables或firewalld）临时或永久地封禁该IP。它的工作成果，通常记录在/var/log/fail2ban.log这个文件里，里面包含了被封禁IP的地址、封禁时间、触发服务（jail）等关键信息。我们的分析，就从解析这个日志文件开始。

2.2 MaxMind GeoLite2：IP的“身份证”数据库

MaxMind公司提供的GeoLite2数据库，是一个免费的、精度稍低于其商业版GeoIP2的IP地理位置数据库。它通过将IP地址段映射到具体的物理位置和网络属性，来回答关于一个IP的“灵魂三问”：你在哪（国家、城市、经纬度）？你是谁（自治系统号ASN、组织名称）？你用什么上网（网络类型，如移动网络、托管主机）？

GeoLite2主要包含两个我们最常用的数据库文件：

1. GeoLite2-Country.mmdb: 提供IP到国家级别的映射。
2. GeoLite2-City.mmdb: 提供更详细的信息，包括国家、细分（省/州）、城市、邮政编码、经纬度、时区等。
3. GeoLite2-ASN.mmdb: 提供IP到自治系统号（ASN）和所属组织（如“中国电信”、“Amazon.com”）的映射。

这些数据库文件是二进制的，采用了高效的MaxMind DB格式（.mmdb），查询速度非常快。我们需要通过编程接口来读取它们。

2.3 Python：粘合剂与数据分析器

Python在这里扮演了核心角色。我们需要用它完成三件事：

1. 日志解析：读取并解析fail2ban.log，提取出我们关心的IP地址、封禁时间等信息。这涉及到文本处理、正则表达式等。
2. 数据库查询：使用Python的geoip2库，加载.mmdb文件，并根据提取出的IP地址进行查询，获取地理位置和网络信息。
3. 数据整合与输出：将原始日志信息和查询结果整合在一起，并以一种易于阅读和分析的格式输出，比如控制台表格、CSV文件，甚至生成一个简单的统计图表。

整个流程的架构非常清晰：fail2ban产生日志 -> Python脚本解析日志 -> 查询GeoLite2数据库 -> 生成分析报告。

3. 环境准备与数据库获取

3.1 安装必要的Python库

首先，确保你的Python环境（建议Python 3.6以上）已经安装了必需的库。最核心的是geoip2，它是MaxMind官方提供的数据库读取库。

pip install geoip2

为了后续更好地格式化输出和分析，我们通常还会安装pandas和tabulate。pandas用于数据处理，tabulate用于在终端打印漂亮的表格。

pip install pandas tabulate

3.2 下载并部署GeoLite2数据库

MaxMind已经将GeoLite2数据库转为需要注册并获取许可证密钥才能下载。虽然免费，但步骤比过去直接下载文件多了一步。

1. 注册账号：访问 MaxMind 官网（https://www.maxmind.com），点击“Sign Up”注册一个免费账户。
2. 创建许可证密钥：登录后，进入“My License Key”页面，点击“Generate new license key”。为这个密钥起个名字，比如“My Server GeoIP”。
3. 下载数据库：有了许可证密钥后，你可以使用MaxMind提供的geoipupdate工具自动下载和更新数据库，这是推荐的方式。
   • 首先安装geoipupdate。在Ubuntu/Debian上：

     sudo apt update sudo apt install geoipupdate

   • 编辑配置文件/etc/GeoIP.conf。你需要修改以下几个关键配置：

     # 你的MaxMind账户ID（在“My Account”页面查看） AccountID YOUR_ACCOUNT_ID # 你刚才生成的许可证密钥 LicenseKey YOUR_LICENSE_KEY # 要下载的数据库ID列表，我们至少需要国家和城市 EditionIDs GeoLite2-Country GeoLite2-City GeoLite2-ASN

   • 运行更新命令下载数据库：

     sudo geoipupdate

   • 成功后，数据库文件默认会下载到/usr/share/GeoIP/目录下，文件名类似GeoLite2-City.mmdb。

注意：geoipupdate工具非常方便，它可以设置成定时任务（如每周一次），自动保持数据库最新。IP地理位置信息是动态变化的，定期更新很重要。

备选方案（手动下载）：如果你不想在服务器上配置geoipupdate，也可以在MaxMind官网登录后，手动下载.mmdb文件，然后上传到服务器的某个目录，比如/home/yourname/geoip_db/。只需在后续的Python脚本中指定正确的文件路径即可。

4. Python脚本核心代码实现与解析

接下来，我们一步步构建这个分析脚本。我会将完整脚本拆解成几个函数，并详细解释每一部分的作用和注意事项。

4.1 解析fail2ban日志文件

fail2ban的日志行格式通常如下：

2023-10-27 14:35:22,123 fail2ban.actions [12345]: NOTICE [sshd] Ban 192.168.1.100 2023-10-27 15:40:11,456 fail2ban.actions [12345]: NOTICE [nginx-botsearch] Ban 203.0.113.5

我们需要从中提取出时间戳、触发封禁的jail名称（如sshd）和被Ban的IP地址。

import re from datetime import datetime def parse_fail2ban_log(log_file_path): """ 解析fail2ban.log文件，提取封禁记录。 参数: log_file_path (str): fail2ban.log文件的完整路径。 返回: list: 包含字典的列表，每个字典是一条记录，格式为： {'timestamp': datetime对象, 'jail': str, 'ip': str} """ ban_records = [] # 匹配封禁行的正则表达式 # 匹配格式：日期时间 组件: 级别 [jail名] Ban IP地址 ban_pattern = re.compile( r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+\s+\S+\s+NOTICE\s+\[(?P<jail>\S+)\]\s+Ban\s+(?P<ip>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' ) try: with open(log_file_path, 'r', encoding='utf-8') as f: for line in f: match = ban_pattern.search(line) if match: time_str = match.group('timestamp') jail = match.group('jail') ip = match.group('ip') # 将字符串时间转换为datetime对象，便于后续处理 timestamp = datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S') ban_records.append({ 'timestamp': timestamp, 'jail': jail, 'ip': ip }) except FileNotFoundError: print(f"错误：找不到日志文件 {log_file_path}") return [] except Exception as e: print(f"解析日志文件时出错：{e}") return [] return ban_records

实操心得：fail2ban的日志格式在不同版本或自定义配置下可能有细微差别。上述正则表达式覆盖了最常见的情况。如果你的日志格式不同（例如包含了unban行），需要调整正则表达式。一个调试技巧是：先手动查看几行日志，然后用在线正则测试工具（如regex101.com）验证你的表达式是否能准确匹配。

4.2 查询GeoLite2数据库获取IP信息

有了IP列表，下一步就是查询数据库。我们使用geoip2.database.Reader来读取.mmdb文件。

import geoip2.database import socket def query_geoip_info(ip_list, city_db_path, asn_db_path): """ 批量查询IP的地理位置和ASN信息。 参数: ip_list (list): 需要查询的IP地址字符串列表。 city_db_path (str): GeoLite2-City.mmdb 文件路径。 asn_db_path (str): GeoLite2-ASN.mmdb 文件路径。 返回: dict: 以IP为键，查询结果为值的字典。 结果包含：国家、城市、经纬度、ASN编号、组织名称。 """ ip_info_map = {} # 验证IP地址基本格式，避免无效查询 def is_valid_ip(ip): try: socket.inet_pton(socket.AF_INET, ip) return True except socket.error: try: socket.inet_pton(socket.AF_INET6, ip) return True except socket.error: return False # 初始化数据库读取器 try: city_reader = geoip2.database.Reader(city_db_path) asn_reader = geoip2.database.Reader(asn_db_path) except FileNotFoundError as e: print(f"错误：找不到数据库文件。请检查路径。{e}") return ip_info_map except Exception as e: print(f"初始化数据库读取器失败：{e}") return ip_info_map for ip in ip_list: if not is_valid_ip(ip): ip_info_map[ip] = {'error': 'Invalid IP address'} continue info = {} try: # 查询城市信息（包含国家、城市等） city_response = city_reader.city(ip) info['country'] = city_response.country.name info['country_code'] = city_response.country.iso_code info['subdivision'] = city_response.subdivisions.most_specific.name if city_response.subdivisions else None info['city'] = city_response.city.name info['latitude'] = city_response.location.latitude info['longitude'] = city_response.location.longitude info['timezone'] = city_response.location.time_zone # 查询ASN信息 asn_response = asn_reader.asn(ip) info['asn'] = asn_response.autonomous_system_number info['organization'] = asn_response.autonomous_system_organization except geoip2.errors.AddressNotFoundError: # 数据库中没有该IP的记录（例如私有IP、保留IP） info['error'] = 'Address not found in database' except Exception as e: info['error'] = f'Query error: {str(e)}' ip_info_map[ip] = info # 关闭读取器，释放资源 city_reader.close() asn_reader.close() return ip_info_map

注意事项：

1. 数据库路径：务必提供正确的.mmdb文件路径。如果使用geoipupdate，默认路径是/usr/share/GeoIP/。
2. 异常处理：不是所有IP都能在数据库中找到对应记录（例如127.0.0.1、192.168.x.x等内网地址）。AddressNotFoundError是正常情况，我们的脚本需要妥善处理，避免崩溃。
3. 资源管理：Reader对象在使用完毕后必须调用.close()方法关闭，以释放文件句柄和内存。更优雅的做法是使用with语句上下文管理器。

4.3 整合数据并生成分析报告

现在，我们将日志记录和IP信息整合起来，并生成易于阅读的报告。这里我们选择两种输出方式：控制台打印和CSV文件导出。

import pandas as pd from tabulate import tabulate def generate_report(ban_records, ip_info_map, output_csv=None): """ 生成分析报告。 参数: ban_records (list): parse_fail2ban_log函数返回的记录列表。 ip_info_map (dict): query_geoip_info函数返回的IP信息字典。 output_csv (str, optional): 如果要导出CSV，提供文件名。 返回: pandas.DataFrame: 包含所有整合数据的数据框。 """ report_data = [] for record in ban_records: ip = record['ip'] geo_info = ip_info_map.get(ip, {}) row = { 'Timestamp': record['timestamp'].strftime('%Y-%m-%d %H:%M:%S'), 'Jail': record['jail'], 'IP Address': ip, 'Country': geo_info.get('country', 'N/A'), 'Country Code': geo_info.get('country_code', 'N/A'), 'Region/State': geo_info.get('subdivision', 'N/A'), 'City': geo_info.get('city', 'N/A'), 'ASN': geo_info.get('asn', 'N/A'), 'Organization': geo_info.get('organization', 'N/A'), 'Error': geo_info.get('error', '') # 如果有错误信息，记录在此 } report_data.append(row) # 创建DataFrame df = pd.DataFrame(report_data) # 1. 控制台输出摘要 if not df.empty: print("=== Fail2Ban封禁IP地理位置分析报告 ===") print(f"分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"总封禁记录数: {len(df)}") print(f"涉及独立IP数: {df['IP Address'].nunique()}") print("\n--- 攻击来源国家/地区TOP 5 ---") country_stats = df['Country'].value_counts().head() print(tabulate(country_stats.reset_index().values, headers=['Country', 'Count'], tablefmt='grid')) print("\n--- 攻击来源组织(ISP) TOP 5 ---") org_stats = df['Organization'].value_counts().head() print(tabulate(org_stats.reset_index().values, headers=['Organization', 'Count'], tablefmt='grid')) print("\n--- 详细记录 (前10条) ---") # 选择要显示的列，避免控制台输出过宽 display_columns = ['Timestamp', 'Jail', 'IP Address', 'Country', 'City', 'Organization'] print(tabulate(df[display_columns].head(10), headers='keys', tablefmt='grid', showindex=False)) else: print("未找到任何封禁记录。") # 2. 导出到CSV文件（如果指定了输出路径） if output_csv and not df.empty: try: df.to_csv(output_csv, index=False, encoding='utf-8-sig') # utf-8-sig支持Excel直接打开显示中文 print(f"\n详细报告已导出至: {output_csv}") except Exception as e: print(f"导出CSV文件失败: {e}") return df

4.4 主函数：串联整个流程

最后，我们编写一个主函数来调用上述所有功能，并允许通过命令行参数指定日志文件和数据库路径。

import argparse def main(): parser = argparse.ArgumentParser(description='分析fail2ban日志，并查询IP地理位置信息。') parser.add_argument('--log', default='/var/log/fail2ban.log', help='fail2ban日志文件路径 (默认: /var/log/fail2ban.log)') parser.add_argument('--city-db', default='/usr/share/GeoIP/GeoLite2-City.mmdb', help='GeoLite2-City数据库路径') parser.add_argument('--asn-db', default='/usr/share/GeoIP/GeoLite2-ASN.mmdb', help='GeoLite2-ASN数据库路径') parser.add_argument('--output', help='输出CSV报告的文件路径 (可选)') args = parser.parse_args() print("开始解析fail2ban日志...") ban_records = parse_fail2ban_log(args.log) if not ban_records: print("未解析到封禁记录，程序退出。") return # 提取唯一的IP地址列表，避免重复查询 unique_ips = list(set([record['ip'] for record in ban_records])) print(f"共发现 {len(ban_records)} 条封禁记录，涉及 {len(unique_ips)} 个独立IP。") print("开始查询IP地理位置信息...") ip_info_map = query_geoip_info(unique_ips, args.city_db, args.asn_db) print("生成分析报告...") _ = generate_report(ban_records, ip_info_map, args.output) print("\n分析完成。") if __name__ == '__main__': main()

4.5 完整脚本与使用示例

将以上所有代码块按顺序保存到一个文件中，例如analyze_fail2ban_geo.py。这样就得到了一个完整的分析工具。

基本使用方法：

1. 直接运行（使用默认路径）：假设数据库已安装在默认位置，且你有权限读取fail2ban日志。

   sudo python3 analyze_fail2ban_geo.py

   （需要sudo是因为/var/log/fail2ban.log通常只有root可读）

2. 指定自定义路径：

   sudo python3 analyze_fail2ban_geo.py --log /path/to/your/fail2ban.log --city-db /path/to/GeoLite2-City.mmdb --asn-db /path/to/GeoLite2-ASN.mmdb --output ./attack_report.csv

3. 查看帮助：

   python3 analyze_fail2ban_geo.py -h

运行后，你将在终端看到类似下面的摘要报告，并且会在当前目录生成一个包含所有详细信息的attack_report.csv文件，可以用Excel或文本编辑器打开。

=== Fail2Ban封禁IP地理位置分析报告 === 分析时间: 2023-10-27 16:20:05 总封禁记录数: 142 涉及独立IP数: 89 --- 攻击来源国家/地区TOP 5 --- +----------------+---------+ | Country | Count | +================+=========+ | China | 67 | | United States | 32 | | Netherlands | 12 | | Germany | 9 | | Russia | 7 | +----------------+---------+ --- 攻击来源组织(ISP) TOP 5 --- +-----------------------------------------------+---------+ | Organization | Count | +===============================================+=========+ | CHINA UNICOM China169 Backbone | 25 | | Amazon.com, Inc. | 18 | | CHINANET-BACKBONE | 15 | | Alibaba.com LLC | 8 | | Hetzner Online GmbH | 7 | +-----------------------------------------------+---------+

5. 高级分析与可视化扩展

基础报告已经很有用，但我们还可以更进一步，挖掘更多信息。

5.1 按时间维度进行分析

攻击往往不是均匀发生的。我们可以按小时或日期对攻击进行聚合，找出攻击的高峰时段。

def analyze_time_pattern(df): """分析攻击的时间模式""" if df.empty: return # 确保Timestamp列是datetime类型 df['Timestamp_dt'] = pd.to_datetime(df['Timestamp']) # 按小时统计攻击次数 df['Hour'] = df['Timestamp_dt'].dt.hour hourly_attacks = df.groupby('Hour').size() print("\n--- 24小时内攻击频率分布 ---") # 打印一个简单的柱状图（文本） max_count = hourly_attacks.max() for hour in range(24): count = hourly_attacks.get(hour, 0) bar = '█' * int((count / max_count) * 20) if max_count > 0 else '' print(f"{hour:02d}:00 - {hour:02d}:59 | {count:3d} {bar}") # 按星期统计 df['Weekday'] = df['Timestamp_dt'].dt.day_name() weekday_stats = df.groupby('Weekday').size().reindex(['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'], fill_value=0) print("\n--- 一周内攻击分布 ---") print(tabulate(weekday_stats.reset_index().values, headers=['Weekday', 'Count'], tablefmt='grid'))

在主函数generate_report中调用这个函数，可以立即看到攻击是否集中在某个特定时间段，例如是否总是在UTC时间凌晨（对应攻击者所在地的白天）出现高峰。

5.2 生成简单的地理分布图

虽然纯文本报告已经足够，但一张地图能更直观地展示攻击来源。我们可以使用folium库（基于Leaflet.js）生成一个交互式的HTML地图。

首先安装folium：

pip install folium

然后添加地图生成函数：

import folium from folium.plugins import HeatMap def generate_attack_map(df, output_html='attack_map.html'): """生成攻击源地理分布热力图""" if df.empty or 'latitude' not in df.columns or 'longitude' not in df.columns: print("无法生成地图：缺少经纬度数据。") return # 过滤掉无效的经纬度数据 valid_locations = df.dropna(subset=['latitude', 'longitude']) if valid_locations.empty: print("没有有效的经纬度数据用于生成地图。") return # 以所有攻击点的平均位置作为地图中心 center_lat = valid_locations['latitude'].mean() center_lon = valid_locations['longitude'].mean() # 创建地图对象 attack_map = folium.Map(location=[center_lat, center_lon], zoom_start=2) # 准备热力图数据：[[lat1, lon1, intensity1], ...] # 这里用每个IP出现的次数作为强度 ip_counts = df['IP Address'].value_counts().to_dict() heat_data = [] for _, row in valid_locations.iterrows(): ip = row['IP Address'] intensity = ip_counts.get(ip, 1) # 强度为该IP的攻击次数 heat_data.append([row['latitude'], row['longitude'], intensity]) # 添加热力图层 HeatMap(heat_data, radius=15, blur=10, max_zoom=1).add_to(attack_map) # 也可以添加标记点，显示详细信息 for _, row in valid_locations.head(20).iterrows(): # 只显示前20个点避免过于拥挤 popup_text = f""" <b>IP:</b> {row['IP Address']}<br> <b>时间:</b> {row['Timestamp']}<br> <b>攻击服务:</b> {row['Jail']}<br> <b>位置:</b> {row['City']}, {row['Country']}<br> <b>ISP:</b> {row['Organization']} """ folium.CircleMarker( location=[row['latitude'], row['longitude']], radius=5, popup=popup_text, color='red', fill=True ).add_to(attack_map) # 保存地图 attack_map.save(output_html) print(f"攻击源分布地图已生成: {output_html}") print(f"请用浏览器打开此HTML文件查看交互式地图。")

在主流程中，你需要确保将经纬度信息从ip_info_map整合到最终的df中，然后调用此函数。生成的地图会以热力图的形式高亮显示攻击密集的区域，点击标记点还能看到详细信息。

5.3 关联威胁情报（进阶思路）

对于安全要求更高的场景，可以进一步将查询到的IP、ASN组织与公开的威胁情报源进行关联。例如：

• 检查IP是否在已知的恶意IP黑名单中：可以调用AbuseIPDB、Virustotal等平台的API（通常有免费额度）。
• 分析ASN组织：频繁出现攻击的特定云服务商（如某个VPS提供商）或ISP，可能意味着攻击者偏好使用该平台。

这需要额外的API调用和数据处理，但能极大提升分析的深度。一个简单的实现是，在query_geoip_info函数后，增加一个enrich_with_threat_intel函数，对高风险国家的IP或特定ASN的IP进行标记。

6. 常见问题、优化与部署实践

在实际使用中，你可能会遇到一些问题。下面是一些常见情况的排查和优化建议。

6.1 常见问题与解决方案

6.2 性能优化与生产部署建议

当你的服务器非常繁忙，fail2ban日志每天产生数万条记录时，基础脚本可能需要优化。

1. 使用缓存：对于重复出现的IP（很多扫描器IP会反复尝试），每次运行脚本都重新查询数据库是浪费。可以引入一个简单的缓存文件（如JSON或SQLite），记录IP和其地理信息的映射。下次查询时先检查缓存。
2. 增量分析：脚本每次都是分析全部历史日志。可以修改脚本，只分析某个时间点之后的新日志（例如，记录上次分析的最后一条日志的时间戳）。
3. 定时任务与自动化：将脚本设置为cron定时任务（例如每天凌晨2点运行），并自动将报告通过邮件发送给管理员，或保存到特定的监控目录。

   # 编辑crontab -e 0 2 * * * /usr/bin/python3 /path/to/analyze_fail2ban_geo.py --output /var/log/fail2ban_geo_report_$(date +\%Y\%m\%d).csv

4. 与监控系统集成：可以将脚本的输出（如某个国家的攻击激增）与Zabbix、Prometheus等监控系统联动，触发告警。
5. 使用更高效的数据库：对于超大规模分析，可以考虑将GeoLite2数据导入到PostgreSQL+PostGIS或Elasticsearch中，利用数据库的索引进行快速关联查询。

6.3 安全与隐私考量

• 数据库准确性：GeoLite2是免费数据库，其准确性（尤其是城市级别）可能不如商业版GeoIP2。对于关键的法律或取证场景，建议评估或购买更精确的数据源。
• 日志隐私：分析报告中包含了IP地址。这份报告本身属于敏感信息，应妥善保管，避免公开泄露。
• 合规性：在某些地区，收集和处理IP地址信息可能受到隐私法规（如GDPR）的约束。请确保你的操作符合所在组织的规定和当地法律。

这个项目从一个小小的需求点出发，串联起了日志处理、外部数据库查询、数据分析和可视化的完整链条。它提供的不仅仅是一个“在哪里”的答案，更是你理解服务器所面临威胁态势的一扇窗口。通过定期运行和分析，你可以逐渐摸清攻击的模式，从而更有针对性地调整你的安全策略，比如针对高频攻击来源国的IP段进行预封禁，或者重点关注来自某些特定ISP的异常流量。