如何通过rsyslog模块高效将auditD日志转换为JSON格式

问题背景 #

需要通过rsyslog把audit daemon的默认日志转换成JSON格式，但auditd存在多种审计记录类型，逐个为每种类型编写规则效率极低。已尝试过rulebase.rb和mmpstructdata，希望找到更高效的模块或实现方法。

当前日志示例 #

type=USER_AUTH msg=audit(1576848185.440:407340): user pid=22571 
    uid=0 auid=1000 ses=3 msg='op=PAM:authentication acct="root" 
    exe="/usr/bin/sudo" hostname=? addr=? terminal=pts/0 
    res=success'

type=CWD msg=audit(1364481363.243:24287):  cwd="/home/shadowman"

type=PATH msg=audit(1364481363.243:24287): item=0 
    name="/etc/ssh/sshd_config" inode=409248 dev=fd:00 
    mode=0100600 ouid=0 ogid=0 rdev=00:00 
    obj=system_u:object_r:etc_t:s0  objtype=NORMAL cap_fp=none 
    cap_fi=none cap_fe=0 cap_fver=0

期望JSON格式 #

{
  "type": "USER_AUTH",
  "time_stamp": 1576848185.44,
  "msg_id": 407340,
  "pid": "22571",
  "uid": 0,
  "auid": 1000,
  "ses": 3,
  "msg": {
    "op": "PAM:authentication",
    "acct": "root",
    "exe": "/usr/bin/sudo",
    "hostname": "?",
    "addr": "?",
    "terminal": "pts/0",
    "res": "success"
  }
}

高效解决方案 #

1. 使用mmnormalize模块 + liblognorm规则 #

rsyslog的mmnormalize模块基于liblognorm，可通过通用规则匹配auditd各类日志，无需为每种记录类型单独编写规则。

操作步骤：

• 安装依赖：确保系统安装rsyslog-mmnormalize和liblognorm包（不同发行版包名略有差异，Debian/Ubuntu、RHEL/CentOS均可用同名包安装）。
• 创建liblognorm规则文件（如/etc/rsyslog.d/auditd.lua），编写通用匹配规则：

  rule=:%type:word% msg=audit(:%timestamp:float%:%msg_id:number%): %rest:rest%
  rule=:%type:word% msg=audit(:%timestamp:float%:%msg_id:number%): user pid=%pid:number% uid=%uid:number% auid=%auid:number% ses=%ses:number% msg='%msg_content:rest%'
  rule=:%type:word% msg=audit(:%timestamp:float%:%msg_id:number%): cwd="%cwd:string%"
  rule=:%type:word% msg=audit(:%timestamp:float%:%msg_id:number%): item=%item:number% name="%name:string%" inode=%inode:number% dev=%dev:string% mode=%mode:string% ouid=%ouid:number% ogid=%ogid:number% rdev=%rdev:string% obj=%obj:string% objtype=%objtype:word% cap_fp=%cap_fp:word% cap_fi=%cap_fi:word% cap_fe=%cap_fe:number% cap_fver=%cap_fver:number%
  

  上述规则可覆盖常见auditd记录类型，liblognorm会自动匹配对应日志行。
