VPC Flow Logs Analysis for Security

VPC Flow Logs capture network traffic information for security analysis, troubleshooting, and compliance. This guide covers advanced analysis techniques for threat detection.

Configuring Flow Logs

Enhanced Format

VPCFlowLog:
  Type: AWS::EC2::FlowLog
  Properties:
    ResourceType: VPC
    ResourceId: !Ref VPC
    TrafficType: ALL
    LogDestinationType: s3
    LogDestination: !Sub arn:aws:s3:::${FlowLogBucket}
    LogFormat: >-
      ${version} ${account-id} ${interface-id} ${srcaddr} 
      ${dstaddr} ${srcport} ${dstport} ${protocol} ${packets} 
      ${bytes} ${start} ${end} ${action} ${log-status}
      ${vpc-id} ${subnet-id} ${instance-id} ${tcp-flags}
      ${type} ${pkt-srcaddr} ${pkt-dstaddr} ${region}
      ${az-id} ${sublocation-type} ${sublocation-id}
    MaxAggregationInterval: 60
    DestinationOptions:
      FileFormat: parquet
      HiveCompatiblePartitions: true
      PerHourPartition: true

Athena Analysis

Create Table

CREATE EXTERNAL TABLE vpc_flow_logs (
  version int,
  account_id string,
  interface_id string,
  srcaddr string,
  dstaddr string,
  srcport int,
  dstport int,
  protocol int,
  packets bigint,
  bytes bigint,
  start_time bigint,
  end_time 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,
  region string,
  az_id string,
  sublocation_type string,
  sublocation_id string
)
PARTITIONED BY (
  date_partition string,
  hour_partition string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS PARQUET
LOCATION 's3://flow-logs-bucket/AWSLogs/123456789012/vpcflowlogs/'
TBLPROPERTIES (
  'projection.enabled' = 'true',
  'projection.date_partition.type' = 'date',
  'projection.date_partition.format' = 'yyyy/MM/dd',
  'projection.date_partition.range' = '2024/01/01,NOW',
  'projection.hour_partition.type' = 'integer',
  'projection.hour_partition.range' = '0,23',
  'storage.location.template' = 
    's3://flow-logs-bucket/AWSLogs/123456789012/vpcflowlogs/${date_partition}/${hour_partition}'
);

Security Queries

Detect Port Scans

SELECT 
  srcaddr,
  COUNT(DISTINCT dstport) as unique_ports,
  COUNT(*) as connection_attempts,
  array_agg(DISTINCT dstport) as scanned_ports
FROM vpc_flow_logs
WHERE 
  action = 'REJECT'
  AND date_partition >= date_format(current_date - interval '1' day, '%Y/%m/%d')
GROUP BY srcaddr
HAVING COUNT(DISTINCT dstport) > 100
ORDER BY unique_ports DESC
LIMIT 50;

Unusual Outbound Traffic

SELECT 
  instance_id,
  dstaddr,
  dstport,
  SUM(bytes) as total_bytes,
  SUM(packets) as total_packets,
  COUNT(*) as flow_count
FROM vpc_flow_logs
WHERE 
  action = 'ACCEPT'
  AND type = 'IPv4'
  AND srcaddr LIKE '10.%'  -- Internal IPs
  AND dstaddr NOT LIKE '10.%'  -- External destinations
  AND dstport NOT IN (443, 80, 53)  -- Unusual ports
  AND date_partition >= date_format(current_date - interval '1' day, '%Y/%m/%d')
GROUP BY instance_id, dstaddr, dstport
HAVING SUM(bytes) > 1000000000  -- More than 1GB
ORDER BY total_bytes DESC;

Detect Lateral Movement

WITH internal_connections AS (
  SELECT 
    srcaddr,
    dstaddr,
    dstport,
    SUM(packets) as total_packets
  FROM vpc_flow_logs
  WHERE 
    action = 'ACCEPT'
    AND srcaddr LIKE '10.%'
    AND dstaddr LIKE '10.%'
    AND dstport IN (22, 3389, 5985, 5986)  -- SSH, RDP, WinRM
    AND date_partition >= date_format(current_date - interval '1' day, '%Y/%m/%d')
  GROUP BY srcaddr, dstaddr, dstport
)
SELECT 
  srcaddr,
  COUNT(DISTINCT dstaddr) as unique_destinations,
  array_agg(DISTINCT dstaddr) as target_ips,
  SUM(total_packets) as total_packets
FROM internal_connections
GROUP BY srcaddr
HAVING COUNT(DISTINCT dstaddr) > 10
ORDER BY unique_destinations DESC;

Real-Time Analysis

CloudWatch Logs Insights

fields @timestamp, srcAddr, dstAddr, dstPort, action
| filter action = "REJECT"
| stats count(*) as rejected_count by srcAddr
| sort rejected_count desc
| limit 20

Lambda for Alerts

import boto3
import json

def analyze_flow_logs(event, context):
    suspicious_patterns = []
    
    for record in event['Records']:
        log_data = json.loads(record['body'])
        
        # Check for suspicious patterns
        if is_port_scan(log_data):
            suspicious_patterns.append({
                'type': 'PORT_SCAN',
                'source': log_data['srcaddr'],
                'details': log_data
            })
        
        if is_data_exfiltration(log_data):
            suspicious_patterns.append({
                'type': 'DATA_EXFILTRATION',
                'source': log_data['srcaddr'],
                'destination': log_data['dstaddr'],
                'bytes': log_data['bytes']
            })
    
    if suspicious_patterns:
        send_security_alert(suspicious_patterns)

def is_port_scan(log):
    # Implement port scan detection logic
    return log.get('action') == 'REJECT' and log.get('packets', 0) < 3

def is_data_exfiltration(log):
    # Implement data exfiltration detection
    return log.get('bytes', 0) > 100000000 and log.get('dstport') not in [443, 80]

Working with Warqline

We are a cloud engineering consultancy and an official AWS and Google Cloud partner. If you are running this in production and want a second pair of eyes, we scope work in a free 45-minute technical call: you describe what you are running and what worries you, and we tell you what we would look at first.

Talk to an engineer

Conclusion

VPC Flow Logs provide essential visibility into network traffic. Combine Athena for historical analysis, CloudWatch for real-time monitoring, and custom Lambda functions for automated threat detection.