mirror of
https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git
synced 2026-09-05 07:50:51 +03:00
feat: add 5 cybersecurity skills - CloudTrail anomalies, SSL/TLS assessment, Wazuh detection, Prefetch analysis, WMI lateral movement
This commit is contained in:
mukul975
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2025 Mahipal
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,40 @@
|
||||
---
|
||||
name: detecting-aws-cloudtrail-anomalies
|
||||
description: Detect unusual API call patterns in AWS CloudTrail logs using boto3, statistical baselining, and behavioral analysis to identify credential compromise, privilege escalation, and unauthorized resource access.
|
||||
domain: cybersecurity
|
||||
subdomain: cloud-security
|
||||
tags: [cloud-security, aws, cloudtrail, anomaly-detection, threat-detection, boto3]
|
||||
version: "1.0"
|
||||
author: mahipal
|
||||
license: MIT
|
||||
---
|
||||
# Detecting AWS CloudTrail Anomalies
|
||||
|
||||
## Overview
|
||||
|
||||
AWS CloudTrail records API calls across AWS services. This skill covers querying CloudTrail events with boto3's `lookup_events` API, building statistical baselines of normal API activity, detecting anomalies such as unusual event sources, geographic anomalies, high-frequency API calls, and first-time API usage patterns that indicate compromised credentials or insider threats.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Python 3.9+ with `boto3` library
|
||||
- AWS credentials with CloudTrail read permissions (cloudtrail:LookupEvents)
|
||||
- Understanding of AWS IAM and common API patterns
|
||||
- CloudTrail enabled in target AWS account (management events at minimum)
|
||||
|
||||
## Steps
|
||||
|
||||
### Step 1: Query CloudTrail Events
|
||||
Use boto3 CloudTrail client's lookup_events to retrieve recent API activity with pagination.
|
||||
|
||||
### Step 2: Build Activity Baseline
|
||||
Aggregate events by user, source IP, event source, and event name to establish normal behavior patterns.
|
||||
|
||||
### Step 3: Detect Anomalies
|
||||
Flag unusual patterns: new event sources per user, first-time API calls, geographic IP changes, high error rates, and sensitive API usage (IAM, KMS, S3 policy changes).
|
||||
|
||||
### Step 4: Generate Detection Report
|
||||
Produce a JSON report with anomaly scores, top suspicious users, and recommended investigation actions.
|
||||
|
||||
## Expected Output
|
||||
|
||||
JSON report with event statistics, baseline deviations, anomalous users/IPs, sensitive API calls, and error rate analysis.
|
||||
@@ -0,0 +1,59 @@
|
||||
# API Reference: Detecting AWS CloudTrail Anomalies
|
||||
|
||||
## boto3 CloudTrail API
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
client = boto3.client("cloudtrail", region_name="us-east-1")
|
||||
|
||||
# Paginated event lookup
|
||||
paginator = client.get_paginator("lookup_events")
|
||||
pages = paginator.paginate(
|
||||
StartTime=datetime(2025, 1, 1),
|
||||
EndTime=datetime.utcnow(),
|
||||
LookupAttributes=[{"AttributeKey": "EventName", "AttributeValue": "ConsoleLogin"}],
|
||||
PaginationConfig={"MaxItems": 500, "PageSize": 50},
|
||||
)
|
||||
for page in pages:
|
||||
for event in page["Events"]:
|
||||
ct = json.loads(event["CloudTrailEvent"])
|
||||
print(ct["sourceIPAddress"], event["EventName"])
|
||||
```
|
||||
|
||||
## CloudTrail Event Fields
|
||||
|
||||
| Field | Location | Description |
|
||||
|-------|----------|-------------|
|
||||
| EventName | Event | API action name |
|
||||
| EventSource | Event | AWS service (e.g. iam.amazonaws.com) |
|
||||
| Username | Event | IAM user or assumed role |
|
||||
| sourceIPAddress | CloudTrailEvent JSON | Caller IP address |
|
||||
| errorCode | CloudTrailEvent JSON | Error type if failed |
|
||||
| userAgent | CloudTrailEvent JSON | Client SDK/browser |
|
||||
| awsRegion | CloudTrailEvent JSON | Region of API call |
|
||||
|
||||
## Sensitive API Calls to Monitor
|
||||
|
||||
| Event Name | Risk | Reason |
|
||||
|------------|------|--------|
|
||||
| StopLogging | Critical | Disabling CloudTrail |
|
||||
| DeleteTrail | Critical | Removing audit trail |
|
||||
| CreateAccessKey | High | New credentials for user |
|
||||
| AttachUserPolicy | High | Privilege escalation |
|
||||
| PutBucketPolicy | High | S3 access change |
|
||||
| ConsoleLogin | Medium | Interactive access |
|
||||
| RunInstances | Medium | Resource creation |
|
||||
| AssumeRole | Medium | Role switching |
|
||||
|
||||
## Rate Limits
|
||||
|
||||
- lookup_events: 2 requests/second per account per region
|
||||
- Maximum lookback: 90 days
|
||||
- Max results per page: 50 events
|
||||
|
||||
## References
|
||||
|
||||
- boto3 CloudTrail: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cloudtrail.html
|
||||
- CloudTrail Insights: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/logging-insights-events-with-cloudtrail.html
|
||||
- LookupEvents API: https://docs.aws.amazon.com/awscloudtrail/latest/APIReference/API_LookupEvents.html
|
||||
@@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Agent for detecting anomalies in AWS CloudTrail logs.
|
||||
|
||||
Queries CloudTrail events via boto3, builds behavioral baselines,
|
||||
and detects unusual API patterns indicating credential compromise,
|
||||
privilege escalation, or unauthorized access.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from collections import Counter, defaultdict
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import boto3
|
||||
except ImportError:
|
||||
boto3 = None
|
||||
|
||||
SENSITIVE_EVENTS = {
|
||||
"CreateUser", "CreateAccessKey", "AttachUserPolicy", "AttachRolePolicy",
|
||||
"PutUserPolicy", "PutRolePolicy", "CreateRole", "AssumeRole",
|
||||
"ConsoleLogin", "PutBucketPolicy", "PutBucketAcl",
|
||||
"CreateKeyPair", "RunInstances", "StopLogging", "DeleteTrail",
|
||||
"DisableKey", "ScheduleKeyDeletion", "CreateGrante",
|
||||
"AuthorizeSecurityGroupIngress", "ModifyInstanceAttribute",
|
||||
}
|
||||
|
||||
ERROR_INDICATORS = {"AccessDenied", "UnauthorizedAccess", "Client.UnauthorizedAccess"}
|
||||
|
||||
|
||||
class CloudTrailAnomalyDetector:
|
||||
"""Detects anomalies in AWS CloudTrail API activity."""
|
||||
|
||||
def __init__(self, profile=None, region=None, lookback_hours=24,
|
||||
output_dir="./cloudtrail_anomalies"):
|
||||
self.output_dir = Path(output_dir)
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.lookback_hours = lookback_hours
|
||||
self.findings = []
|
||||
self.client = None
|
||||
if boto3:
|
||||
session = boto3.Session(profile_name=profile, region_name=region or "us-east-1")
|
||||
self.client = session.client("cloudtrail")
|
||||
|
||||
def fetch_events(self, max_results=1000):
|
||||
"""Fetch CloudTrail events using lookup_events with pagination."""
|
||||
if not self.client:
|
||||
return []
|
||||
start_time = datetime.utcnow() - timedelta(hours=self.lookback_hours)
|
||||
events = []
|
||||
paginator = self.client.get_paginator("lookup_events")
|
||||
page_iter = paginator.paginate(
|
||||
StartTime=start_time,
|
||||
EndTime=datetime.utcnow(),
|
||||
PaginationConfig={"MaxItems": max_results, "PageSize": 50},
|
||||
)
|
||||
for page in page_iter:
|
||||
for event in page.get("Events", []):
|
||||
ct_event = json.loads(event.get("CloudTrailEvent", "{}"))
|
||||
events.append({
|
||||
"event_name": event.get("EventName", ""),
|
||||
"event_source": event.get("EventSource", ""),
|
||||
"event_time": event.get("EventTime", "").isoformat()
|
||||
if hasattr(event.get("EventTime", ""), "isoformat")
|
||||
else str(event.get("EventTime", "")),
|
||||
"username": event.get("Username", ""),
|
||||
"source_ip": ct_event.get("sourceIPAddress", ""),
|
||||
"user_agent": ct_event.get("userAgent", ""),
|
||||
"error_code": ct_event.get("errorCode", ""),
|
||||
"error_message": ct_event.get("errorMessage", ""),
|
||||
"aws_region": ct_event.get("awsRegion", ""),
|
||||
"read_only": event.get("ReadOnly", ""),
|
||||
})
|
||||
return events
|
||||
|
||||
def build_baseline(self, events):
|
||||
"""Build behavioral baseline from events."""
|
||||
user_events = defaultdict(list)
|
||||
user_ips = defaultdict(set)
|
||||
user_sources = defaultdict(set)
|
||||
for e in events:
|
||||
user = e["username"]
|
||||
user_events[user].append(e["event_name"])
|
||||
user_ips[user].add(e["source_ip"])
|
||||
user_sources[user].add(e["event_source"])
|
||||
return {
|
||||
"user_event_counts": {u: len(evts) for u, evts in user_events.items()},
|
||||
"user_unique_ips": {u: len(ips) for u, ips in user_ips.items()},
|
||||
"user_unique_sources": {u: len(srcs) for u, srcs in user_sources.items()},
|
||||
}
|
||||
|
||||
def detect_anomalies(self, events):
|
||||
"""Detect anomalous patterns in CloudTrail events."""
|
||||
user_events = defaultdict(list)
|
||||
for e in events:
|
||||
user_events[e["username"]].append(e)
|
||||
|
||||
sensitive_calls = [e for e in events if e["event_name"] in SENSITIVE_EVENTS]
|
||||
for e in sensitive_calls:
|
||||
self.findings.append({
|
||||
"severity": "high", "type": "Sensitive API Call",
|
||||
"detail": f"{e['username']} called {e['event_name']} from {e['source_ip']}",
|
||||
})
|
||||
|
||||
error_events = [e for e in events if e["error_code"] in ERROR_INDICATORS]
|
||||
error_by_user = Counter(e["username"] for e in error_events)
|
||||
for user, count in error_by_user.items():
|
||||
if count >= 5:
|
||||
self.findings.append({
|
||||
"severity": "high", "type": "High Access Denied Rate",
|
||||
"detail": f"{user} received {count} AccessDenied errors",
|
||||
})
|
||||
|
||||
for user, evts in user_events.items():
|
||||
ips = {e["source_ip"] for e in evts}
|
||||
if len(ips) >= 5:
|
||||
self.findings.append({
|
||||
"severity": "medium", "type": "Multiple Source IPs",
|
||||
"detail": f"{user} accessed from {len(ips)} distinct IPs",
|
||||
})
|
||||
|
||||
trail_tampering = [e for e in events
|
||||
if e["event_name"] in ("StopLogging", "DeleteTrail", "UpdateTrail")]
|
||||
for e in trail_tampering:
|
||||
self.findings.append({
|
||||
"severity": "critical", "type": "CloudTrail Tampering",
|
||||
"detail": f"{e['username']} called {e['event_name']}",
|
||||
})
|
||||
|
||||
return {
|
||||
"sensitive_api_calls": len(sensitive_calls),
|
||||
"access_denied_events": len(error_events),
|
||||
"trail_tampering_events": len(trail_tampering),
|
||||
}
|
||||
|
||||
def generate_report(self):
|
||||
events = self.fetch_events()
|
||||
baseline = self.build_baseline(events)
|
||||
anomalies = self.detect_anomalies(events)
|
||||
|
||||
event_name_counts = Counter(e["event_name"] for e in events).most_common(20)
|
||||
source_counts = Counter(e["event_source"] for e in events).most_common(10)
|
||||
|
||||
report = {
|
||||
"report_date": datetime.utcnow().isoformat(),
|
||||
"lookback_hours": self.lookback_hours,
|
||||
"total_events": len(events),
|
||||
"top_event_names": event_name_counts,
|
||||
"top_event_sources": source_counts,
|
||||
"baseline": baseline,
|
||||
"anomaly_summary": anomalies,
|
||||
"findings": self.findings,
|
||||
"total_findings": len(self.findings),
|
||||
}
|
||||
out = self.output_dir / "cloudtrail_anomaly_report.json"
|
||||
with open(out, "w") as f:
|
||||
json.dump(report, f, indent=2, default=str)
|
||||
print(json.dumps(report, indent=2, default=str))
|
||||
return report
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Detect anomalies in AWS CloudTrail API activity"
|
||||
)
|
||||
parser.add_argument("--profile", default=None, help="AWS CLI profile name")
|
||||
parser.add_argument("--region", default="us-east-1", help="AWS region")
|
||||
parser.add_argument("--hours", type=int, default=24, help="Lookback window in hours")
|
||||
parser.add_argument("--output-dir", default="./cloudtrail_anomalies")
|
||||
args = parser.parse_args()
|
||||
|
||||
os.makedirs(args.output_dir, exist_ok=True)
|
||||
detector = CloudTrailAnomalyDetector(
|
||||
profile=args.profile, region=args.region,
|
||||
lookback_hours=args.hours, output_dir=args.output_dir,
|
||||
)
|
||||
detector.generate_report()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user