diff --git a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py new file mode 100755 index 00000000..3c58e3da --- /dev/null +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -0,0 +1,145 @@ +import boto3 +import argparse +import os +import sys +import logging +import time +from botocore.exceptions import ClientError, ProfileNotFound + +# --- VERSIONING --- +__version__ = "1.0.17" + +def get_args(): + parser = argparse.ArgumentParser(description=f"AWS Org Connectivity Test v{__version__}") + parser.add_argument("--profile", default=os.environ.get("AWS_PROFILE"), help="AWS CLI profile") + parser.add_argument("--region", default=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), help="AWS Region") + parser.add_argument("--role-name", required=True, help="Role name to assume in member accounts") + parser.add_argument("--sort", choices=['id', 'name'], default='name', help="Field to sort results by") + parser.add_argument("--debug", action="store_true", help="Enable verbose wire-trace logging") + return parser.parse_args() + +def get_full_path(org_client, entity_id, cache): + """Resolves hierarchy path and caches results. Strips 'Root'.""" + if entity_id in cache: + return cache[entity_id] + + if entity_id.startswith('r-'): + cache[entity_id] = (None, entity_id) + return None, entity_id + + try: + parent_resp = org_client.list_parents(ChildId=entity_id) + parents = parent_resp.get('Parents', []) + if not parents: + return "Unknown", entity_id + + parent_id = parents[0]['Id'] + ou_desc = org_client.describe_organizational_unit(OrganizationalUnitId=entity_id) + ou_name = ou_desc['OrganizationalUnit']['Name'] + + parent_path, _ = get_full_path(org_client, parent_id, cache) + full_path = f"{parent_path}:{ou_name}" if parent_path else ou_name + + cache[entity_id] = (full_path, entity_id) + return full_path, entity_id + except Exception: + return f"Error({entity_id})", entity_id + +def test_connectivity(args): + # Start high-precision timer + start_time = time.perf_counter() + + if args.debug: + boto3.set_stream_logger(name='botocore', level=logging.DEBUG) + + try: + session = boto3.Session(profile_name=args.profile, region_name=args.region) + sts_client = session.client('sts') + org_client = session.client('organizations') + + caller = sts_client.get_caller_identity() + partition = caller['Arn'].split(':')[1] + + print("Gathering Organization accounts and hierarchy...") + all_accounts = [] + paginator = org_client.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 args.sort == 'name' else 'Id' + all_accounts.sort(key=lambda x: x[sort_key].lower()) + + table_width = 215 + print("-" * table_width) + print(f"AWS ORG CONNECTIVITY TEST - v{__version__} | Total Accounts: {len(all_accounts)}") + print("-" * table_width) + print(f"{'#':<4} | {'Account Name':<25} | {'Alias':<64} | {'Account ID':<15} | {'Status':<12} | {'Full OU Path'}") + print("-" * table_width) + + hierarchy_cache = {} + ou_summary = {} + + for i, acc in enumerate(all_accounts, 1): + acc_id = acc['Id'] + acc_name = acc['Name'] + role_arn = f"arn:{partition}:iam::{acc_id}:role/{args.role_name}" + + parent_resp = org_client.list_parents(ChildId=acc_id) + parents = parent_resp.get('Parents', []) + + if parents: + parent_id = parents[0]['Id'] + raw_path, ou_id = get_full_path(org_client, parent_id, hierarchy_cache) + ou_path = raw_path if raw_path else "N/A (Root)" + else: + ou_path = "Orphaned" + ou_id = "N/A" + + summary_key = (ou_path, ou_id) + if summary_key not in ou_summary: + ou_summary[summary_key] = {"success": 0, "fail": 0} + + alias = "None" + status_text = "PENDING" + + try: + assumed = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="ConnTest") + creds = assumed['Credentials'] + m_sess = boto3.Session( + aws_access_key_id=creds['AccessKeyId'], + aws_secret_access_key=creds['SecretAccessKey'], + aws_session_token=creds['SessionToken'], + region_name=session.region_name + ) + aliases = m_sess.client('iam').list_account_aliases().get('AccountAliases', []) + if aliases: + alias = aliases[0] + status_text = "SUCCESS" + ou_summary[summary_key]["success"] += 1 + except ClientError as e: + status_text = f"FAIL({e.response['Error']['Code']})" + ou_summary[summary_key]["fail"] += 1 + + print(f"{i:<4} | {acc_name[:25]:<25} | {alias:<64} | {acc_id:<15} | {status_text:<12} | {ou_path}") + + # Stop timer + end_time = time.perf_counter() + elapsed_seconds = end_time - start_time + minutes, seconds = divmod(int(elapsed_seconds), 60) + + # Output Summary with Elapsed Time + print("\n" + "=" * 85) + print(f"{'OU PATH SUMMARY':<45} | {'OU ID':<18} | {'SUCC':<6} | {'FAIL':<6}") + print("=" * 85) + for (path, ou_id), counts in sorted(ou_summary.items(), key=lambda x: x[0][0] if x[0][0] else ""): + print(f"{path[:45]:<45} | {ou_id:<18} | {counts['success']:<6} | {counts['fail']:<6}") + print("=" * 85) + print(f"Total Elapsed Time: {minutes}m {seconds}s") + print("=" * 85) + + except Exception as e: + print(f"CRITICAL ERROR: {e}") + +if __name__ == "__main__": + test_connectivity(get_args()) +