Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

When to Use

When building security monitoring pipelines for AWS API activity
When investigating security incidents to trace attacker actions across AWS services
When compliance requires audit logging of all administrative and data access operations
When creating detection rules for known attack patterns in AWS environments
When establishing baseline API behavior for anomaly detection

Do not use for real-time threat detection (use GuardDuty which already analyzes CloudTrail), for application-level logging (use CloudWatch Application Logs), or for network traffic analysis (use VPC Flow Logs).

Prerequisites

CloudTrail enabled with management events and optionally data events across all accounts
S3 bucket configured as CloudTrail delivery channel with appropriate retention policies
Amazon Athena configured with CloudTrail log table for ad-hoc queries
CloudWatch Logs subscription for real-time analysis with Logs Insights
SIEM integration (Splunk, Elastic, or Security Lake) for production monitoring

Workflow

Step 1: Configure CloudTrail for Comprehensive Logging

Ensure CloudTrail captures all relevant event types across the organization.

# Create an organization trail (captures all accounts)
aws cloudtrail create-trail \
  --name org-security-trail \
  --s3-bucket-name cloudtrail-logs-org-ACCOUNT \
  --is-organization-trail \
  --is-multi-region-trail \
  --include-global-service-events \
  --enable-log-file-validation \
  --kms-key-id alias/cloudtrail-key \
  --cloud-watch-logs-log-group-arn arn:aws:logs:us-east-1:ACCOUNT:log-group:cloudtrail-org:* \
  --cloud-watch-logs-role-arn arn:aws:iam::ACCOUNT:role/CloudTrailCloudWatchRole
 
# Start logging
aws cloudtrail start-logging --name org-security-trail
 
# Enable data events for S3 and Lambda
aws cloudtrail put-event-selectors \
  --trail-name org-security-trail \
  --advanced-event-selectors '[
    {
      "Name": "S3DataEvents",
      "FieldSelectors": [
        {"Field": "eventCategory", "Equals": ["Data"]},
        {"Field": "resources.type", "Equals": ["AWS::S3::Object"]}
      ]
    },
    {
      "Name": "LambdaDataEvents",
      "FieldSelectors": [
        {"Field": "eventCategory", "Equals": ["Data"]},
        {"Field": "resources.type", "Equals": ["AWS::Lambda::Function"]}
      ]
    }
  ]'
 
# Verify trail configuration
aws cloudtrail describe-trails --trail-name-list org-security-trail

Step 2: Set Up Athena for CloudTrail Query Analysis

Create an Athena table for querying CloudTrail logs with SQL.

-- Create CloudTrail Athena table
CREATE EXTERNAL TABLE 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>
    >
  >,
  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,
  sharedEventId STRING,
  vpcEndpointId STRING
)
PARTITIONED BY (region STRING, year STRING, month STRING, day STRING)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION 's3://cloudtrail-logs-org-ACCOUNT/AWSLogs/ORG_ID/';
 
-- Add partitions for recent data
ALTER TABLE cloudtrail_logs ADD
  PARTITION (region='us-east-1', year='2026', month='02', day='23')
  LOCATION 's3://cloudtrail-logs-org-ACCOUNT/AWSLogs/ORG_ID/ACCOUNT/CloudTrail/us-east-1/2026/02/23/';

Step 3: Run Security-Focused Athena Queries

Execute queries to detect common attack patterns and suspicious activity.

-- Detect console logins without MFA
SELECT eventtime, useridentity.username, sourceipaddress, useridentity.arn
FROM cloudtrail_logs
WHERE eventname = 'ConsoleLogin'
  AND additionalEventData LIKE '%"MFAUsed":"No"%'
  AND errorcode IS NULL
ORDER BY eventtime DESC;
 
-- Find IAM privilege escalation attempts
SELECT eventtime, useridentity.arn, eventname, errorcode, sourceipaddress
FROM cloudtrail_logs
WHERE eventname IN (
  'CreatePolicyVersion', 'SetDefaultPolicyVersion', 'AttachUserPolicy',
  'AttachRolePolicy', 'PutUserPolicy', 'PutRolePolicy',
  'CreateAccessKey', 'CreateLoginProfile', 'UpdateLoginProfile',
  'PassRole', 'AssumeRole'
)
ORDER BY eventtime DESC
LIMIT 100;
 
