Comment utiliser Athena pour analyser les journaux de flux Amazon VPC ?

Brève description

Vous pouvez utiliser l'éditeur de requête de la console Athena pour créer une base de données, créer une table pour les journaux de flux VPC et exécuter des exemples de requêtes. Utilisez les journaux de flux pour analyser les modèles de trafic réseau et identifier les menaces et les risques qui pèsent sur votre réseau Amazon VPC.

Résolution

Créer une base de données avec l'éditeur de requête de la console Athena

Procédez comme suit :

1. Ouvrez l'éditeur de requête de la console Athena.
2. Exécutez la commande suivante :

   CREATE DATABASE test_db_vpclogs;

   Remarque : Remplacez test_db_vpclogs par le nom de votre base de données.

Important : Il est recommandé de créer la base de données dans la même région AWS que le compartiment Amazon Simple Storage Service (Amazon S3) dans lequel les journaux de flux sont stockés.

Créer une table pour les journaux de flux dans la base de données

Procédez comme suit :

1. Dans l'éditeur de requête de la console Athena, exécutez la commande suivante :

   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}')

   Remarque : Remplacez test_table_vpclogs par le nom de votre table.

Modifiez le paramètre EMPLACEMENT afin qu’il pointe vers le compartiment Amazon S3 qui contient vos données de journal. Définissez la première valeur de projection.log_time.range en fonction des premiers journaux disponibles dans vos données. Par exemple, si vos journaux commencent le 1er mars 2023, définissez la plage de temps sur '2025/03/01,NOW'. Modifiez la valeur en fonction de la date de début réelle de vos journaux pour une projection précise des données. La commande précédente utilise la projection de partition pour créer une table, la partitionner et remplir automatiquement les partitions. Si aucune partition projetée n’existe dans Amazon S3, Athena projette toujours la partition. Il est recommandé d’utiliser des attributs partitionnés dans vos requêtes.

Exécuter des instructions SQL sur la table pour les journaux de flux

Utilisez l’éditeur de requête de la console Athena pour exécuter des instructions SQL sur la table. Vous pouvez enregistrer les requêtes, afficher les requêtes précédentes ou télécharger les résultats des requêtes au format .csv.

Exemples de requêtes

Remarque : Dans les exemples de requêtes suivants, remplacez test_table_vpclogs par le nom de votre table. Modifiez les valeurs des colonnes et les autres variables en fonction de votre cas d'utilisation.

Pour afficher les 100 premières entrées du journal de flux dans l’ordre chronologique pour une période donnée, exécutez la requête suivante :

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

Pour afficher le serveur qui reçoit le plus grand nombre de paquets HTTP pendant une période donnée, exécutez la requête suivante :

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;

Remarque : La requête précédente compte le nombre de paquets reçus sur le port HTTPS 443 et les regroupe par adresse IP de destination. Puis, la requête renvoie les 10 principales entrées de la semaine précédente.

Pour vérifier les journaux créés pour une période donnée, exécutez la requête suivante :

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';

Pour afficher les journaux de flux pour une adresse IP source spécifique entre une plage de temps spécifiée, exécutez la requête suivante :

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

Pour répertorier les connexions TCP rejetées pendant une période spécifiée, exécutez la requête suivante :

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;

Pour afficher les journaux de flux pour la plage d’adresses IP commençant par 10.117, exécutez la requête suivante :

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

Pour afficher les journaux de flux pour une adresse IP de destination spécifique sur une période donnée, exécutez la requête suivante :

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

Informations connexes

Comment puis-je surveiller le trafic dans mon VPC à l’aide de journaux de flux ?

Analyse des journaux de flux VPC à l’aide d’Amazon Athena et d’Amazon Quick Sight