1. 项目概述与核心价值最近在梳理我们内部安全运营流程时发现一个痛点安全设备告警很多但真正需要人工介入去封禁IP的往往就那么几类高危攻击。每次登录防火墙、查日志、确认、再手动添加黑名单一套流程下来攻击可能已经持续了十几分钟。对于深信服AF这类下一代防火墙其日志里其实已经包含了非常精准的攻击判定信息如果能自动处理效率会高得多。于是我花了点时间把Zabbix监控和深信服AF 8.0的防火墙联动了起来实现了一个自动化闭环AF将攻击日志实时推给ZabbixZabbix触发告警并调用一个Python脚本脚本解析出攻击源IP后通过AF的API自动将其加入黑名单整个过程从攻击发生到IP被封锁可以控制在30分钟以内告警周期可调。这套方案特别适合应对那些持续性的端口扫描、暴力破解和Web常见漏洞攻击把安全运维人员从重复的“救火”工作中解放出来专注于更复杂的威胁分析。整个方案的核心就三块AF的日志外发配置、Zabbix的监控项与告警联动以及一个“承上启下”的Python脚本。下面我就把详细的配置步骤、脚本原理以及我踩过的几个坑毫无保留地分享出来。2. 整体架构与联动原理拆解在动手配置之前我们先要把整个数据流和自动化逻辑理清楚。这就像盖房子先画图纸理解了原理后面配置时遇到问题才知道往哪个方向排查。2.1 核心数据流向与组件角色整个联动体系涉及三个核心组件它们各自扮演着明确的角色深信服AF防火墙 (攻击检测与执行端)它是整个链条的起点和终点。AF内置了IPS、WAF、防暴力破解等安全模块当它检测到预设的高危攻击行为例如SQL注入、Struts2漏洞利用、SSH暴力破解成功时会生成一条详细的安全日志。我们的目标就是让AF把这些特定的日志消息发送出去。同时AF也提供了丰富的RESTful API允许我们通过编程方式管理其策略包括添加黑名单。Zabbix监控服务器 (日志汇聚与告警中枢)它扮演着“消息总线”和“触发器”的角色。我们会在Zabbix Server上开启一个Syslog接收服务用来接收AF发过来的日志。Zabbix通过“监控项”来捕获这些日志文本并利用“触发器”的正则表达式功能对日志内容进行匹配。一旦匹配到代表高危攻击的关键字触发器状态就会变为“PROBLEM”进而触发“动作”。“动作”里定义了要执行的操作——也就是调用我们的Python脚本。Python处理脚本 (逻辑解析与联动执行器)这是整个自动化的“大脑”。它由Zabbix在告警时调用。脚本需要完成几个关键任务从Zabbix传递的参数中解析出原始的日志信息使用正则表达式从日志中精准提取出攻击源IP地址然后构造合法的HTTP请求通过AF的API接口将这个IP添加到防火墙的“临时黑名单”或“地址组”中实现即时封禁。整个流程可以概括为AF检测攻击 - 发送Syslog - Zabbix接收并告警 - 调用Python脚本 - 脚本调用AF API封禁IP。这是一个标准的“检测-告警-响应”自动化闭环。2.2 为什么选择Syslog Zabbix API的方式你可能会问AF本身有没有自动封禁功能或者有没有更简单的方案这里说说我的选型考虑AF自身的自动封禁AF确实有“联动封堵”功能但通常阈值和策略比较固定灵活性不足且难以与外部威胁情报或其他监控系统如Zabbix监控的业务异常进行复杂联动。纯Syslog服务器分析如果只用一台Syslog服务器如rsyslog, syslog-ng收集日志然后写脚本分析也可以。但这意味着你要自己搭建一套日志存储、解析和告警的轮子在可靠性和可视化上不如Zabbix成熟。Zabbix的独特优势Zabbix本身就是一个强大的监控平台。用它来做有几个好处告警管理成熟告警升级、通知邮件、钉钉、企业微信、确认、历史记录等功能开箱即用。集中监控可以在同一个平台看到网络设备、服务器、应用以及现在的安全事件告警便于关联分析。灵活可控通过触发器表达式可以非常精细地定义何种日志产生告警。例如可以设定只在上班时间对某些攻击进行自动封禁避免误操作影响业务。易于调试所有接收到的日志、触发的告警都可以在Zabbix前端直观查看调试阶段非常方便。所以这套组合拳在灵活性、功能性和复用现有监控体系之间取得了很好的平衡。3. 深信服AF防火墙侧关键配置要让AF把日志吐出来需要完成两步第一配置日志服务器地址第二定义哪些日志需要发送。AF的界面逻辑比较清晰跟着做就行。3.1 配置Syslog日志外发服务器这里的目标是告诉AF“请把你生成的安全日志发送到Zabbix服务器假设IP是192.168.1.100的514端口。”登录AF防火墙的管理界面。进入【系统】-【日志设置】-【日志发送】-【新增】。在弹出窗口中填写以下关键信息日志类型选择“安全日志”。这是最关键的一步攻击事件日志都在这个类别里。流量日志、管理日志等不需要发送。发送方式选择“SYSLOG”。服务器地址填写你的Zabbix服务器的IP地址例如192.168.1.100。端口默认是514。确保与Zabbix Server上配置的Syslog监听端口一致。协议选择“UDP”。在局域网内UDP性能更好且足够可靠。如果对可靠性要求极高可以考虑TCP但需要在Zabbix侧也做相应配置。格式选择“设备事件格式”或“标准格式”。建议先选“标准格式”它的可读性更好字段排列规律便于后续用正则表达式解析。AF的“设备事件格式”是JSON虽然结构化好但可能需要调整Zabbix的监控项类型。其他选项如“设备编码”、“日志级别”保持默认即可。然后点击【确定】保存。注意配置完成后最好在AF上模拟一次攻击例如从外网扫描一下AF的WAN口然后立即在【系统】-【日志设置】-【日志发送】-【发送测试】中使用相同的配置发送一条测试日志以确认网络可达、端口开放。3.2 精准筛选需要外发的高危攻击日志AF的安全日志非常多全量发送会给Zabbix和网络带来不必要的压力。我们需要聚焦在那些真正需要自动化响应的高危事件上。在AF管理界面进入【安全日志】查看页面。你会看到各种攻击事件如“HTTP攻击”、“暴力破解”、“入侵攻击”等。花点时间浏览一下确定你想要自动封锁的攻击类型。常见的候选包括入侵攻击特别是那些已利用成功的漏洞攻击如Apache Log4j2、Spring Framework RCE等。暴力破解SSH、RDP、FTP等协议的暴力破解成功事件。注意是“成功”而不是“尝试”。封锁尝试IP可能会误伤但封锁成功IP则非常必要。WEB攻击严重的SQL注入、跨站脚本XSS、命令注入等。记住这些事件的特征比如日志中会包含的关键字如“入侵攻击”、“暴力破解成功”。这些关键字将用于后续在Zabbix中配置触发器。实际上AF在发送日志时如果你在【日志发送】配置中只选了“安全日志”它默认就会发送所有安全事件。我们无法在AF侧做更细粒度的过滤。过滤的动作我们后移到Zabbix的触发器里去做这样更灵活。比如今天你想封禁SQL注入明天想增加对某新型漏洞的封禁只需要修改Zabbix的触发器表达式无需来回调整AF的配置。4. Zabbix服务器侧配置详解Zabbix是联动的中枢配置稍显复杂但一步步来很清楚。主要分为开启Syslog接收、创建监控项捕获日志、设置触发器识别攻击、定义动作调用脚本。4.1 启用并配置Zabbix Server的Syslog监听默认情况下Zabbix Server可能没有开启Syslog监听功能。检查Zabbix Server配置文件登录Zabbix Server所在Linux服务器编辑其主配置文件zabbix_server.conf。vim /etc/zabbix/zabbix_server.conf寻找并修改参数找到以下两个参数确保它们如下设置ListenPort10051 # 这是Zabbix Trapper端口保持默认 # 添加或取消注释以下行来启用Syslog监听 ListenPort10514 # 这是一个自定义端口用于监听Syslog。避免使用标准的514端口防止与系统rsyslog服务冲突。实际上更常见的做法是让系统自带的rsyslog或syslog-ng接收日志然后转发给Zabbix。但对于轻量级或希望简化的部署直接让Zabbix Server监听一个端口也是可行的。更推荐的方法是使用rsyslog中转因为更稳定、功能更强如日志文件落盘。使用rsyslog中转的标准做法推荐编辑/etc/rsyslog.conf 启用UDP/TCP监听取消注释# 提供UDP接收 module(loadimudp) input(typeimudp port514) # 提供TCP接收 module(loadimtcp) input(typeimtcp port514)在文件末尾添加规则将来自AF的日志转发到Zabbix Server的trapper端口10051并指定主机和监控项键值# 假设AF的IP是192.168.1.1 Zabbix Server IP是127.0.0.1 if $fromhost-ip 192.168.1.1 then { action(typeomfwd Target127.0.0.1 Port10051 Protocoltcp TemplateRSYSLOG_TraditionalFileFormat) stop }重启rsyslog服务systemctl restart rsyslog。这样AF的日志先到系统514端口再由rsyslog转发给Zabbix Server进程处理。在Zabbix Web界面创建监控项时类型选择“Zabbix trapper”即可。4.2 创建主机与监控项捕获AF日志在Zabbix Web界面上操作。创建主机进入【配置】-【主机】-【创建主机】。主机名称填写一个易于识别的名字如Sangfor_AF。可见的主机名称同上。群组放入一个合适的组如“Firewalls”或“Security Devices”。Agent接口的IP/DNS这里不是填AF的IP而是填Zabbix Server能接收到日志的IP。如果你用rsyslog中转就填Zabbix Server本机的IP如127.0.0.1。关键点这个IP是用来关联监控项的不是设备的管理IP。创建监控项在刚创建的主机上点击“监控项”-“创建监控项”。名称AF Security Logs。类型如果你让Zabbix Server直接监听端口如10514则选择“Zabbix采集器主动式”或“简单检查”可能不适用。正确做法是选择“Zabbix trapper”。因为无论直接监听还是通过rsyslog转发日志都是以“trapper”方式主动发送到Zabbix的。键值输入一个自定义的键名例如af.security.log。这个键值很重要后续在rsyslog转发规则或测试发送日志时要用到。类型选择“文本”。其他更新间隔等信息对trapper类型无效保持默认即可。4.3 配置触发器识别高危攻击事件触发器是定义“什么情况算告警”的核心。在Sangfor_AF主机的“触发器”标签页点击“创建触发器”。名称一个清晰的描述如AF检测到高危入侵攻击。严重性选择“灾难”或“严重”因为这是触发自动封禁的条件。表达式这是最关键的部分。点击“添加”构建表达式。监控项选择刚才创建的AF Security Logs。功能选择regexp(正则表达式匹配) 或find(查找子串)。对于简单关键字find更高效。N秒内最新的值这里填0。表示检查最新收到的一条日志。模式输入你要匹配的关键字。例如要匹配入侵攻击AF日志里可能有“事件类型: 入侵攻击”这样的字段。那么模式可以写成入侵攻击。为了更精确可以匹配“事件子类型: Apache Log4j2”等。一个简单的表达式示例{Sangfor_AF:af.security.log.regexp(入侵攻击)}1。这个表达式的意思是如果主机Sangfor_AF的监控项af.security.log的最新值中正则匹配到了“入侵攻击”这个文本则触发告警。可以创建多个触发器针对不同类型的攻击创建不同的触发器。例如触发器1表达式匹配入侵攻击。触发器2表达式匹配暴力破解成功。触发器3表达式匹配SQL注入。 这样在告警和动作中可以区分处理也便于统计。4.4 定义告警动作调用Python脚本动作定义了“告警时做什么”。我们将在这里调用Python脚本。进入【配置】-【动作】。事件源选择“触发器”。创建动作点击“创建动作”。名称自动封锁AF攻击IP。条件这里可以细化什么情况下执行动作。例如触发器AF检测到高危入侵攻击(选择你创建的特定触发器)并且维护状态非在维护期内并且时间在工作日 09:00-18:00(避免非工作时间误操作) 条件可以让你更精准地控制自动化范围。操作这是核心配置页签。默认操作步骤持续时间例如1m。默认信息这里可以自定义告警消息。关键点必须包含{ITEM.VALUE}这个宏这个宏包含了触发告警的那条原始日志信息我们的Python脚本需要它来解析IP。 示例消息高危攻击告警防火墙AF检测到攻击事件。 原始日志{ITEM.VALUE} 触发条件{TRIGGER.NAME} 正在执行自动封禁脚本...操作细节点击“新的”来添加操作。操作类型选择“远程命令”。目标列表选择“Zabbix服务器”或运行脚本的特定主机。脚本必须放在Zabbix Server或你指定的Proxy服务器上。类型选择“自定义脚本”。执行在选择“Zabbix服务器”。命令这里填写调用Python脚本的命令并传递参数。例如python3 /usr/lib/zabbix/alertscripts/block_ip.py {TRIGGER.NAME} {ITEM.VALUE} {HOST.IP}block_ip.py你的Python脚本路径。{TRIGGER.NAME}传递触发器名称便于脚本区分攻击类型。{ITEM.VALUE}最重要的参数传递原始日志文本。{HOST.IP}传递主机的IP这里指AF的IP注意这里的主机IP是Zabbix主机配置里的IP可能不是AF的管理IP。更可靠的做法是将AF的管理IP、API账号等信息直接写在脚本的配置部分或者通过宏传递。 更稳妥的命令是python3 /usr/lib/zabbix/alertscripts/block_ip.py {ITEM.VALUE}其他所需信息如AF地址、API密钥在脚本内部配置文件中写死。恢复操作可以配置当触发器恢复正常时即一段时间内没有新攻击日志执行什么操作例如调用另一个脚本将IP从黑名单中释放需谨慎建议设置较长的自动封锁时间如24小时到期自动删除。5. Python联动脚本开发与解析这个脚本是自动化的执行引擎它需要完成日志解析和API调用两个核心任务。下面我提供一个功能完整、带有错误处理和日志记录的脚本示例并逐段解释。5.1 脚本核心功能与依赖库脚本block_ip.py需要安装requests库。确保在Zabbix Server上安装pip3 install requests。脚本主要逻辑接收参数从Zabbix动作的命令行参数中获取原始日志字符串。解析IP使用正则表达式从日志中提取攻击源IP地址。这是最容易出错的一步因为日志格式可能变化。调用AF API使用AF提供的REST API将提取到的IP地址添加到防火墙的黑名单规则中。记录与反馈将操作结果成功或失败记录到本地日志文件方便排查问题。5.2 完整Python脚本示例与注释将以下脚本保存到Zabbix Server的告警脚本目录通常是/usr/lib/zabbix/alertscripts/并赋予执行权限 (chmod x block_ip.py)。#!/usr/bin/env python3 # -*- coding: utf-8 -*- 深信服AF防火墙自动封禁IP脚本 调用方式python3 block_ip.py 原始日志文本 import sys import re import json import logging from datetime import datetime, timedelta import requests from requests.packages.urllib3.exceptions import InsecureRequestWarning # 禁用SSL警告如果AF使用自签名证书 requests.packages.urllib3.disable_warnings(InsecureRequestWarning) # 配置区域 - 根据你的实际环境修改 AF_MGMT_IP 192.168.1.1 # AF防火墙管理IP AF_PORT 443 # AF管理端口通常是443 AF_USERNAME api_user # API账号建议在AF上创建一个仅用于API操作的只读账号 AF_PASSWORD YourStrongPassword # API账号密码 BLOCK_DURATION_HOURS 24 # 封禁时长小时 BLOCK_ADDRESS_GROUP 临时黑名单 # AF上已有的地址组名称用于存放封禁IP # 日志配置 LOG_FILE /var/log/zabbix/af_auto_block.log logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stdout) # 同时输出到控制台方便Zabbix查看 ] ) logger logging.getLogger(__name__) def extract_ip_from_log(log_text): 从AF安全日志中提取源IP地址。 这是最需要根据你的实际日志格式调整的函数。 示例日志片段 “日期2023-10-27 时间15:30:25 事件类型入侵攻击 事件子类型Apache Log4j2 源IP202.100.1.100 目的IP10.0.0.100 ...” # 方法1使用正则匹配常见的IP字段模式 # 假设日志中有“源IPxxx.xxx.xxx.xxx”这样的格式 ip_patterns [ r源IP[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“源IP” r源地址[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“源地址” rsrc[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“src” rfrom\s(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“from x.x.x.x” ] for pattern in ip_patterns: match re.search(pattern, log_text) if match: ip match.group(1) # 简单的IP格式验证 if re.match(r^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$, ip): parts ip.split(.) if all(0 int(part) 255 for part in parts): logger.info(f从日志中提取到IP: {ip}) return ip # 如果上述模式都没匹配到尝试更通用的IP匹配可能误匹配目的IP # 这是一个备选方案不推荐首选使用 generic_ip_match re.findall(r\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b, log_text) if generic_ip_match: logger.warning(f使用通用正则匹配到多个IP: {generic_ip_match}请检查日志格式。默认取第一个。) # 这里需要更复杂的逻辑来判断哪个是源IP例如结合“目的IP”字段的位置 # 简单返回第一个风险较高 return generic_ip_match[0] logger.error(f无法从日志中提取有效的源IP。日志内容: {log_text[:500]}...) # 只打印前500字符 return None def af_api_login(ip, port, username, password): 登录AF获取认证Cookie login_url fhttps://{ip}:{port}/api/v1/auth login_data { username: username, password: password, # client_id: API, # 某些版本可能需要 # client_secret: API # 某些版本可能需要 } try: # AF 8.0 API可能需要特定的Content-Type headers {Content-Type: application/json} resp requests.post(login_url, jsonlogin_data, headersheaders, verifyFalse, timeout10) resp.raise_for_status() # 检查HTTP错误 result resp.json() if result.get(success) True: # AF API通常返回一个cookie或token后续请求需要携带 # 例如cookie可能在resp.cookies里或者返回的json里有token auth_token result.get(data, {}).get(token) if auth_token: logger.info(AF API登录成功) return {Authorization: fBearer {auth_token}} # 假设是Bearer Token else: # 也可能是通过Cookie认证 cookie_jar resp.cookies logger.info(AF API登录成功 (Cookie认证)) return cookie_jar else: logger.error(fAF API登录失败: {result.get(msg, Unknown error)}) return None except requests.exceptions.RequestException as e: logger.error(f登录AF API时发生网络错误: {e}) return None except json.JSONDecodeError as e: logger.error(f解析AF登录响应JSON失败: {e}) return None def af_api_add_address_object(ip, auth_info, address_group_name): 在AF上创建一个地址对象并将其添加到指定的地址组。 步骤 1. 创建地址对象如果不存在。 2. 将地址对象添加到地址组。 base_url fhttps://{AF_MGMT_IP}:{AF_PORT}/api/v1 headers {Content-Type: application/json} if isinstance(auth_info, dict) and Authorization in auth_info: headers.update(auth_info) auth_cookies None else: auth_cookies auth_info auth_headers {} # 1. 创建地址对象 obj_name fAutoBlock_{ip.replace(., _)}_{datetime.now().strftime(%Y%m%d_%H%M)} create_obj_url f{base_url}/address/address_object obj_data { name: obj_name, ip_netmask: f{ip}/32, comment: f自动封禁 - 来自Zabbix联动 - {datetime.now()} } try: resp requests.post(create_obj_url, jsonobj_data, headersheaders, cookiesauth_cookies, verifyFalse, timeout15) resp.raise_for_status() create_result resp.json() if create_result.get(success) ! True: # 可能地址对象已存在尝试查询 logger.warning(f创建地址对象可能失败或已存在: {create_result.get(msg)}。尝试直接添加到地址组。) # 这里简化处理假设对象名就是基于IP生成的直接进行下一步 pass except Exception as e: logger.error(f创建地址对象时出错: {e}) # 继续尝试添加也许对象已存在 # 2. 将地址对象添加到地址组 # 首先需要获取地址组的ID。这里假设你知道地址组名称需要先查询其ID。 # 为简化这里演示直接使用“地址组名称”进行添加的API请根据AF实际API文档调整 # 以下为示例逻辑AF API可能不同 # add_to_group_url f{base_url}/address/address_group_member # group_data { # group_name: address_group_name, # member_names: [obj_name] # } # 更常见的做法AF的“黑名单”可能是一个“策略”或“安全规则”。 # 实际上更直接的方式是调用“封堵IP”的特定API。 # 根据AF 8.0文档封堵IP的API可能类似于 block_url f{base_url}/block/ip block_data { ip: ip, duration: BLOCK_DURATION_HOURS * 3600, # 转换为秒 reason: Zabbix自动封禁-高危攻击, direction: both # 双向封禁 } try: resp requests.post(block_url, jsonblock_data, headersheaders, cookiesauth_cookies, verifyFalse, timeout15) resp.raise_for_status() block_result resp.json() if block_result.get(success) True: logger.info(f成功封禁IP: {ip}, 时长: {BLOCK_DURATION_HOURS}小时) return True else: logger.error(f封禁IP失败API返回错误: {block_result}) return False except Exception as e: logger.error(f调用封禁API时出错: {e}) return False def main(): if len(sys.argv) 2: logger.error(脚本需要至少1个参数原始日志文本。) print(Usage: python3 block_ip.py \log text\) sys.exit(1) raw_log sys.argv[1] logger.info(f接收到告警日志开始处理。日志片段: {raw_log[:200]}...) # 步骤1: 提取IP attack_ip extract_ip_from_log(raw_log) if not attack_ip: logger.error(无法提取攻击IP流程终止。) sys.exit(2) # 步骤2: 登录AF获取认证信息 auth_info af_api_login(AF_MGMT_IP, AF_PORT, AF_USERNAME, AF_PASSWORD) if not auth_info: logger.error(AF API登录失败无法继续。) sys.exit(3) # 步骤3: 调用API封禁IP success af_api_add_address_object(attack_ip, auth_info, BLOCK_ADDRESS_GROUP) if success: logger.info(f自动化封禁流程执行完毕。IP: {attack_ip} 已被加入封锁列表。) sys.exit(0) # 成功退出码 else: logger.error(f自动化封禁流程执行失败。IP: {attack_ip}) sys.exit(4) # 失败退出码 if __name__ __main__: main()5.3 脚本关键点解析与自定义调整日志解析函数extract_ip_from_log这是脚本最脆弱的部分。你必须根据你AF实际发出的Syslog日志格式来调整正则表达式。最佳调试方法先在Zabbix的“最新数据”里找到一条真实的告警日志复制出来然后在Python交互环境里测试你的正则表达式是否能准确提取源IP。注意区分源IPsrc和目的IPdst。AF API调用脚本中的af_api_add_address_object函数是一个示例。深信服AF不同大版本的API可能有较大差异。你需要查阅你所使用的AF 8.0具体版本如8.0.5, 8.0.10的官方REST API文档。关键是要找到“添加黑名单”或“封堵IP”对应的API端点Endpoint、请求方法POST/GET、请求体和认证方式。脚本中使用了简单的用户名密码登录有些版本可能需要使用API Key。错误处理与日志脚本中加入了详细的logging记录所有操作和错误都会写入/var/log/zabbix/af_auto_block.log文件。务必定期检查这个日志这是排查联动失败问题的最重要依据。安全警告在脚本中我们verifyFalse跳过了SSL证书验证这是因为很多内网设备的证书是自签名的。在生产环境中如果条件允许建议为AF配置有效的证书并启用验证以提高安全性。6. 联动测试与全流程验证配置完成后必须进行端到端的测试确保链条每个环节都畅通。6.1 模拟攻击生成测试日志最直接的测试方法就是模拟一次真实的攻击。从互联网或测试机对AF保护的一台服务器发起一次明显的攻击扫描。例如使用nmap对服务器的SSH端口进行扫描或者使用sqlmap对一个测试页面进行注入确保在授权和隔离的环境中进行。观察AF的“安全日志”界面确认产生了对应的攻击日志如“入侵攻击”或“暴力破解”。等待几十秒到一分钟取决于Zabbix监控项的更新间隔和触发器检查周期。6.2 在Zabbix中检查日志接收与告警登录Zabbix Web进入【监控】-【最新数据】选择你创建的Sangfor_AF主机查看af.security.log这个监控项。你应该能看到最新的值就是AF发过来的那条攻击日志的全文。进入【监控】-【问题】界面你应该能看到一条新的问题触发器名称就是你设置的“AF检测到高危入侵攻击”状态为“PROBLEM”。点击该问题查看“历史”和“动作日志”确认触发的动作“自动封锁AF攻击IP”已经执行并且远程命令的状态如果Zabbix有记录的话。6.3 验证脚本执行与AF封锁结果查看脚本日志立刻登录Zabbix Server执行tail -f /var/log/zabbix/af_auto_block.log。在触发告警后你应该能看到脚本被调用的日志记录包括“接收到告警日志”、“提取到IP: x.x.x.x”、“AF API登录成功”、“成功封禁IP”等信息。验证AF黑名单登录AF防火墙管理界面进入【策略】-【安全策略】-【黑名单】或类似路径不同版本可能叫“封堵列表”、“地址组”等。你应该能看到一个新的条目IP地址就是攻击源的IP备注信息里可能有“Zabbix自动封禁”等字样并且生效时间符合你设置的BLOCK_DURATION_HOURS。验证封锁效果尝试从被封锁的IP再次访问AF保护的服务应该会被直接拒绝或超时。6.4 常见测试失败场景与排查如果测试失败按照以下顺序排查AF日志未发出检查AF上的“日志发送”配置IP和端口是否正确。在Zabbix Server上用tcpdump -i any port 514 -n或nc -luk 514命令监听看是否能收到AF发来的UDP报文。Zabbix未收到日志检查Zabbix Server上rsyslog配置是否正确服务是否重启。检查Zabbix主机配置中“Agent接口”的IP是否与rsyslog转发目标IP一致。检查Zabbix监控项的“键值”是否与rsyslog转发规则中指定的键值一致。触发器未触发检查触发器的表达式是否正确。在“最新数据”里复制一条日志去测试你的正则表达式是否匹配。检查触发器的“严重性”是否足够高以及动作的“条件”是否满足如时间条件。动作未执行或脚本执行失败在Zabbix动作的“操作”里可以尝试在“命令”中写一个简单的测试命令如echo test /tmp/zabbix_test.log看文件是否生成以验证远程命令执行功能是否正常。手动在Zabbix Server上切换到zabbix用户su - zabbix然后执行你的Python脚本并传入参数看是否有错误输出。Zabbix默认以zabbix用户身份执行脚本权限和环境变量可能受限这是最常见的坑。检查脚本本身的权限 (chmod 755)以及Python路径和依赖库 (requests) 是否对zabbix用户可用。仔细查看/var/log/zabbix/af_auto_block.log日志里面的错误信息是定位问题的关键。7. 生产环境部署的注意事项与优化建议当测试通过准备上生产环境时以下几点需要特别注意权限最小化原则在AF上专门创建一个用于API操作的账号权限只赋予“策略管理”或“黑名单管理”相关的最小权限集绝不能使用管理员账号。将Python脚本中的AF密码等敏感信息从明文改为从加密的配置文件或环境变量中读取。可以考虑使用ansible-vault加密或专门的密钥管理服务。防止误封与误操作设置IP白名单在脚本解析IP后、调用API前加入一个白名单检查。将公司出口IP、合作伙伴IP、云服务IP等加入白名单列表避免误封。设置内网IP过滤通常不需要封禁内网IP如10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16。在脚本中判断如果是内网IP则记录日志并跳过封禁。谨慎设置触发器初期只对确凿无疑的高危事件如“暴力破解成功”、“已知严重漏洞利用”进行自动封禁。对于“端口扫描”、“WEB攻击尝试”等可以仅告警不自动封禁。设置冷却时间在Zabbix动作中可以设置“默认操作步骤持续时间”和“问题更新间隔”避免在短时间内对同一IP的重复攻击产生大量重复的封禁操作给AF API造成压力。脚本健壮性增强加入重试机制对于AF API调用可以加入简单的重试逻辑如最多重试3次以应对网络瞬时波动。加入锁机制如果脚本可能被并发执行多个告警同时触发应考虑使用文件锁fcntl或数据库锁防止对同一个IP的重复封禁操作冲突。更精细的日志格式处理可以编写一个日志解析类针对不同的事件类型从触发器名称或日志内容判断采用不同的正则表达式提取IP提高准确率。监控与维护监控脚本执行除了脚本自身的日志可以在Zabbix中创建一个监控项监控/var/log/zabbix/af_auto_block.log文件的大小或最后修改时间如果长时间未更新可能意味着联动失效。定期审计封禁列表每周或每月检查一次AF上的自动封禁列表确认封禁的IP是否合理及时清理过期的或误封的条目。备份配置将AF的API配置、Zabbix的触发器表达式、动作命令以及Python脚本本身进行版本管理如Git。这套联动方案实施后我们团队处理高频、低复杂度攻击事件的效率得到了质的提升。它就像一位不知疲倦的初级安全分析师7x24小时值守将我们从中重复性的工作中解脱出来。当然自动化不是万能的它依赖于准确的检测AF的规则和精确的解析脚本的正则在部署初期需要一段时间的观察和调优。一旦稳定运行它将成为你安全防御体系中一个非常可靠且高效的自动化响应节点。
1. 项目概述与核心价值最近在梳理我们内部安全运营流程时发现一个痛点安全设备告警很多但真正需要人工介入去封禁IP的往往就那么几类高危攻击。每次登录防火墙、查日志、确认、再手动添加黑名单一套流程下来攻击可能已经持续了十几分钟。对于深信服AF这类下一代防火墙其日志里其实已经包含了非常精准的攻击判定信息如果能自动处理效率会高得多。于是我花了点时间把Zabbix监控和深信服AF 8.0的防火墙联动了起来实现了一个自动化闭环AF将攻击日志实时推给ZabbixZabbix触发告警并调用一个Python脚本脚本解析出攻击源IP后通过AF的API自动将其加入黑名单整个过程从攻击发生到IP被封锁可以控制在30分钟以内告警周期可调。这套方案特别适合应对那些持续性的端口扫描、暴力破解和Web常见漏洞攻击把安全运维人员从重复的“救火”工作中解放出来专注于更复杂的威胁分析。整个方案的核心就三块AF的日志外发配置、Zabbix的监控项与告警联动以及一个“承上启下”的Python脚本。下面我就把详细的配置步骤、脚本原理以及我踩过的几个坑毫无保留地分享出来。2. 整体架构与联动原理拆解在动手配置之前我们先要把整个数据流和自动化逻辑理清楚。这就像盖房子先画图纸理解了原理后面配置时遇到问题才知道往哪个方向排查。2.1 核心数据流向与组件角色整个联动体系涉及三个核心组件它们各自扮演着明确的角色深信服AF防火墙 (攻击检测与执行端)它是整个链条的起点和终点。AF内置了IPS、WAF、防暴力破解等安全模块当它检测到预设的高危攻击行为例如SQL注入、Struts2漏洞利用、SSH暴力破解成功时会生成一条详细的安全日志。我们的目标就是让AF把这些特定的日志消息发送出去。同时AF也提供了丰富的RESTful API允许我们通过编程方式管理其策略包括添加黑名单。Zabbix监控服务器 (日志汇聚与告警中枢)它扮演着“消息总线”和“触发器”的角色。我们会在Zabbix Server上开启一个Syslog接收服务用来接收AF发过来的日志。Zabbix通过“监控项”来捕获这些日志文本并利用“触发器”的正则表达式功能对日志内容进行匹配。一旦匹配到代表高危攻击的关键字触发器状态就会变为“PROBLEM”进而触发“动作”。“动作”里定义了要执行的操作——也就是调用我们的Python脚本。Python处理脚本 (逻辑解析与联动执行器)这是整个自动化的“大脑”。它由Zabbix在告警时调用。脚本需要完成几个关键任务从Zabbix传递的参数中解析出原始的日志信息使用正则表达式从日志中精准提取出攻击源IP地址然后构造合法的HTTP请求通过AF的API接口将这个IP添加到防火墙的“临时黑名单”或“地址组”中实现即时封禁。整个流程可以概括为AF检测攻击 - 发送Syslog - Zabbix接收并告警 - 调用Python脚本 - 脚本调用AF API封禁IP。这是一个标准的“检测-告警-响应”自动化闭环。2.2 为什么选择Syslog Zabbix API的方式你可能会问AF本身有没有自动封禁功能或者有没有更简单的方案这里说说我的选型考虑AF自身的自动封禁AF确实有“联动封堵”功能但通常阈值和策略比较固定灵活性不足且难以与外部威胁情报或其他监控系统如Zabbix监控的业务异常进行复杂联动。纯Syslog服务器分析如果只用一台Syslog服务器如rsyslog, syslog-ng收集日志然后写脚本分析也可以。但这意味着你要自己搭建一套日志存储、解析和告警的轮子在可靠性和可视化上不如Zabbix成熟。Zabbix的独特优势Zabbix本身就是一个强大的监控平台。用它来做有几个好处告警管理成熟告警升级、通知邮件、钉钉、企业微信、确认、历史记录等功能开箱即用。集中监控可以在同一个平台看到网络设备、服务器、应用以及现在的安全事件告警便于关联分析。灵活可控通过触发器表达式可以非常精细地定义何种日志产生告警。例如可以设定只在上班时间对某些攻击进行自动封禁避免误操作影响业务。易于调试所有接收到的日志、触发的告警都可以在Zabbix前端直观查看调试阶段非常方便。所以这套组合拳在灵活性、功能性和复用现有监控体系之间取得了很好的平衡。3. 深信服AF防火墙侧关键配置要让AF把日志吐出来需要完成两步第一配置日志服务器地址第二定义哪些日志需要发送。AF的界面逻辑比较清晰跟着做就行。3.1 配置Syslog日志外发服务器这里的目标是告诉AF“请把你生成的安全日志发送到Zabbix服务器假设IP是192.168.1.100的514端口。”登录AF防火墙的管理界面。进入【系统】-【日志设置】-【日志发送】-【新增】。在弹出窗口中填写以下关键信息日志类型选择“安全日志”。这是最关键的一步攻击事件日志都在这个类别里。流量日志、管理日志等不需要发送。发送方式选择“SYSLOG”。服务器地址填写你的Zabbix服务器的IP地址例如192.168.1.100。端口默认是514。确保与Zabbix Server上配置的Syslog监听端口一致。协议选择“UDP”。在局域网内UDP性能更好且足够可靠。如果对可靠性要求极高可以考虑TCP但需要在Zabbix侧也做相应配置。格式选择“设备事件格式”或“标准格式”。建议先选“标准格式”它的可读性更好字段排列规律便于后续用正则表达式解析。AF的“设备事件格式”是JSON虽然结构化好但可能需要调整Zabbix的监控项类型。其他选项如“设备编码”、“日志级别”保持默认即可。然后点击【确定】保存。注意配置完成后最好在AF上模拟一次攻击例如从外网扫描一下AF的WAN口然后立即在【系统】-【日志设置】-【日志发送】-【发送测试】中使用相同的配置发送一条测试日志以确认网络可达、端口开放。3.2 精准筛选需要外发的高危攻击日志AF的安全日志非常多全量发送会给Zabbix和网络带来不必要的压力。我们需要聚焦在那些真正需要自动化响应的高危事件上。在AF管理界面进入【安全日志】查看页面。你会看到各种攻击事件如“HTTP攻击”、“暴力破解”、“入侵攻击”等。花点时间浏览一下确定你想要自动封锁的攻击类型。常见的候选包括入侵攻击特别是那些已利用成功的漏洞攻击如Apache Log4j2、Spring Framework RCE等。暴力破解SSH、RDP、FTP等协议的暴力破解成功事件。注意是“成功”而不是“尝试”。封锁尝试IP可能会误伤但封锁成功IP则非常必要。WEB攻击严重的SQL注入、跨站脚本XSS、命令注入等。记住这些事件的特征比如日志中会包含的关键字如“入侵攻击”、“暴力破解成功”。这些关键字将用于后续在Zabbix中配置触发器。实际上AF在发送日志时如果你在【日志发送】配置中只选了“安全日志”它默认就会发送所有安全事件。我们无法在AF侧做更细粒度的过滤。过滤的动作我们后移到Zabbix的触发器里去做这样更灵活。比如今天你想封禁SQL注入明天想增加对某新型漏洞的封禁只需要修改Zabbix的触发器表达式无需来回调整AF的配置。4. Zabbix服务器侧配置详解Zabbix是联动的中枢配置稍显复杂但一步步来很清楚。主要分为开启Syslog接收、创建监控项捕获日志、设置触发器识别攻击、定义动作调用脚本。4.1 启用并配置Zabbix Server的Syslog监听默认情况下Zabbix Server可能没有开启Syslog监听功能。检查Zabbix Server配置文件登录Zabbix Server所在Linux服务器编辑其主配置文件zabbix_server.conf。vim /etc/zabbix/zabbix_server.conf寻找并修改参数找到以下两个参数确保它们如下设置ListenPort10051 # 这是Zabbix Trapper端口保持默认 # 添加或取消注释以下行来启用Syslog监听 ListenPort10514 # 这是一个自定义端口用于监听Syslog。避免使用标准的514端口防止与系统rsyslog服务冲突。实际上更常见的做法是让系统自带的rsyslog或syslog-ng接收日志然后转发给Zabbix。但对于轻量级或希望简化的部署直接让Zabbix Server监听一个端口也是可行的。更推荐的方法是使用rsyslog中转因为更稳定、功能更强如日志文件落盘。使用rsyslog中转的标准做法推荐编辑/etc/rsyslog.conf 启用UDP/TCP监听取消注释# 提供UDP接收 module(loadimudp) input(typeimudp port514) # 提供TCP接收 module(loadimtcp) input(typeimtcp port514)在文件末尾添加规则将来自AF的日志转发到Zabbix Server的trapper端口10051并指定主机和监控项键值# 假设AF的IP是192.168.1.1 Zabbix Server IP是127.0.0.1 if $fromhost-ip 192.168.1.1 then { action(typeomfwd Target127.0.0.1 Port10051 Protocoltcp TemplateRSYSLOG_TraditionalFileFormat) stop }重启rsyslog服务systemctl restart rsyslog。这样AF的日志先到系统514端口再由rsyslog转发给Zabbix Server进程处理。在Zabbix Web界面创建监控项时类型选择“Zabbix trapper”即可。4.2 创建主机与监控项捕获AF日志在Zabbix Web界面上操作。创建主机进入【配置】-【主机】-【创建主机】。主机名称填写一个易于识别的名字如Sangfor_AF。可见的主机名称同上。群组放入一个合适的组如“Firewalls”或“Security Devices”。Agent接口的IP/DNS这里不是填AF的IP而是填Zabbix Server能接收到日志的IP。如果你用rsyslog中转就填Zabbix Server本机的IP如127.0.0.1。关键点这个IP是用来关联监控项的不是设备的管理IP。创建监控项在刚创建的主机上点击“监控项”-“创建监控项”。名称AF Security Logs。类型如果你让Zabbix Server直接监听端口如10514则选择“Zabbix采集器主动式”或“简单检查”可能不适用。正确做法是选择“Zabbix trapper”。因为无论直接监听还是通过rsyslog转发日志都是以“trapper”方式主动发送到Zabbix的。键值输入一个自定义的键名例如af.security.log。这个键值很重要后续在rsyslog转发规则或测试发送日志时要用到。类型选择“文本”。其他更新间隔等信息对trapper类型无效保持默认即可。4.3 配置触发器识别高危攻击事件触发器是定义“什么情况算告警”的核心。在Sangfor_AF主机的“触发器”标签页点击“创建触发器”。名称一个清晰的描述如AF检测到高危入侵攻击。严重性选择“灾难”或“严重”因为这是触发自动封禁的条件。表达式这是最关键的部分。点击“添加”构建表达式。监控项选择刚才创建的AF Security Logs。功能选择regexp(正则表达式匹配) 或find(查找子串)。对于简单关键字find更高效。N秒内最新的值这里填0。表示检查最新收到的一条日志。模式输入你要匹配的关键字。例如要匹配入侵攻击AF日志里可能有“事件类型: 入侵攻击”这样的字段。那么模式可以写成入侵攻击。为了更精确可以匹配“事件子类型: Apache Log4j2”等。一个简单的表达式示例{Sangfor_AF:af.security.log.regexp(入侵攻击)}1。这个表达式的意思是如果主机Sangfor_AF的监控项af.security.log的最新值中正则匹配到了“入侵攻击”这个文本则触发告警。可以创建多个触发器针对不同类型的攻击创建不同的触发器。例如触发器1表达式匹配入侵攻击。触发器2表达式匹配暴力破解成功。触发器3表达式匹配SQL注入。 这样在告警和动作中可以区分处理也便于统计。4.4 定义告警动作调用Python脚本动作定义了“告警时做什么”。我们将在这里调用Python脚本。进入【配置】-【动作】。事件源选择“触发器”。创建动作点击“创建动作”。名称自动封锁AF攻击IP。条件这里可以细化什么情况下执行动作。例如触发器AF检测到高危入侵攻击(选择你创建的特定触发器)并且维护状态非在维护期内并且时间在工作日 09:00-18:00(避免非工作时间误操作) 条件可以让你更精准地控制自动化范围。操作这是核心配置页签。默认操作步骤持续时间例如1m。默认信息这里可以自定义告警消息。关键点必须包含{ITEM.VALUE}这个宏这个宏包含了触发告警的那条原始日志信息我们的Python脚本需要它来解析IP。 示例消息高危攻击告警防火墙AF检测到攻击事件。 原始日志{ITEM.VALUE} 触发条件{TRIGGER.NAME} 正在执行自动封禁脚本...操作细节点击“新的”来添加操作。操作类型选择“远程命令”。目标列表选择“Zabbix服务器”或运行脚本的特定主机。脚本必须放在Zabbix Server或你指定的Proxy服务器上。类型选择“自定义脚本”。执行在选择“Zabbix服务器”。命令这里填写调用Python脚本的命令并传递参数。例如python3 /usr/lib/zabbix/alertscripts/block_ip.py {TRIGGER.NAME} {ITEM.VALUE} {HOST.IP}block_ip.py你的Python脚本路径。{TRIGGER.NAME}传递触发器名称便于脚本区分攻击类型。{ITEM.VALUE}最重要的参数传递原始日志文本。{HOST.IP}传递主机的IP这里指AF的IP注意这里的主机IP是Zabbix主机配置里的IP可能不是AF的管理IP。更可靠的做法是将AF的管理IP、API账号等信息直接写在脚本的配置部分或者通过宏传递。 更稳妥的命令是python3 /usr/lib/zabbix/alertscripts/block_ip.py {ITEM.VALUE}其他所需信息如AF地址、API密钥在脚本内部配置文件中写死。恢复操作可以配置当触发器恢复正常时即一段时间内没有新攻击日志执行什么操作例如调用另一个脚本将IP从黑名单中释放需谨慎建议设置较长的自动封锁时间如24小时到期自动删除。5. Python联动脚本开发与解析这个脚本是自动化的执行引擎它需要完成日志解析和API调用两个核心任务。下面我提供一个功能完整、带有错误处理和日志记录的脚本示例并逐段解释。5.1 脚本核心功能与依赖库脚本block_ip.py需要安装requests库。确保在Zabbix Server上安装pip3 install requests。脚本主要逻辑接收参数从Zabbix动作的命令行参数中获取原始日志字符串。解析IP使用正则表达式从日志中提取攻击源IP地址。这是最容易出错的一步因为日志格式可能变化。调用AF API使用AF提供的REST API将提取到的IP地址添加到防火墙的黑名单规则中。记录与反馈将操作结果成功或失败记录到本地日志文件方便排查问题。5.2 完整Python脚本示例与注释将以下脚本保存到Zabbix Server的告警脚本目录通常是/usr/lib/zabbix/alertscripts/并赋予执行权限 (chmod x block_ip.py)。#!/usr/bin/env python3 # -*- coding: utf-8 -*- 深信服AF防火墙自动封禁IP脚本 调用方式python3 block_ip.py 原始日志文本 import sys import re import json import logging from datetime import datetime, timedelta import requests from requests.packages.urllib3.exceptions import InsecureRequestWarning # 禁用SSL警告如果AF使用自签名证书 requests.packages.urllib3.disable_warnings(InsecureRequestWarning) # 配置区域 - 根据你的实际环境修改 AF_MGMT_IP 192.168.1.1 # AF防火墙管理IP AF_PORT 443 # AF管理端口通常是443 AF_USERNAME api_user # API账号建议在AF上创建一个仅用于API操作的只读账号 AF_PASSWORD YourStrongPassword # API账号密码 BLOCK_DURATION_HOURS 24 # 封禁时长小时 BLOCK_ADDRESS_GROUP 临时黑名单 # AF上已有的地址组名称用于存放封禁IP # 日志配置 LOG_FILE /var/log/zabbix/af_auto_block.log logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stdout) # 同时输出到控制台方便Zabbix查看 ] ) logger logging.getLogger(__name__) def extract_ip_from_log(log_text): 从AF安全日志中提取源IP地址。 这是最需要根据你的实际日志格式调整的函数。 示例日志片段 “日期2023-10-27 时间15:30:25 事件类型入侵攻击 事件子类型Apache Log4j2 源IP202.100.1.100 目的IP10.0.0.100 ...” # 方法1使用正则匹配常见的IP字段模式 # 假设日志中有“源IPxxx.xxx.xxx.xxx”这样的格式 ip_patterns [ r源IP[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“源IP” r源地址[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“源地址” rsrc[:]\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“src” rfrom\s(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}), # 匹配“from x.x.x.x” ] for pattern in ip_patterns: match re.search(pattern, log_text) if match: ip match.group(1) # 简单的IP格式验证 if re.match(r^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$, ip): parts ip.split(.) if all(0 int(part) 255 for part in parts): logger.info(f从日志中提取到IP: {ip}) return ip # 如果上述模式都没匹配到尝试更通用的IP匹配可能误匹配目的IP # 这是一个备选方案不推荐首选使用 generic_ip_match re.findall(r\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b, log_text) if generic_ip_match: logger.warning(f使用通用正则匹配到多个IP: {generic_ip_match}请检查日志格式。默认取第一个。) # 这里需要更复杂的逻辑来判断哪个是源IP例如结合“目的IP”字段的位置 # 简单返回第一个风险较高 return generic_ip_match[0] logger.error(f无法从日志中提取有效的源IP。日志内容: {log_text[:500]}...) # 只打印前500字符 return None def af_api_login(ip, port, username, password): 登录AF获取认证Cookie login_url fhttps://{ip}:{port}/api/v1/auth login_data { username: username, password: password, # client_id: API, # 某些版本可能需要 # client_secret: API # 某些版本可能需要 } try: # AF 8.0 API可能需要特定的Content-Type headers {Content-Type: application/json} resp requests.post(login_url, jsonlogin_data, headersheaders, verifyFalse, timeout10) resp.raise_for_status() # 检查HTTP错误 result resp.json() if result.get(success) True: # AF API通常返回一个cookie或token后续请求需要携带 # 例如cookie可能在resp.cookies里或者返回的json里有token auth_token result.get(data, {}).get(token) if auth_token: logger.info(AF API登录成功) return {Authorization: fBearer {auth_token}} # 假设是Bearer Token else: # 也可能是通过Cookie认证 cookie_jar resp.cookies logger.info(AF API登录成功 (Cookie认证)) return cookie_jar else: logger.error(fAF API登录失败: {result.get(msg, Unknown error)}) return None except requests.exceptions.RequestException as e: logger.error(f登录AF API时发生网络错误: {e}) return None except json.JSONDecodeError as e: logger.error(f解析AF登录响应JSON失败: {e}) return None def af_api_add_address_object(ip, auth_info, address_group_name): 在AF上创建一个地址对象并将其添加到指定的地址组。 步骤 1. 创建地址对象如果不存在。 2. 将地址对象添加到地址组。 base_url fhttps://{AF_MGMT_IP}:{AF_PORT}/api/v1 headers {Content-Type: application/json} if isinstance(auth_info, dict) and Authorization in auth_info: headers.update(auth_info) auth_cookies None else: auth_cookies auth_info auth_headers {} # 1. 创建地址对象 obj_name fAutoBlock_{ip.replace(., _)}_{datetime.now().strftime(%Y%m%d_%H%M)} create_obj_url f{base_url}/address/address_object obj_data { name: obj_name, ip_netmask: f{ip}/32, comment: f自动封禁 - 来自Zabbix联动 - {datetime.now()} } try: resp requests.post(create_obj_url, jsonobj_data, headersheaders, cookiesauth_cookies, verifyFalse, timeout15) resp.raise_for_status() create_result resp.json() if create_result.get(success) ! True: # 可能地址对象已存在尝试查询 logger.warning(f创建地址对象可能失败或已存在: {create_result.get(msg)}。尝试直接添加到地址组。) # 这里简化处理假设对象名就是基于IP生成的直接进行下一步 pass except Exception as e: logger.error(f创建地址对象时出错: {e}) # 继续尝试添加也许对象已存在 # 2. 将地址对象添加到地址组 # 首先需要获取地址组的ID。这里假设你知道地址组名称需要先查询其ID。 # 为简化这里演示直接使用“地址组名称”进行添加的API请根据AF实际API文档调整 # 以下为示例逻辑AF API可能不同 # add_to_group_url f{base_url}/address/address_group_member # group_data { # group_name: address_group_name, # member_names: [obj_name] # } # 更常见的做法AF的“黑名单”可能是一个“策略”或“安全规则”。 # 实际上更直接的方式是调用“封堵IP”的特定API。 # 根据AF 8.0文档封堵IP的API可能类似于 block_url f{base_url}/block/ip block_data { ip: ip, duration: BLOCK_DURATION_HOURS * 3600, # 转换为秒 reason: Zabbix自动封禁-高危攻击, direction: both # 双向封禁 } try: resp requests.post(block_url, jsonblock_data, headersheaders, cookiesauth_cookies, verifyFalse, timeout15) resp.raise_for_status() block_result resp.json() if block_result.get(success) True: logger.info(f成功封禁IP: {ip}, 时长: {BLOCK_DURATION_HOURS}小时) return True else: logger.error(f封禁IP失败API返回错误: {block_result}) return False except Exception as e: logger.error(f调用封禁API时出错: {e}) return False def main(): if len(sys.argv) 2: logger.error(脚本需要至少1个参数原始日志文本。) print(Usage: python3 block_ip.py \log text\) sys.exit(1) raw_log sys.argv[1] logger.info(f接收到告警日志开始处理。日志片段: {raw_log[:200]}...) # 步骤1: 提取IP attack_ip extract_ip_from_log(raw_log) if not attack_ip: logger.error(无法提取攻击IP流程终止。) sys.exit(2) # 步骤2: 登录AF获取认证信息 auth_info af_api_login(AF_MGMT_IP, AF_PORT, AF_USERNAME, AF_PASSWORD) if not auth_info: logger.error(AF API登录失败无法继续。) sys.exit(3) # 步骤3: 调用API封禁IP success af_api_add_address_object(attack_ip, auth_info, BLOCK_ADDRESS_GROUP) if success: logger.info(f自动化封禁流程执行完毕。IP: {attack_ip} 已被加入封锁列表。) sys.exit(0) # 成功退出码 else: logger.error(f自动化封禁流程执行失败。IP: {attack_ip}) sys.exit(4) # 失败退出码 if __name__ __main__: main()5.3 脚本关键点解析与自定义调整日志解析函数extract_ip_from_log这是脚本最脆弱的部分。你必须根据你AF实际发出的Syslog日志格式来调整正则表达式。最佳调试方法先在Zabbix的“最新数据”里找到一条真实的告警日志复制出来然后在Python交互环境里测试你的正则表达式是否能准确提取源IP。注意区分源IPsrc和目的IPdst。AF API调用脚本中的af_api_add_address_object函数是一个示例。深信服AF不同大版本的API可能有较大差异。你需要查阅你所使用的AF 8.0具体版本如8.0.5, 8.0.10的官方REST API文档。关键是要找到“添加黑名单”或“封堵IP”对应的API端点Endpoint、请求方法POST/GET、请求体和认证方式。脚本中使用了简单的用户名密码登录有些版本可能需要使用API Key。错误处理与日志脚本中加入了详细的logging记录所有操作和错误都会写入/var/log/zabbix/af_auto_block.log文件。务必定期检查这个日志这是排查联动失败问题的最重要依据。安全警告在脚本中我们verifyFalse跳过了SSL证书验证这是因为很多内网设备的证书是自签名的。在生产环境中如果条件允许建议为AF配置有效的证书并启用验证以提高安全性。6. 联动测试与全流程验证配置完成后必须进行端到端的测试确保链条每个环节都畅通。6.1 模拟攻击生成测试日志最直接的测试方法就是模拟一次真实的攻击。从互联网或测试机对AF保护的一台服务器发起一次明显的攻击扫描。例如使用nmap对服务器的SSH端口进行扫描或者使用sqlmap对一个测试页面进行注入确保在授权和隔离的环境中进行。观察AF的“安全日志”界面确认产生了对应的攻击日志如“入侵攻击”或“暴力破解”。等待几十秒到一分钟取决于Zabbix监控项的更新间隔和触发器检查周期。6.2 在Zabbix中检查日志接收与告警登录Zabbix Web进入【监控】-【最新数据】选择你创建的Sangfor_AF主机查看af.security.log这个监控项。你应该能看到最新的值就是AF发过来的那条攻击日志的全文。进入【监控】-【问题】界面你应该能看到一条新的问题触发器名称就是你设置的“AF检测到高危入侵攻击”状态为“PROBLEM”。点击该问题查看“历史”和“动作日志”确认触发的动作“自动封锁AF攻击IP”已经执行并且远程命令的状态如果Zabbix有记录的话。6.3 验证脚本执行与AF封锁结果查看脚本日志立刻登录Zabbix Server执行tail -f /var/log/zabbix/af_auto_block.log。在触发告警后你应该能看到脚本被调用的日志记录包括“接收到告警日志”、“提取到IP: x.x.x.x”、“AF API登录成功”、“成功封禁IP”等信息。验证AF黑名单登录AF防火墙管理界面进入【策略】-【安全策略】-【黑名单】或类似路径不同版本可能叫“封堵列表”、“地址组”等。你应该能看到一个新的条目IP地址就是攻击源的IP备注信息里可能有“Zabbix自动封禁”等字样并且生效时间符合你设置的BLOCK_DURATION_HOURS。验证封锁效果尝试从被封锁的IP再次访问AF保护的服务应该会被直接拒绝或超时。6.4 常见测试失败场景与排查如果测试失败按照以下顺序排查AF日志未发出检查AF上的“日志发送”配置IP和端口是否正确。在Zabbix Server上用tcpdump -i any port 514 -n或nc -luk 514命令监听看是否能收到AF发来的UDP报文。Zabbix未收到日志检查Zabbix Server上rsyslog配置是否正确服务是否重启。检查Zabbix主机配置中“Agent接口”的IP是否与rsyslog转发目标IP一致。检查Zabbix监控项的“键值”是否与rsyslog转发规则中指定的键值一致。触发器未触发检查触发器的表达式是否正确。在“最新数据”里复制一条日志去测试你的正则表达式是否匹配。检查触发器的“严重性”是否足够高以及动作的“条件”是否满足如时间条件。动作未执行或脚本执行失败在Zabbix动作的“操作”里可以尝试在“命令”中写一个简单的测试命令如echo test /tmp/zabbix_test.log看文件是否生成以验证远程命令执行功能是否正常。手动在Zabbix Server上切换到zabbix用户su - zabbix然后执行你的Python脚本并传入参数看是否有错误输出。Zabbix默认以zabbix用户身份执行脚本权限和环境变量可能受限这是最常见的坑。检查脚本本身的权限 (chmod 755)以及Python路径和依赖库 (requests) 是否对zabbix用户可用。仔细查看/var/log/zabbix/af_auto_block.log日志里面的错误信息是定位问题的关键。7. 生产环境部署的注意事项与优化建议当测试通过准备上生产环境时以下几点需要特别注意权限最小化原则在AF上专门创建一个用于API操作的账号权限只赋予“策略管理”或“黑名单管理”相关的最小权限集绝不能使用管理员账号。将Python脚本中的AF密码等敏感信息从明文改为从加密的配置文件或环境变量中读取。可以考虑使用ansible-vault加密或专门的密钥管理服务。防止误封与误操作设置IP白名单在脚本解析IP后、调用API前加入一个白名单检查。将公司出口IP、合作伙伴IP、云服务IP等加入白名单列表避免误封。设置内网IP过滤通常不需要封禁内网IP如10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16。在脚本中判断如果是内网IP则记录日志并跳过封禁。谨慎设置触发器初期只对确凿无疑的高危事件如“暴力破解成功”、“已知严重漏洞利用”进行自动封禁。对于“端口扫描”、“WEB攻击尝试”等可以仅告警不自动封禁。设置冷却时间在Zabbix动作中可以设置“默认操作步骤持续时间”和“问题更新间隔”避免在短时间内对同一IP的重复攻击产生大量重复的封禁操作给AF API造成压力。脚本健壮性增强加入重试机制对于AF API调用可以加入简单的重试逻辑如最多重试3次以应对网络瞬时波动。加入锁机制如果脚本可能被并发执行多个告警同时触发应考虑使用文件锁fcntl或数据库锁防止对同一个IP的重复封禁操作冲突。更精细的日志格式处理可以编写一个日志解析类针对不同的事件类型从触发器名称或日志内容判断采用不同的正则表达式提取IP提高准确率。监控与维护监控脚本执行除了脚本自身的日志可以在Zabbix中创建一个监控项监控/var/log/zabbix/af_auto_block.log文件的大小或最后修改时间如果长时间未更新可能意味着联动失效。定期审计封禁列表每周或每月检查一次AF上的自动封禁列表确认封禁的IP是否合理及时清理过期的或误封的条目。备份配置将AF的API配置、Zabbix的触发器表达式、动作命令以及Python脚本本身进行版本管理如Git。这套联动方案实施后我们团队处理高频、低复杂度攻击事件的效率得到了质的提升。它就像一位不知疲倦的初级安全分析师7x24小时值守将我们从中重复性的工作中解脱出来。当然自动化不是万能的它依赖于准确的检测AF的规则和精确的解析脚本的正则在部署初期需要一段时间的观察和调优。一旦稳定运行它将成为你安全防御体系中一个非常可靠且高效的自动化响应节点。
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
