Dicas e solução de problemas ao escrever analisadores

Compatível com:

Google SecOps SIEM

Este documento descreve os problemas que você pode encontrar ao escrever o código do analisador.

Ao escrever o código do analisador, você pode encontrar erros quando as instruções de análise não funcionam como esperado. As situações que podem gerar erros incluem:

• Um padrão Grok falha
• Uma operação rename ou replace falha
• Erros de sintaxe no código do analisador

Práticas comuns no código do analisador

As seções a seguir descrevem as práticas recomendadas, dicas e soluções para ajudar a resolver problemas.

Evite usar pontos ou hifens em nomes de variáveis

O uso de hifens e pontos em nomes de variáveis pode causar um comportamento inesperado, geralmente ao realizar operações merge para armazenar valores em campos de UDM. Também pode haver problemas intermitentes de análise.

Por exemplo, não use os seguintes nomes de variáveis:

• my.variable.result
• my-variable-result

Em vez disso, use o seguinte nome de variável: my_variable_result.

Não use termos com significado especial como nome de variável

Algumas palavras, como event e timestamp, podem ter um significado especial no código do parser.

A string event é usada com frequência para representar um único registro do UDM e é usada na instrução @output. Se uma mensagem de registro incluir um campo chamado event ou se você definir uma variável intermediária chamada event, e o código do analisador usar a palavra event na instrução @output, você vai receber uma mensagem de erro sobre um conflito de nome.

Renomeie a variável intermediária ou use o termo event1 como um prefixo nos nomes de campo da UDM e na instrução @output.

A palavra timestamp representa o carimbo de data/hora criado do registro bruto original. Um valor definido nessa variável intermediária é salvo no campo UDM metadata.event_timestamp. O termo @timestamp representa a data e a hora em que o registro bruto foi analisado para criar um registro de UDM.

O exemplo a seguir define o campo UDM metadata.event_timestamp como a data e a hora em que o registro bruto foi analisado.

 # Save the log parse date and time to the timestamp variable
  mutate {
     rename => {
       "@timestamp" => "timestamp"
     }
   }

O exemplo a seguir define o campo UDM metadata.event_timestamp como a data e a hora extraídas do registro bruto original e armazenadas na variável intermediária when.

Observação: este exemplo não mostra as instruções que extraem a data e a hora do registro bruto original para o campo intermediário when e transformam o valor no formato correto.

   # Save the event timestamp to timestamp variable
   mutate {
     rename => {
       "when" => "timestamp"
     }
   }

Não use os seguintes termos como variáveis:

• collectiontimestamp
• createtimestamp
• evento
• filename
• mensagem
• namespace
• output
• onerrorcount
• timestamp
• timezone

Armazenar cada valor de dados em um campo de UDM separado

Não armazene vários campos em um único campo do UDM concatenando-os com um delimitador. Veja um exemplo abaixo.

"principal.user.first_name" => "first:%{first_name},last:%{last_name}"

Em vez disso, armazene cada valor em um campo UDM separado.

"principal.user.first_name" => "%{first_name}"
"principal.user.last_name" => "%{last_name}"

Usar espaços em vez de tabulações no código

Não use guias no código do analisador. Use apenas espaços e recue dois espaços por vez.

Não realize várias ações de mesclagem em uma única operação

Se você mesclar vários campos em uma única operação, isso poderá resultar em resultados inconsistentes. Em vez disso, coloque instruções merge em operações separadas.

Por exemplo, substitua o seguinte exemplo:

mutate {
  merge => {
      "security_result.category_details" => "category_details"
      "security_result.category_details" => "super_category_details"
  }
}

Por este código:

mutate {
  merge => {
    "security_result.category_details" => "category_details"
  }
}

mutate {
  merge => {
    "security_result.category_details" => "super_category_details"
  }
}

Como escolher entre as expressões condicionais if e if else

Se o valor condicional que você está testando só puder ter uma correspondência, use a instrução condicional if else. Essa abordagem é um pouco mais eficiente. No entanto, se você tiver um cenário em que o valor testado possa corresponder mais de uma vez, use várias instruções if distintas e ordene as instruções do caso mais genérico ao mais específico.

