Como uso o Amazon Athena para analisar logs de fluxo da VPC?

Breve descrição

Use os Logs de fluxos da Amazon VPC para analisar padrões de tráfego de rede e identificar ameaças e riscos em toda a sua rede da Amazon VPC.

Resolução

Usar o Athena para analisar Logs de fluxos da Amazon VPC

Para usar o Athena para analisar os logs de fluxos da Amazon VPC, conclua as seguintes etapas:

1. Use o editor de consultas do console do Amazon Athena para executar o seguinte comando para criar um banco de dados. Substitua test_db_vpclogs por um nome para seu banco de dados:

   CREATE DATABASE test_db_vpclogs;
   

   Importante: é uma prática recomendada criar o banco de dados na mesma região da AWS que o bucket do Amazon Simple Storage Service (Amazon S3) no qual os logs de fluxo estão armazenados.

2. No banco de dados, execute o comando a seguir para criar uma tabela para os logs de fluxo da VPC. Substitua test_table_vpclogs pelo nome da sua tabela. Além disso, modifique o parâmetro LOCATION para apontar para o bucket do Amazon S3 que contém seus dados de 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}'
     )

   Observação: o comando anterior usa a projeção de partição para criar uma tabela, particionar a tabela e preencher automaticamente as partições. Se uma partição projetada não existir no Amazon S3, o Athena ainda projetará a partição. A prática recomendada é usar atributos particionados em suas consultas.

3. Use o editor de consultas do console do Amazon Athena para executar instruções SQL na tabela. Você pode salvar consultas, visualizar consultas anteriores ou baixar resultados de consultas no formato de arquivo CSV.

Exemplos de consultas

Observação: substitua test_table_vpclogs na consulta pelo nome da sua tabela. Modifique os valores da coluna e outras variáveis de acordo com sua consulta.

Para visualizar as primeiras 100 entradas do log de fluxo em ordem cronológica para um período de tempo especificado, execute uma consulta semelhante à seguinte:

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

Para ver qual servidor recebe os dez maiores números de pacotes HTTP para um determinado período de tempo, execute uma consulta semelhante à seguinte:

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;

A consulta anterior conta o número de pacotes recebidos na porta HTTPS 443 e os agrupa por endereço IP de destino. Em seguida, ela retorna as 10 principais entradas da semana anterior.

Para verificar os logs que foram criados em um intervalo de tempo especificado, execute uma consulta semelhante à seguinte:

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 visualizar os logs de fluxo de um endereço IP de origem específico entre um intervalo de tempo especificado, execute uma consulta semelhante à seguinte:

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

Para listar as conexões TCP rejeitadas entre um intervalo de tempo especificado, execute uma consulta semelhante à seguinte:

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 visualizar os logs de fluxo do intervalo de endereços IP que começa com "10.117", execute uma consulta semelhante à seguinte:

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

Para visualizar os logs de fluxo de um endereço IP de destino específico entre um intervalo de tempo, execute uma consulta semelhante à seguinte:

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

Informações relacionadas

Como monitorar o tráfego na minha VPC com logs de fluxo?

Analisar logs de fluxo da VPC usando o Amazon Athena e o Amazon QuickSight