performing-cloud-log-forensics-with-athena skill (Anthropic-Cybersecurity-Skills)

From Public Agent Wiki

What it does. 'Uses AWS Athena to query CloudTrail, VPC Flow Logs, S3 access logs, Part of mukul975/Anthropic-Cybersecurity-Skills (817 security skills) (mukul975/Anthropic-Cybersecurity-Skills).
Upstream mukul975/Anthropic-Cybersecurity-Skills
Skill file skills/performing-cloud-log-forensics-with-athena/SKILL.md
License Apache-2.0 (skill folder LICENSE)
Author mukul975
Fetched 2026-09-10

Install

  • npx skills add mukul975/Anthropic-Cybersecurity-Skills --skill performing-cloud-log-forensics-with-athena, or copy the skill folder into ~/.claude/skills/performing-cloud-log-forensics-with-athena/.
  • Raw file: curl -sL https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/performing-cloud-log-forensics-with-athena/SKILL.md

SKILL.md (verbatim)

name: performing-cloud-log-forensics-with-athena
description: 'Uses AWS Athena to query CloudTrail, VPC Flow Logs, S3 access logs,
  and ALB logs for forensic investigation. Covers CREATE TABLE DDL with partition
  projection, forensic SQL queries for detecting unauthorized access, data exfiltration,
  lateral movement, and privilege escalation. Use when investigating AWS security
  incidents or building cloud-native forensic workflows at scale.

  '
domain: cybersecurity
subdomain: cloud-security
tags:
- cloud
- forensics
- athena
- aws
- cloudtrail
- vpc-flow-logs
- s3
- alb
version: '1.0'
author: mukul975
license: Apache-2.0
nist_csf:
- PR.IR-01
- ID.AM-08
- GV.SC-06
- DE.CM-01
mitre_attack:
- T1078.004
- T1530
- T1537
- T1580
- T1021

Performing Cloud Log Forensics with AWS Athena

When to Use

  • When investigating AWS security incidents that require querying massive volumes of cloud logs
  • When performing forensic analysis across CloudTrail, VPC Flow Logs, S3 access logs, and ALB logs
  • When building reusable Athena tables with partition projection for ongoing incident response
  • When hunting for indicators of compromise across multiple AWS log sources simultaneously
  • When creating evidence-grade SQL queries for compliance audits or legal proceedings

Prerequisites

  • AWS account with Athena, S3, and Glue permissions
  • CloudTrail configured to deliver logs to an S3 bucket
  • VPC Flow Logs enabled and publishing to S3
  • S3 server access logging enabled on target buckets
  • ALB access logging enabled and publishing to S3
  • Python 3.8+ with boto3 installed
  • Appropriate IAM permissions for Athena queries and S3 access

Instructions

Phase 1: Create Athena Database and CloudTrail Table

Create a dedicated forensics database and CloudTrail table using partition projection to automatically discover partitions without manual ALTER TABLE statements.

CREATE DATABASE IF NOT EXISTS cloud_forensics;

