diff --git a/local-app/python-tools/cross-organization/assess_check_cloudtrail.py b/local-app/python-tools/cross-organization/assess_check_cloudtrail.py new file mode 100644 index 00000000..31e14e66 --- /dev/null +++ b/local-app/python-tools/cross-organization/assess_check_cloudtrail.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python + +import json +import argparse +import sys + +__version__ = "1.0.0" + +def main(): + parser = argparse.ArgumentParser(description="AWS CloudTrail Audit Assessor") + parser.add_argument("--input", required=True, help="JSON file from check_cloudtrail") + args = parser.parse_args() + + with open(args.input, 'r') as f: + data = json.load(f) + + print("-" * 120) + print(f"{'Account ID':<15} | {'Global Summary':<25} | {'Issues/Best Practices'}") + print("-" * 120) + + for account in data: + acc_id = account.get("account_id") + checks = account.get("data", {}) + summary = checks.get("account_summary", {}).get("_summary", "UNKNOWN") + + issues = [] + + # Logic for Summary Field examination + if "DUPLICATIVE" in summary: + issues.append("Redundant: Local trails exist alongside Org trail (Extra Cost)") + elif "NO_TRAILS" in summary: + issues.append("CRITICAL: No CloudTrail logging detected") + elif "LOCAL_ONLY" in summary: + issues.append("Note: Using local trails; consider migrating to Org Trail") + + # Detail Checks + for key, val in checks.items(): + if key.startswith("trail:"): + t_name = val.get("trail_name") + if val.get("is_logging") == "False": + issues.append(f"Trail {t_name} is STOPPED") + if val.get("is_multi_region") == "False": + issues.append(f"Trail {t_name} is Single-Region only") + if val.get("log_file_validation") == "False": + issues.append(f"Trail {t_name} missing Log File Validation") + if val.get("kms_key_id") == "SSE-S3 (Default)": + issues.append(f"Trail {t_name} not using CMK encryption") + if val.get("cw_logs_retention_days") == "Never Expire": + issues.append(f"Trail {t_name} CloudWatch Logs set to Never Expire (Cost Risk)") + + issue_str = " | ".join(issues) if issues else "Compliant" + print(f"{acc_id:<15} | {summary:<25} | {issue_str}") + +if __name__ == "__main__": + main() diff --git a/local-app/python-tools/cross-organization/check_cloudtrail.py b/local-app/python-tools/cross-organization/check_cloudtrail.py new file mode 100644 index 00000000..812bd4e2 --- /dev/null +++ b/local-app/python-tools/cross-organization/check_cloudtrail.py @@ -0,0 +1,102 @@ +import boto3 +import time +from datetime import datetime, timedelta + +__version__ = "1.0.0" + +def get_s3_metrics(session, bucket_name, region): + """Fetches bucket storage metrics from CloudWatch.""" + cw = session.client('cloudwatch', region_name=region) + metrics = {"bucket_size_bytes": 0, "object_count": 0} + end = datetime.utcnow() + start = end - timedelta(days=2) + try: + r1 = cw.get_metric_statistics( + Namespace='AWS/S3', MetricName='BucketSizeBytes', + Dimensions=[{'Name': 'BucketName', 'Value': bucket_name}, {'Name': 'StorageType', 'Value': 'StandardStorage'}], + StartTime=start, EndTime=end, Period=86400, Statistics=['Average'] + ) + if r1['Datapoints']: metrics["bucket_size_bytes"] = int(r1['Datapoints'][-1]['Average']) + + r2 = cw.get_metric_statistics( + Namespace='AWS/S3', MetricName='NumberOfObjects', + Dimensions=[{'Name': 'BucketName', 'Value': bucket_name}, {'Name': 'StorageType', 'Value': 'AllStorageTypes'}], + StartTime=start, EndTime=end, Period=86400, Statistics=['Average'] + ) + if r2['Datapoints']: metrics["object_count"] = int(r2['Datapoints'][-1]['Average']) + except: pass + return metrics + +def get_log_group_details(session, group_arn, region): + """Fetches retention and size for CloudWatch Log Groups.""" + if not group_arn: return {} + logs = session.client('logs', region_name=region) + group_name = group_arn.split(':')[-2] # Extract name from ARN + try: + resp = logs.describe_log_groups(logGroupNamePrefix=group_name) + for g in resp.get('logGroups', []): + if g['logGroupName'] == group_name: + return { + "cw_logs_retention_days": g.get('retentionInDays', 'Never Expire'), + "cw_logs_size_bytes": g.get('storedBytes', 0) + } + except: pass + return {} + +def account_task(account_session, account_id, account_name, region): + results = {"alias": "N/A", "data": {}} + org_trail_count = 0 + local_trail_count = 0 + + try: + results["alias"] = account_session.client('iam').list_account_aliases().get('AccountAliases', ["N/A"])[0] + ct = account_session.client('cloudtrail', region_name=region) + + # describe_trails retrieves trails for the current region (+ shadow trails if multi-region) + trails = ct.describe_trails(includeShadowTrails=True).get('trailList', []) + + for trail in trails: + trail_name = trail['Name'] + status = ct.get_trail_status(Name=trail_name) + + is_org = trail.get('IsOrganizationTrail', False) + if is_org: org_trail_count += 1 + else: local_trail_count += 1 + + trail_data = { + "trail_name": trail_name, + "is_logging": str(status.get('IsLogging', False)), + "is_org_trail": str(is_org), + "is_multi_region": str(trail.get('IsMultiRegionTrail', False)), + "log_file_validation": str(trail.get('LogFileValidationEnabled', False)), + "s3_bucket": trail.get('S3BucketName', 'N/A'), + "kms_key_id": trail.get('KmsKeyId', 'SSE-S3 (Default)'), + "sns_topic": trail.get('SNSTopicName', 'N/A'), + "home_region": trail.get('HomeRegion', 'N/A') + } + + # Fetch CloudWatch Logs Integration + if 'CloudWatchLogsLogGroupArn' in trail: + trail_data.update(get_log_group_details(account_session, trail['CloudWatchLogsLogGroupArn'], region)) + + # Fetch S3 Metrics (if bucket is in the same account) + if trail_data['s3_bucket'] != 'N/A': + trail_data.update(get_s3_metrics(account_session, trail_data['s3_bucket'], region)) + + results["data"][f"trail:{trail_name}"] = trail_data + + # Add Per-Account Summary + summary = "OK" + if org_trail_count > 0 and local_trail_count > 0: + summary = f"DUPLICATIVE/{local_trail_count}_LOCAL_WITH_ORG" + elif org_trail_count == 0 and local_trail_count > 0: + summary = f"LOCAL_ONLY/{local_trail_count}" + elif org_trail_count == 0 and local_trail_count == 0: + summary = "NO_TRAILS" + + results["data"]["account_summary"] = {"_summary": summary} + + except Exception as e: + results["error"] = str(e) + + return results