v1.0.7

infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py

@@ -2,102 +2,60 @@
 import argparse
 import os
 import sys
-import logging
 from botocore.exceptions import ClientError, ProfileNotFound
 
 def get_args():
-    parser = argparse.ArgumentParser(description="Test AWS Org Connectivity with assuming roles.")
-
-    # Defaults pull from environment variables (AWS_PROFILE / AWS_DEFAULT_REGION)
-    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 (default: us-east-1)")
-
-    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 Boto3/Botocore HTTP wire-trace logging")
-
+    parser = argparse.ArgumentParser(description="Debug AWS Org Cross-Account Connectivity.")
+    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")
     return parser.parse_args()
 
 def test_connectivity(args):
-    # 1. Handle Debug Logging immediately
-    if args.debug:
-        # This is the 'canonical' way to debug Boto3; 
-        # it logs everything (headers, payloads, etc.) to stdout.
-        boto3.set_stream_logger(name='botocore', level=logging.DEBUG)
-        print("!!! DEBUG LOGGING ENABLED (Botocore wire-trace) !!!\n")
-
     try:
-        # Initialize the base session
         session = boto3.Session(profile_name=args.profile, region_name=args.region)
-
-        # 2. Print Configuration Summary
-        current_profile = args.profile or "Default/Environment"
-        current_region = session.region_name
-
-        print("-" * 60)
-        print(f"PRE-FLIGHT CONFIGURATION:")
-        print(f"  Profile:      {current_profile}")
-        print(f"  Base Region:  {current_region}")
-        print(f"  Target Role:  {args.role_name}")
-        print(f"  Debug Mode:   {'ON' if args.debug else 'OFF'}")
-        print("-" * 60)
-
-        # 3. Get Management Account Identity
         sts_client = session.client('sts')
-        caller = sts_client.get_caller_identity()
-        print(f"Sourcing identity from: {caller['Arn']}\n")
-
-        # 4. Start Organization loop
         org_client = session.client('organizations')
-        print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}")
-        print("-" * 60)
+
+        # Identity Discovery
+        caller = sts_client.get_caller_identity()
+        source_arn = caller['Arn']
+
+        print("="*60)
+        print("DEBUG PRE-FLIGHT INFORMATION")
+        print("="*60)
+        print(f"SOURCE IDENTITY (WHO YOU ARE): {source_arn}")
+        print(f"SOURCE REGION:               {session.region_name}")
+        print(f"TARGET ROLE NAME:            {args.role_name}")
+        print(f"PROFILE BEING USED:          {args.profile or 'Environment/Default'}")
+        print("="*60 + "\n")
 
         paginator = org_client.get_paginator('list_accounts')
         for page in paginator.paginate():
             for account in page['Accounts']:
-                if account['Status'] != 'ACTIVE':
-                    continue
+                if account['Status'] != 'ACTIVE': continue
 
                 acc_id = account['Id']
-                acc_name = account['Name']
                 role_arn = f"arn:aws:iam::{acc_id}:role/{args.role_name}"
 
+                print(f"Attempting: {acc_id} | Role: {role_arn} | Region: {session.region_name}")
+
                 try:
-                    # Attempting the AssumeRole
                     assumed = sts_client.assume_role(
                         RoleArn=role_arn,
-                        RoleSessionName="ConnectivityTest"
-                    )
-
-                    # Verify identity inside the member account
-                    creds = assumed['Credentials']
-                    member_sts = boto3.client(
-                        'sts',
-                        aws_access_key_id=creds['AccessKeyId'],
-                        aws_secret_access_key=creds['SecretAccessKey'],
-                        aws_session_token=creds['SessionToken'],
-                        region_name=current_region
+                        RoleSessionName="DebugConnectivityTest"
                     )
-
-                    member_sts.get_caller_identity()
-                    print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success")
-
+                    print(f"  ✅ SUCCESS: Successfully assumed role in {acc_id}")
                 except ClientError as e:
-                    print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ {e.response['Error']['Code']}")
+                    code = e.response['Error']['Code']
+                    print(f"  ❌ FAILED: {code}")
+                    if code == "AccessDenied":
+                        print(f"     [!] Check that the Trust Policy in Account {acc_id} allows:")
+                        print(f"         Principal: {source_arn}")
+                print("-" * 60)
 
-    except ProfileNotFound:
-        print(f"Error: Profile '{args.profile}' not found.")
     except Exception as e:
-        print(f"An unexpected error occurred: {e}")
+        print(f"CRITICAL ERROR: {e}")
 
 if __name__ == "__main__":
     test_connectivity(get_args())