modularize

local-app/python-tools/test-cross-organization/README.md

@@ -1,4 +1,4 @@
-# README: AWS Organization Task Runner
+# AWS Organization Task Runner
 
 ## Overview
 

local-app/python-tools/test-cross-organization/README.org_runner.md

@@ -0,0 +1,276 @@
+To ensure you have a clean, production-ready setup, I have split the code into two distinct files: the **Core Runner** and an **Example Task Module**. I have also provided a comprehensive Markdown README to document the new plugin architecture.
+
+### 1. The Core Runner: `org_runner.py`
+
+This script handles the Organization hierarchy, credential assumption, and dynamic loading of external tasks.
+
+```python
+#!/usr/bin/env python3
+import boto3
+import argparse
+import os
+import sys
+import logging
+import time
+import importlib
+from datetime import datetime
+from pathlib import Path
+from botocore.exceptions import ClientError, ProfileNotFound
+
+# --- VERSIONING ---
+__version__ = "1.2.0"
+
+class OrgTaskRunner:
+    def __init__(self, args):
+        self.args = args
+        self.output_log = []
+        self.hierarchy_cache = {}
+        self.ou_summary = {}
+        self.start_time = 0
+        self.partition = "aws"
+        self.session = None
+        self.org_id = "unknown"
+
+    def log_print(self, message=""):
+        """Prints to console and buffers for file output."""
+        print(message)
+        self.output_log.append(message)
+
+    def get_full_path(self, org_client, entity_id):
+        """Resolves hierarchy path and caches results. Strips 'Root'."""
+        if entity_id in self.hierarchy_cache:
+            return self.hierarchy_cache[entity_id]
+        if entity_id.startswith('r-'):
+            self.hierarchy_cache[entity_id] = (None, entity_id)
+            return None, entity_id
+        try:
+            parents = org_client.list_parents(ChildId=entity_id)['Parents']
+            p_id = parents[0]['Id'] if parents else None
+            ou_desc = org_client.describe_organizational_unit(OrganizationalUnitId=entity_id)
+            ou_name = ou_desc['OrganizationalUnit']['Name']
+            p_path, _ = self.get_full_path(org_client, p_id) if p_id else (None, None)
+            full_path = f"{p_path}:{ou_name}" if p_path else ou_name
+            self.hierarchy_cache[entity_id] = (full_path, entity_id)
+            return full_path, entity_id
+        except:
+            return f"Error({entity_id})", entity_id
+
+    def load_tasks(self):
+        """Dynamically imports account_task from provided module names."""
+        tasks = []
+        if self.args.enable_checks:
+            for module_name in self.args.enable_checks:
+                try:
+                    module = importlib.import_module(module_name)
+                    if hasattr(module, 'account_task'):
+                        tasks.append((module_name, getattr(module, 'account_task')))
+                    else:
+                        self.log_print(f"Warning: {module_name} missing 'account_task' function.")
+                except ImportError as e:
+                    self.log_print(f"Error importing {module_name}: {e}")
+        return tasks
+
+    def run(self):
+        self.start_time = time.perf_counter()
+        if self.args.debug:
+            boto3.set_stream_logger(name='botocore', level=logging.DEBUG)
+
+        try:
+            self.session = boto3.Session(profile_name=self.args.profile, region_name=self.args.region)
+            sts = self.session.client('sts')
+            org = self.session.client('organizations')
+
+            caller = sts.get_caller_identity()
+            self.partition = caller['Arn'].split(':')[1]
+
+            try:
+                self.org_id = org.describe_organization()['Organization']['Id']
+            except: pass
+
+            tasks = self.load_tasks()
+
+            # Filename Logic
+            final_filename = None
+            if self.args.output:
+                ds = datetime.now().strftime("%Y-%m-%dT%H%M%S")
+                if self.args.output == 'DEFAULT':
+                    final_filename = f"results_{self.org_id}.{ds}.txt"
+                else:
+                    p = Path(self.args.output)
+                    final_filename = f"{p.stem}_{ds}{p.suffix}"
+
+            self.log_print("-" * 100)
+            self.log_print(f"AWS ORG TASK RUNNER - Version {__version__}")
+            self.log_print("-" * 100)
+            self.log_print(f"  Profile:      {self.args.profile or 'Default'}")
+            self.log_print(f"  Identity:     {caller['Arn']}")
+            self.log_print(f"  Checks:       {[t[0] for t in tasks] if tasks else 'Connectivity Only'}")
+            if final_filename: self.log_print(f"  Output File:  {final_filename}")
+            self.log_print("-" * 100)
+
+            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'])
+
+            sort_key = 'Name' if self.args.sort == 'name' else 'Id'
+            all_accounts.sort(key=lambda x: x[sort_key].lower())
+
+            self.log_print(f"{'#':<4} | {'Account Name':<25} | {'Alias':<40} | {'Account ID':<15} | {'Status':<12} | {'Full OU Path'}")
+            self.log_print("-" * 180)
+
+            for i, acc in enumerate(all_accounts, 1):
+                acc_id, acc_name = acc['Id'], acc['Name']
+                role_arn = f"arn:{self.partition}:iam::{acc_id}:role/{self.args.role_name}"
+
+                parents = org.list_parents(ChildId=acc_id).get('Parents', [])
+                raw_path, ou_id = self.get_full_path(org, parents[0]['Id']) if parents else ("Orphaned", "N/A")
+                ou_path = raw_path if raw_path else "N/A (Root)"
+
+                sum_key = (ou_path, ou_id)
+                if sum_key not in self.ou_summary: self.ou_summary[sum_key] = {"success": 0, "fail": 0}
+
+                alias, status_text = "None", "PENDING"
+                try:
+                    assumed = sts.assume_role(RoleArn=role_arn, RoleSessionName="OrgRunnerTask")
+                    m_sess = boto3.Session(
+                        aws_access_key_id=assumed['Credentials']['AccessKeyId'],
+                        aws_secret_access_key=assumed['Credentials']['SecretAccessKey'],
+                        aws_session_token=assumed['Credentials']['SessionToken'],
+                        region_name=self.session.region_name
+                    )
+
+                    # Default connectivity check + dynamic tasks
+                    if not tasks:
+                        aliases = m_sess.client('iam').list_account_aliases().get('AccountAliases', [])
+                        alias = aliases[0] if aliases else "None"
+                        status_text = "SUCCESS"
+                    else:
+                        for t_name, t_func in tasks:
+                            res = t_func(m_sess, acc_id, acc_name, self.session.region_name)
+                            status_text = res.get('status', 'SUCCESS')
+                            alias = res.get('alias', 'N/A')
+
+                    self.ou_summary[sum_key]["success"] += 1
+                except ClientError as e:
+                    status_text = f"FAIL({e.response['Error']['Code']})"
+                    self.ou_summary[sum_key]["fail"] += 1
+
+                self.log_print(f"{i:<4} | {acc_name[:25]:<25} | {alias[:40]:<40} | {acc_id:<15} | {status_text:<12} | {ou_path}")
+
+            # Summary Table
+            end_time = time.perf_counter()
+            el = divmod(int(end_time - self.start_time), 60)
+
+            self.log_print("\n" + "=" * 85)
+            self.log_print(f"{'OU PATH SUMMARY':<45} | {'OU ID':<18} | {'SUCC':<6} | {'FAIL':<6}")
+            self.log_print("=" * 85)
+            for (path, o_id), counts in sorted(self.ou_summary.items()):
+                self.log_print(f"{path[:45]:<45} | {o_id:<18} | {counts['success']:<6} | {counts['fail']:<6}")
+            self.log_print("=" * 85)
+            self.log_print(f"Time: {el[0]}m {el[1]}s | Accounts: {len(all_accounts)}")
+            self.log_print("=" * 85)
+
+            if final_filename:
+                with open(final_filename, 'w') as f:
+                    f.write("\n".join(self.output_log))
+
+        except Exception as e:
+            print(f"CRITICAL ERROR: {e}")
+
+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("--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("--debug", action="store_true")
+    OrgTaskRunner(p.parse_args()).run()
+
+```
+
+### 2. Example Task Module: `check_config.py`
+
+Save this in the same directory as the runner.
+
+```python
+def account_task(account_session, account_id, account_name, region):
+    """
+    Checks for AWS Config Global Resource Recording.
+    """
+    res = {"alias": "None", "status": "SUCCESS"}
+    try:
+        iam = account_session.client('iam')
+        aliases = iam.list_account_aliases().get('AccountAliases', [])
+        res["alias"] = aliases[0] if aliases else "None"
+
+        config = account_session.client('config', region_name=region)
+        recs = config.describe_configuration_recorders().get('ConfigurationRecorders', [])
+
+        if any(r.get('recordingGroup', {}).get('includeGlobalResourceTypes') for r in recs):
+            res["status"] = "GLOBAL_ON"
+        else:
+            res["status"] = "GLOBAL_OFF"
+    except Exception as e:
+        res["status"] = "ERROR"
+    return res
+
+```
+
+---
+
+# README: AWS Organization Task Runner (v1.2.0)
+
+## Overview
+
+A high-performance CLI utility for executing modular tasks across an AWS Organization. It automates cross-account authentication, handles multi-partition ARNs, and maps the results to your Organizational Unit (OU) hierarchy.
+
+## Core Features
+
+* **Dynamic Plugin System**: Load any Python module with an `account_task` function at runtime.
+* **Partition Agnostic**: Works across `aws`, `aws-us-gov`, and `aws-cn`.
+* **Automatic Reporting**: Generates real-time console tables and timestamped summary files.
+* **Smart Caching**: Minimizes API calls to AWS Organizations by caching OU paths.
+
+## Installation
+
+1. Ensure you have Python 3.8+ and `boto3` installed.
+2. Save `org_runner.py` and make it executable: `chmod +x org_runner.py`.
+3. Create a task module (e.g., `check_config.py`) in the same folder.
+
+## Usage
+
+Run a basic connectivity check:
+
+```bash
+./org_runner.py --role-name OrganizationAccountAccessRole
+
+```
+
+Run a specific logic module:
+
+```bash
+./org_runner.py --role-name MyAdminRole --enable-checks check_config --output
+
+```
+
+## Plugin Development
+
+To create a new check, define a function with this exact signature in a `.py` file:
+
+```python
+def account_task(account_session, account_id, account_name, region):
+    # account_session is a pre-authenticated boto3 Session for the target account
+    return {"status": "SUCCESS", "alias": "my-alias"}
+
+```
+
+## Changelog Summary
+
+* **v1.0.9**: Added Partition support.
+* **v1.0.14**: Added OU Hierarchy labels and Summary tables.
+* **v1.0.18**: Added dual console/file logging.
+* **v1.2.0**: Implemented `importlib` for dynamic task loading and refactored into a class-based runner.
+