Escolha um conjunto representativo de arquivos de registro para testar as mudanças do analisador

Uma prática recomendada é testar o código do analisador usando amostras de registros brutos com uma ampla variação de formatos. Isso permite encontrar registros exclusivos ou casos extremos que o parser precisa processar.

Adicionar comentários descritivos ao código do analisador

Adicione comentários ao código do analisador que expliquem por que a instrução é importante, em vez de o que ela faz. O comentário ajuda qualquer pessoa que mantenha o analisador a seguir o fluxo. Veja um exemplo abaixo.

# only assign a Namespace if the source address is RFC 1918 or Loopback IP address
if [jsonPayload][id][orig_h] =~ /^(127(?:\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?))\{3\}$)|(10(?:\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?))\{3\}$)|(192\.168(?:\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?))\{2\}$)|(172\.(?:1[6-9]|2\d|3[0-1])(?:\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?))\{2\}$)/ {
  mutate {
    replace => {
      "event1.idm.read_only_udm.principal.namespace" => "%{resource.labels.project_id}"
    }
  }
}

Inicializar variáveis intermediárias com antecedência

Antes de extrair valores do registro bruto original, inicialize variáveis intermediárias que serão usadas para armazenar valores de teste.

Isso evita que um erro seja retornado indicando que a variável intermediária não existe.

A instrução a seguir atribui o valor na variável product ao campo UDM metadata.product_name.

mutate{
  replace => {
    "event1.idm.read_only_udm.metadata.product_name" => "%{product}"
  }
}

Se a variável product não existir, você vai receber o seguinte erro:

"generic::invalid_argument: pipeline failed: filter mutate (4) failed: replace failure: field \"event1.idm.read_only_udm.metadata.product_name\": source field \"product\": field not set"

Você pode adicionar uma instrução on_error para detectar o erro. Veja um exemplo abaixo.

mutate{
  replace => {
    "event1.idm.read_only_udm.metadata.product_name" => "%{product}"
    }
  on_error => "_error_does_not_exist"
  }

A instrução de exemplo anterior captura o erro de análise em uma variável intermediária booleana, chamada _error_does_not_exist. Ele não permite que você use a variável product em uma instrução condicional, por exemplo, if. Veja um exemplo abaixo.

if [product] != "" {
  mutate{
    replace => {
      "event1.idm.read_only_udm.metadata.product_name" => "%{product}"
    }
  }
  on_error => "_error_does_not_exist"
}

O exemplo anterior retorna o seguinte erro porque a cláusula condicional if não oferece suporte a instruções on_error:

"generic::invalid_argument: pipeline failed: filter conditional (4) failed: failed to evaluate expression: generic::invalid_argument: "product" not found in state data"

Para resolver isso, adicione um bloco de instruções separado que inicialize as variáveis intermediárias antes de executar as instruções do filtro de extração (json, csv, xml, kv ou grok). Veja um exemplo.

filter {
  # Initialize intermediate variables for any field you will use for a conditional check
  mutate {
    replace => {
      "timestamp" => ""
      "does_not_exist" => ""
    }
  }

  # load the logs fields from the message field
  json {
    source         => "message"
    array_function => "split_columns"
    on_error       => "_not_json"
  }
}

O snippet atualizado do código do analisador processa os vários cenários usando uma declaração condicional para verificar se o campo existe. Além disso, a instrução on_error processa os erros encontrados.

Converter SHA-256 em base64

O exemplo a seguir extrai o valor SHA-256, o codifica em base64, converte os dados codificados em uma string hexadecimal e substitui campos específicos pelos valores extraídos e processados.

if [Sha256] != "" 
{
  base64
  {
  encoding => "RawStandard"
  source => "Sha256"
  target => "base64_sha256"
  on_error => "base64_message_error"
  }
  mutate
  {
    convert =>
    {
      "base64_sha256" => "bytestohex"
    }
    on_error => "already_a_string"
  }
  mutate
  {
    replace => 
  {
     "event.idm.read_only_udm.network.tls.client.certificate.sha256" => "%{base64_sha256}"
     "event.idm.read_only_udm.target.resource.name" => "%{Sha256}"
  }
  }
}

