Descripción general de los analizadores
En este documento, se proporciona una descripción general de cómo Google Security Operations analiza los registros sin procesar en el formato de modelo de datos unificado (UDM).
Google SecOps puede recibir datos de registro que provienen de las siguientes fuentes de transferencia:
- Reenviador de Google SecOps
- Feed de la API de Chronicle
- API de transferencia de Chronicle
- Socio tecnológico externo
En general, los clientes envían datos como registros sin procesar originales. Google SecOps identifica de forma única el dispositivo que generó los registros con LogType. LogType identifica lo siguiente:
- El proveedor y el dispositivo que generaron el registro, como Cisco Firewall, Linux DHCP Server o Bro DNS
- Qué analizador convierte el registro sin procesar en UDM estructurado Existe una relación de uno a uno entre un analizador y un LogType. Cada analizador convierte los datos recibidos por un solo LogType.
Google SecOps proporciona un conjunto de analizadores predeterminados que leen registros sin procesar originales y generan registros UDM estructurados con los datos del registro sin procesar original. Google SecOps mantiene estos analizadores. Los clientes también pueden definir instrucciones de asignación de datos personalizadas mediante la creación de un analizador específico para el cliente. Si envías una carga útil de varias líneas, el sistema interpreta cada línea como una entrada de registro independiente.
El analizador contiene instrucciones de asignación de datos. Define cómo se asignan los datos del registro sin procesar original a uno o más campos en la estructura de datos UDM.
Si no hay errores de análisis, Google SecOps crea un registro estructurado de UDM con los datos del registro sin procesar. El proceso de conversión de un registro sin procesar en un registro UDM se denomina normalización.
Un analizador predeterminado puede asignar un subconjunto de valores principales del registro sin procesar. Por lo general, estos campos principales son los más importantes para proporcionar estadísticas de seguridad en Google SecOps. Los valores no asignados permanecen en el registro sin procesar, pero no se almacenan en el registro UDM.
Un cliente también puede usar la API de transferencia, para enviar datos en formato UDM estructurado.
Personaliza cómo se analizan los datos transferidos
Google SecOps proporciona las siguientes capacidades que permiten a los clientes personalizar el análisis de datos en los datos de registro originales entrantes. Para obtener más detalles sobre la administración de estos analizadores, consulta Administra analizadores precompilados y personalizados.
- Analizadores específicos para el cliente: Los clientes crean una configuración de analizador personalizado para un tipo de registro específico que cumpla con sus requisitos específicos. Un analizador específico para el cliente reemplaza al analizador predeterminado para el LogType específico. Para obtener más detalles, consulta Administra analizadores precompilados y personalizados.
- Extensiones de analizador: Los clientes pueden agregar instrucciones de asignación personalizadas, además de la configuración del analizador predeterminado. Cada cliente puede crear su propio conjunto único de instrucciones de asignación personalizadas. Estas instrucciones de asignación definen cómo extraer y transformar campos adicionales de registros sin procesar originales a campos UDM. Una extensión de analizador no reemplaza al analizador predeterminado ni al analizador específico para el cliente.
Un ejemplo con un registro de proxy web de Squid
En esta sección, se proporciona un ejemplo de registro de proxy web de Squid y se describe cómo se asignan los valores a un registro UDM. Para obtener una descripción de todos los campos del esquema UDM, consulta Lista de campos del modelo de datos unificado.
El registro de proxy web de Squid de ejemplo contiene valores separados por espacios. Cada registro representa un evento y almacena los siguientes datos: marca de tiempo, duración, cliente, código de resultado/estado del resultado, bytes transmitidos, método de solicitud, URL, usuario, código de jerarquía y tipo de contenido. En este ejemplo, se extraen y asignan los siguientes campos a un registro UDM: hora, cliente, estado del resultado, bytes, método de solicitud y URL.
1588059648.129 23 192.168.23.4 TCP_HIT/200 904 GET www.google.com/images/sunlogo.png - HIER_DIRECT/203.0.113.52 image/jpeg
A medida que comparas estas estructuras, observa que solo un subconjunto de los datos de registro originales se incluye en el registro UDM. Algunos campos son obligatorios y otros son opcionales. Además, solo un subconjunto de las secciones del registro UDM contiene datos. Si el analizador no asigna datos del registro original al registro UDM, no verás esa sección del registro UDM en Google SecOps.
La sección metadata almacena la marca de tiempo del evento. Observa que el valor se convirtió del formato Epoch al formato RFC 3339. Esta conversión es opcional. La marca de tiempo se puede almacenar como formato Epoch, con un procesamiento previo para separar las partes de segundos y milisegundos en campos independientes.
El campo metadata.event_type
almacena el valor NETWORK_HTTP, que es un valor enumerado
que identifica el tipo de evento. El valor de metadata.event_type determina qué campos UDM adicionales son obligatorios en comparación con los opcionales. Los valores product_name y vendor_name contienen descripciones fáciles de usar del dispositivo que registró el registro original.
El metadata.event_type en un registro de evento UDM no es el mismo que el log_type definido cuando se transfieren datos con la API de transferencia. Estos dos atributos almacenan información diferente.
La sección network contiene valores del evento de registro original. Observa en este ejemplo que el valor de estado del registro original se analizó desde el campo "código/estado del resultado" antes de escribirse en el registro UDM. Solo se incluyó el result_code en el registro UDM.
La sección principal almacena la información del cliente del registro original. La sección target almacena la URL completamente calificada y la dirección IP.
La sección security_result almacena uno de los valores enum para representar la acción que se registró en el registro original.
Este es el registro UDM con formato JSON. Observa que solo se incluyen las secciones que contienen datos. No se incluyen las secciones src, observer, intermediary, about y extensions.
{
"metadata": {
"event_timestamp": "2020-04-28T07:40:48.129Z",
"event_type": "NETWORK_HTTP",
"product_name": "Squid Proxy",
"vendor_name": "Squid"
},
"principal": {
"ip": "192.168.23.4"
},
"target": {
"url": "www.google.com/images/sunlogo.png",
"ip": "203.0.113.52"
},
"network": {
"http": {
"method": "GET",
"response_code": 200,
"received_bytes": 904
}
},
"security_result": {
"action": "UNKNOWN_ACTION"
}
}
Pasos dentro de las instrucciones del analizador
Las instrucciones de asignación de datos dentro de un analizador siguen un patrón común, como se indica a continuación:
- Analiza y extrae datos del registro original.
- Manipula los datos extraídos. Esto incluye el uso de lógica condicional para analizar valores de forma selectiva, convertir tipos de datos, reemplazar subcadenas en un valor, convertir a mayúsculas o minúsculas, etcétera.
- Asigna valores a los campos UDM.
- Envía el registro UDM asignado a la clave @output.
Analiza y extrae datos del registro original
Establece la instrucción de filtro
La instrucción filter es la primera instrucción del conjunto de instrucciones de análisis.
Todas las instrucciones de análisis adicionales se encuentran dentro de la instrucción filter.
filter {
}
Inicializa las variables que almacenarán los valores extraídos
Dentro de la instrucción filter, inicializa las variables intermedias que el analizador usará para almacenar los valores extraídos del registro.
Estas variables se usan cada vez que se analiza un registro individual. El valor de cada variable intermedia se establecerá en uno o más campos UDM más adelante en las instrucciones de análisis.
mutate {
replace => {
"event.idm.read_only_udm.metadata.product_name" => "Webproxy"
"event.idm.read_only_udm.metadata.vendor_name" => "Squid"
"not_valid_log" => "false"
"when" => ""
"srcip" => ""
"action" => ""
"username" => ""
"url" => ""
"tgtip" => ""
"method" => ""
}
}
Extrae valores individuales del registro
Google SecOps proporciona un conjunto de filtros, basados en Logstash, para extraer campos de archivos de registro originales. Según el formato del registro, usas uno o varios filtros de extracción para extraer todos los datos del registro. Si la cadena es la siguiente:
- JSON nativo, la sintaxis del analizador es similar al filtro JSON que admite registros con formato JSON. No se admite JSON anidado.
- Formato XML, la sintaxis del analizador es similar al filtro XML que admite registros con formato XML.
- Pares clave-valor, la sintaxis del analizador es similar al filtro Kv que admite mensajes con formato clave-valor.
- Formato CSV, la sintaxis del analizador es similar al filtro Csv que admite mensajes con formato CSV.
- Todos los demás formatos, la sintaxis del analizador es similar al filtro GROK con patrones integrados de GROK . Esto usa instrucciones de extracción de estilo Regex.
Google SecOps proporciona un subconjunto de las capacidades disponibles en cada filtro. Google SecOps también proporciona una sintaxis de asignación de datos personalizada que no está disponible en los filtros. Consulta la referencia de sintaxis del analizador para obtener una descripción de las funciones compatibles y las funciones personalizadas.
Continuando con el ejemplo de registro de proxy web de Squid, la siguiente instrucción de extracción de datos incluye una combinación de sintaxis Grok de Logstash y expresiones regulares.
La siguiente instrucción de extracción almacena valores en las siguientes variables intermedias:
whensrcipactionreturnCodesizemethodusernameurltgtip
Esta instrucción de ejemplo también usa la palabra clave overwrite para almacenar los valores extraídos en cada variable. Si el proceso de extracción muestra un error, la instrucción on_error establece not_valid_log en true.
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+).*"
]
}
overwrite => ["when","srcip","action","returnCode","size","method","url","username","tgtip"]
on_error => "not_valid_log"
}
Manipula y transforma los valores extraídos
Google SecOps aprovecha las capacidades del complemento de filtro mutate de Logstash para permitir la manipulación de los valores extraídos del registro original. Google SecOps proporciona un subconjunto de las capacidades disponibles en el complemento. Consulta la sintaxis del analizador para obtener una descripción de las funciones compatibles y las funciones personalizadas, como las siguientes:
- Convertir valores a un tipo de datos diferente
- Reemplazar valores en la cadena
- Combinar dos arrays o agregar una cadena a un array (los valores de cadena se convierten en un array antes de la combinación)
- Convertir a minúsculas o mayúsculas
En esta sección, se proporcionan ejemplos de transformación de datos que se basan en el registro de proxy web de Squid presentado anteriormente.
Transforma la marca de tiempo del evento
Todos los eventos almacenados como un registro UDM deben tener una marca de tiempo del evento. En este ejemplo, se verifica si se extrajo un valor para los datos del registro. Luego, usa la
función de fecha
Grok para hacer coincidir el valor con el formato de hora UNIX.
if [when] != "" {
date {
match => [
"when", "UNIX"
]
}
}
Transforma el valor username
En la siguiente instrucción de ejemplo, se convierte el valor de la variable username a minúsculas.
mutate {
lowercase => [ "username"]
}
Transforma el valor action
En el siguiente ejemplo, se evalúa el valor de la variable intermedia action y se cambia el valor a ALLOW, BLOCK o UNKNOWN_ACTION, que son
valores válidos para el campo UDM security_result.action. El campo UDM security_result.action es un tipo enumerado que almacena solo valores específicos.
if ([action] == "TCP_DENIED" or [action] == "TCP_MISS" or [action] == "Denied" or [action] == "denied" or [action] == "Dropped") {
mutate {
replace => {
"action" => "BLOCK"
}
}
} else if ([action] == "TCP_TUNNEL" or [action] == "Accessed" or [action] == "Built" or [action] == "Retrieved" or [action] == "Stored") {
mutate {
replace => {
"action" => "ALLOW"
}
}
} else {
mutate {
replace => {
"action" => "UNKNOWN_ACTION" }
}
}
Transforma la dirección IP de destino
En el siguiente ejemplo, se busca un valor en la variable intermedia tgtip.
Si se encuentra, el valor coincide con un patrón de dirección IP mediante un patrón Grok predefinido. Si hay un error que hace coincidir el valor con un patrón de dirección IP, la función on_error establece la propiedad not_valid_tgtip en true. Si la coincidencia es exitosa, no se establece la propiedad not_valid_tgtip.
if [tgtip] not in [ "","-" ] {
grok {
match => {
"tgtip" => [ "%{IP:tgtip}" ]
}
overwrite => ["tgtip"]
on_error => "not_valid_tgtip"
}
Cambia el tipo de datos de returnCode y size
En el siguiente ejemplo, se convierte el valor de la variable size a uinteger y el valor de la variable returnCode a uinteger. Esto es obligatorio porque la variable size se guardará en el campo UDM network.received_bytes, que almacena un tipo de datos int64. La variable returnCode se guardará en el campo UDM network.http.response_code, que almacena un tipo de datos int32.
mutate {
convert => {
"returnCode" => "integer"
"size" => "uinteger"
}
}
Asigna valores a los campos UDM en un evento
Después de extraer y procesar previamente los valores, asígnalos a los campos de un registro de evento UDM. Puedes asignar valores extraídos y valores estáticos a un campo UDM.
Si propagas event.disambiguation_key, asegúrate de que este campo sea único para cada evento que se genere para el registro determinado. Si dos eventos diferentes tienen la misma disambiguation_key, esto generará un comportamiento inesperado en el sistema.
Los ejemplos de analizadores de esta sección se basan en el ejemplo anterior de registro de proxy web de Squid.
Guarda la marca de tiempo del evento
Cada registro de evento UDM debe tener un valor establecido para el campo UDM metadata.event_timestamp. En el siguiente ejemplo, se guarda la marca de tiempo del evento extraída del registro en la variable integrada @timestamp. Google Security Operations guarda esto en el campo UDM metadata.event_timestamp de forma predeterminada.
mutate {
rename => {
"when" => "timestamp"
}
}
Establece el tipo de evento
Cada registro de evento UDM debe tener un valor establecido para el campo UDM metadata.event_type. Este campo es un tipo enumerado. El valor de este campo determina qué campos UDM adicionales se deben propagar para que se guarde el registro UDM.
El proceso de análisis y normalización fallará si alguno de los campos obligatorios no contiene datos válidos.
replace => {
"event.idm.read_only_udm.metadata.event_type" => "NETWORK_HTTP"
}
}
Guarda los valores username y method con la instrucción replace
Los valores de los campos intermedios username y method son cadenas. En el siguiente ejemplo, se verifica si existe un valor válido y, si es así, se almacena el valor username en el campo UDM principal.user.userid y el valor method en el campo UDM network.http.method.
if [username] not in [ "-" ,"" ] {
mutate {
replace => {
"event.idm.read_only_udm.principal.user.userid" => "%{username}"
}
}
}
if [method] != "" {
mutate {
replace => {
"event.idm.read_only_udm.network.http.method" => "%{method}"
}
}
}
Guarda el action en el campo UDM security_result.action
En la sección anterior, se evaluó el valor de la variable intermedia action y se transformó en uno de los valores estándar para el campo UDM security_result.action.
Los campos UDM security_result y action almacenan un array de elementos, lo que significa que debes seguir un enfoque ligeramente diferente cuando guardes este valor.
Primero, guarda el valor transformado en un campo security_result.action intermedio. El campo security_result es un elemento superior del campo action.
mutate {
merge => {
"security_result.action" => "action"
}
}
Luego, guarda el campo intermedio security_result.action en el
security_result campo UDM. El campo UDM security_result almacena un array de elementos, por lo que el valor se agrega a este campo.
# save the security_result field
mutate {
merge => {
"event.idm.read_only_udm.security_result" => "security_result"
}
}
Almacena la dirección IP de destino y la dirección IP de origen con la instrucción merge
Almacena los siguientes valores en el registro de evento UDM:
- Valor en la variable intermedia
srcipen el campo UDMprincipal.ip - Valor en la variable intermedia
tgtipen el campo UDMtarget.ip
Los campos UDM principal.ip y target.ip almacenan un array de elementos, por lo que los valores se agregan a cada campo.
En los siguientes ejemplos, se muestran diferentes enfoques para guardar estos valores.
Durante el paso de transformación, la variable intermedia tgtip coincidió con una dirección IP mediante un patrón Grok predefinido. En la siguiente instrucción de ejemplo,
se verifica si la propiedad not_valid_tgtipes verdadera, lo que indica que tgtip
el valor no pudo coincidir con un patrón de dirección IP. Si es falso, guarda el valor tgtip en el campo UDM target.ip.
if ![not_valid_tgtip] {
mutate {
merge => {
"event.idm.read_only_udm.target.ip" => "tgtip"
}
}
}
La variable intermedia srcip no se transformó. En la siguiente instrucción, se verifica si se extrajo un valor del registro original y, si es así, se guarda el valor en el campo UDM principal.ip.
if [srcip] != "" {
mutate {
merge => {
"event.idm.read_only_udm.principal.ip" => "srcip"
}
}
}
Guarda url, returnCode y size con la instrucción rename
En la siguiente instrucción de ejemplo, se almacenan los valores con la instrucción rename:
- La variable
urlguardada en el campo UDMtarget.url - La variable intermedia
returnCodeguardada en el campo UDMnetwork.http.response_code - La variable intermedia
sizeguardada en el campo UDMnetwork.received_bytes
mutate {
rename => {
"url" => "event.idm.read_only_udm.target.url"
"returnCode" => "event.idm.read_only_udm.network.http.response_code"
"size" => "event.idm.read_only_udm.network.received_bytes"
}
}
Vincula el registro UDM al resultado
La instrucción final de la instrucción de asignación de datos envía los datos procesados a un registro de evento UDM.
mutate {
merge => {
"@output" => "event"
}
}
El código completo del analizador
Este es el ejemplo de código completo del analizador. El orden de las instrucciones no sigue el mismo orden que las secciones anteriores de este documento, pero genera el mismo resultado.
filter {
# initialize variables
mutate {
replace => {
"event.idm.read_only_udm.metadata.product_name" => "Webproxy"
"event.idm.read_only_udm.metadata.vendor_name" => "Squid"
"not_valid_log" => "false"
"when" => ""
"srcip" => ""
"action" => ""
"username" => ""
"url" => ""
"tgtip" => ""
"method" => ""
}
}
# Extract fields from the raw log.
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+).*"
]
}
overwrite => ["when","srcip","action","returnCode","size","method","url","username","tgtip"]
on_error => "not_valid_log"
}
# Parse event timestamp
if [when] != "" {
date {
match => [
"when", "UNIX"
]
}
}
# Save the value in "when" to the event timestamp
mutate {
rename => {
"when" => "timestamp"
}
}
# Transform and save username
if [username] not in [ "-" ,"" ] {
mutate {
lowercase => [ "username"]
}
}
mutate {
replace => {
"event.idm.read_only_udm.principal.user.userid" => "%{username}"
}
}
if ([action] == "TCP_DENIED" or [action] == "TCP_MISS" or [action] == "Denied" or [action] == "denied" or [action] == "Dropped") {
mutate {
replace => {
"action" => "BLOCK"
}
}
} else if ([action] == "TCP_TUNNEL" or [action] == "Accessed" or [action] == "Built" or [action] == "Retrieved" or [action] == "Stored") {
mutate {
replace => {
"action" => "ALLOW"
}
}
} else {
mutate {
replace => {
"action" => "UNKNOWN_ACTION" }
}
}
# save transformed value to an intermediary field
mutate {
merge => {
"security_result.action" => "action"
}
}
# save the security_result field
mutate {
merge => {
"event.idm.read_only_udm.security_result" => "security_result"
}
}
# check for presence of target ip. Extract and store target IP address.
if [tgtip] not in [ "","-" ] {
grok {
match => {
"tgtip" => [ "%{IP:tgtip}" ]
}
overwrite => ["tgtip"]
on_error => "not_valid_tgtip"
}
# store target IP address
if ![not_valid_tgtip] {
mutate {
merge => {
"event.idm.read_only_udm.target.ip" => "tgtip"
}
}
}
}
# convert the returnCode and size to integer data type
mutate {
convert => {
"returnCode" => "integer"
"size" => "uinteger"
}
}
# save url, returnCode, and size
mutate {
rename => {
"url" => "event.idm.read_only_udm.target.url"
"returnCode" => "event.idm.read_only_udm.network.http.response_code"
"size" => "event.idm.read_only_udm.network.received_bytes"
}
# set the event type to NETWORK_HTTP
replace => {
"event.idm.read_only_udm.metadata.event_type" => "NETWORK_HTTP"
}
}
# validate and set source IP address
if [srcip] != "" {
mutate {
merge => {
"event.idm.read_only_udm.principal.ip" => "srcip"
}
}
}
# save event to @output
mutate {
merge => {
"@output" => "event"
}
}
} #end of filter
¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.