diff --git a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py index bc7f912a..3c58e3da 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -3,106 +3,143 @@ import os import sys import logging +import time from botocore.exceptions import ClientError, ProfileNotFound # --- VERSIONING --- -__version__ = "1.0.10" +__version__ = "1.0.17" def get_args(): - parser = argparse.ArgumentParser(description=f"Multi-Partition AWS Org Connectivity Test v{__version__}") - - parser.add_argument("--profile", - default=os.environ.get("AWS_PROFILE"), - help="AWS CLI profile name") - - 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="The IAM role name to assume in member accounts") - - parser.add_argument("--debug", - action="store_true", - help="Enable verbose wire-trace logging") - + 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: - # Initialize management session session = boto3.Session(profile_name=args.profile, region_name=args.region) sts_client = session.client('sts') + org_client = session.client('organizations') - # Identify Partition and Identity caller = sts_client.get_caller_identity() - source_arn = caller['Arn'] - partition = source_arn.split(':')[1] + partition = caller['Arn'].split(':')[1] - print("-" * 100) - print(f"AWS ORG CONNECTIVITY TEST - Version {__version__}") - print("-" * 100) - print(f"CONFIGURATION:") - print(f" Partition: {partition}") - print(f" Profile: {args.profile or 'Environment/Default'}") - print(f" Region: {session.region_name}") - print(f" Source ARN: {source_arn}") - print(f" Target Role: {args.role_name}") - print("-" * 100) - - org_client = session.client('organizations') + 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']) - # Table Header - print(f"{'Account Name':<25} | {'Alias':<25} | {'Account ID':<15} | {'Status'}") - print("-" * 100) + 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) - for page in paginator.paginate(): - for account in page['Accounts']: - if account['Status'] != 'ACTIVE': - continue - - acc_id = account['Id'] - acc_name = account['Name'] - role_arn = f"arn:{partition}:iam::{acc_id}:role/{args.role_name}" - alias = "N/A" - - try: - # Assume Role - assumed = sts_client.assume_role( - RoleArn=role_arn, - RoleSessionName=f"ConnectivityTest-v{__version__.replace('.', '-')}" - ) - - # Create temporary session for member account - creds = assumed['Credentials'] - member_session = boto3.Session( - aws_access_key_id=creds['AccessKeyId'], - aws_secret_access_key=creds['SecretAccessKey'], - aws_session_token=creds['SessionToken'], - region_name=session.region_name - ) - - # Fetch Account Alias - iam_client = member_session.client('iam') - aliases = iam_client.list_account_aliases().get('AccountAliases', []) - if aliases: - alias = aliases[0] - - print(f"{acc_name[:25]:<25} | {alias[:25]:<25} | {acc_id:<15} | SUCCESS") - - except ClientError as e: - error_code = e.response['Error']['Code'] - print(f"{acc_name[:25]:<25} | {'ERROR':<25} | {acc_id:<15} | FAILED ({error_code})") - - except ProfileNotFound: - print(f"ERROR: Profile '{args.profile}' not found.") except Exception as e: print(f"CRITICAL ERROR: {e}") if __name__ == "__main__": test_connectivity(get_args()) +