如何用脚本分析防火墙日志？

wen 实用脚本 2026-07-22 3

本文目录导读：

如何用脚本分析防火墙日志？

核心分析思路
常用脚本工具对比
Python 脚本示例（以 Syslog 格式为例）
处理 JSON 格式的日志（如 Palo Alto / AWS WAF）
进阶技巧与注意事项

分析防火墙日志是网络安全管理中非常重要的一环，可以帮助你识别攻击、异常流量、配置错误等，使用脚本（如 Python、Shell、PowerShell）来解析和筛选日志,能大幅提高效率。

以下是一个完整的指南，包含思路分析、常用工具以及一个可运行的 Python 脚本示例。

核心分析思路

在写脚本之前,你需要明确几个关键点：

日志格式是什么？ 常见格式：
- Syslog （最常见）：由时间戳、主机名、进程名、消息组成。
- W3C/CSV （如各类Web防火墙）：逗号或制表符分隔,有列头。
- 键值对 （如Fortinet、部分Palo Alto）：key1=value1 key2=value2。
- 固定宽度 （较老的防火墙）。
你想分析什么？ （即脚本的筛选条件）
- 源IP/目的IP：你的 IP 是否被扫描/攻击？谁在访问危险地址？
- 端口/协议：是否有人在扫 SSH（22）或 RDP（3389）？
- 动作：Deny（阻止）、Allow（允许）、Drop（丢弃），重点关注 Deny 和 Drop。
- 时间范围：攻击通常集中在某个时间段。
- 攻击特征：SQL注入、XSS、暴力破解等。
数据量有多大？
- 小日志（< 1GB）：Python Pandas 非常方便。
- 大日志（> 10GB）：建议使用 grep/awk（Linux）或 Logstash，或者 Python 的流式读取。
期望的输出是什么？
- Top 10 攻击源 IP
- 被阻断最多的目标 IP
- 异常时间段内的请求统计
- 告警通知（如钉钉/邮件）

常用脚本工具对比

工具	适用场景	优点	缺点
grep + awk + sort	Linux环境，快速排查	无需安装，处理文本极快	不支持复杂逻辑和正则
Python + re	任何环境，格式复杂	灵活，正则强大，可统计入库	慢于 awk，内存消耗大
Python + pandas	结构化日志（CSV/JSON）	数据分析极强，分组聚合简单	处理超大文件较慢
PowerShell	Windows 环境	内置 `Get-WinEvent`，管道处理	跨平台差，语法较怪

Python 脚本示例（以 Syslog 格式为例）

假设你的日志长这样（来自 Cisco ASA / iptables 的典型 syslog）：

<134>2023-10-27T10:15:30Z firewall.example.com %ASA-4-106023: Deny tcp src inside:192.168.1.10/54321 dst outside:203.0.113.5/80 by access-group "ACL_IN"
<134>2023-10-27T10:15:35Z firewall.example.com %ASA-4-106023: Deny tcp src inside:192.168.1.10/54322 dst outside:198.51.100.25/22 by access-group "ACL_IN"
<134>2023-10-27T10:16:01Z firewall.example.com %ASA-6-302013: Built outbound TCP connection 12345 for outside:10.0.0.1/443 to inside:192.168.1.1/50001

目标： 找出被拒绝次数最多的前 10 个源 IP，并统计它们攻击了多少个不同目标 IP。

import re
from collections import defaultdict
import sys
def parse_firewall_log(file_path):
    """
    解析防火墙 Syslog 日志，分析 Deny 事件。
    """
    # 正则：匹配 Deny 记录并提取源IP、目标IP、协议、端口
    # 注意：这需要根据你的具体设备格式进行调整
    # 示例格式: Deny tcp src inside:源IP/源端口 dst outside:目标IP/目标端口 by ...
    pattern = re.compile(
        r'Deny\s+(?P<protocol>\w+)\s+src\s+\w+:(?P<src_ip>\d+\.\d+\.\d+\.\d+)/(?P<src_port>\d+)\s+'
        r'dst\s+\w+:(?P<dst_ip>\d+\.\d+\.\d+\.\d+)/(?P<dst_port>\d+)'
    )
    # 数据结构
    src_ip_stats = defaultdict(lambda: {"count": 0, "target_ips": set(), "target_ports": set()})
    total_deny = 0
    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            for line in f:
                match = pattern.search(line)
                if match:
                    total_deny += 1
                    data = match.groupdict()
                    src = data['src_ip']
                    dst = data['dst_ip']
                    port = data['dst_port']
                    src_ip_stats[src]["count"] += 1
                    src_ip_stats[src]["target_ips"].add(dst)
                    src_ip_stats[src]["target_ports"].add(port)
    except FileNotFoundError:
        print(f"错误：文件 '{file_path}' 未找到。")
        sys.exit(1)
    return total_deny, src_ip_stats