Processar erros em instruções do analisador

Não é incomum que os registros recebidos estejam em um formato inesperado ou tenham dados mal formatados.

Você pode criar o analisador para processar esses erros. Uma prática recomendada é adicionar gerenciadores on_error ao filtro de extração e, em seguida, testar a variável intermediária antes de prosseguir para o próximo segmento da lógica do analisador.

O exemplo a seguir usa o filtro de extração json com uma instrução on_error para definir a variável booleana _not_json. Se _not_json estiver definido como true, isso significa que a entrada de registro recebida não estava no formato JSON válido e não foi analisada. Se a variável _not_json for false, a entrada de registro recebida estará no formato JSON válido.

 # load the incoming log from the default message field
  json {
    source         => "message"
    array_function => "split_columns"
    on_error       => "_not_json"
  }

Também é possível testar se um campo está no formato correto. O exemplo a seguir verifica se _not_json está definido como true, indicando que o registro não estava no formato esperado.

 # Test that the received log matches the expected format
  if [_not_json] {
    drop { tag => "TAG_MALFORMED_MESSAGE" }
  } else {
    # timestamp is always expected
    if [timestamp] != "" {

      # ...additional parser logic goes here …

    } else {

      # if the timestamp field does not exist, it's not a log source
      drop { tag => "TAG_UNSUPPORTED" }
    }
  }

Isso garante que a análise não falhe se os registros forem ingeridos com um formato incorreto para o tipo de registro especificado.

Use o filtro drop com a variável tag para que a condição seja capturada na tabela de métricas de ingestão no BigQuery.

• TAG_UNSUPPORTED
• TAG_MALFORMED_ENCODING
• TAG_MALFORMED_MESSAGE
• TAG_NO_SECURITY_VALUE

O filtro drop impede que o analisador processe o registro bruto, normalize os campos e crie um registro do UDM. O registro bruto original ainda é processado no Google Security Operations e pode ser pesquisado usando a pesquisa de registro bruto no Google Security Operations.

O valor transmitido para a variável tag é armazenado no campo drop_reason_code na tabela de métricas de transferência. É possível executar uma consulta ad hoc na tabela de forma semelhante a esta:

SELECT
  log_type,
  drop_reason_code,
  COUNT(drop_reason_code) AS count
FROM `datalake.ingestion_metrics`
GROUP BY 1,2
ORDER BY 1 ASC

Resolver erros de validação

Ao criar um analisador, você pode encontrar erros relacionados à validação. Por exemplo, um campo obrigatório não está definido no registro da UDM. O erro pode ser semelhante a este:

Error: generic::unknown: invalid event 0: LOG_PARSING_GENERATED_INVALID_EVENT: "generic::invalid_argument: udm validation failed: target field is not set"

O código do analisador é executado, mas o registro de UDM gerado não inclui todos os campos de UDM necessários, conforme definido pelo valor definido para metadata.event_type. Confira outros exemplos que podem causar esse erro:

• Se metadata.event_type for USER_LOGIN e o campo UDM target.user value não estiver definido.
• Se metadata.event_type for NETWORK_CONNECTION e o campo target.hostnameUDM não estiver definido.

Para mais informações sobre o campo metadata.event_type do UDM e os campos obrigatórios, consulte o guia de uso do UDM.

Uma opção para resolver esse tipo de erro é começar definindo valores estáticos para campos de UDM. Depois de definir todos os campos da UDM necessários, examine o registro bruto original para saber quais valores analisar e salvar no registro da UDM. Se o registro bruto original não contiver determinados campos, talvez seja necessário definir valores padrão.

Observação: não é possível enviar um analisador com apenas valores estáticos, porque cada registro do UDM pode ter os mesmos dados. No entanto, essa abordagem é útil para resolver erros de validação do UDM.

Confira a seguir um exemplo de modelo específico para um tipo de evento USER_LOGIN que ilustra essa abordagem.

Observe o seguinte:

• O modelo inicializa variáveis intermediárias e define cada uma como uma string estática.
• O código na seção Atribuição de campo define os valores em variáveis intermediárias para campos do UDM.

