FBI-Analyzer Web日志安全分析系统教学文档

一、系统概述

FBI-Analyzer是一个基于Golang和Lua开发的灵活日志分析系统，采用插件化架构设计，主要用于Web安全分析场景。该系统可作为WAF的辅助系统，通过旁路分析减轻WAF性能压力。

核心特点

• 插件化架构：类似ngx-lua的设计风格，使用Lua编写分析逻辑
• 高性能处理：支持十万级QPS的日志分析
• 旁路分析：与业务系统解耦，不影响线上服务性能
• 快速生效：插件可秒级更新生效
• 丰富函数库：内置多种实用函数和模块

二、系统架构与原理

技术栈

• 主语言：Golang（处理高性能并发和基础功能）
• 脚本语言：Lua（实现业务逻辑和插件）
• 数据源：Kafka（日志输入）
• 缓存：Redis（数据存储和状态保持）

工作原理

1. 主程序通过Golang启动多个Lua虚拟机实例（每个CPU核心一个）
2. 从Kafka消费日志数据并分发给各个Lua虚拟机
3. Lua插件处理日志数据并执行安全分析逻辑
4. 分析结果存储到Redis供WAF查询使用

三、核心功能模块

1. Lua插件系统

插件编写规范

• 使用标准Lua语法（非LuaJIT，不支持FFI）
• 通过.调用方法而非:（避免隐式self参数）
• 可访问内置全局变量和函数库

内置全局变量

fbi = {
    var = {  -- 访问日志变量
        status = 200,  -- 状态码示例
        host = "www.example.com"  -- 主机名示例
    },
    log = logFunction,  -- 日志记录函数
    ERROR = 1,          -- 错误级别
    DEBUG = 2,          -- 调试级别
    INFO = 3            -- 信息级别
}

2. 内置函数库

Redis模块

local redis = require("redis")

-- 单条操作
local result, err = redis.hmget(key, field)
local ok, err = redis.hmset(key, field, value)
local result, err = redis.incr(key, field)
local ok, err = redis.expire(key, seconds)
local ok, err = redis.delete(key)

-- 批量操作(Pipeline)
local pipeline = redis.pipeline
pipeline.new()
local result, err = pipeline.hmget(key, field)
-- ...其他操作
local err = pipeline.exec()
pipeline.close()

正则模块

local re = require("re")
local ok, err = re.match("abcabcd", "^ab")  -- 返回布尔值
local str, err = re.find("abcabcd", "^ab")  -- 返回匹配字符串

时间模块

local time = require("time")
local tu = time.unix()      -- 获取时间戳
local tf = time.format()    -- 格式化时间
local zero = time.zero      -- 基准时间(用于时间段计算)

3. 日志处理流程

1. 从Kafka获取JSON格式的日志数据
2. 在Golang中反序列化为AccessLog结构体
3. 传递给Lua虚拟机进行处理
4. Lua插件通过fbi.var访问日志字段

四、实战应用示例

示例需求

"对5分钟内的访问状态码40x的IP进行统计，5分钟内超过100次的打上标签锁定10分钟，供WAF进行拦截"

实现代码

local redis = require("redis")
local time = require("time")
local var = fbi.var
local log = fbi.log
local ERROR = fbi.ERROR

-- 获取客户端IP(考虑XFF头)
local ip = var.XFF or var.remote_addr

-- 只处理40x状态码
if var.status >= 400 and var.status < 500 then
    -- 计算当前5分钟时间段
    local now = time.unix()
    local period = math.floor((now - time.zero) / 300)  -- 每300秒一个时间段
    
    -- 构造Redis键
    local key = "ip:40x:" .. ip .. ":" .. tostring(period)
    
    -- 递增计数
    local count, err = redis.incr("40x_counter", key)
    if err then
        log(ERROR, "redis incr error:", err)
        return
    end
    
    -- 如果是第一次设置，设置过期时间
    if tonumber(count) == 1 then
        redis.expire("40x_counter", key, 300)  -- 5分钟过期
    end
    
    -- 检查是否超过阈值
    if tonumber(count) > 100 then
        -- 标记恶意IP，锁定10分钟
        redis.hmset("malicious_ips", ip, now + 600)  -- 10分钟=600秒
        log(ERROR, "IP blocked:", ip, "count:", count)
    end
end

五、系统部署与配置

1. 环境准备

依赖安装

# 安装librdkafka(Kafka依赖)
# 参照 https://github.com/confluentinc/confluent-kafka-go#installing-librdkafka