CREATE EXTERNAL TABLE cloud_forensics.cloudtrail_logs (
    eventVersion STRING,
    userIdentity STRUCT<
        type: STRING,
        principalId: STRING,
        arn: STRING,
        accountId: STRING,
        invokedBy: STRING,
        accessKeyId: STRING,
        userName: STRING,
        sessionContext: STRUCT<
            attributes: STRUCT<
                mfaAuthenticated: STRING,
                creationDate: STRING>,
            sessionIssuer: STRUCT<
                type: STRING,
                principalId: STRING,
                arn: STRING,
                accountId: STRING,
                userName: STRING>,
            ec2RoleDelivery: STRING,
            webIdFederationData: STRUCT<
                federatedProvider: STRING,
                attributes: MAP<STRING, STRING>>>>,
    eventTime STRING,
    eventSource STRING,
    eventName STRING,
    awsRegion STRING,
    sourceIPAddress STRING,
    userAgent STRING,
    errorCode STRING,
    errorMessage STRING,
    requestParameters STRING,
    responseElements STRING,
    additionalEventData STRING,
    requestId STRING,
    eventId STRING,
    readOnly STRING,
    resources ARRAY<STRUCT<
        arn: STRING,
        accountId: STRING,
        type: STRING>>,
    eventType STRING,
    apiVersion STRING,
    recipientAccountId STRING,
    serviceEventDetails STRING,
    sharedEventID STRING,
    vpcEndpointId STRING,
    tlsDetails STRUCT<
        tlsVersion: STRING,
        cipherSuite: STRING,
        clientProvidedHostHeader: STRING>
)
COMMENT 'CloudTrail logs with partition projection for forensic analysis'
PARTITIONED BY (
    `account` STRING,
    `region` STRING,
    `timestamp` STRING
)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://YOUR-CLOUDTRAIL-BUCKET/AWSLogs/'
TBLPROPERTIES (
    'projection.enabled' = 'true',
    'projection.account.type' = 'enum',
    'projection.account.values' = 'YOUR_ACCOUNT_ID',
    'projection.region.type' = 'enum',
    'projection.region.values' = 'us-east-1,us-west-2,eu-west-1',
    'projection.timestamp.type' = 'date',
    'projection.timestamp.format' = 'yyyy/MM/dd',
    'projection.timestamp.range' = '2023/01/01,NOW',
    'projection.timestamp.interval' = '1',
    'projection.timestamp.interval.unit' = 'DAYS',
    'storage.location.template' = 's3://YOUR-CLOUDTRAIL-BUCKET/AWSLogs/${account}/CloudTrail/${region}/${timestamp}'
);

Phase 2: Create VPC Flow Logs Table

