v1.0.8

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

@@ -5,57 +5,60 @@
 from botocore.exceptions import ClientError, ProfileNotFound
 
 def get_args():
-    parser = argparse.ArgumentParser(description="Debug AWS Org Cross-Account Connectivity.")
+    parser = argparse.ArgumentParser(description="Multi-Partition AWS Org Connectivity Test.")
     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("--region", default=os.environ.get("AWS_DEFAULT_REGION"), help="AWS Region")
     parser.add_argument("--role-name", required=True, help="Role name to assume in member accounts")
+    parser.add_argument("--debug", action="store_true", help="Enable debug logging")
     return parser.parse_args()
 
 def test_connectivity(args):
     try:
         session = boto3.Session(profile_name=args.profile, region_name=args.region)
         sts_client = session.client('sts')
-        org_client = session.client('organizations')
 
-        # Identity Discovery
+        # 1. Identify Partition and Identity
         caller = sts_client.get_caller_identity()
         source_arn = caller['Arn']
 
+        # Parse partition from the source ARN (format is arn:PARTITION:service:...)
+        partition = source_arn.split(':')[1]
+
         print("="*60)
-        print("DEBUG PRE-FLIGHT INFORMATION")
+        print("PRE-FLIGHT CONFIGURATION")
         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(f"  PARTITION DETECTED: {partition}")
+        print(f"  SOURCE IDENTITY:    {source_arn}")
+        print(f"  TARGET ROLE NAME:   {args.role_name}")
         print("="*60 + "\n")
 
+        org_client = session.client('organizations')
         paginator = org_client.get_paginator('list_accounts')
+
+        print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}")
+        print("-" * 65)
+
         for page in paginator.paginate():
             for account in page['Accounts']:
                 if account['Status'] != 'ACTIVE': continue
 
                 acc_id = account['Id']
-                role_arn = f"arn:aws:iam::{acc_id}:role/{args.role_name}"
-
-                print(f"Attempting: {acc_id} | Role: {role_arn} | Region: {session.region_name}")
+                # 2. Dynamically construct the ARN using the detected partition
+                role_arn = f"arn:{partition}:iam::{acc_id}:role/{args.role_name}"
 
                 try:
                     assumed = sts_client.assume_role(
                         RoleArn=role_arn,
-                        RoleSessionName="DebugConnectivityTest"
+                        RoleSessionName="MultiPartitionTest"
                     )
-                    print(f"  ✅ SUCCESS: Successfully assumed role in {acc_id}")
+                    print(f"{account['Name'][:25]:<25} | {acc_id:<15} | ✅ Success")
                 except ClientError as e:
-                    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)
+                    print(f"{account['Name'][:25]:<25} | {acc_id:<15} | ❌ {e.response['Error']['Code']}")
 
+    except ProfileNotFound:
+        print(f"Error: Profile '{args.profile}' not found.")
     except Exception as e:
-        print(f"CRITICAL ERROR: {e}")
+        print(f"Error: {e}")
 
 if __name__ == "__main__":
     test_connectivity(get_args())