# 获取项目
git clone https://github.com/C4o/FBI-Analyzer
cd FBI-Analyzer
go build main.go

2. 配置文件示例(config.yaml)

# Redis配置
redis: "127.0.0.1:6379"
password: ""
db: 9

# Kafka配置
broker: 192.168.1.1:9092
groupid: group-access-test-v1
topic: 
  - waflog
offset: latest

# 日志配置
path: Analyzer.log

3. 运行模式

生产模式

// 初始化redis
red := db.Redis{
    RedisAddr: conf.Cfg.RedAddr,
    RedisPass: conf.Cfg.RedPass,
    RedisDB: conf.Cfg.DB,
}

// 初始化kafka
kaf := db.Kafka{
    Broker: conf.Cfg.Broker,
    GroupID: conf.Cfg.GroupID,
    Topic: conf.Cfg.Topic,
    Offset: conf.Cfg.Offset,
}

// 启动lua进程
for i := 0; i < runtime.NumCPU(); i++ {
    go lua.LuaThread(i)
    go kaf.Consumer(lua.Kchan, i)
}

// redis健康检查
red.Health()

测试模式(无Redis/Kafka)

// 注释掉Redis和Kafka初始化

// 启动lua进程
for i := 0; i < runtime.NumCPU(); i++ {
    go lua.LuaThread(i)
}

// 使用本地模拟消费者
lua.TestConsumer()

六、插件开发指南

1. 插件编写要点

1. 日志访问：通过fbi.var获取日志字段

   local status = var.status
   local host = var.host
   local uri = var.uri
   

2. 日志记录：使用内置log函数

   log(fbi.ERROR, "error message")
   log(fbi.INFO, "info message")
   

3. 错误处理：检查Redis等操作的返回值

   local ok, err = redis.hmset(key, field, value)
   if err then
       log(fbi.ERROR, "redis error:", err)
       return
   end
   

2. 性能优化建议

1. 使用Pipeline：批量执行Redis命令减少网络往返
2. 合理设置过期时间：避免Redis数据无限增长
3. 缓存编译结果：插件更新时会自动缓存编译结果
4. 避免阻塞操作：Lua脚本中不要执行耗时操作

七、应用场景扩展

1. WAF辅助系统

• 实现复杂的风控逻辑
• 统计分析恶意请求模式
• 维护IP黑白名单

2. 安全分析场景

• 扫描器识别与拦截
• 暴力破解防护
• API滥用检测

3. 业务监控

• 异常访问模式检测
• API调用统计
• 业务指标监控

八、故障排查

常见错误处理

1. Lua脚本错误

   cat Analyzer.log | grep "#" | head -n 5
   

   示例输出：

   [error] coroutines failed : scripts/counter.lua:5: bad argument #3 to incr (value expected).
   

2. Kafka连接问题

   [error] Consumer error: 10.205.241.146:9092/bootstrap: Connect to ipv4#10.205.241.146:9092 failed: No route to host
   

3. Redis操作错误

   • 检查Redis服务是否正常运行
   • 验证配置中的密码和数据库编号

九、项目定制开发

1. 自定义日志格式

1. 修改rule/struct.go定义日志结构体

   type AccessLog struct {
       Host    string `json:"host"`
       Status  int    `json:"status"`
       XFF     string `json:"XFF"`
       // 添加自定义字段...
   }
   

2. 更新lua/http.go中的字段映射

   func GetReqVar(L *lua.LState) int {
       access := L.GetGlobal("access").(*lua.LUserData).Value.(*rule.AccessLog)
       switch L.CheckString(2) {
           // 添加自定义字段处理...
       }
   }
   

2. 添加自定义函数

1. 在Golang中实现函数

   func customFunction(L *lua.LState) int {
       // 实现逻辑...
       return 0
   }
   

2. 注册到Lua虚拟机

   // 在初始化时添加
   L.SetGlobal("customFunc", L.NewFunction(customFunction))
   

十、总结

FBI-Analyzer通过Golang和Lua的结合，实现了高性能、灵活的Web日志安全分析系统。其核心优势在于：

1. 性能与灵活性的平衡：Golang处理高性能部分，Lua实现业务逻辑
2. 无缝集成现有WAF：作为旁路系统减轻WAF压力
3. 快速响应安全需求：通过Lua插件快速实现新的分析规则
4. 易于扩展：支持自定义日志格式和函数库

通过本教学文档，您应该能够理解系统架构、掌握插件开发方法，并根据实际需求进行定制开发。