实战从Fail2ban日志到全球攻击地图——用Python与MaxMind GeoIP构建自动化威胁分析系统每天登录服务器看着fail2ban的日志里又多了几十条来自陌生IP的登录尝试你是否也曾好奇过这些攻击者究竟来自何方是随机的脚本小子还是有组织的定向扫描仅仅封禁IP地址似乎只是被动防御的终点。但如果我们能将这些冰冷的IP地址转化为一张可视化的全球攻击热力图洞察攻击来源的地理分布、网络归属甚至时间规律那么安全运维就从“救火”迈向了“预警与情报分析”的新阶段。本文面向具备一定Linux运维和Python基础的安全工程师、系统管理员及技术爱好者。我们将手把手带你构建一个完整的自动化分析流水线从fail2ban的SQLite数据库深处提取攻击IP利用MaxMind的GeoIP数据库为每个IP赋予地理位置“身份”最后通过Python脚本进行批量处理、统计分析与可视化呈现。整个过程不仅是一篇教程更是一个可立即部署到生产环境、持续输出安全情报的实用工具集。你会发现安全日志不再是负担而是洞察威胁态势的宝贵矿藏。1. 基石构建理解GeoIP数据与获取MaxMind数据库在开始编码之前我们必须先理解手中的“地图”从何而来以及它的精度和局限性。GeoIP技术的核心是将一个数字形式的IP地址映射到现实世界中的物理位置。这种映射关系并非通过GPS卫星定位而是基于互联网号码分配机构IANA将大段IP地址块分配给全球各地的互联网服务提供商ISP再结合这些ISP的注册信息、网络路由拓扑以及众包数据如Wi-Fi定位综合推算而成。MaxMind是这一领域的领先者它提供商业版和免费的GeoLite2版本数据库。对于个人研究、学习或非商业用途GeoLite2完全够用。它主要包含三种类型的数据GeoLite2 Country: 提供IP地址所属的国家信息。文件最小查询最快。GeoLite2 City: 提供国家、细分行政区如省/州、城市、经纬度、邮编部分区域等信息。这是我们进行攻击源地理分析最常用的数据库。GeoLite2 ASN: 提供自治系统号ASN及对应的组织名称通常是ISP。这对于判断攻击是来自某个数据中心、大型企业网络还是家庭宽带非常有价值。注意由于数据收集方式和隐私政策的限制免费版GeoLite2数据库的精度尤其在城市级别可能无法达到商业版的水平。对于中国境内的IP其城市级别的位置信息可能较为模糊或仅精确到省级。1.1 注册MaxMind账户并下载数据库首先你需要访问MaxMind官网进行注册。这个过程是免费的旨在遵循其最终用户许可协议。访问注册页面打开浏览器前往https://www.maxmind.com/en/geolite2/signup。填写注册信息使用你的邮箱完成注册流程。登录并生成许可证密钥注册成功后登录进入Account-Manage License Keys页面。点击Generate new license key为你的应用例如“Fail2ban GeoIP Analysis”创建一个密钥。务必妥善保存这个密钥后续的自动化下载脚本会用到它。下载数据库登录后你可以在Downloads-Download Files部分找到GeoLite2-City和GeoLite2-ASN的下载链接。但更推荐使用其提供的API进行自动化下载。为了便于后续自动化更新我们直接使用官方提供的下载API。以下是一个使用curl命令下载最新版GeoLite2-City数据库的示例# 替换YOUR_LICENSE_KEY为你在上一步获取的实际密钥 LICENSE_KEYYOUR_LICENSE_KEY_HERE wget https://download.maxmind.com/app/geoip_download?edition_idGeoLite2-Citylicense_key${LICENSE_KEY}suffixtar.gz -O GeoLite2-City.tar.gz # 解压下载的压缩包 tar -xzf GeoLite2-City.tar.gz # 通常解压后会得到一个以日期命名的文件夹如GeoLite2-City_20250131 # 进入该文件夹找到最大的.mmdb文件这就是我们需要的数据库文件 find . -name *.mmdb -type f | head -1解压后你会得到一个后缀为.mmdb的二进制文件这就是GeoIP2数据库文件。将其移动到一个固定的目录例如/usr/local/share/GeoIP/。sudo mkdir -p /usr/local/share/GeoIP sudo mv GeoLite2-City_*/GeoLite2-City.mmdb /usr/local/share/GeoIP/1.2 数据库的维护与更新MaxMind会定期更新其数据库通常每周二。为了确保分析的准确性建立一个自动更新机制至关重要。你可以创建一个简单的Shell脚本结合cron定时任务来实现。创建一个脚本文件例如/usr/local/bin/update_geoip.sh#!/bin/bash LICENSE_KEYYOUR_LICENSE_KEY_HERE GEOIP_DIR/usr/local/share/GeoIP BACKUP_DIR${GEOIP_DIR}/backup DATE$(date %Y%m%d) # 创建备份目录 mkdir -p $BACKUP_DIR # 备份旧数据库 cp ${GEOIP_DIR}/GeoLite2-City.mmdb ${BACKUP_DIR}/GeoLite2-City.mmdb.${DATE} 2/dev/null # 下载新版数据库 wget -q https://download.maxmind.com/app/geoip_download?edition_idGeoLite2-Citylicense_key${LICENSE_KEY}suffixtar.gz -O /tmp/geoip.tar.gz if [ $? -eq 0 ]; then # 解压并替换 tar -xzf /tmp/geoip.tar.gz -C /tmp NEW_DB$(find /tmp -name GeoLite2-City.mmdb -type f | head -1) if [ -f $NEW_DB ]; then sudo mv $NEW_DB ${GEOIP_DIR}/ echo $(date): GeoIP database updated successfully. else echo $(date): Failed to find .mmdb file in the archive. fi # 清理临时文件 rm -rf /tmp/GeoLite2-City_* /tmp/geoip.tar.gz else echo $(date): Failed to download GeoIP database. fi然后赋予执行权限并添加到cron每周执行一次sudo chmod x /usr/local/bin/update_geoip.sh # 编辑crontab例如每周一凌晨3点更新 sudo crontab -e # 添加一行0 3 * * 1 /usr/local/bin/update_geoip.sh2. 深入敌后解析Fail2ban的SQLite日志数据库fail2ban默认使用SQLite数据库来持久化存储封禁记录、日志和IP信息。相比直接解析纯文本日志文件直接查询数据库更加高效和结构化。数据库文件通常位于/var/lib/fail2ban/fail2ban.sqlite3。2.1 探索Fail2ban数据库结构首先让我们用sqlite3命令行工具探索一下这个数据库里有什么宝藏。# 连接到数据库 sudo sqlite3 /var/lib/fail2ban/fail2ban.sqlite3 # 查看所有表 sqlite .tables # 你可能会看到bans fail2banDb jails logs bips对于攻击IP分析最核心的表通常是bipsBad IPs和bans。bips表存储了所有被识别为“坏”的IP地址及其累计失败次数等信息。这是进行长期趋势分析的最佳数据源。bans表存储了历史上所有执行过的封禁动作记录包括IP、封禁时间、解禁时间、监狱jail名称等。适合分析封禁事件的时间线。让我们查看一下bips表的结构和部分数据-- 查看bips表结构 sqlite .schema bips -- 输出可能类似CREATE TABLE bips (ip TEXT PRIMARY KEY, failures INTEGER, timeofban INTEGER, ...) -- 查看最近记录的10个恶意IP及其失败次数 sqlite SELECT ip, failures FROM bips ORDER BY timeofban DESC LIMIT 10;一个典型的查询结果可能如下所示ipfailures203.0.113.4512198.51.100.238192.0.2.156252.2 设计高效的数据提取SQL为了给后续的Python脚本提供数据我们需要编写一个能够提取关键信息的SQL查询。除了IP地址我们可能还关心攻击发生的频率、最近发生的时间等。-- 提取所有恶意IP并按照失败次数降序排列便于优先分析最活跃的攻击源 SELECT ip, failures, datetime(timeofban, unixepoch) as last_ban_time FROM bips WHERE ip IS NOT NULL AND ip ! ORDER BY failures DESC;如果你希望分析特定时间段例如最近7天内新出现的攻击IP可以结合bans表进行更复杂的查询-- 查询最近7天内被首次封禁的IP地址 SELECT DISTINCT b.ip, COUNT(*) as ban_count, MAX(b.timeofban) as latest_ban FROM bans b WHERE b.timeofban strftime(%s, now, -7 days) GROUP BY b.ip ORDER BY ban_count DESC;将查询结果导出为CSV文件便于Python处理sqlite3 /var/lib/fail2ban/fail2ban.sqlite3 EOF .headers on .mode csv .output attack_ips.csv SELECT ip, failures FROM bips ORDER BY failures DESC; EOF这样我们就得到了一个名为attack_ips.csv的数据文件其中包含了我们需要分析的所有IP地址。3. 核心引擎用Python脚本桥接数据与地理位置现在我们有了IP列表数据源和GeoIP数据库地图。接下来就是用Python编写“翻译”引擎将IP地址转换为地理位置信息。我们将使用geoip2这个官方Python库。3.1 环境准备与库安装首先确保你的Python环境建议使用Python 3.6以上版本并安装必要的库。pip install geoip2 pandasgeoip2: 用于查询MaxMind的.mmdb数据库。pandas: 一个强大的数据分析库用于方便地处理和操作我们的IP数据及查询结果。3.2 编写GeoIP查询函数创建一个Python脚本例如geoip_analyzer.py。我们首先编写核心的查询函数。#!/usr/bin/env python3 # -*- coding: utf-8 -*- import geoip2.database import sqlite3 import pandas as pd from pathlib import Path import logging # 配置日志便于调试和记录错误 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class Fail2banGeoAnalyzer: def __init__(self, geoip_db_path, fail2ban_db_path): 初始化分析器加载GeoIP数据库和连接Fail2ban数据库。 :param geoip_db_path: GeoLite2-City.mmdb 文件路径 :param fail2ban_db_path: fail2ban.sqlite3 文件路径 self.geoip_db_path Path(geoip_db_path) if not self.geoip_db_path.is_file(): raise FileNotFoundError(fGeoIP database not found at {geoip_db_path}) self.fail2ban_db_path Path(fail2ban_db_path) if not self.fail2ban_db_path.is_file(): raise FileNotFoundError(fFail2ban database not found at {fail2ban_db_path}) logger.info(fLoading GeoIP database from {self.geoip_db_path}) self.reader geoip2.database.Reader(str(self.geoip_db_path)) logger.info(fConnecting to Fail2ban database at {self.fail2ban_db_path}) self.f2b_conn sqlite3.connect(str(self.fail2ban_db_path)) def lookup_ip(self, ip_address): 查询单个IP地址的地理位置信息。 :param ip_address: 字符串格式的IP地址 :return: 包含地理位置信息的字典查询失败返回None try: response self.reader.city(ip_address) geo_info { ip: ip_address, country_code: response.country.iso_code, country_name: response.country.name, subdivision: response.subdivisions.most_specific.name if response.subdivisions.most_specific else None, city: response.city.name, latitude: response.location.latitude, longitude: response.location.longitude, time_zone: response.location.time_zone, isp_asn: None # 如果需要ASN信息需使用GeoLite2-ASN数据库单独查询 } return geo_info except geoip2.errors.AddressNotFoundError: logger.warning(fIP address {ip_address} not found in GeoIP database.) return None except Exception as e: logger.error(fError looking up IP {ip_address}: {e}) return None def extract_ips_from_fail2ban(self, limitNone): 从fail2ban的bips表中提取IP地址和失败次数。 :param limit: 限制提取的记录数用于测试。None表示提取全部。 :return: pandas DataFrame包含ip和failures列 query SELECT ip, failures FROM bips WHERE ip IS NOT NULL AND ip ! ORDER BY failures DESC if limit: query f LIMIT {limit} df pd.read_sql_query(query, self.f2b_conn) logger.info(fExtracted {len(df)} IP records from fail2ban database.) return df这个类封装了核心功能初始化数据库连接、查询单个IP、从fail2ban数据库提取IP列表。注意lookup_ip方法中我们处理了IP未在数据库中找到的异常情况这在分析全球IP时很常见例如一些私有地址、保留地址或非常新的IP段。3.3 批量处理与数据整合接下来我们添加批量处理IP列表并生成分析报告的方法。# 在 Fail2banGeoAnalyzer 类中继续添加方法 def batch_lookup_and_analyze(self, ip_dataframe): 批量查询IP地理位置并生成分析数据。 :param ip_dataframe: 包含ip列的DataFrame :return: 合并了地理位置信息的DataFrame results [] total_ips len(ip_dataframe) for idx, row in ip_dataframe.iterrows(): ip row[ip] geo_data self.lookup_ip(ip) if geo_data: # 将原始行数据如failures和地理数据合并 merged_row {**row.to_dict(), **geo_data} results.append(merged_row) # 可选每处理100个IP打印一次进度 if (idx 1) % 100 0: logger.info(fProcessed {idx 1}/{total_ips} IPs...) result_df pd.DataFrame(results) logger.info(fSuccessfully resolved {len(result_df)} out of {total_ips} IPs.) return result_df def generate_summary_report(self, geo_df): 根据地理信息DataFrame生成统计摘要。 :param geo_df: 包含地理位置信息的DataFrame :return: 包含各类统计结果的字典 if geo_df.empty: return {error: No data to summarize} report { total_attacks: int(geo_df[failures].sum()) if failures in geo_df.columns else len(geo_df), unique_ips: geo_df[ip].nunique(), top_countries: geo_df[country_name].value_counts().head(10).to_dict(), top_cities: geo_df[geo_df[city].notna()][city].value_counts().head(10).to_dict(), attacks_by_country: {} } # 按国家统计攻击次数如果存在failures列 if failures in geo_df.columns: attacks_by_country geo_df.groupby(country_name)[failures].sum().sort_values(ascendingFalse).head(10) report[attacks_by_country] attacks_by_country.to_dict() else: attacks_by_country geo_df[country_name].value_counts().head(10) report[attacks_by_country] attacks_by_country.to_dict() return report def close(self): 关闭数据库连接释放资源。 self.reader.close() self.f2b_conn.close() logger.info(Resources closed.)3.4 主程序与输出最后编写主函数来串联整个流程并将结果输出为结构化的文件如JSON和CSV。def main(): # 配置路径 GEOIP_DB /usr/local/share/GeoIP/GeoLite2-City.mmdb FAIL2BAN_DB /var/lib/fail2ban/fail2ban.sqlite3 OUTPUT_PREFIX fail2ban_geo_analysis analyzer None try: analyzer Fail2banGeoAnalyzer(GEOIP_DB, FAIL2BAN_DB) # 1. 提取IP logger.info(Step 1: Extracting IPs from fail2ban...) ip_df analyzer.extract_ips_from_fail2ban() # 去掉limit参数以处理全部数据 # 对于首次测试可以加上limit100 if ip_df.empty: logger.warning(No IP addresses found in fail2ban database.) return # 2. 批量查询地理位置 logger.info(Step 2: Performing GeoIP lookup...) geo_df analyzer.batch_lookup_and_analyze(ip_df) # 3. 生成报告 logger.info(Step 3: Generating analysis report...) report analyzer.generate_summary_report(geo_df) # 4. 保存结果 # 保存详细的CSV数据 detailed_csv f{OUTPUT_PREFIX}_detailed.csv geo_df.to_csv(detailed_csv, indexFalse, encodingutf-8-sig) logger.info(fDetailed data saved to {detailed_csv}) # 保存JSON格式的摘要报告 import json summary_json f{OUTPUT_PREFIX}_summary.json with open(summary_json, w, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2) logger.info(fSummary report saved to {summary_json}) # 5. 在控制台打印关键发现 print(\n *50) print(攻击分析摘要) print(*50) print(f总计攻击尝试次数: {report.get(total_attacks, N/A)}) print(f涉及唯一IP地址数: {report.get(unique_ips, N/A)}) print(\n攻击来源TOP 10国家/地区:) for country, count in report.get(top_countries, {}).items(): print(f {country}: {count} 次) print(\n攻击最频繁的城市:) for city, count in report.get(top_cities, {}).items(): print(f {city}: {count} 次) except Exception as e: logger.error(fAn error occurred during analysis: {e}, exc_infoTrue) finally: if analyzer: analyzer.close() if __name__ __main__: main()运行这个脚本你将在当前目录得到两个文件一个包含每个IP详细地理信息的CSV文件和一个汇总了关键统计指标的JSON报告。控制台也会输出一份简洁的摘要。4. 从数据到洞察可视化分析与自动化仪表板得到结构化的数据后将其可视化是产生洞察的关键一步。一张地图或一个图表远比成行的数据更具说服力。这里我们介绍两种主流的可视化方式使用matplotlib/seaborn生成静态图表以及使用folium库生成交互式地图。4.1 使用Matplotlib/Seaborn进行统计图表可视化基于上一节生成的geo_df(DataFrame)我们可以轻松绘制攻击来源国家的分布柱状图。首先安装必要的可视化库pip install matplotlib seaborn然后在之前的脚本后追加以下函数或在新的脚本中调用保存的CSV文件import matplotlib.pyplot as plt import seaborn as sns def plot_attack_analysis(geo_df, output_imageattack_analysis.png): 生成攻击来源国家分布和城市分布图表。 plt.figure(figsize(15, 6)) # 子图1攻击来源国家TOP 10 plt.subplot(1, 2, 1) top_countries geo_df[country_name].value_counts().head(10) sns.barplot(xtop_countries.values, ytop_countries.index, paletteviridis) plt.title(Top 10 Attack Source Countries) plt.xlabel(Number of Unique IPs) plt.tight_layout() # 子图2攻击来源城市TOP 10 (排除空值) plt.subplot(1, 2, 2) city_data geo_df[geo_df[city].notna()] if not city_data.empty: top_cities city_data[city].value_counts().head(10) sns.barplot(xtop_cities.values, ytop_cities.index, paletterocket) plt.title(Top 10 Attack Source Cities (with Geo Data)) plt.xlabel(Number of Unique IPs) else: plt.text(0.5, 0.5, No city-level data available, hacenter, vacenter) plt.title(City Data Unavailable) plt.tight_layout() plt.savefig(output_image, dpi300, bbox_inchestight) plt.show() print(fChart saved as {output_image}) # 在主函数调用后添加 # plot_attack_analysis(geo_df)4.2 使用Folium创建交互式攻击地图静态图表很棒但交互式地图能提供更直观的空间感知。folium是一个基于Leaflet.js的Python库可以轻松创建地图。pip install folium创建一个新的脚本create_attack_map.pyimport folium from folium.plugins import HeatMap import pandas as pd def create_interactive_map(csv_file_path, output_mapattack_heatmap.html): 根据包含经纬度的CSV文件创建攻击源热力图。 df pd.read_csv(csv_file_path) # 清理数据确保经纬度存在且有效 df_clean df.dropna(subset[latitude, longitude]) # 过滤掉经纬度为0的点通常是数据缺失的默认值 df_clean df_clean[(df_clean[latitude] ! 0) (df_clean[longitude] ! 0)] if df_clean.empty: print(No valid geographic coordinates found for mapping.) return # 以数据点的平均经纬度为中心创建地图 map_center [df_clean[latitude].mean(), df_clean[longitude].mean()] m folium.Map(locationmap_center, zoom_start2, tilesCartoDB dark_matter) # 准备热力图数据格式为[纬度, 经度, 权重]。权重可以用失败次数这里用1表示每个IP点。 heat_data [[row[latitude], row[longitude], row.get(failures, 1)] for _, row in df_clean.iterrows()] # 添加热力图层 HeatMap(heat_data, radius15, blur10, max_zoom1).add_to(m) # 可选添加标记点点击显示IP和位置信息 for _, row in df_clean.head(50).iterrows(): # 只添加前50个点避免卡顿 popup_text f bIP:/b {row[ip]}br bLocation:/b {row.get(city, N/A)}, {row.get(country_name, N/A)}br bFailures:/b {row.get(failures, N/A)} folium.CircleMarker( location[row[latitude], row[longitude]], radius3, colorred, fillTrue, popuppopup_text ).add_to(m) # 保存为HTML文件 m.save(output_map) print(fInteractive map saved to {output_map}. Open it in your browser.) if __name__ __main__: # 使用之前脚本生成的详细CSV文件 create_interactive_map(fail2ban_geo_analysis_detailed.csv)运行此脚本后会生成一个HTML文件。用浏览器打开它你将看到一张全球地图攻击密集的区域会以高亮的热力图形式显示。你可以缩放、拖动地图并点击标记点查看具体IP的详细信息。4.3 构建自动化分析流水线与简易仪表板将以上所有步骤整合并加入定时任务你就拥有了一个完整的自动化威胁情报系统。思路如下定时数据更新使用cron每周运行一次update_geoip.sh脚本更新GeoIP数据库。定时分析任务使用另一个cron任务例如每天凌晨2点运行你的Python分析脚本geoip_analyzer.py。结果发布脚本运行后将生成的JSON摘要、CSV详细数据和HTML地图文件通过scp同步到一台内部Web服务器或者直接放在服务器本地的Web目录如/var/www/html/secure_dashboard/下。简易仪表板创建一个简单的HTML页面index.html使用JavaScript如Chart.js读取JSON摘要文件动态生成图表并嵌入生成的folium地图HTML。这样你每天只需打开一个内部网页就能看到最新的攻击态势全景。这个自建的仪表板不仅能让你对攻击来源一目了然长期积累的数据更能帮助你发现规律例如是否在特定时段如下班后攻击增多攻击是否主要来自某些特定的数据中心ASN可能预示着僵尸网络这些洞察是配置更精准的fail2ban规则如针对特定国家IP段进行更严格的检查或调整其他安全策略的宝贵依据。当我第一次在自己的服务器上运行起这套脚本看到地图上闪烁的攻击源从世界各地汇聚而来时那种感觉非常奇妙。它把抽象的日志条目变成了具象的威胁画像。最让我意外的是通过分析ASN信息我发现大量攻击并非来自分散的家庭IP而是集中在少数几个大型云服务商和数据中心的IP段。这直接促使我调整了安全组的入口规则在边界上就对已知的“高危”ASN进行更严格的流量限制。安全运维的视角从此从被动响应转向了主动的、基于数据的决策。