-- Detect CloudTrail tampering
SELECT eventtime, useridentity.arn, eventname, requestparameters, sourceipaddress
FROM cloudtrail_logs
WHERE eventname IN ('StopLogging', 'DeleteTrail', 'UpdateTrail', 'PutEventSelectors')
ORDER BY eventtime DESC;
 
-- Find API calls from Tor exit nodes or unusual IPs
SELECT eventtime, useridentity.arn, eventname, sourceipaddress, awsregion
FROM cloudtrail_logs
WHERE sourceipaddress NOT LIKE '10.%'
  AND sourceipaddress NOT LIKE '172.%'
  AND sourceipaddress NOT LIKE '192.168.%'
  AND useridentity.type = 'IAMUser'
  AND errorcode IS NULL
GROUP BY eventtime, useridentity.arn, eventname, sourceipaddress, awsregion
ORDER BY eventtime DESC
LIMIT 200;
 
-- Detect unauthorized API calls (AccessDenied patterns)
SELECT useridentity.arn, eventname, COUNT(*) as denied_count
FROM cloudtrail_logs
WHERE errorcode IN ('AccessDenied', 'UnauthorizedAccess', 'Client.UnauthorizedAccess')
  AND eventtime > date_format(date_add('day', -7, now()), '%Y-%m-%dT%H:%i:%sZ')
GROUP BY useridentity.arn, eventname
HAVING COUNT(*) > 10
ORDER BY denied_count DESC;

Step 4: Build Real-Time Detection with CloudWatch Logs Insights

Create real-time queries for active security monitoring.

# Detect root account usage
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, eventName, sourceIPAddress, userAgent
    | filter userIdentity.type = "Root"
    | sort @timestamp desc
  '
 
# Detect security group changes
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, userIdentity.arn, eventName, requestParameters.groupId, sourceIPAddress
    | filter eventName in ["AuthorizeSecurityGroupIngress", "AuthorizeSecurityGroupEgress", "RevokeSecurityGroupIngress", "CreateSecurityGroup"]
    | sort @timestamp desc
  '
 
# Detect new IAM users or access keys created
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, userIdentity.arn, eventName, requestParameters.userName, sourceIPAddress
    | filter eventName in ["CreateUser", "CreateAccessKey", "CreateLoginProfile"]
    | sort @timestamp desc
  '

Step 5: Create CloudWatch Metric Filters and Alarms

Set up automated alerting for critical security events based on CIS Benchmark recommendations.

# CIS 3.1: Unauthorized API calls alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name unauthorized-api-calls \
  --filter-pattern '{($.errorCode = "*UnauthorizedAccess") || ($.errorCode = "AccessDenied*")}' \
  --metric-transformations '[{"metricName":"UnauthorizedAPICalls","metricNamespace":"CISBenchmark","metricValue":"1"}]'
 
aws cloudwatch put-metric-alarm \
  --alarm-name cis-unauthorized-api-calls \
  --metric-name UnauthorizedAPICalls --namespace CISBenchmark \
  --statistic Sum --period 300 --threshold 10 \
  --comparison-operator GreaterThanThreshold --evaluation-periods 1 \
  --alarm-actions arn:aws:sns:us-east-1:ACCOUNT:security-alerts
 
# CIS 3.3: Root account usage alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name root-account-usage \
  --filter-pattern '{$.userIdentity.type = "Root" && $.userIdentity.invokedBy NOT EXISTS && $.eventType != "AwsServiceEvent"}' \
  --metric-transformations '[{"metricName":"RootAccountUsage","metricNamespace":"CISBenchmark","metricValue":"1"}]'
 
# CIS 3.4: IAM policy changes alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name iam-policy-changes \
  --filter-pattern '{($.eventName=CreatePolicy) || ($.eventName=DeletePolicy) || ($.eventName=AttachRolePolicy) || ($.eventName=DetachRolePolicy) || ($.eventName=AttachUserPolicy) || ($.eventName=DetachUserPolicy)}' \
  --metric-transformations '[{"metricName":"IAMPolicyChanges","metricNamespace":"CISBenchmark","metricValue":"1"}]'
 