• 在rsyslog配置文件（如/etc/rsyslog.d/auditd.conf）中加载模块并配置：

  module(load="mmnormalize")
  module(load="mmjsonparse") # 辅助解析嵌套的msg内容
  module(load="omfile")
  
  # 读取auditd日志文件
  input(type="imfile" File="/var/log/audit/audit.log" Tag="auditd:" Severity="info" Facility="local6")
  
  # 处理auditd日志
  if $syslogtag == 'auditd:' then {
    action(type="mmnormalize" rulebase="/etc/rsyslog.d/auditd.lua")
    action(type="mmjsonparse" field="msg_content") # 解析msg_content中的键值对为嵌套结构
    # 定义JSON输出模板
    template(name="auditdJson" type="list") {
      constant(value="{")
      constant(value="\"type\":\"") field(name="$!type") constant(value="\",")
      constant(value="\"time_stamp\":") field(name="$!timestamp") constant(value=",")
      constant(value="\"msg_id\":") field(name="$!msg_id") constant(value=",")
      # 动态添加存在的字段
      if $!pid then { constant(value="\"pid\":\"") field(name="$!pid") constant(value="\",") }
      if $!uid then { constant(value="\"uid\":") field(name="$!uid") constant(value=",") }
      if $!auid then { constant(value="\"auid\":") field(name="$!auid") constant(value=",") }
      if $!ses then { constant(value="\"ses\":") field(name="$!ses") constant(value=",") }
      if $!cwd then { constant(value="\"cwd\":\"") field(name="$!cwd") constant(value="\",") }
      if $!item then { constant(value="\"item\":") field(name="$!item") constant(value=",") }
      if $!name then { constant(value="\"name\":\"") field(name="$!name") constant(value="\",") }
      # 处理嵌套的msg字段
      if $!msg_content!op then {
        constant(value="\"msg\":{")
        constant(value="\"op\":\"") field(name="$!msg_content!op") constant(value="\",")
        constant(value="\"acct\":\"") field(name="$!msg_content!acct") constant(value="\",")
        constant(value="\"exe\":\"") field(name="$!msg_content!exe") constant(value="\",")
        constant(value="\"hostname\":\"") field(name="$!msg_content!hostname") constant(value="\",")
        constant(value="\"addr\":\"") field(name="$!msg_content!addr") constant(value="\",")
        constant(value="\"terminal\":\"") field(name="$!msg_content!terminal") constant(value="\",")
        constant(value="\"res\":\"") field(name="$!msg_content!res") constant(value="\"")
        constant(value="}")
      }
      constant(value="}")
      constant(value="\n")
    }
    # 输出到JSON日志文件
    action(type="omfile" File="/var/log/audit/audit.json" Template="auditdJson")
    stop
  }
  

• 重启rsyslog服务：systemctl restart rsyslog

2. 使用rsyslog模板结合正则批量提取字段 #

若不想依赖liblognorm，可直接用rsyslog的模板功能，通过正则表达式批量提取所有可能的字段，不存在的字段会自动忽略。

示例配置片段：

module(load="omfile")

input(type="imfile" File="/var/log/audit/audit.log" Tag="auditd:")

template(name="auditdJsonTemplate" type="string" string="{"
  "\"type\":\"%msg:R,ERE,1,FIELD:(^type=)(\\w+)--end%\", "
  "\"time_stamp\":%msg:R,ERE,1,FIELD:(audit\\()([0-9]+\\.[0-9]+)--end%, "
  "\"msg_id\":%msg:R,ERE,1,FIELD:(:[0-9]+\\))--end::drop-last%, "
  "%msg:R,ERE,0,FIELD:(pid=)([0-9]+)--end:\"pid\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(uid=)([0-9]+)--end:\"uid\":%2%, %"
  "%msg:R,ERE,0,FIELD:(auid=)([0-9]+)--end:\"auid\":%2%, %"
  "%msg:R,ERE,0,FIELD:(ses=)([0-9]+)--end:\"ses\":%2%, %"
  "%msg:R,ERE,0,FIELD:(cwd=")([^&]+)--end:\"cwd\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(msg='op=)([^ ]+)--end:\"msg\":{\"op\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(acct=")([^&]+)--end:\"acct\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(exe=")([^&]+)--end:\"exe\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(hostname=)([^ ]+)--end:\"hostname\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(addr=)([^ ]+)--end:\"addr\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(terminal=)([^ ]+)--end:\"terminal\":\"%2%\", %"
  "%msg:R,ERE,0,FIELD:(res=)([^']+)--end:\"res\":\"%2%\"}%\""
  "}\n")

if $syslogtag == 'auditd:' then {
  action(type="omfile" File="/var/log/audit/audit.json" Template="auditdJsonTemplate")
  stop
}

3. 最简方案：直接配置auditd输出JSON #

若系统auditd版本≥3.0，可跳过rsyslog，直接让auditd输出JSON格式日志。修改/etc/audit/auditd.conf：

log_format = json

重启auditd服务：systemctl restart auditd

内容的提问来源于stack exchange，提问作者Logesh R