如何使用logstash解析audit.log

Question

我想用logstash收集一个日志文件，文件格式是这样的：

type=USER_START msg=audit(1404170401.294:157): user pid=29228 uid=0 auid=0 ses=7972 subj=system_u:system_r:crond_t:s0-s0:c0.c1023 msg='op=PAM:session_open acct="root" exe="/usr/sbin/crond" hostname=? addr=? terminal=cron res=success'

我应该使用哪个过滤器来匹配该行？或者有另一种方法来处理它。

任何帮助，将不胜感激。

---

使用下面的模式将行与grok debugger匹配，但仍然收到一条No matches消息。

type=%{WORD:audit_type} msg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\): user pid=%{NUMBER:audit_pid} uid=%{NUMBER:audit_uid} auid=%{NUMBER:audit_audid} subj=%{WORD:audit_subject} msg=%{GREEDYDATA:audit_message}

但是当我删除时subj=%{WORD:audit_subject} msg=%{GREEDYDATA:audit_message}，它成功并得到了一个这样的 JSON 对象。

{
  "audit_type": [
    [
      "USER_END"
    ]
  ],
  "audit_epoch": [
    [
      "1404175981.491"
    ]
  ],
  "BASE10NUM": [
    [
      "1404175981.491",
      "524",
      "1465",
      "0",
      "0"
    ]
  ],
  "audit_counter": [
    [
      "524"
    ]
  ],
  "audit_pid": [
    [
      "1465"
    ]
  ],
  "audit_uid": [
    [
      "0"
    ]
  ],
  "audit_audid": [
    [
      "0"
    ]
  ]
}

不知道为什么subj，msg无法继续工作。

Answer 1

审计日志被写成一系列键=值对，可以使用 kv 过滤器轻松提取。但是我注意到该键msg有时会使用两次，并且也是一系列键=值对。

首先神交用于获取领域audit_type，audit_epoch，audit_counter和sub_msg（2号MSG字段）

grok {
  pattern => [ "type=%{DATA:audit_type}\smsg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\):.*?( msg=\'(?<sub_msg>.*?)\')?$" ]
  named_captures_only => true
}

kv 用于提取除 msg 和 type 之外的所有键=值对，因为我们已经使用 grok 获取了这些数据：

kv {
  exclude_keys => [ "msg", "type" ]
}

kv 再次用于解析 sub_msg 中的 key=value 对（如果存在）：

kv {
  source => "sub_msg"
}

date 用于将日期设置为 audit_epoch 中的值，使用日期格式UNIX将解析浮点或整数时间戳：

date {
  match => [ "audit_epoch", "UNIX" ]
}

最后 mutate 用于删除冗余字段：

mutate {
  remove_field => ['sub_msg', 'audit_epoch']
}

您还可以重命名像 sysadmin1138 建议的字段：

mutate {
  rename => [
    "auid", "uid_audit",
    "fsuid", "uid_fs",
    "suid", "uid_set",
    "ses", "session_id"
  ]
  remove_field => ['sub_msg', 'audit_epoch']
}

所有组合过滤器看起来像这样：

filter {
  grok {
    pattern => [ "type=%{DATA:audit_type}\smsg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\):.*?( msg=\'(?<sub_msg>.*?)\')?$" ]
    named_captures_only => true
  }
  kv {
    exclude_keys => [ "msg", "type" ]
  }
  kv {
    source => "sub_msg"
  }
  date {
    match => [ "audit_epoch", "UNIX" ]
  }
  mutate {
    rename => [
      "auid", "uid_audit",
      "fsuid", "uid_fs",
      "suid", "uid_set",
      "ses", "session_id"
    ]
    remove_field => ['sub_msg', 'audit_epoch']
  }
}