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

Descripción corta

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. 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.

Resolución

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

Sigue estos pasos:

1. Abre el editor de consultas de la consola de Athena.
2. Ejecuta el siguiente comando:

   CREATE DATABASE test_db_vpclogs;

   Nota: Sustituye test_db_vpclogs por el nombre de tu 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

Sigue estos pasos:

1. En el editor de consultas de la consola de Athena, ejecuta el siguiente comando:

   CREATE EXTERNAL TABLE `vpcflow_logs`(  
    `version` int,  
    `resource_type` string,  
    `account_id` string,  
    `tgw_id` string,  
    `tgw_attachment_id` string,  
    `tgw_src_vpc_account_id` string,  
    `tgw_dst_vpc_account_id` string,  
    `tgw_src_vpc_id` string,  
    `tgw_dst_vpc_id` string,  
    `tgw_src_subnet_id` string,  
    `tgw_dst_subnet_id` string,  
    `tgw_src_eni` string,  
    `tgw_dst_eni` string,  
    `tgw_src_az_id` string,  
    `tgw_dst_az_id` string,  
    `tgw_pair_attachment_id` string,  
    `srcaddr` string,  
    `dstaddr` string,  
    `srcport` int,  
    `dstport` int,  
    `protocol` bigint,  
    `packets` bigint,  
    `bytes` bigint,  
    `start` bigint,  
    `end` bigint,  
    `log_status` string,  
    `type` string,  
    `packets_lost_no_route` bigint,  
    `packets_lost_blackhole` bigint,  
    `packets_lost_mtu_exceeded` bigint,  
    `packets_lost_ttl_expired` bigint,  
    `tcp_flags` int,  
    `region` string,  
    `flow_direction` string,  
    `pkt_src_aws_service` string,  
    `pkt_dst_aws_service` string)  
   PARTITIONED BY (  
    `aws_region` string,  
    `log_time` 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://AWS_S3_BUCKET/awsexampleprefix/AWSLogs/AWS_ACCOUNT_NUMBER/vpcflowlogs/'  
   TBLPROPERTIES (  
    'projection.aws_region.type'='enum',  
    'projection.aws_region.values'='us-east-1,us-west-2,ap-southeast-2',  
    'projection.log_time.format'='yyyy/MM/dd',  
    'projection.log_time.range'='2025/01/01,NOW',  
    'projection.log_time.type'='date',  
    'projection.enabled'='true',  
    'skip.header.line.count'='1',  
    'storage.location.template'='s3://AWS_S3_BUCKET/awsexampleprefix/AWSLogs/AWS_ACCOUNT_NUMBER/vpcflowlogs/${aws_region}/${log_time}')

   Nota: Sustituye test_table_vpclogs por el nombre de tu tabla.

Modifica el parámetro LOCATION en la consulta para que señale al bucket de Amazon S3 que contiene los datos de tu registro. Establece el primer valor de projection.log_time.range en función de los primeros registros disponibles en tus datos. Por ejemplo, si tus registros comienzan el 1 de marzo de 2023, establece el intervalo de tiempo como «2025/03/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

Utiliza 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 el servidor que recibe los diez primeros paquetes HTTP durante un periodo de tiempo especificado, 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;

Nota: 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 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

¿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