### 解决方案 为了满足需求，可以从以下几个方面入手： #### 数据提取与正则表达式匹配 使用 Python 的 `re` 模块来解析日志文件中的 IP 地址。通常情况下，安全日志 (secure) 文件会记录访问者的 IP 和其他相关信息。假设日志格式类似于以下形式： ``` Failed password for invalid user admin from 192.168.1.1 port 22 ssh2 ``` 可以定义一个正则表达式模式用于匹配 IP 地址[^1]。 ```python import re from collections import Counter import geoip2.database def extract_ips_from_log(log_file_path): ip_pattern = r'\b(?:\d{1,3}\.){3}\d{1,3}\b' # 正则表达式匹配 IPv4 地址 with open(log_file_path, 'r') as file: log_data = file.read() ips = re.findall(ip_pattern, log_data) return ips ``` #### 统计所有出现过的 IP 地址 通过集合操作去重并统计所有的唯一 IP 地址。 ```python def get_unique_ips(ips): unique_ips = set(ips) return list(unique_ips) ``` #### 过滤出登录失败相关的 IP 及总失败次数 遍历日志文件，查找包含关键词（如 "Failed password" 或类似的错误消息）的日志条目，并从中提取对应的 IP 地址及其频率。 ```python def filter_failed_login_ips(log_file_path): failed_ip_pattern = r'Failed.*?(\b(?:\d{1,3}\.){3}\d{1,3}\b)' with open(log_file_path, 'r') as file: lines = file.readlines() failed_ips = [] for line in lines: match = re.search(failed_ip_pattern, line) if match: failed_ips.append(match.group(1)) return failed_ips ``` #### 对每个 IP 计算其重复次数 利用 `collections.Counter` 来统计每个 IP 出现的次数。 ```python def count_ip_occurrences(ips): counter = Counter(ips) return dict(counter) ``` #### 筛选出登录失败次数最多的前 5 个 IP 并附上它们所属的国家 借助 GeoLite2 数据库查询 IP 所属的地理位置信息。 ```python def get_top_n_failing_ips_with_countries(counts, n=5): top_ips = sorted(counts.items(), key=lambda item: item[1], reverse=True)[:n] reader = geoip2.database.Reader('GeoLite2-Country.mmdb') result = {} for ip, count in top_ips: try: response = reader.country(ip) country_name = response.country.name or "Unknown" except Exception: country_name = "Unknown" result[ip] = {'count': count, 'country': country_name} reader.close() return result ``` --- ### 完整流程示例代码 以下是完整的脚本实现上述功能： ```python import re from collections import Counter import geoip2.database def extract_ips_from_log(log_file_path): ip_pattern = r'\b(?:\d{1,3}\.){3}\d{1,3}\b' with open(log_file_path, 'r') as file: log_data = file.read() ips = re.findall(ip_pattern, log_data) return ips def get_unique_ips(ips): return list(set(ips)) def filter_failed_login_ips(log_file_path): failed_ip_pattern = r'Failed.*?(\b(?:\d{1,3}\.){3}\d{1,3}\b)' with open(log_file_path, 'r') as file: lines = file.readlines() failed_ips = [] for line in lines: match = re.search(failed_ip_pattern, line) if match: failed_ips.append(match.group(1)) return failed_ips def count_ip_occurrences(ips): return dict(Counter(ips)) def get_top_n_failing_ips_with_countries(counts, n=5): top_ips = sorted(counts.items(), key=lambda item: item[1], reverse=True)[:n] reader = geoip2.database.Reader('GeoLite2-Country.mmdb') result = {} for ip, count in top_ips: try: response = reader.country(ip) country_name = response.country.name or "Unknown" except Exception: country_name = "Unknown" result[ip] = {'count': count, 'country': country_name} reader.close() return result if __name__ == "__main__": log_file_path = "/var/log/secure" all_ips = extract_ips_from_log(log_file_path) unique_ips = get_unique_ips(all_ips) failed_ips = filter_failed_login_ips(log_file_path) counts = count_ip_occurrences(failed_ips) top_5_ips = get_top_n_failing_ips_with_countries(counts) print("Unique IPs:", unique_ips[:10]) # 显示部分结果 print("Top 5 Failing IPs:") for ip, data in top_5_ips.items(): print(f"{ip}: {data['count']} failures, Country: {data['country']}") ``` --- ### 注意事项 - **日志文件路径**：需确认 `/var/log/secure` 是否存在或替换为实际的安全日志路径。 - **GeoLite2 数据库**：下载 MaxMind 提供的免费数据库文件 (`GeoLite2-Country.mmdb`) 并放置于适当位置。 - **权限问题**：读取系统日志可能需要管理员权限，请确保运行环境具备相应权限。 ---