# CIS 3.5: CloudTrail configuration changes alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name cloudtrail-changes \
  --filter-pattern '{($.eventName = StopLogging) || ($.eventName = DeleteTrail) || ($.eventName = UpdateTrail)}' \
  --metric-transformations '[{"metricName":"CloudTrailChanges","metricNamespace":"CISBenchmark","metricValue":"1"}]'

Key Concepts

Term	Definition
CloudTrail	AWS service that records API calls made to AWS services, providing an audit trail of actions taken by users, roles, and services
Management Events	CloudTrail events for control plane operations like creating resources, modifying IAM, and configuring services
Data Events	CloudTrail events for data plane operations like S3 object access and Lambda function invocations, providing granular activity logging
Log File Validation	CloudTrail feature that creates a digest file for verifying that log files have not been tampered with after delivery
CloudTrail Lake	Managed data lake for CloudTrail events enabling SQL-based queries without managing Athena tables or S3 data
Organization Trail	Single trail that captures API activity across all accounts in an AWS Organization to a central S3 bucket

Tools & Systems

Amazon Athena: Serverless SQL query engine for analyzing CloudTrail logs stored in S3 at scale
CloudWatch Logs Insights: Real-time log query service for interactive CloudTrail analysis within the last 30 days
CloudTrail Lake: Managed event data lake with built-in SQL query capabilities and 7-year retention
Amazon Security Lake: Centralized security data lake that normalizes CloudTrail data into OCSF format for SIEM consumption
AWS CloudTrail: Core audit logging service capturing all API activity across AWS accounts and services

Common Scenarios

Scenario: Investigating an IAM Credential Compromise Through CloudTrail

Context: GuardDuty alerts on UnauthorizedAccess:IAMUser/MaliciousIPCaller for a developer's access key. The security team needs to trace all actions taken by the compromised credential.

Approach:

Query CloudTrail for all events by the compromised AccessKeyId across all regions
Build a timeline of API calls to understand the attack sequence
Identify the initial access point (when did the key first appear from a malicious IP)
Map all resources created, modified, or accessed by the attacker
Check for persistence mechanisms (new users, access keys, Lambda functions, EC2 instances)
Verify CloudTrail was not tampered with (check for StopLogging or UpdateTrail events)
Document the full attack chain and scope of impact for the incident response report

Pitfalls: CloudTrail events can take up to 15 minutes to appear in S3 and CloudWatch Logs. For real-time visibility during active incidents, use CloudTrail Lake or CloudWatch Logs Insights rather than Athena queries against S3. Cross-region attacks require querying multiple region partitions in Athena.

Output Format

CloudTrail Security Analysis Report
======================================
Account: 123456789012
Analysis Period: 2026-02-16 to 2026-02-23
Trail: org-security-trail (organization-wide)
 
SECURITY EVENTS DETECTED:
  Root account logins:                  2
  Console logins without MFA:           7
  Privilege escalation attempts:       12
  CloudTrail configuration changes:     0
  Security group modifications:        34
  Unauthorized API calls:             156
 
HIGH-PRIORITY FINDINGS:
[CT-001] Console Login Without MFA
  User: admin-user
  Time: 2026-02-22T14:30:00Z
  IP: 203.0.113.50
  Action Required: Enforce MFA via IAM policy
 
[CT-002] IAM Privilege Escalation
  User: dev-user
  Time: 2026-02-23T03:15:00Z
  Events: CreatePolicyVersion -> AttachRolePolicy
  IP: 185.x.x.x (suspicious)
  Action Required: Investigate credential compromise
 
ALERTING STATUS:
  CIS metric filters configured: 14 / 14
  CloudWatch alarms active: 14 / 14
  Alerts fired (last 7 days): 8

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.2 KB

API Reference: Implementing CloudTrail Log Analysis

Libraries

boto3 -- AWS CloudTrail

Install: pip install boto3
Docs: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cloudtrail.html

Key Methods

Method	Description
`lookup_events()`	Search recent CloudTrail events with filters
`describe_trails()`	List configured trails
`get_trail_status()`	Check if trail is actively logging
`create_trail()`	Create a new CloudTrail trail
`start_logging()` / `stop_logging()`	Control trail recording
`get_event_selectors()`	View event type configuration
`put_event_selectors()`	Configure management/data event capture

Lookup Attributes

