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 GeoLite2IP的“身份证”数据库MaxMind公司提供的GeoLite2数据库是一个免费的、精度稍低于其商业版GeoIP2的IP地理位置数据库。它通过将IP地址段映射到具体的物理位置和网络属性来回答关于一个IP的“灵魂三问”你在哪国家、城市、经纬度你是谁自治系统号ASN、组织名称你用什么上网网络类型如移动网络、托管主机GeoLite2主要包含两个我们最常用的数据库文件GeoLite2-Country.mmdb: 提供IP到国家级别的映射。GeoLite2-City.mmdb: 提供更详细的信息包括国家、细分省/州、城市、邮政编码、经纬度、时区等。GeoLite2-ASN.mmdb: 提供IP到自治系统号ASN和所属组织如“中国电信”、“Amazon.com”的映射。这些数据库文件是二进制的采用了高效的MaxMind DB格式.mmdb查询速度非常快。我们需要通过编程接口来读取它们。2.3 Python粘合剂与数据分析器Python在这里扮演了核心角色。我们需要用它完成三件事日志解析读取并解析fail2ban.log提取出我们关心的IP地址、封禁时间等信息。这涉及到文本处理、正则表达式等。数据库查询使用Python的geoip2库加载.mmdb文件并根据提取出的IP地址进行查询获取地理位置和网络信息。数据整合与输出将原始日志信息和查询结果整合在一起并以一种易于阅读和分析的格式输出比如控制台表格、CSV文件甚至生成一个简单的统计图表。整个流程的架构非常清晰fail2ban产生日志 - Python脚本解析日志 - 查询GeoLite2数据库 - 生成分析报告。3. 环境准备与数据库获取3.1 安装必要的Python库首先确保你的Python环境建议Python 3.6以上已经安装了必需的库。最核心的是geoip2它是MaxMind官方提供的数据库读取库。pip install geoip2为了后续更好地格式化输出和分析我们通常还会安装pandas和tabulate。pandas用于数据处理tabulate用于在终端打印漂亮的表格。pip install pandas tabulate3.2 下载并部署GeoLite2数据库MaxMind已经将GeoLite2数据库转为需要注册并获取许可证密钥才能下载。虽然免费但步骤比过去直接下载文件多了一步。注册账号访问 MaxMind 官网https://www.maxmind.com点击“Sign Up”注册一个免费账户。创建许可证密钥登录后进入“My License Key”页面点击“Generate new license key”。为这个密钥起个名字比如“My Server GeoIP”。下载数据库有了许可证密钥后你可以使用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(?Ptimestamp\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d\s\S\sNOTICE\s\[(?Pjail\S)\]\sBan\s(?Pip\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) ) try: with open(log_file_path, r, encodingutf-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] fQuery error: {str(e)} ip_info_map[ip] info # 关闭读取器释放资源 city_reader.close() asn_reader.close() return ip_info_map注意事项数据库路径务必提供正确的.mmdb文件路径。如果使用geoipupdate默认路径是/usr/share/GeoIP/。异常处理不是所有IP都能在数据库中找到对应记录例如127.0.0.1、192.168.x.x等内网地址。AddressNotFoundError是正常情况我们的脚本需要妥善处理避免崩溃。资源管理Reader对象在使用完毕后必须调用.close()方法关闭以释放文件句柄和内存。更优雅的做法是使用with语句上下文管理器。4.3 整合数据并生成分析报告现在我们将日志记录和IP信息整合起来并生成易于阅读的报告。这里我们选择两种输出方式控制台打印和CSV文件导出。import pandas as pd from tabulate import tabulate def generate_report(ban_records, ip_info_map, output_csvNone): 生成分析报告。 参数: 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], tablefmtgrid)) print(\n--- 攻击来源组织(ISP) TOP 5 ---) org_stats df[Organization].value_counts().head() print(tabulate(org_stats.reset_index().values, headers[Organization, Count], tablefmtgrid)) print(\n--- 详细记录 (前10条) ---) # 选择要显示的列避免控制台输出过宽 display_columns [Timestamp, Jail, IP Address, Country, City, Organization] print(tabulate(df[display_columns].head(10), headerskeys, tablefmtgrid, showindexFalse)) else: print(未找到任何封禁记录。) # 2. 导出到CSV文件如果指定了输出路径 if output_csv and not df.empty: try: df.to_csv(output_csv, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel直接打开显示中文 print(f\n详细报告已导出至: {output_csv}) except Exception as e: print(f导出CSV文件失败: {e}) return df4.4 主函数串联整个流程最后我们编写一个主函数来调用上述所有功能并允许通过命令行参数指定日志文件和数据库路径。import argparse def main(): parser argparse.ArgumentParser(description分析fail2ban日志并查询IP地理位置信息。) parser.add_argument(--log, default/var/log/fail2ban.log, helpfail2ban日志文件路径 (默认: /var/log/fail2ban.log)) parser.add_argument(--city-db, default/usr/share/GeoIP/GeoLite2-City.mmdb, helpGeoLite2-City数据库路径) parser.add_argument(--asn-db, default/usr/share/GeoIP/GeoLite2-ASN.mmdb, helpGeoLite2-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。这样就得到了一个完整的分析工具。基本使用方法直接运行使用默认路径假设数据库已安装在默认位置且你有权限读取fail2ban日志。sudo python3 analyze_fail2ban_geo.py需要sudo是因为/var/log/fail2ban.log通常只有root可读指定自定义路径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查看帮助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_value0) print(\n--- 一周内攻击分布 ---) print(tabulate(weekday_stats.reset_index().values, headers[Weekday, Count], tablefmtgrid))在主函数generate_report中调用这个函数可以立即看到攻击是否集中在某个特定时间段例如是否总是在UTC时间凌晨对应攻击者所在地的白天出现高峰。5.2 生成简单的地理分布图虽然纯文本报告已经足够但一张地图能更直观地展示攻击来源。我们可以使用folium库基于Leaflet.js生成一个交互式的HTML地图。首先安装foliumpip install folium然后添加地图生成函数import folium from folium.plugins import HeatMap def generate_attack_map(df, output_htmlattack_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_start2) # 准备热力图数据[[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, radius15, blur10, max_zoom1).add_to(attack_map) # 也可以添加标记点显示详细信息 for _, row in valid_locations.head(20).iterrows(): # 只显示前20个点避免过于拥挤 popup_text f bIP:/b {row[IP Address]}br b时间:/b {row[Timestamp]}br b攻击服务:/b {row[Jail]}br b位置:/b {row[City]}, {row[Country]}br bISP:/b {row[Organization]} folium.CircleMarker( location[row[latitude], row[longitude]], radius5, popuppopup_text, colorred, fillTrue ).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 常见问题与解决方案问题现象可能原因解决方案脚本报错geoip2.errors.AddressNotFoundError或查询结果大量为N/A1. 数据库文件路径错误。2. 查询的IP是私有地址如10.x.x.x,192.168.x.x或保留地址。3. 数据库文件损坏或版本太旧。1. 使用绝对路径并用ls命令检查文件是否存在。2. 这是正常现象内网IP本无地理信息。可在脚本中过滤掉这些IP。3. 运行sudo geoipupdate更新数据库。运行脚本提示Permission denied无法读取日志/var/log/fail2ban.log文件默认只有root用户可读。使用sudo运行脚本sudo python3 analyze_fail2ban_geo.py。或者将当前用户加入有读取权限的组不推荐。查询速度慢尤其是日志文件很大时1. 脚本为每个IP包括重复的都查询了一次数据库。2. 数据库文件在机械硬盘上且脚本频繁打开关闭。1. 像我们代码中那样先提取唯一IP(unique_ips) 再进行批量查询避免重复工作。2. 确保数据库文件在SSD上。在脚本中Reader对象在整个查询周期内只打开一次而不是每次查询都打开。输出报告中的中文城市/组织名显示为乱码终端或CSV文件的编码问题。1. 确保终端支持UTF-8。2. 导出CSV时使用utf-8-sig编码如代码所示该编码兼容Excel。geoipupdate更新失败1. 许可证密钥配置错误。2. 网络问题无法连接MaxMind服务器。1. 检查/etc/GeoIP.conf中的AccountID和LicenseKey是否正确。2. 尝试手动下载数据库文件并替换。6.2 性能优化与生产部署建议当你的服务器非常繁忙fail2ban日志每天产生数万条记录时基础脚本可能需要优化。使用缓存对于重复出现的IP很多扫描器IP会反复尝试每次运行脚本都重新查询数据库是浪费。可以引入一个简单的缓存文件如JSON或SQLite记录IP和其地理信息的映射。下次查询时先检查缓存。增量分析脚本每次都是分析全部历史日志。可以修改脚本只分析某个时间点之后的新日志例如记录上次分析的最后一条日志的时间戳。定时任务与自动化将脚本设置为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与监控系统集成可以将脚本的输出如某个国家的攻击激增与Zabbix、Prometheus等监控系统联动触发告警。使用更高效的数据库对于超大规模分析可以考虑将GeoLite2数据导入到PostgreSQLPostGIS或Elasticsearch中利用数据库的索引进行快速关联查询。6.3 安全与隐私考量数据库准确性GeoLite2是免费数据库其准确性尤其是城市级别可能不如商业版GeoIP2。对于关键的法律或取证场景建议评估或购买更精确的数据源。日志隐私分析报告中包含了IP地址。这份报告本身属于敏感信息应妥善保管避免公开泄露。合规性在某些地区收集和处理IP地址信息可能受到隐私法规如GDPR的约束。请确保你的操作符合所在组织的规定和当地法律。这个项目从一个小小的需求点出发串联起了日志处理、外部数据库查询、数据分析和可视化的完整链条。它提供的不仅仅是一个“在哪里”的答案更是你理解服务器所面临威胁态势的一扇窗口。通过定期运行和分析你可以逐渐摸清攻击的模式从而更有针对性地调整你的安全策略比如针对高频攻击来源国的IP段进行预封禁或者重点关注来自某些特定ISP的异常流量。