local-app/python-tools/test-cross-organization/check_config.py

@@ -1,36 +1,20 @@
-def config_check_account_task(account_session, account_id, account_name, region):
+def account_task(account_session, account_id, account_name, region):
     """
-    Task: Checks for AWS Config Recorders in the specified region.
-    Identifies if 'IncludeGlobalResourceTypes' is enabled (potential cost/duplication).
+    Checks for AWS Config Global Resource Recording.
     """
-    results = {"alias": "None", "status": "SUCCESS", "details": ""}
-
+    res = {"alias": "None", "status": "SUCCESS"}
     try:
-        # 1. Get IAM Alias for reporting
         iam = account_session.client('iam')
         aliases = iam.list_account_aliases().get('AccountAliases', [])
-        results["alias"] = aliases[0] if aliases else "None"
-
-        # 2. Check AWS Config Recorder
+        res["alias"] = aliases[0] if aliases else "None"
+
         config = account_session.client('config', region_name=region)
-        recorders = config.describe_configuration_recorders().get('ConfigurationRecorders', [])
+        recs = config.describe_configuration_recorders().get('ConfigurationRecorders', [])
 
-        if not recorders:
-            results["status"] = "WARNING(NoRecorder)"
+        if any(r.get('recordingGroup', {}).get('includeGlobalResourceTypes') for r in recs):
+            res["status"] = "GLOBAL_ON"
         else:
-            # Check for Global Resource recording
-            for r in recorders:
-                group = r.get('recordingGroup', {})
-                if group.get('includeGlobalResourceTypes'):
-                    results["status"] = "ACTION_REQ"
-                    results["details"] = "GlobalResources:True"
-                else:
-                    results["status"] = "OPTIMIZED"
-                    results["details"] = "GlobalResources:False"
-
-    except ClientError as e:
-        results["status"] = f"FAIL({e.response['Error']['Code']})"
+            res["status"] = "GLOBAL_OFF"
     except Exception as e:
-        results["status"] = f"FAIL(Error)"
-
-    return results
+        res["status"] = "ERROR"
+    return res