AttributeKey	Description
`EventName`	API action name (e.g., `RunInstances`)
`Username`	IAM user or role name
`ResourceType`	AWS resource type
`ResourceName`	Specific resource identifier
`EventSource`	AWS service (e.g., `ec2.amazonaws.com`)
`ReadOnly`	Filter read vs write events

Suspicious Event Names

Event	Threat Category
`StopLogging` / `DeleteTrail`	Anti-forensics
`CreateUser` / `CreateAccessKey`	Persistence
`AttachUserPolicy` / `PutUserPolicy`	Privilege escalation
`ConsoleLogin` (failed)	Brute force
`RunInstances`	Resource abuse / cryptomining
`AuthorizeSecurityGroupIngress`	Lateral movement
`DisableKey`	Ransomware indicator

Athena Query Integration

Create Athena table from CloudTrail S3 logs
SQL queries for historical analysis beyond 90-day API limit
Partition by region, year, month for performance

CloudWatch Logs Insights

filter eventName = "ConsoleLogin" -- Login analysis
stats count(*) by eventName -- API call frequency
filter errorCode = "AccessDenied" -- Permission issues

External References

CloudTrail User Guide: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/
CloudTrail Log Events: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-event-reference.html
Athena + CloudTrail: https://docs.aws.amazon.com/athena/latest/ug/cloudtrail-logs.html

Scripts 1

