Come faccio a usare Amazon Athena per analizzare i log di flusso VPC?

Breve descrizione

Usa i log Amazon VPC Flow per analizzare i modelli di traffico di rete e identificare minacce e rischi nella tua rete Amazon VPC.

Risoluzione

Usa Athena per analizzare i log di flusso di Amazon VPC

Per utilizzare Athena per analizzare i log di flusso di Amazon VPC, completa i seguenti passaggi:

1. Usa l'editor di query della console Amazon Athena per eseguire il seguente comando per creare un database. Sostituisci test_db_vpclogs con un nome per il tuo database:

   CREATE DATABASE test_db_vpclogs;
   

   Importante: è consigliabile creare il database nella stessa regione AWS del bucket Amazon Simple Storage Service (Amazon S3) in cui sono archiviati i log di flusso.

2. Nel database, esegui il comando seguente per creare una tabella per i log di flusso VPC. Sostituisci test_table_vpclogs con il nome della tua tabella. Modifica il parametro LOCATION in modo che punti al bucket Amazon S3 che contiene i dati di log:

   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: il comando precedente utilizza la proiezione delle partizioni per creare una tabella, partizionare la tabella e popolare automaticamente le partizioni. Se una partizione proiettata non esiste in Amazon S3, Athena la proietta comunque. È consigliabile utilizzare attributi partizionati nelle query.

3. Usa l'editor di query della console Amazon Athena per eseguire istruzioni SQL sulla tabella. Puoi salvare le query, visualizzare quelle precedenti o scaricare i risultati delle query in formato CSV.

Query di esempio

Nota: sostituisci test_table_vpclogs con il nome della tua tabella. Modifica i valori della colonna e le altre variabili per adattarli alla query.

Per visualizzare le prime 100 voci del log di flusso in ordine cronologico per un periodo di tempo specificato, esegui una query simile alla seguente:

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

Per visualizzare quale server riceve i primi dieci pacchetti HTTP per un periodo di tempo specificato, esegui una query simile alla seguente:

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;

La query precedente conta il numero di pacchetti ricevuti sulla porta HTTPS 443 e li raggruppa per indirizzo IP di destinazione. Quindi, restituisce le prime 10 voci della settimana precedente.

Per controllare i log creati per un intervallo di tempo specificato, esegui una query simile alla seguente:

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

Per visualizzare i log di flusso per un indirizzo IP di origine specifico in un intervallo di tempo specificato, esegui una query simile alla seguente:

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

Per elencare le connessioni TCP rifiutate in un intervallo di tempo specificato, esegui una query simile alla seguente:

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;

Per visualizzare i log di flusso per l'intervallo di indirizzi IP che inizia con "10.117", esegui una query simile alla seguente:

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

Per visualizzare i log di flusso per uno specifico indirizzo IP di destinazione in un determinato intervallo di tempo, esegui una query simile alla seguente:

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

Informazioni correlate

How do I monitor traffic in my VPC with flow logs?

Analyzing VPC Flow Logs using Amazon Athena, and Amazon QuickSight