¿Cómo puedo usar Amazon Athena para analizar los registros de flujo de VPC?

Breve descripción

Utilice los Registros de flujo de Amazon VPC para investigar los patrones de tráfico de la red e identificar las amenazas y los riesgos en la red de Amazon VPC.

Resolución

Utilice Athena para analizar los registros de flujo de Amazon VPC

Para usar Athena para analizar los registros de flujo de Amazon VPC, siga estos pasos:

1. Utilice el editor de consultas de la consola Amazon Athena para ejecutar el siguiente comando y crear una base de datos. Sustituya test_db_vpclogs por un nombre para su base de datos:

   CREATE DATABASE test_db_vpclogs;
   

   Importante: Se recomienda crear la base de datos en la misma región de AWS que el bucket de Amazon Simple Storage Service (Amazon S3) en el que se almacenan los registros de flujo.

2. En la base de datos, ejecute el siguiente comando para crear una tabla para los registros de flujo de VPC. Sustituya test_table_vpclogs por el nombre de su tabla. Modifique el parámetro LOCATION en la consulta para que señale al bucket de Amazon S3 que contiene los datos de su registro:

   CREATE EXTERNAL TABLE `test_table_vpclogs`(
     `version` int,
     `account_id` string,
     `interface_id` string,
     `srcaddr` string,
     `dstaddr` string,
     `srcport` int,
     `dstport` int,
     `protocol` bigint,
     `packets` bigint,
     `bytes` bigint,
     `start` bigint,
     `end` bigint,
     `action` string,
     `log_status` string,
     `vpc_id` string,
     `subnet_id` string,
     `instance_id` string,
     `tcp_flags` int,
     `type` string,
     `pkt_srcaddr` string,
     `pkt_dstaddr` string,
     `az_id` string,
     `sublocation_type` string,
     `sublocation_id` string,
     `pkt_src_aws_service` string,
     `pkt_dst_aws_service` string,
     `flow_direction` string,
     `traffic_path` int)
   PARTITIONED BY (
     `region` string,
     `day` string)
   ROW FORMAT DELIMITED
     FIELDS TERMINATED BY ' '
   STORED AS INPUTFORMAT
     'org.apache.hadoop.mapred.TextInputFormat'
   OUTPUTFORMAT
     'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
   LOCATION
     's3://awsexamplebucket/awsexampleprefix/awsexamplelogs/1111222233334444/vpcflowlogs/'
   TBLPROPERTIES (
     'projection.day.format'='yyyy/MM/dd',
     'projection.day.range'='2021/01/01,NOW',
     'projection.day.type'='date',
     'projection.enabled'='true',
     'projection.region.type'='enum',
     'projection.region.values'='us-east-1,us-west-2,ap-south-1,eu-west-1',
     'skip.header.line.count'='1',
     'storage.location.template'='s3://awsexamplebucket/awsexampleprefix/awsexamplelogs/1111222233334444/vpcflowlogs/${region}/${day}'
     )

   Nota: El comando anterior usa la proyección de particiones para crear una tabla, particionarla y rellenar automáticamente las particiones. Si no existe ninguna partición proyectada en Amazon S3, Athena proyectará la partición de todos modos. Se recomienda utilizar atributos con particiones en las consultas.

3. Utilice la consola de Amazon Athena para ejecutar instrucciones SQL en la tabla. Puede guardar las consultas, ver consultas anteriores o descargar los resultados de las consultas en formato CSV.

Consultas de ejemplo

Nota: Sustituya test_table_vpclogs por el nombre de su tabla. Modifique los valores de las columnas y otras variables para que se ajusten a su consulta.

Para ver las 100 primeras entradas del registro de flujo en orden cronológico durante un período de tiempo especificado, ejecute una consulta similar a la siguiente:

SELECT *
 FROM test_table_vpclogs
 WHERE day >= '2021/02/01' AND day < '2021/02/28'
 ORDER BY day ASC
 LIMIT 100;

Para ver el servidor que recibe los diez primeros paquetes HTTP durante un período de tiempo especificado, ejecute una consulta similar a la siguiente:

SELECT SUM(packets) AS packetcount,
       dstaddr
FROM test_table_vpclogs
WHERE dstport = 443
  AND day >= '2021/03/01'
  AND day < '2021/03/31'
GROUP BY dstaddr
ORDER BY packetcount DESC
LIMIT 10;

Esta anterior consulta cuenta el número de paquetes que se reciben en el puerto HTTPS 443 y los agrupa por dirección IP de destino. A continuación, devuelve las 10 entradas principales de la semana anterior.

Para comprobar los registros que se han creado en un intervalo de tiempo determinado, ejecute una consulta similar a la siguiente:

SELECT interface_id,
       srcaddr,
       action,
       protocol,
       to_iso8601(from_unixtime(start)) AS start_time,
       to_iso8601(from_unixtime("end")) AS end_time
FROM test_table_vpclogs
WHERE DAY >= '2021/04/01'
  AND DAY < '2021/04/30';

Para ver los registros de flujo de una dirección IP de origen concreta entre un intervalo de tiempo especificado, ejecute una consulta similar a la siguiente:

SELECT *
FROM test_table_vpclogs
WHERE srcaddr = '10.117.1.22'
  AND day >= '2021/02/01'
  AND day < '2021/02/28';

Para mostrar un listado con las conexiones TCP rechazadas entre un intervalo de tiempo especificado, ejecute una consulta similar a la siguiente:

SELECT day,
       interface_id,
       srcaddr,
       action,
       protocol
FROM test_table_vpclogs
WHERE action = 'REJECT'
    AND protocol = 6
    AND day >= '2021/02/01' AND day < '2021/02/28'
LIMIT 10;

Para ver los registros de flujo del rango de direcciones IP que comienza por «10.117», ejecute una consulta similar a la siguiente:

SELECT *
FROM test_table_vpclogs
WHERE split_part(srcaddr,'.', 1)='10'
  AND split_part(srcaddr,'.', 2) ='117'

Para ver los registros de flujo de una dirección IP de destino concreta dentro de un intervalo de tiempo, ejecute una consulta similar a la siguiente:

SELECT *
FROM test_table_vpclogs
WHERE dstaddr = '10.0.1.14'
AND day >= '2021/01/01'
AND day < '2021/01/31'

Información relacionada

¿Cómo puedo supervisar el tráfico en mi VPC con registros de flujo?

Análisis de los registros de flujo de VPC mediante Amazon Athena y Amazon QuickSight