agent.py7.4 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""CloudTrail log analysis agent for security monitoring and threat detection."""

import json
import sys
import argparse
from datetime import datetime, timedelta
from collections import Counter

try:
    import boto3
    from botocore.exceptions import ClientError
except ImportError:
    print("Install boto3: pip install boto3")
    sys.exit(1)


SUSPICIOUS_EVENTS = {
    "ConsoleLogin": "Potential unauthorized console access",
    "StopLogging": "CloudTrail logging disabled - cover tracks",
    "DeleteTrail": "CloudTrail trail deleted - evidence destruction",
    "CreateUser": "New IAM user created - possible persistence",
    "CreateAccessKey": "New access key - potential credential theft",
    "AttachUserPolicy": "Policy attached to user - privilege escalation",
    "PutBucketPolicy": "S3 bucket policy changed - data exposure risk",
    "AuthorizeSecurityGroupIngress": "Security group opened - lateral movement",
    "RunInstances": "EC2 instances launched - cryptomining or C2",
    "CreateRole": "New IAM role created - privilege escalation",
    "AssumeRole": "Role assumed - potential lateral movement",
    "PutUserPolicy": "Inline policy added to user",
    "DeleteBucketEncryption": "Bucket encryption removed",
    "DisableKey": "KMS key disabled - ransomware indicator",
    "ModifyInstanceAttribute": "Instance attribute changed",
}


def get_cloudtrail_client(region="us-east-1"):
    """Create CloudTrail client."""
    return boto3.client("cloudtrail", region_name=region)


def lookup_events(client, event_name=None, hours=24, max_results=50):
    """Look up CloudTrail events with optional filtering."""
    start_time = datetime.utcnow() - timedelta(hours=hours)
    kwargs = {"StartTime": start_time, "MaxResults": max_results,
              "LookupAttributes": []}
    if event_name:
        kwargs["LookupAttributes"] = [{"AttributeKey": "EventName", "AttributeValue": event_name}]
    try:
        resp = client.lookup_events(**kwargs)
        events = []
        for e in resp.get("Events", []):
            detail = json.loads(e.get("CloudTrailEvent", "{}"))
            events.append({
                "event_name": e.get("EventName"),
                "event_time": str(e.get("EventTime")),
                "username": e.get("Username", "unknown"),
                "source_ip": detail.get("sourceIPAddress", "unknown"),
                "user_agent": detail.get("userAgent", "unknown"),
                "region": detail.get("awsRegion", "unknown"),
                "error_code": detail.get("errorCode"),
                "error_message": detail.get("errorMessage"),
                "resources": [r.get("ResourceName", "") for r in e.get("Resources", [])],
            })
        return events
    except ClientError as e:
        return [{"error": str(e)}]


def detect_suspicious_activity(client, hours=24):
    """Scan CloudTrail for suspicious API calls."""
    detections = []
    for event_name, description in SUSPICIOUS_EVENTS.items():
        events = lookup_events(client, event_name=event_name, hours=hours)
        for e in events:
            if e.get("error"):
                continue
            severity = "CRITICAL" if event_name in ["StopLogging", "DeleteTrail", "DisableKey"] \
                else "HIGH" if event_name in ["CreateUser", "CreateAccessKey", "AttachUserPolicy"] \
                else "MEDIUM"
            detections.append({
                "event": event_name, "description": description,
                "severity": severity, "user": e["username"],
                "source_ip": e["source_ip"], "time": e["event_time"],
                "resources": e["resources"],
            })
    return sorted(detections, key=lambda x: {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2}.get(x["severity"], 3))


def detect_failed_auth(client, hours=24):
    """Detect failed authentication attempts."""
    events = lookup_events(client, event_name="ConsoleLogin", hours=hours, max_results=100)
    failed = [e for e in events if e.get("error_code")]
    by_ip = Counter(e["source_ip"] for e in failed)
    by_user = Counter(e["username"] for e in failed)
    return {"total_failed": len(failed), "by_source_ip": dict(by_ip.most_common(10)),
            "by_username": dict(by_user.most_common(10))}


def detect_unauthorized_regions(client, authorized_regions, hours=24):
    """Detect API calls from unauthorized AWS regions."""
    events = lookup_events(client, hours=hours, max_results=100)
    unauthorized = [e for e in events if e.get("region") and
                    e["region"] not in authorized_regions and not e.get("error")]
    return unauthorized


def analyze_user_activity(client, username, hours=24):
    """Analyze all activity for a specific user."""
    kwargs = {"StartTime": datetime.utcnow() - timedelta(hours=hours),
              "MaxResults": 50,
              "LookupAttributes": [{"AttributeKey": "Username", "AttributeValue": username}]}
    try:
        resp = client.lookup_events(**kwargs)
        actions = Counter()
        timeline = []
        for e in resp.get("Events", []):
            actions[e.get("EventName")] += 1
            timeline.append({"event": e.get("EventName"), "time": str(e.get("EventTime"))})
        return {"user": username, "total_events": len(timeline),
                "actions": dict(actions.most_common(20)), "timeline": timeline[:20]}
    except ClientError as e:
        return {"error": str(e)}


def run_cloudtrail_analysis(region="us-east-1", hours=24):
    """Run comprehensive CloudTrail security analysis."""
    client = get_cloudtrail_client(region)

    print(f"\n{'='*60}")
    print(f"  CLOUDTRAIL SECURITY ANALYSIS")
    print(f"  Region: {region} | Lookback: {hours}h")
    print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
    print(f"{'='*60}\n")

    detections = detect_suspicious_activity(client, hours)
    print(f"--- SUSPICIOUS ACTIVITY ({len(detections)} detections) ---")
    for d in detections[:15]:
        print(f"  [{d['severity']}] {d['event']}: {d['description']}")
        print(f"    User: {d['user']} | IP: {d['source_ip']} | Time: {d['time']}")

    auth = detect_failed_auth(client, hours)
    print(f"\n--- FAILED AUTHENTICATION ---")
    print(f"  Total failures: {auth['total_failed']}")
    print(f"  Top IPs: {auth['by_source_ip']}")
    print(f"  Top Users: {auth['by_username']}")

    print(f"\n{'='*60}\n")
    return {"detections": detections, "auth_failures": auth}


def main():
    parser = argparse.ArgumentParser(description="CloudTrail Log Analysis Agent")
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--hours", type=int, default=24, help="Lookback period in hours")
    parser.add_argument("--analyze", action="store_true", help="Run full analysis")
    parser.add_argument("--user", help="Analyze specific user activity")
    parser.add_argument("--output", help="Save report to JSON")
    args = parser.parse_args()

    if args.user:
        client = get_cloudtrail_client(args.region)
        result = analyze_user_activity(client, args.user, args.hours)
        print(json.dumps(result, indent=2, default=str))
    elif args.analyze:
        report = run_cloudtrail_analysis(args.region, args.hours)
        if args.output:
            with open(args.output, "w") as f:
                json.dump(report, f, indent=2, default=str)
            print(f"[+] Report saved to {args.output}")
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Keep exploring