Como analisar o audit.log usando o logstash

Question

Eu quero usar o logstash para coletar um arquivo de log, e o formato do arquivo é assim:

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'

Qual filtro devo usar para corresponder à linha? ou há outra maneira de lidar com isso.

Qualquer ajuda seria apreciada.

Usou o padrão abaixo para corresponder à linha com o depurador grok , mas ainda recebeu uma mensagem 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}

Mas quando eu removi subj=%{WORD:audit_subject} msg=%{GREEDYDATA:audit_message} , ele obteve sucesso e obteve um objeto JSON como este.

{
  "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"
    ]
  ]
}

Não sei porque subj e msg não podem funcionar.

    

Answer 1

Uma pesquisa rápida encontra esta no 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}

Uma resenha superficial sugere que é provavelmente o que você está procurando.

    

Answer 2

Os logs de auditoria são gravados como uma série de pares key = value que são facilmente extraídos usando o filtro kv. No entanto, notei que a chave msg às vezes é usada duas vezes e também é uma série de pares chave = valor.

O primeiro grok é usado para obter os campos audit_type , audit_epoch , audit_counter e sub_msg (o campo da segunda msg)

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

kv é usado para extrair todos os pares key = value, exceto msg e type, pois já obtivemos esses dados com grok:

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

kv é usado novamente para analisar os pares chave = valor em sub_msg (se existir):

kv {
  source => "sub_msg"
}

date é usada para definir a data como o valor em audit_epoch, usando o formato de data UNIX para analisar os timestamps de float ou inteiro:

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

Por último, o uso de mutate é usado para remover campos redundantes:

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

Você também pode renomear campos como o sysadmin1138 sugerido:

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

Tudo combinado o filtro se parece com isso:

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']
  }
}

    

Answer 3

Uma solução melhor do que grok pode ser usar o kv filtro. Isso analisa os campos configurados no formato "key = value", que são a maioria dos registros do log de auditoria. Ao contrário do Grok, isso irá lidar com strings com campos sometimes-there-sometimes-not. No entanto, os nomes de campo estão em suas formas curtas menos úteis, portanto, talvez seja necessário fazer alguma renomeação de campo.

filter { 
  kv { }
}

Isso lhe daria a maior parte e os campos corresponderiam ao que aparece nos registros. Todos os tipos de dados seriam string . Para resolver todos os problemas para humanizar os campos:

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

O campo msg , que contém o timestamp e o event-Id, ainda precisará ser grokked. As outras respostas mostram como fazer isso.

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 4

o formato do grok mudou, então dê uma olhada nisso:

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" ]
    }
}

Isso usa a data de audit_epoch como @datetime.