如何使用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调试器匹配行，但仍然得到一条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

快速搜索在github上找到这

AUDIT 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} 
AUDITLOGIN type=%{WORD:audit_type} msg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\): login pid=%{NUMBER:audit_pid} uid=%{NUMBER:audit_uid} old auid=%{NUMBER:old_auid} new auid=%{NUMBER:new_auid} old ses=%{NUMBER:old_ses} new ses=%{NUMBER:new_ses}

粗略的回顾表明，这可能是你要找的东西。

Answer 2

一个比grok更好的解决方案可能是使用kv过滤器。这将分析以"key=value“格式配置的字段，大多数审计日志中心都是这种格式。不像Grok，这将处理字符串有时-那里-有时-而不是字段。然而，字段名是它们不太有用的简短形式，所以您可能需要进行一些字段重命名。

filter { 
  kv { }
}

这将使你得到大部分，而且字段将匹配在日志中显示的内容。所有的数据类型都是string。尽一切努力使田里人性化：

filter {
  kv { }
  mutate {
    rename => { 
      "type" => "audit_type"
      "auid" => "uid_audit"
      "fsuid => "uid_fs"
      "suid" => "uid_set"
      "ses" => "session_id"
    }
  }
}

但是，包含时间戳和事件Id的msg字段仍然需要检查。其他的答案显示了如何做到这一点。

filter {
  kv { }
  grok {
    match => { "msg" => "audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\):"
  }
  mutate {
    rename => { 
      "type" => "audit_type"
      "auid" => "uid_audit"
      "fsuid => "uid_fs"
      "suid" => "uid_set"
      "ses" => "session_id"
    }
  }
}

Answer 3

grok的格式已经更改，因此请看以下内容：

filter {
    grok {
        # example: type=CRED_DISP msg=audit(1431084081.914:298): pid=1807 uid=0 auid=1000 ses=7 msg='op=PAM:setcred acct="user1" exe="/usr/sbin/sshd" hostname=host1 addr=192.168.160.1 terminal=ssh res=success'
        match => { "message" => "type=%{WORD:audit_type} msg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_counter}\): pid=%{NUMBER:audit_pid} uid=%{NUMBER:audit_uid} auid=%{NUMBER:audit_audid} ses=%{NUMBER:ses} msg=\'op=%{WORD:operation}:%{WORD:detail_operation} acct=\"%{WORD:acct_user}\" exe=\"%{GREEDYDATA:exec}\" hostname=%{GREEDYDATA:hostname} addr=%{GREEDYDATA:ipaddr} terminal=%{WORD:terminal} res=%{WORD:result}\'" }
    }
    date {
        match => [ "audit_epoch", "UNIX_MS" ]
    }
}

这使用audit_epoch中的日期作为@datetime。