CREATE EXTERNAL TABLE cloud_forensics.vpc_flow_logs (
    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,
    az_id STRING,
    sublocation_type STRING,
    sublocation_id STRING,
    pkt_srcaddr STRING,
    pkt_dstaddr STRING,
    region STRING,
    pkt_src_aws_service STRING,
    pkt_dst_aws_service STRING,
    flow_direction STRING,
    traffic_path INT
)
PARTITIONED BY (
    `date` STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ' '
LOCATION 's3://YOUR-VPC-FLOW-LOGS-BUCKET/AWSLogs/YOUR_ACCOUNT_ID/vpcflowlogs/'
TBLPROPERTIES (
    'skip.header.line.count' = '1',
    'projection.enabled' = 'true',
    'projection.date.type' = 'date',
    'projection.date.format' = 'yyyy/MM/dd',
    'projection.date.range' = '2023/01/01,NOW',
    'projection.date.interval' = '1',
    'projection.date.interval.unit' = 'DAYS',
    'storage.location.template' = 's3://YOUR-VPC-FLOW-LOGS-BUCKET/AWSLogs/YOUR_ACCOUNT_ID/vpcflowlogs/us-east-1/${date}'
);

Phase 3: Create S3 Access Logs Table

CREATE EXTERNAL TABLE cloud_forensics.s3_access_logs (
    bucket_owner STRING,
    bucket_name STRING,
    request_datetime STRING,
    remote_ip STRING,
    requester STRING,
    request_id STRING,
    operation STRING,
    key STRING,
    request_uri STRING,
    http_status INT,
    error_code STRING,
    bytes_sent BIGINT,
    object_size BIGINT,
    total_time INT,
    turn_around_time INT,
    referrer STRING,
    user_agent STRING,
    version_id STRING,
    host_id STRING,
    signature_version STRING,
    cipher_suite STRING,
    authentication_type STRING,
    host_header STRING,
    tls_version STRING,
    access_point_arn STRING,
    acl_required STRING
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
    'serialization.format' = '1',
    'input.regex' = '([^ ]*) ([^ ]*) \\[(.*?)\\] ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) (\"[^\"]*\"|-) (-|[0-9]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) (\"[^\"]*\"|-) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*)'
)
STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://YOUR-S3-ACCESS-LOGS-BUCKET/logs/';

Phase 4: Create ALB Access Logs Table

CREATE EXTERNAL TABLE cloud_forensics.alb_access_logs (
    type STRING,
    time STRING,
    elb STRING,
    client_ip STRING,
    client_port INT,
    target_ip STRING,
    target_port INT,
    request_processing_time DOUBLE,
    target_processing_time DOUBLE,
    response_processing_time DOUBLE,
    elb_status_code INT,
    target_status_code STRING,
    received_bytes BIGINT,
    sent_bytes BIGINT,
    request_verb STRING,
    request_url STRING,
    request_proto STRING,
    user_agent STRING,
    ssl_cipher STRING,
    ssl_protocol STRING,
    target_group_arn STRING,
    trace_id STRING,
    domain_name STRING,
    chosen_cert_arn STRING,
    matched_rule_priority STRING,
    request_creation_time STRING,
    actions_executed STRING,
    redirect_url STRING,
    lambda_error_reason STRING,
    target_port_list STRING,
    target_status_code_list STRING,
    classification STRING,
    classification_reason STRING,
    conn_trace_id STRING
)
PARTITIONED BY (
    `day` STRING
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
    'serialization.format' = '1',
    'input.regex' = '([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*)[:-]([0-9]*) ([-.0-9]*) ([-.0-9]*) ([-.0-9]*) (|[0-9]*) (-|[0-9]*) ([-0-9]*) ([-0-9]*) \"([^ ]*) (.*) (- |[^ ]*)\" \"([^\"]*)\" ([A-Z0-9-_]+) ([A-Za-z0-9.-]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\" \"([^\"]*)\" ([-.0-9]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\" \"([^ ]*)\" \"([^\"]*)\" \"([^ ]*)\" \"([^ ]*)\" \"([^ ]*)\"'
)
STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://YOUR-ALB-LOGS-BUCKET/AWSLogs/YOUR_ACCOUNT_ID/elasticloadbalancing/us-east-1/'
TBLPROPERTIES (
    'projection.enabled' = 'true',
    'projection.day.type' = 'date',
    'projection.day.format' = 'yyyy/MM/dd',
    'projection.day.range' = '2023/01/01,NOW',
    'projection.day.interval' = '1',
    'projection.day.interval.unit' = 'DAYS',
    'storage.location.template' = 's3://YOUR-ALB-LOGS-BUCKET/AWSLogs/YOUR_ACCOUNT_ID/elasticloadbalancing/us-east-1/${day}'
);

Phase 5: Forensic Investigation Queries

Detect Unauthorized API Calls

SELECT
    eventtime,
    useridentity.arn AS caller_arn,
    useridentity.accountid AS account,
    eventsource,
    eventname,
    errorcode,
    errormessage,
    sourceipaddress,
    useragent
FROM cloud_forensics.cloudtrail_logs
WHERE errorcode IN ('AccessDenied', 'UnauthorizedAccess', 'Client.UnauthorizedAccess')
    AND timestamp BETWEEN '2024/01/01' AND '2024/12/31'
ORDER BY eventtime DESC
LIMIT 1000;

Detect Privilege Escalation Attempts

SELECT
    eventtime,
    useridentity.arn AS actor,
    eventname,
    eventsource,
    json_extract_scalar(requestparameters, '$.policyArn') AS policy_arn,
    json_extract_scalar(requestparameters, '$.roleName') AS role_name,
    json_extract_scalar(requestparameters, '$.userName') AS target_user,
    sourceipaddress
FROM cloud_forensics.cloudtrail_logs
WHERE eventname IN (
    'AttachUserPolicy', 'AttachRolePolicy', 'AttachGroupPolicy',
    'PutUserPolicy', 'PutRolePolicy', 'PutGroupPolicy',
    'CreatePolicyVersion', 'SetDefaultPolicyVersion',
    'AddUserToGroup', 'UpdateAssumeRolePolicy',
    'CreateAccessKey', 'CreateLoginProfile',
    'UpdateLoginProfile', 'AssumeRole'
)
    AND timestamp BETWEEN '2024/01/01' AND '2024/12/31'
ORDER BY eventtime DESC;

Detect Data Exfiltration via S3

SELECT
    eventtime,
    useridentity.arn AS actor,
    eventname,
    json_extract_scalar(requestparameters, '$.bucketName') AS bucket,
    json_extract_scalar(requestparameters, '$.key') AS object_key,
    sourceipaddress,
    useragent
FROM cloud_forensics.cloudtrail_logs
WHERE eventsource = 's3.amazonaws.com'
    AND eventname IN ('GetObject', 'CopyObject', 'PutBucketPolicy',
                      'PutBucketAcl', 'PutObjectAcl', 'SelectObjectContent')
    AND sourceipaddress NOT LIKE '10.%'
    AND sourceipaddress NOT LIKE '172.%'
    AND sourceipaddress NOT LIKE '192.168.%'
    AND timestamp BETWEEN '2024/01/01' AND '2024/12/31'
ORDER BY eventtime DESC;

Detect Lateral Movement via VPC Flow Logs

SELECT
    srcaddr,
    dstaddr,
    dstport,
    protocol,
    SUM(packets) AS total_packets,
    SUM(bytes) AS total_bytes,
    COUNT(*) AS connection_count,
    MIN(from_unixtime(start)) AS first_seen,
    MAX(from_unixtime("end")) AS last_seen
FROM cloud_forensics.vpc_flow_logs
WHERE action = 'ACCEPT'
    AND srcaddr LIKE '10.%'
    AND dstport IN (22, 3389, 5985, 5986, 445, 135, 139)
    AND date BETWEEN '2024/06/01' AND '2024/06/30'
GROUP BY srcaddr, dstaddr, dstport, protocol
HAVING COUNT(*) > 100
ORDER BY connection_count DESC;

Detect Port Scanning Activity

SELECT
    srcaddr,
    COUNT(DISTINCT dstport) AS unique_ports_scanned,
    COUNT(DISTINCT dstaddr) AS unique_targets,
    SUM(packets) AS total_packets,
    MIN(from_unixtime(start)) AS first_seen,
    MAX(from_unixtime("end")) AS last_seen
FROM cloud_forensics.vpc_flow_logs
WHERE action = 'REJECT'
    AND date BETWEEN '2024/06/01' AND '2024/06/30'
GROUP BY srcaddr
HAVING COUNT(DISTINCT dstport) > 25
ORDER BY unique_ports_scanned DESC;

Detect Suspicious S3 Bulk Downloads

SELECT
    remote_ip,
    requester,
    bucket_name,
    COUNT(*) AS request_count,
    SUM(bytes_sent) AS total_bytes_downloaded,
    COUNT(DISTINCT key) AS unique_objects,
    MIN(request_datetime) AS first_request,
    MAX(request_datetime) AS last_request
FROM cloud_forensics.s3_access_logs
WHERE operation LIKE '%GET%'
    AND http_status = 200
GROUP BY remote_ip, requester, bucket_name
HAVING COUNT(*) > 500
ORDER BY total_bytes_downloaded DESC;

Detect ALB-Level Injection Attempts

SELECT
    time,
    client_ip,
    request_verb,
    request_url,
    elb_status_code,
    target_status_code,
    user_agent
FROM cloud_forensics.alb_access_logs
WHERE (
    request_url LIKE '%UNION%SELECT%'
    OR request_url LIKE '%<script%'
    OR request_url LIKE '%../../../%'
    OR request_url LIKE '%/etc/passwd%'
    OR request_url LIKE '%cmd.exe%'
    OR request_url LIKE '%/proc/self%'
    OR request_url LIKE '%SLEEP(%'
    OR request_url LIKE '%WAITFOR%'
)
    AND day BETWEEN '2024/06/01' AND '2024/06/30'
ORDER BY time DESC;

Phase 6: Cross-Log Correlation

Correlate findings across log sources for comprehensive incident timelines.

-- Correlate suspicious CloudTrail actor with VPC Flow Logs
WITH suspicious_ips AS (
    SELECT DISTINCT sourceipaddress AS ip
    FROM cloud_forensics.cloudtrail_logs
    WHERE errorcode = 'AccessDenied'
        AND timestamp BETWEEN '2024/06/01' AND '2024/06/30'
)
SELECT
    v.srcaddr,
    v.dstaddr,
    v.dstport,
    v.protocol,
    SUM(v.bytes) AS total_bytes,
    COUNT(*) AS flow_count
FROM cloud_forensics.vpc_flow_logs v
JOIN suspicious_ips s ON v.srcaddr = s.ip
WHERE v.date BETWEEN '2024/06/01' AND '2024/06/30'
GROUP BY v.srcaddr, v.dstaddr, v.dstport, v.protocol
ORDER BY total_bytes DESC;

Examples

# Quick-start: run the forensics agent for a full investigation
python agent.py \
    --action full_investigation \
    --database cloud_forensics \
    --start-date 2024-06-01 \
    --end-date 2024-06-30 \
    --output forensics_report.json

# Run specific queries only
python agent.py \
    --action privilege_escalation \
    --database cloud_forensics \
    --start-date 2024-06-15 \
    --end-date 2024-06-16

# Create all forensic tables from scratch
python agent.py \
    --action setup_tables \
    --cloudtrail-bucket my-cloudtrail-logs \
    --vpc-flow-bucket my-vpc-flow-logs \
    --s3-access-bucket my-s3-access-logs \
    --alb-bucket my-alb-logs \
    --account-id 123456789012 \
    --regions us-east-1,us-west-2

Other files in this skill

references/api-reference.md (verbatim)

AWS Athena API Reference

This reference covers the Amazon Athena API as used for cloud log forensics, primarily through the AWS SDK for Python (boto3) and the AWS CLI. Athena is a serverless, interactive query service that runs ANSI SQL (Trino/Presto engine) directly against data in Amazon S3.

Authentication

Athena uses standard AWS authentication — there is no separate Athena API key. Credentials are resolved by the AWS SDK credential provider chain, in order:

  1. Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
  2. Shared credentials file: ~/.aws/credentials (profile via AWS_PROFILE)
  3. Shared config file: ~/.aws/config
  4. IAM role for Amazon EC2 / ECS task role / EKS IRSA / Lambda execution role
  5. SSO / aws sso login
import boto3

# Default credential chain
athena = boto3.client("athena", region_name="us-east-1")

# Explicit profile / assumed role session
session = boto3.Session(profile_name="ir-forensics", region_name="us-east-1")
athena = session.client("athena")

Required IAM permissions for forensic querying (least privilege):

Action Purpose
athena:StartQueryExecution Submit a query
athena:GetQueryExecution Poll query status
athena:GetQueryResults Fetch result rows
athena:StopQueryExecution Cancel a running query
athena:GetWorkGroup / athena:ListWorkGroups Workgroup discovery
glue:GetTable, glue:GetDatabase, glue:GetPartitions Read table metadata (Glue Data Catalog)
s3:GetObject, s3:ListBucket Read source log data
s3:PutObject, s3:GetObject on the results bucket Write/read query output

Key Methods (boto3 athena client / Athena API)

Method Description Key Parameters
start_query_execution Submit a SQL query (DDL or DML). Asynchronous — returns immediately. QueryString (required), QueryExecutionContext={Database, Catalog}, ResultConfiguration={OutputLocation, EncryptionConfiguration}, WorkGroup, ClientRequestToken (idempotency, ≤128 chars), ExecutionParameters (list for ? placeholders), ResultReuseConfiguration
get_query_execution Poll a query's status, statistics, and engine details. QueryExecutionId (required)
get_query_results Retrieve result rows (paginated, max 1000 rows/page). QueryExecutionId (required), MaxResults (1–1000), NextToken, QueryResultType
stop_query_execution Cancel a running query. QueryExecutionId (required)
batch_get_query_execution Get details for up to 50 query IDs at once. QueryExecutionIds (list, ≤50)
list_query_executions List query IDs (most recent first). WorkGroup, MaxResults (≤50), NextToken
get_query_runtime_statistics Detailed per-stage execution stats. QueryExecutionId
create_work_group / get_work_group Manage workgroups (cost controls, result location, encryption). Name, Configuration
create_named_query / list_named_queries Save/list reusable saved queries. Name, Database, QueryString, WorkGroup
get_database / list_databases / list_table_metadata Inspect Data Catalog metadata. CatalogName, DatabaseName

start_query_execution parameter detail

  • QueryString — the SQL text. Up to 262,144 bytes (256 KB).
  • QueryExecutionContext — {"Database": "cloud_forensics", "Catalog": "AwsDataCatalog"}. Sets the default database so unqualified table names resolve.
  • ResultConfiguration.OutputLocation — s3://aws-athena-query-results-.../ where the CSV result and metadata are written. Required unless the workgroup enforces an output location.
  • WorkGroup — defaults to primary. Use a dedicated forensics workgroup to enforce encryption, a per-query data-scanned limit (BytesScannedCutoffPerQuery), and a fixed result location.
  • ExecutionParameters — positional values for parameterized queries using ? placeholders (prevents SQL injection when interpolating IOCs).
  • ResultReuseConfiguration — {"ResultReuseByAgeConfiguration": {"Enabled": true, "MaxAgeInMinutes": 60}} reuses prior results to cut cost/latency.

Python SDK

# Installation
pip install boto3

import boto3
import time

athena = boto3.client("athena", region_name="us-east-1")

def run_query(sql, database="cloud_forensics",
              output="s3://aws-athena-query-results-acct-region/forensics/",
              workgroup="forensics", params=None):
    """Submit a query, poll to completion, return result rows."""
    kwargs = {
        "QueryString": sql,
        "QueryExecutionContext": {"Database": database},
        "ResultConfiguration": {"OutputLocation": output},
        "WorkGroup": workgroup,
    }
    if params:
        kwargs["ExecutionParameters"] = params  # for ? placeholders

    qid = athena.start_query_execution(**kwargs)["QueryExecutionId"]

    # Poll status
    while True:
        resp = athena.get_query_execution(QueryExecutionId=qid)
        state = resp["QueryExecution"]["Status"]["State"]
        if state in ("SUCCEEDED", "FAILED", "CANCELLED"):
            break
        time.sleep(1)

    if state != "SUCCEEDED":
        reason = resp["QueryExecution"]["Status"].get("StateChangeReason", "")
        raise RuntimeError(f"Query {state}: {reason}")

    # Paginate results
    rows = []
    paginator = athena.get_paginator("get_query_results")
    for page in paginator.paginate(QueryExecutionId=qid):
        rows.extend(page["ResultSet"]["Rows"])
    return rows

# Parameterized query — safe IOC lookup
run_query(
    "SELECT eventtime, eventname, sourceipaddress "
    "FROM cloudtrail_logs WHERE sourceipaddress = ? LIMIT 100",
    params=["203.0.113.45"],
)

CLI equivalents:

aws athena start-query-execution \
  --query-string "SELECT count(*) FROM cloud_forensics.cloudtrail_logs" \
  --query-execution-context Database=cloud_forensics \
  --result-configuration OutputLocation=s3://my-athena-results/ \
  --work-group forensics

aws athena get-query-execution --query-execution-id <id>
aws athena get-query-results   --query-execution-id <id>

Common Response Fields

get_query_execution → QueryExecution:

Field Meaning
QueryExecutionId Unique query ID
Status.State QUEUED | RUNNING | SUCCEEDED | FAILED | CANCELLED
Status.StateChangeReason Failure/cancel reason text
Statistics.DataScannedInBytes Bytes scanned (drives cost — $5/TB scanned)
Statistics.EngineExecutionTimeInMillis Execution time
Statistics.TotalExecutionTimeInMillis Wall-clock including queue time
ResultConfiguration.OutputLocation S3 path to the result CSV

get_query_results → ResultSet.Rows (each Row.Data is a list of {"VarCharValue": ...}); the first row is the column header. ResultSetMetadata.ColumnInfo describes column names/types.

Rate Limits / Service Quotas

These are default, adjustable AWS account-level quotas (per Region):

Quota Default
StartQueryExecution call rate (DML) 20 calls/sec (burst), then throttled
GetQueryExecution call rate 100 calls/sec
GetQueryResults call rate 100 calls/sec
Active DML queries (running + queued) 200 (Engine v3)
Active DDL queries 20
Query timeout (DML) 30 minutes
DDL query timeout 600 minutes
QueryString max size 256 KB
Result page (GetQueryResults) 1000 rows max

Throttling surfaces as TooManyRequestsException / ThrottlingException. boto3 retries these automatically with exponential backoff (adaptive retry mode recommended for high-volume forensic batch jobs). Cost is billed by bytes scanned, so partition pruning and columnar formats (Parquet/ORC) drastically reduce both cost and the chance of hitting the per-query data-scan cutoff.

Error Codes

Error Meaning
InvalidRequestException Malformed request / invalid parameter
TooManyRequestsException API call rate or concurrent-query quota exceeded
ThrottlingException Service throttling; back off and retry
ResourceNotFoundException Workgroup, catalog, or named query not found
MetadataException Glue Data Catalog metadata error
Query FAILED with HIVE_BAD_DATA Row doesn't match table schema/SerDe
Query FAILED with HIVE_CURSOR_ERROR S3 object unreadable (permissions, corrupt file)
Query FAILED with HIVE_PARTITION_SCHEMA_MISMATCH Partition schema differs from table
AccessDeniedException Missing IAM permission for Athena, Glue, or S3

Resources

references/athena-forensics-reference.md (verbatim)

Reference: Cloud Log Forensics with AWS Athena

Athena Partition Projection

Partition projection eliminates the need for ALTER TABLE ADD PARTITION by automatically inferring partition values at query time based on declared ranges. This is critical for forensic tables that span long date ranges across multiple accounts and regions.

Key TBLPROPERTIES

'projection.enabled' = 'true'
'projection.<column>.type' = 'date|enum|integer|injected'
'projection.<column>.range' = '<start>,<end>'   -- for date/integer
'projection.<column>.format' = 'yyyy/MM/dd'     -- for date
'projection.<column>.interval' = '1'            -- for date/integer
'projection.<column>.interval.unit' = 'DAYS'    -- DAYS|HOURS|MINUTES|SECONDS
'projection.<column>.values' = 'val1,val2'      -- for enum
'storage.location.template' = 's3://bucket/path/${column1}/${column2}'

CloudTrail Log Structure

CloudTrail JSON fields relevant to forensics:

Field Description Forensic Use
userIdentity.arn Caller identity Attribute actions to actors
eventName API call name Identify suspicious operations
eventSource AWS service Scope investigation
sourceIPAddress Origin IP Detect external access
errorCode AccessDenied etc. Find unauthorized attempts
requestParameters API parameters Understand intent
responseElements API response Confirm impact
userAgent Client software Detect unusual tooling
tlsDetails TLS version/cipher Detect weak crypto

VPC Flow Log Fields

Field Type Forensic Use
srcaddr IP Identify source of traffic
dstaddr IP Identify destination
srcport INT Source port (ephemeral = client)
dstport INT Destination port (service identification)
protocol INT 6=TCP, 17=UDP, 1=ICMP
action STRING ACCEPT or REJECT
bytes BIGINT Volume of data transferred
packets BIGINT Packet count
start/end BIGINT Unix epoch timestamps
flow_direction STRING ingress or egress

S3 Access Log Fields

Field Forensic Use
remote_ip Source of S3 requests
requester IAM identity or anonymous
operation REST API operation (REST.GET.OBJECT, etc.)
key S3 object path accessed
http_status Success/failure indicator
bytes_sent Data volume exfiltrated
total_time Request duration

ALB Access Log Fields

Field Forensic Use
client_ip Source of web requests
request_url Full URL with potential injection payloads
elb_status_code ALB response (5xx = server-side issues)
target_status_code Backend response
request_processing_time ALB processing delay
user_agent Client identification

Forensic Query Patterns

Lateral Movement Indicators (VPC Flow Logs)

  • Internal-to-internal traffic on management ports (22, 3389, 5985, 445)
  • High connection counts between internal hosts
  • Unusual protocol usage (ICMP tunneling)
  • Traffic to honeypot IPs

Privilege Escalation Indicators (CloudTrail)

  • IAM policy attachment events
  • CreateAccessKey for other users
  • AssumeRole to high-privilege roles
  • ConsoleLogin without MFA
  • Security group modifications opening ingress

Data Exfiltration Indicators (S3 + CloudTrail)

  • Bulk GetObject from sensitive buckets
  • PutBucketPolicy making buckets public
  • CopyObject to external accounts
  • DeleteBucketEncryption
  • Large bytes_sent volumes from S3 access logs

Web Attack Indicators (ALB)

  • SQL injection patterns in URLs (UNION SELECT, SLEEP, WAITFOR)
  • Path traversal (../../, /etc/passwd)
  • XSS payloads (<script>, javascript:)
  • Command injection (cmd.exe, /bin/sh)

Protocol Number Reference

Protocol Number Name
1 ICMP
6 TCP
17 UDP
47 GRE
50 ESP
58 ICMPv6

Common Suspicious Ports

Port Service Concern
22 SSH Lateral movement
445 SMB Lateral movement, ransomware
3389 RDP Lateral movement
5985/5986 WinRM Lateral movement
4444 Metasploit default C2 channel
8080 Alt HTTP Proxy, backdoor
1433 MSSQL Database access
3306 MySQL Database access
5432 PostgreSQL Database access
6379 Redis Cache access

References

Back to mukul975/Anthropic-Cybersecurity-Skills (817 security skills) or Agent skills.