Para expandir esse código, adicione outras variáveis intermediárias e campos do UDM. Depois de identificar todos os campos do UDM que precisam ser preenchidos, faça o seguinte:

• Na seção Configuração de entrada, adicione um código que extrai campos do registro bruto original e define os valores para as variáveis intermediárias.

• Na seção Extração de data, adicione um código que extraia o carimbo de data/hora do evento do registro bruto original, o transforme e o defina como a variável intermediária.

• Conforme necessário, substitua o valor inicializado definido em cada variável intermediária por uma string vazia.

filter {
 mutate {
   replace => {
     # UDM > Metadata
     "metadata_event_timestamp"    => ""
     "metadata_vendor_name"        => "Example"
     "metadata_product_name"       => "Example SSO"
     "metadata_product_version"    => "1.0"
     "metadata_product_event_type" => "login"
     "metadata_product_log_id"     => "12345678"
     "metadata_description"        => "A user logged in."
     "metadata_event_type"         => "USER_LOGIN"

     # UDM > Principal
     "principal_ip"       => "192.168.2.10"

     # UDM > Target
     "target_application"            => "Example Connect"
     "target_user_user_display_name" => "Mary Smith"
     "target_user_userid"            => "[e-mail]"

     # UDM > Extensions
     "auth_type"          => "SSO"
     "auth_mechanism"     => "USERNAME_PASSWORD"

     # UDM > Security Results
     "securityResult_action"         => "ALLOW"
     "security_result.severity"       => "LOW"

   }
 }

 # ------------ Input Configuration  --------------
  # Extract values from the message using one of the extraction filters: json, kv, grok

 # ------------ Date Extract  --------------
 # If the  date {} function is not used, the default is the normalization process time

  # ------------ Field Assignment  --------------
  # UDM Metadata
  mutate {
    replace => {
      "event1.idm.read_only_udm.metadata.vendor_name"        =>  "%{metadata_vendor_name}"
      "event1.idm.read_only_udm.metadata.product_name"       =>  "%{metadata_product_name}"
      "event1.idm.read_only_udm.metadata.product_version"    =>  "%{metadata_product_version}"
      "event1.idm.read_only_udm.metadata.product_event_type" =>  "%{metadata_product_event_type}"
      "event1.idm.read_only_udm.metadata.product_log_id"     =>  "%{metadata_product_log_id}"
      "event1.idm.read_only_udm.metadata.description"        =>  "%{metadata_description}"
      "event1.idm.read_only_udm.metadata.event_type"         =>  "%{metadata_event_type}"
    }
  }

  # Set the UDM > auth fields
  mutate {
    replace => {
      "event1.idm.read_only_udm.extensions.auth.type"        => "%{auth_type}"
    }
    merge => {
      "event1.idm.read_only_udm.extensions.auth.mechanism"   => "auth_mechanism"
    }
  }

  # Set the UDM > principal fields
  mutate {
    merge => {
      "event1.idm.read_only_udm.principal.ip"                => "principal_ip"
    }
  }

  # Set the UDM > target fields
  mutate {
    replace => {
      "event1.idm.read_only_udm.target.user.userid"             =>  "%{target_user_userid}"
      "event1.idm.read_only_udm.target.user.user_display_name"  =>  "%{target_user_user_display_name}"
      "event1.idm.read_only_udm.target.application"             =>  "%{target_application}"
    }
  }

  # Set the UDM > security_results fields
  mutate {
    merge => {
      "security_result.action" => "securityResult_action"
    }
  }

  # Set the security result
  mutate {
    merge => {
      "event1.idm.read_only_udm.security_result" => "security_result"
    }
  }

 # ------------ Output the event  --------------
  mutate {
    merge => {
      "@output" => "event1"
    }
  }

}

Analisar texto não estruturado usando uma função Grok

Ao usar uma função Grok para extrair valores de texto não estruturado, você pode usar padrões predefinidos do Grok e instruções de expressão regular. Os padrões Grok facilitam a leitura do código. Se a expressão regular não incluir caracteres abreviados (como \w, \s), copie e cole a instrução diretamente no código do analisador.

Como os padrões Grok são uma camada de abstração adicional na instrução, eles podem tornar a solução de problemas mais complexa quando você encontra um erro. Confira a seguir um exemplo de função Grok que contém padrões predefinidos do Grok e expressões regulares.

