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

Resolución

Puedes usar el editor de consultas de la consola Athena para crear una base de datos, crear una tabla para los registros de flujo de la VPC y ejecutar consultas de ejemplo. A continuación, utiliza los registros de flujo para investigar los patrones de tráfico de la red e identificar las amenazas y los riesgos en la red de Amazon VPC.

Creación de una base de datos con el editor de consultas de la consola de Athena

Abre el editor de consultas de la consola de Athena y ejecuta el siguiente comando:

CREATE DATABASE test_db_vpclogs;

Nota: Sustituye test_db_vpclogs por el nombre que desees usar para la base de datos.

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.

Creación de una tabla para los registros de flujo en la base de datos

Abre la consola de Athena. A continuación, en el panel de navegación, selecciona Query editor (Editor de consultas). En el editor de consultas de la consola de Athena, ejecuta un comando similar al siguiente ejemplo:

CREATE EXTERNAL TABLE IF NOT EXISTS 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, date string, hour string)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/'
TBLPROPERTIES (
"EXTERNAL"="true",
"skip.header.line.count" = "1",
"projection.enabled" = "true",
"projection.region.type" = "enum",
"projection.region.values" = "us-east-1,us-west-2,ap-south-1,eu-west-1",
"projection.date.type" = "date",
"projection.date.range" = "2021/01/01,NOW",
"projection.date.format" = "yyyy/MM/dd",
"projection.hour.type" = "integer",
"projection.hour.range" = "00,23",
"projection.hour.digits" = "2",
"storage.location.template" = "s3://amzn-s3-demo-bucket/prefix/AWSLogs/${account_id}/vpcflowlogs/${region}/${date}/${hour}"
)

Nota: Sustituye test_table_vpclogs por el nombre que desees usar para la tabla.

Modifica el parámetro LOCATION en la consulta para que señale al bucket de Amazon S3 que contiene los datos de registro. Actualiza projection.region.values a las regiones de AWS en las que tengas registros de flujo de VPC. Establece el valor de projection.date.range en los primeros registros disponibles en los datos. Por ejemplo, si los registros comienzan el 1 de enero de 2021, establece projection.date.range en '2021/01/01,NOW'. Modifica el valor de acuerdo con la fecha de inicio real de tus registros para una proyección de datos precisa. El comando anterior usa la proyección de partición 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.

Ejecución de instrucciones SQL en la tabla para los registros de flujo

Abre la consola de Athena. A continuación, en el panel de navegación, selecciona Query editor (Editor de consultas). Utiliza el editor de consultas de la consola de Amazon Athena para ejecutar instrucciones SQL en la tabla. Puedes guardar las consultas, ver consultas anteriores o descargar los resultados de las consultas en formato .csv.

Consultas de ejemplo

Nota: En las consultas de ejemplo siguientes, sustituye test_table_vpclogs por el nombre de la tabla. Modifica los valores de las columnas y otras variables para tu caso de uso.

Para ver las 100 primeras entradas del registro de flujo en orden cronológico durante un periodo de tiempo especificado, ejecuta la siguiente consulta:

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

Para ver los 10 servidores de destino que reciben la mayor cantidad de paquetes HTTPS durante un periodo de tiempo específico, ejecuta la siguiente consulta:

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;

Para comprobar los registros que se crean para un intervalo de tiempo específico, ejecuta la siguiente consulta:

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, ejecuta la siguiente consulta:

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, ejecuta la siguiente consulta:

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 intervalo de direcciones IP que comienza por 10.117, ejecuta la siguiente consulta:

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, ejecuta la siguiente consulta:

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

Información relacionada

How do I monitor traffic in my VPC with flow logs? (¿Cómo puedo supervisar el tráfico en mi VPC con registros de flujo?)

Analyzing VPC Flow Logs using Amazon Athena and Amazon QuickSight (Análisis de los registros de flujo de VPC mediante Amazon Athena y Amazon QuickSight)