win8建立网站,django做的电子商务网站,中国建设银行复核网站,wordpress和帝国谁快场景解析在企业网络安全日志处理场景中#xff0c;防火墙、入侵检测系统#xff08;IDS#xff09;等设备会持续产生大量日志#xff0c;记录网络流量、访问请求、异常事件等基础信息#xff0c;但这些原始日志仅能呈现表面现象#xff0c;难以全面剖析安全威胁#xff…场景解析在企业网络安全日志处理场景中防火墙、入侵检测系统IDS等设备会持续产生大量日志记录网络流量、访问请求、异常事件等基础信息但这些原始日志仅能呈现表面现象难以全面剖析安全威胁需要在日志处理过程中引入外部数据增强安全分析能力。例如某天 IDS入侵检测系统 日志记录到多个来自同一 IP 地址的异常端口扫描行为原始日志仅显示时间、源 IP、扫描端口等信息无法判断该 IP 是否为恶意攻击源也不清楚其背后的攻击意图。此时就需要引用外部数据来丰富日志内容。可以从威胁情报平台获取该 IP 地址的历史攻击记录、所属的恶意组织标签等数据从地理位置数据库获取其所在地区、网络服务提供商等信息并将这些外部数据与原始日志进行关联整合。经过数据融合后原本孤立的日志事件就有了更丰富的背景信息。安全人员可以通过这些丰富的日志数据快速判断该 IP 地址是否属于已知的恶意攻击源是否存在特定的攻击目标偏好进而采取更精准的安全防护措施如封禁 IP、加强对应端口的防护策略等。同时还能基于这些数据对攻击模式进行分析预测潜在的安全风险提前完善安全防御体系提升企业整体的网络安全防护能力。方案解析观测云 Pipeline 是一个可编程数据处理器使用观测云开源的 Platypus 语言作为运行时能够在边缘节点进行大规模数据分析和特征提取。Datakit Pipeline 提供以下两个内置函数用于从外部表引用数据query_refer_table()函数原型为 fn query_refer_table(table_name: str, key: str, value)它能够查询 table_name 表中 key 列为 value 的行将返回的首行数据丰富至日志中mquery_refer_table()函数原型为 fn mquery_refer_table(table_name: str, keys: list, values: list)相比 query_refer_table()该函数能够使用多个列和值对 table_name 表进行查询。在以上函数的支持下Pipeline 能够使用外部表对安全日志进行丰富以 Zeek conn.log 为例丰富前的日志如下
{ts: 1591367999.3059881,uid: CMdzit1AMNsmfAIiQc,id.orig_h: 192.168.4.76, # 源 IPid.orig_p: 36844,id.resp_h: 192.168.4.1,id.resp_p: 53,proto: udp,service: dns,duration: 0.06685185432434082,orig_bytes: 62,resp_bytes: 141,conn_state: SF,missed_bytes: 0,history: Dd,orig_pkts: 2,orig_ip_bytes: 118,resp_pkts: 2,resp_ip_bytes: 197,ip_proto: 17
}
假设在 Pipeline 中引用了风险情报表此表中记录了危险 IP包含 IP 列和信息列当源 IP id.orig_h 字段的值能够匹配到风险表中 IP 列的值时就会为此日志丰富风险信息字段丰富后的日志如下
{ts: 1591367999.3059881,uid: CMdzit1AMNsmfAIiQc,id.orig_h: 192.168.4.76, # 源 IPid.orig_p: 36844,id.resp_h: 192.168.4.1,id.resp_p: 53,proto: udp,service: dns,duration: 0.06685185432434082,orig_bytes: 62,resp_bytes: 141,conn_state: SF,missed_bytes: 0,history: Dd,orig_pkts: 2,orig_ip_bytes: 118,resp_pkts: 2,resp_ip_bytes: 197,ip_proto: 17,risk_ip: 192.168.4.76, # 丰富风险信息字段risk_info: 此 IP 近期发起大量攻击 # 丰富风险信息字段
}
在对日志完成丰富后就可以在观测云过滤存在 risk_info 字段的日志进行告警和特征分析具体的分析方法和场景根据丰富的字段扩展例如丰富了风险 IP 的地理位置和运营商信息后就能在观测云仪表盘中以地图的方式呈现攻击来源。在外部表管理方面Datakit 从 refer_table_url 中以指定间隔拉取外部表数据供 Pipeline 使用外部表数据必须组织为以下格式
[{table_name: table_abc,column_name: [col, col2, col3, col4],column_type: [string, float, int, bool],row_data: [[a, 123, 123, false],[ab, 1234., 123, true],[ab, 1234., 1235, false]]},{table_name: table_ijk,column_name: [name, id],column_type: [string, string],row_data: [[a, 12],[a, 123],[ab, 1234]]}
]
也就是说必须提供一个 HTTP/HTTPS 端点暴露表数据可以使用 Nginx 托管 JSON 文件的方式但是考虑到更灵活的集成能力推荐内网部署观测云 DataFlux FuncFunc 是一个函数开发、管理、执行平台可将集成了威胁平台的 Python 函数暴露为拉取表数据的端点也就是说当 Datakit 从此端点同步数据时会触发脚本运行脚本将从一个或者多个平台获取数据并组装为指定的格式。整体架构如下注意该功能内存消耗较高以 150 万行refer_table 行数、磁盘占用约 200MBJSON 文件的不重复数据string 类型两列intfloatbool 各一列为例内存占用维持在 950MB 1.2GB更新时的峰值内存 2.2GB ~ 2.7GB可通过配置 use_sqlite true将数据保存到磁盘上即使用 SQLite 存储数据而不是内存。演示用例前置条件假设用户具备以下条件部署了 DataKit 的 Linux 主机部署了 DataFlux Func。配置 Func在 Func 中新建脚本集 “Pipeline 外部表 Demo”新建 main 文件写入以下函数后发布
DFF.API(外部表, cache_result3000, timeout10)
def refer_table():返回符合 Pipeline query_refer_table() 和 mquery_refer_table() 函数格式要求的表数据。data [{table_name: risk_ip,column_name: [risk_ip, risk_info],column_type: [string, string],row_data: [[180.173.79.213, 属于 xxx 组织的恶意 IP存在端口和漏洞扫描行为],[180.173.79.214, 属于 xxx 组织的恶意 IP存在病毒传播行为],]}]print(执行同步请求)return data
在 Func 管理页面中新建同步 API将此函数暴露为接口为 DataKit 提供外部数据点击“示例”即可查看接口 URL在 Shell 中请求此 URL 即可获得数据配置 DataKit 拉取外部表编辑 DataKit 配置文件
vim /usr/local/datakit/conf.d/datakit.conf
修改以下配置
[pipeline]# 将 YOUR-FUNC-API-URL 替换为 Func 同步 API 的 URLrefer_table_url YOUR-FUNC-API-URLrefer_table_pull_interval 5muse_sqlite truesqlite_mem_mode false
重启 DataKit 使配置生效
datakit service -R
配置示例日志执行以下命令使用脚本生成测试日志
# 创建测试目录
mkdir -p ~/workspace/log_demo cd $_# 创建脚本
cat gen_log.sh EOF
#!/bin/bashwhile true; dotimestamp$(date %s.%N | cut -c1-17)ips(180.173.79.213 180.173.79.214)log{\ts\:${timestamp},\uid\:\CMdzit1AMNsmfAIiQc\,\id.orig_h\:\${ips[$((RANDOM % 2))]}\,\id.orig_p\:36844,\id.resp_h\:\192.168.4.1\,\id.resp_p\:53,\proto\:\udp\,\service\:\dns\,\duration\:0.06685185432434082,\orig_bytes\:62,\resp_bytes\:141,\conn_state\:\SF\,\missed_bytes\:0,\history\:\Dd\,\orig_pkts\:2,\orig_ip_bytes\:118,\resp_pkts\:2,\resp_ip_bytes\:197,\ip_proto\:17}echo ${log} ./demo.logecho gen log: ${log}sleep 1
done
EOF# 运行脚本将在当前目录下生成日志文件 demo.log
bash gen_log.sh
配置 DataKit 采集示例日志配置 DataKit 日志采集插件
cd /usr/local/datakit/conf.d/log
cp logging.conf.sample demo.conf
vim demo.conf
修改以下配置
[[inputs.logging]]# 日志文件路径logfiles [/root/workspace/log_demo/demo.log,]# 日志来源source demo
重启 DataKit 使配置生效
datakit service -R
登录观测云点击【日志】可见 demo.log 已经被采集。配置 Pipeline点击【Pipelines】-【新建 Pipeline】运行模式选择“本地 Pipeline”索引选择 “default”日志来源选择 “demo”Pipeline 名称填写 “demo”在“定义解析规则”中输入以下脚本后点击保存
# 将 JSON 字符串转换为对象
obj load_json(_)# 从 JSON 对象中提取字段并处理
# 原始日志使用时间戳标记日志时间可读性差提取此字段并转换为人类易读的格式
pt_kvs_set(ts_date, obj[ts]*1000000)
datetime(ts_date, us, RFC3339Nano, Asia/Shanghai)# 从日志中提取源 IP并根据源 IP 从外部表中丰富字段
pt_kvs_set(src_ip, obj[id.orig_h])
# query_refer_table 函数的参数分别为外部表名、列名、列值
query_refer_table(risk_ip, risk_ip, src_ip)
效果确认Pipeline 保存后约 1 分钟后生效生效后新增的字段需等待服务端刷新字段后才可查询刷新完成后查看日志列表已经丰富了相关字段。可以快速分析风险的趋势和占比。配置告警后可在风险发生时及时告警。如果在告警中关联创建异常追踪可闭环对安全风险的处理过程。