def print_top_attackers(stats, top_n=10):
    """
    打印攻击者排行榜。
    """
    # 按被拒绝次数排序
    sorted_stats = sorted(stats.items(), key=lambda x: x[1]["count"], reverse=True)[:top_n]
    print(f"\n=== 攻击者 Top {top_n} ===")
    print(f"{'排名':<6}{'源IP':<20}{'拒绝次数':<10}{'攻击目标数':<12}{'攻击端口数':<12}")
    print("="*60)
    for rank, (ip, data) in enumerate(sorted_stats, start=1):
        print(f"{rank:<6}{ip:<20}{data['count']:<10}{len(data['target_ips']):<12}{len(data['target_ports']):<12}")
def main():
    if len(sys.argv) != 2:
        print("用法: python firewall_analysis.py <日志文件路径>")
        sys.exit(1)
    log_file = sys.argv[1]
    total_deny, ip_stats = parse_firewall_log(log_file)
    print(f"总共发现 {total_deny} 条 Deny 记录。")
    if total_deny > 0:
        print_top_attackers(ip_stats, top_n=10)
    else:
        print("未找到任何匹配的 Deny 记录，请检查正则表达式或日志格式。")
if __name__ == "__main__":
    main()

如何使用：

保存脚本 为 fw_analysis.py。
运行：
```
python fw_analysis.py ./firewall.log
```

输出示例：

总共发现 1500 条 Deny 记录。
=== 攻击者 Top 10 ===
排名   源IP                拒绝次数     攻击目标数    攻击端口数  
============================================================
1      192.168.1.10        500          10           5          
2      10.0.0.55           300          3            1          
...

处理 JSON 格式的日志（如 Palo Alto / AWS WAF）

如果你的日志是 JSON 格式，使用 Python 的 json 模块或 Pandas 会更简单。

# 使用 pandas 分析 JSON 日志
import pandas as pd
# 读取 JSON Lines 文件（每行一个 JSON）
df = pd.read_json('firewall_logs.json', lines=True)
# 假设字段是 'src_ip', 'action', 'dst_port'
# 1. 筛选出所有 'Deny' 动作
df_deny = df[df['action'] == 'deny']
# 2. 统计源 IP 出现次数 (Top 10)
top_ips = df_deny['src_ip'].value_counts().head(10)
print(top_ips)
# 3. 统计被阻断最多的端口
top_ports = df_deny['dst_port'].value_counts().head(5)
print(top_ports)
# 4. 按小时统计阻断量
df_deny['timestamp'] = pd.to_datetime(df_deny['timestamp'])
df_deny.set_index('timestamp', inplace=True)
hourly_blocked = df_deny.resample('1H').size()
print(hourly_blocked)

进阶技巧与注意事项

正则表达式调试：如果日志格式复杂，先手动用 Regex101 调试正则。
性能优化：如果文件非常大（几GB），不要一次性读取，用 for line in f: 逐行读取（如上Python示例）。
异常处理：脚本需要处理文件不存在、编码错误（errors='ignore'）、字段缺失等情况。
自动化：
- 配合 crontab（Linux）或 任务计划程序（Windows）定时运行。
- 输出结果后，通过 smtplib 发送邮件告警，或写入数据库（如 SQLite / Elasticsearch）。
分析别忘了 Allow：虽然 Deny 通常更可疑，但异常的 Allow（比如允许了高危端口出站）同样需要关注。
区分设备：
- iptables/nftables：可用 awk 快速提取 SRC= 和 DST=。
- Palo Alto：通常是 CSV 或 Syslog 带 TRAFFIC
- Fortigate：logver=... date=... time=... devname=... logid=...



小数据量 + 结构固定：awk / grep 最快。
格式复杂/需要统计：Python + re 是万能钥匙。
需要可视化/看趋势：Python + Pandas 或 ElasticSearch + Kibana。

从最基础的解析、筛选、统计开始，然后逐步添加告警、入库、可视化功能,就能构建一个实用的防火墙日志分析工具。