grok {
  match => {
    "message" => [
      "%{NUMBER:when}\\s+\\d+\\s%{SYSLOGHOST:srcip} %{WORD:action}\\/%{NUMBER:returnCode} %{NUMBER:size} %{WORD:method} (?P<url>\\S+) (?P<username>.*?) %{WORD}\\/(?P<tgtip>\\S+).*"
    ]
  }
}

Uma instrução de extração sem padrões Grok pode ter melhor desempenho. Por exemplo, o exemplo a seguir leva menos da metade das etapas de processamento para corresponder. Para uma fonte de registro de volume potencialmente alta, essa é uma consideração importante.

Entender as diferenças entre as expressões regulares RE2 e PCRE

Os analisadores das Operações de segurança do Google usam o RE2 como mecanismo de expressão regular. Se você conhece a sintaxe do PCRE, pode perceber diferenças. Confira um exemplo abaixo:

Confira a seguir uma instrução PCRE: (?<_custom_field>\w+)\s

Confira a seguir uma instrução RE2 para o código do analisador: (?P<_custom_field>\\w+)\\s

Use o escape de caracteres

O Google Security Operations armazena os dados de registro brutos recebidos no formato JSON codificado. Isso é para garantir que strings de caracteres que pareçam ser abreviações de expressões regulares sejam interpretadas como a string literal. Por exemplo, \t é interpretado como a string literal, e não como um caractere de tabulação.

O exemplo a seguir é um registro bruto original e o registro de formato codificado em JSON. Observe o caractere de escape adicionado antes de cada caractere de barra invertida que envolve o termo entry.

Confira a seguir o registro bruto original:

field=\entry\

Confira a seguir o registro convertido para o formato codificado em JSON:

field=\\entry\\

Ao usar uma expressão regular no código do analisador, é necessário adicionar outros caracteres de escape se você quiser extrair apenas o valor. Para corresponder a uma barra invertida no registro bruto original, use quatro barras invertidas na instrução de extração.

Confira a seguir uma expressão regular para o código do analisador:

^field=\\\\(?P<_value>.*)\\\\$

Confira o resultado gerado abaixo. O grupo com o nome _value armazena o termo entry:

"_value": "entry"

Ao mover uma instrução de expressão regular padrão para o código do analisador, use a barra invertida para caracteres de abreviação de expressão regular na instrução de extração. Por exemplo, mude \s para \\s.

Deixe os caracteres especiais da expressão regular inalterados quando forem usados dois caracteres de escape na instrução de extração. Por exemplo, \\ permanece inalterado como \\.

Confira a seguir uma expressão regular padrão:

^.*?\\\"(?P<_user>[^\\]+)\\\"\s(?:(logged\son|logged\soff))\s.*?\\\"(?P<_device>[^\\]+)\\\"\.$

A expressão regular a seguir é modificada para funcionar no código do analisador.

^.*?\\\"(?P<_user>[^\\\\]+)\\\"\\s(?:(logged\\son|logged\\soff))\\s.*?\\\"(?P<_device>[^\\\\]+)\\\"\\.$

A tabela a seguir resume quando uma expressão regular padrão precisa incluir outros caracteres de escape antes de ser incluída no código do analisador.

Expressão regular	Expressão regular modificada para o código do analisador	Descrição da mudança
\s	\\s	Os caracteres abreviados precisam ter escape.
\.	\\.	Os caracteres reservados precisam ter escape.
\\"	\\\"	Os caracteres reservados precisam ter escape.
\]	\\]	Os caracteres reservados precisam ter escape.
\|	\\|	Os caracteres reservados precisam ter escape.
[^\\]+	[^\\\\]+	Os caracteres especiais em um grupo de classe de caracteres precisam ter códigos de escape.
\\\\	\\\\	Caracteres especiais fora de um grupo de classe de caracteres ou caracteres abreviados não exigem um escape extra.

As expressões regulares precisam incluir um grupo de captura nomeado

Uma expressão regular, como "^.*$", é uma sintaxe RE2 válida. No entanto, no código do analisador, ele falha com o seguinte erro:

