diff --git a/local-app/python-tools/cross-organization/check_config.py b/local-app/python-tools/cross-organization/check_config.py index 0b052216..2395b4e8 100644 --- a/local-app/python-tools/cross-organization/check_config.py +++ b/local-app/python-tools/cross-organization/check_config.py @@ -1,13 +1,13 @@ -import re +import boto3 from datetime import datetime, timedelta +__version__ = "1.0.5" + def get_s3_metrics(session, bucket_name, region): - """Fetches NumberOfObjects and BucketSizeBytes 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) + start = end - timedelta(days=2) # Config metrics are once per day try: for metric_name in ['BucketSizeBytes', 'NumberOfObjects']: @@ -18,22 +18,16 @@ def get_s3_metrics(session, bucket_name, region): {'Name': 'BucketName', 'Value': bucket_name}, {'Name': 'StorageType', 'Value': 'StandardStorage'} ], - StartTime=start, - EndTime=end, - Period=86400, - Statistics=['Average'] + StartTime=start, EndTime=end, Period=86400, Statistics=['Average'] ) - if response['Datapoints']: key = "bucket_size_bytes" if metric_name == 'BucketSizeBytes' else "object_count" metrics[key] = int(response['Datapoints'][-1]['Average']) - except: - pass # Return 0s if metrics fail (e.g., bucket in different region) + except: pass return metrics -def account_task(account_session, account_id, account_name, region, bucket_regex=".*"): +def account_task(account_session, account_id, account_name, region): results = {"alias": "None", "data": {}} - try: iam = account_session.client('iam') aliases = iam.list_account_aliases().get('AccountAliases', []) @@ -45,27 +39,23 @@ def account_task(account_session, account_id, account_name, region, bucket_regex for reg in regions: config = account_session.client('config', region_name=reg) channels = config.describe_delivery_channels().get('DeliveryChannels', []) + recorders = config.describe_configuration_recorders().get('ConfigurationRecorders', []) + retention = config.describe_retention_configurations().get('RetentionConfigurations', []) - reg_data = { - "s3_bucket": "N/A", - "bucket_regex_match": "N/A", - "bucket_size_bytes": 0, - "object_count": 0 - } + reg_data = {"s3_bucket": "N/A", "recorder_status": "OFF", "retention_days": "Default"} + if recorders: reg_data["recorder_status"] = "ON" + if retention: reg_data["retention_days"] = str(retention[0]['RetentionPeriodInDays']) if channels: bucket = channels[0].get('s3BucketName', 'N/A') - reg_data["s3_bucket"] = bucket - reg_data["bucket_regex_match"] = str(bool(re.match(bucket_regex, bucket))) - - # Fetch metrics if bucket exists + reg_data.update({ + "s3_bucket": bucket, + "sns_topic": channels[0].get('snsTopicARN', 'N/A').split(':')[-1], + "delivery_freq": channels[0].get('configSnapshotDeliveryProperties', {}).get('deliveryFrequency', 'N/A') + }) if bucket != "N/A": - metrics = get_s3_metrics(account_session, bucket, reg) - reg_data.update(metrics) - - results["data"][reg] = reg_data + reg_data.update(get_s3_metrics(account_session, bucket, reg)) - except Exception as e: - results["error"] = str(e) - + results["data"][reg] = reg_data + except Exception as e: results["error"] = str(e) return results diff --git a/local-app/python-tools/cross-organization/org_runner.py b/local-app/python-tools/cross-organization/org_runner.py index b308ce93..d1fc45e2 100755 --- a/local-app/python-tools/cross-organization/org_runner.py +++ b/local-app/python-tools/cross-organization/org_runner.py @@ -7,22 +7,19 @@ import json import csv import time -import re import importlib from datetime import datetime from pathlib import Path # --- VERSIONING --- -__version__ = "1.4.0" +__version__ = "1.3.0" class OrgTaskRunner: def __init__(self, args): self.args = args self.output_log = [] self.full_results = [] - self.hierarchy_cache = {} self.start_time = 0 - self.org_id = "unknown" def log_print(self, message=""): print(message) @@ -34,10 +31,7 @@ def run(self): sts = session.client('sts') org = session.client('organizations') - try: - self.org_id = org.describe_organization()['Organization']['Id'] - except: pass - + # Load tasks tasks = [] if self.args.enable_checks: sys.path.append(os.getcwd()) @@ -45,14 +39,11 @@ def run(self): module = importlib.import_module(m.replace('.py', '')) tasks.append(getattr(module, 'account_task')) - all_accounts = [] paginator = org.get_paginator('list_accounts') - for page in paginator.paginate(): - all_accounts.extend([acc for acc in page['Accounts'] if acc['Status'] == 'ACTIVE']) - + all_accounts = [acc for page in paginator.paginate() for acc in page['Accounts'] if acc['Status'] == 'ACTIVE'] all_accounts.sort(key=lambda x: x['Name' if self.args.sort == 'name' else 'Id'].lower()) - self.log_print(f"--- Org Run v{__version__} | Regex: {self.args.central_bucket_regex} ---") + self.log_print(f"--- Starting Org Run v{__version__} ---") for i, acc in enumerate(all_accounts, 1): acc_id, acc_name = acc['Id'], acc['Name'] @@ -68,23 +59,19 @@ def run(self): ) account_data = {"account_id": acc_id, "account_name": acc_name, "checks": {}} - for t_func in tasks: - # Pass the regex to the task function - res = t_func(m_sess, acc_id, acc_name, self.args.region, self.args.central_bucket_regex) + res = t_func(m_sess, acc_id, acc_name, self.args.region) account_data["alias"] = res.get("alias") - account_data["checks"] = res.get("data", {}) + account_data["checks"].update(res.get("data", {})) self.full_results.append(account_data) self.log_print(f"[{i}/{len(all_accounts)}] Processed {acc_name}") - except Exception as e: self.log_print(f"[{i}/{len(all_accounts)}] FAILED {acc_name}: {e}") - # EXPORTS if self.args.output: ds = datetime.now().strftime("%Y-%m-%dT%H%M%S") - base = self.args.output if self.args.output != 'DEFAULT' else f"audit_{self.org_id}" + base = self.args.output if self.args.output != 'DEFAULT' else "audit_results" with open(f"{base}.{ds}.json", 'w') as f: json.dump(self.full_results, f, indent=2) @@ -94,18 +81,15 @@ def run(self): writer.writerow(["account_id", "account_alias", "region", "field_name", "field_value"]) for acc in self.full_results: for reg, fields in acc["checks"].items(): - for key, val in fields.items(): - writer.writerow([acc["account_id"], acc.get("alias"), reg, key, val]) - - self.log_print(f"\nExports complete: {base}.{ds}.json and .csv") + for k, v in fields.items(): + writer.writerow([acc["account_id"], acc.get("alias"), reg, k, v]) if __name__ == "__main__": p = argparse.ArgumentParser() p.add_argument("--profile", default=os.environ.get("AWS_PROFILE")) - p.add_argument("--region", default=os.environ.get("AWS_DEFAULT_REGION", "us-east-1")) + p.add_argument("--region", default="us-east-1") p.add_argument("--role-name", required=True) p.add_argument("--sort", choices=['id', 'name'], default='name') p.add_argument("--output", nargs='?', const='DEFAULT') p.add_argument("--enable-checks", nargs='+') - p.add_argument("--central-bucket-regex", default=".*", help="Regex to validate Config S3 buckets") OrgTaskRunner(p.parse_args()).run()