Cloud Flow Logs

#!/bin/bash

echo "=== VPC Flow Log format (v2 default) ==="
cat << 'LOG'
# Fields: version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status
2 123456789 eni-abc123 10.0.1.5 52.1.2.3 45123 443 6 15 15360 1700000000 1700000120 ACCEPT OK
2 123456789 eni-abc123 10.0.1.5 10.0.2.10 52145 445 6 3 180 1700000200 1700000205 ACCEPT OK
LOG

echo ""
echo "=== Enable VPC Flow Logs via AWS CLI ==="
aws ec2 create-flow-logs \
    --resource-type VPC \
    --resource-ids vpc-12345678 \
    --traffic-type ALL \
    --log-destination-type s3 \
    --log-destination "arn:aws:s3:::my-flow-logs-bucket/vpc-logs/"

echo ""
echo "=== Query VPC Flow Logs with Athena ==="
cat << 'SQL'
-- Create Athena table for VPC Flow Logs in S3
CREATE EXTERNAL TABLE vpc_flow_logs (
    version int, account string, interfaceid string,
    sourceaddress string, destinationaddress string,
    sourceport int, destinationport int, protocol int,
    numpackets bigint, numbytes bigint,
    starttime int, endtime int,
    action string, logstatus string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  "input.regex" = "^(\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+) (\\S+)$"
)
LOCATION 's3://my-flow-logs-bucket/vpc-logs/';

-- Find large outbound transfers (potential exfil)
SELECT
    sourceaddress,
    destinationaddress,
    destinationport,
    SUM(numbytes) as total_bytes,
    COUNT(*) as flow_count
FROM vpc_flow_logs
WHERE action = 'ACCEPT'
    AND from_unixtime(starttime) > current_timestamp - interval '24' hour
    AND destinationaddress NOT LIKE '10.%'
    AND destinationaddress NOT LIKE '172.1%'
    AND destinationaddress NOT LIKE '192.168.%'
GROUP BY 1, 2, 3
HAVING SUM(numbytes) > 100000000   -- >100MB
ORDER BY total_bytes DESC
LIMIT 20;

-- Detect beaconing: regular connections from same src→dst→port
SELECT
    sourceaddress, destinationaddress, destinationport,
    COUNT(*) as connection_count,
    MIN(from_unixtime(starttime)) as first_seen,
    MAX(from_unixtime(starttime)) as last_seen,
    STDDEV(starttime - LAG(starttime, 1, starttime) OVER
        (PARTITION BY sourceaddress, destinationaddress, destinationport
         ORDER BY starttime)) as interval_stddev
FROM vpc_flow_logs
WHERE action = 'ACCEPT' AND protocol = 6
    AND from_unixtime(starttime) > current_timestamp - interval '24' hour
GROUP BY 1, 2, 3
HAVING COUNT(*) > 10
    AND (MAX(from_unixtime(starttime)) - MIN(from_unixtime(starttime))) > interval '30' minute
ORDER BY connection_count DESC;
SQL

echo ""
echo "=== Python: analyze VPC Flow Logs from S3 ==="
cat << 'PYTHON'
import boto3
import gzip
import io
import pandas as pd
from collections import defaultdict

s3 = boto3.client('s3')

def load_flow_logs(bucket: str, prefix: str) -> pd.DataFrame:
    """Load VPC Flow Log files from S3 into a DataFrame."""
    paginator = s3.get_paginator('list_objects_v2')
    records = []

    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            key = obj['Key']
            response = s3.get_object(Bucket=bucket, Key=key)
            body = response['Body'].read()

            if key.endswith('.gz'):
                body = gzip.decompress(body)

            for line in body.decode('utf-8').splitlines():
                if line.startswith('#') or not line.strip():
                    continue
                parts = line.split()
                if len(parts) >= 14 and parts[13] == 'ACCEPT':
                    records.append({
                        'src': parts[3], 'dst': parts[4],
                        'sport': int(parts[5]), 'dport': int(parts[6]),
                        'proto': int(parts[7]),
                        'bytes': int(parts[9]),
                        'ts': int(parts[10]),
                    })

    return pd.DataFrame(records)

df = load_flow_logs('my-flow-logs', 'vpc-logs/2024/01/15/')
print(f"Loaded {len(df)} accepted flows")

# Top external destinations by byte count
external = df[~df.dst.str.startswith(('10.', '172.', '192.168.'))]
print(external.groupby(['src', 'dst', 'dport'])['bytes'].sum().sort_values(ascending=False).head(10))
PYTHON