back to 1.3.0, version check_config

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

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,25 +31,19 @@ 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())
             for m in self.args.enable_checks:
                 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()