"ParseLogEntry failed: pipeline failed: filter grok (0) failed: failed to parse data with all match
patterns"

É necessário adicionar um grupo de captura válido à expressão. Se você usar padrões Grok, eles vão incluir um grupo de captura nomeado por padrão. Ao usar substituições de expressão regular, inclua um grupo nomeado.

Confira a seguir um exemplo de expressão regular no código do analisador:

"^(?P<_catchall>.*$)"

O resultado a seguir mostra o texto atribuído ao grupo chamado _catchall.

"_catchall": "User \"BOB\" logged on to workstation \"DESKTOP-01\"."

Use um grupo com nome genérico para começar a criar a expressão

Ao criar uma instrução de extração, comece com uma expressão que capture mais do que você quer. Em seguida, abra a expressão um campo por vez.

O exemplo a seguir começa usando um grupo nomeado (_catchall) que corresponde a toda a mensagem. Em seguida, ele cria a expressão em etapas, correspondendo a outras partes do texto. Em cada etapa, o grupo chamado _catchall contém menos do texto original. Continue e itere uma etapa por vez para corresponder à mensagem até não precisar mais do grupo com o nome _catchall.

Etapa	Expressão regular no código do analisador	Saída do grupo de captura com o nome _catchall
1	"^(?P<_catchall>.*$)"	User \"BOB\" logged on to workstation \"DESKTOP-01\".
2	^User\s\\\"(?P<_catchall>.*$)	BOB\" logged on to workstation \"DESKTOP-01\".
3	^User\s\\\"(?P<_user>.*?)\\\"\s(?P<_catchall>.*$)	logged on to workstation \"DESKTOP-01\".
Continue até que a expressão corresponda a toda a string de texto.

Fazer o escape de caracteres abreviados na expressão regular

Lembre-se de usar o escape de caracteres de abreviação de expressões regulares ao usar a expressão no código do analisador. Confira abaixo um exemplo de string de texto e a expressão regular padrão que extrai a primeira palavra, This.

  This is a sample log.

A expressão regular padrão a seguir extrai a primeira palavra, This. No entanto, quando você executa essa expressão no código do analisador, o resultado não tem a letra s.

Expressão regular padrão	Saída do grupo de captura com o nome _firstWord
"^(?P<_firstWord>[^\s]+)\s.*$"	"_firstWord": "Thi",

Isso ocorre porque as expressões regulares no código do analisador exigem um caractere de escape adicional adicionado aos caracteres abreviados. No exemplo anterior, \s precisa ser alterado para \\s.

Expressão regular revisada para o código do analisador	Saída do grupo de captura com o nome _firstWord
"^(?P<_firstWord>[^\\s]+)\\s.*$"	"_firstWord": "This",

Isso se aplica apenas aos caracteres abreviados, como \s, \r e \t. Outros caracteres, como ``, não precisam de escape.

Um exemplo completo

Esta seção descreve as regras anteriores como um exemplo completo. Aqui está uma string de texto não estruturada e a expressão regular padrão gravada para analisar a string. Por fim, ele inclui a expressão regular modificada que funciona no código do analisador.

Confira a seguir a string de texto original.

User "BOB" logged on to workstation "DESKTOP-01".

A seguir, há uma expressão regular RE2 padrão que analisa a string de texto.

^.*?\\\"(?P<_user>[^\\]+)\\\"\s(?:(logged\son|logged\soff))\s.*?\\\"(?P<_device>[^\\]+)\\\"\.$

Essa expressão extrai os seguintes campos.

Grupo de correspondência	Posição do caractere	String de texto
Correspondência total	0-53	User \"BOB\" logged on to workstation \"DESKTOP-01\".
Grupo "_user"	7-10	BOB
Grupo 2.	13-22	logged on
Grupo "_device"	40-50	DESKTOP-01

Essa é a expressão modificada. A expressão regular padrão RE2 foi modificada para funcionar no código do analisador.

^.*?\\\"(?P<_user>[^\\\\]+)\\\"\\s(?:(logged\\son|logged\\soff))\\s.*?\\\"(?P<_device>[^\\\\]+)\\\"\\.$

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.