v1.0.9

terraform · Dec 30, 2025 · f0b3629 · f0b3629
1 parent aea550e
commit f0b3629
Showing 1 changed file with 62 additions and 20 deletions.
diff --git a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py
@@ -2,35 +2,59 @@
 import argparse
 import os
 import sys
+import logging
 from botocore.exceptions import ClientError, ProfileNotFound
 
+# --- VERSIONING ---
+__version__ = "1.0.9"
+
 def get_args():
-    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"), 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")
+    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")
+
     return parser.parse_args()
 
 def test_connectivity(args):
+    if args.debug:
+        boto3.set_stream_logger(name='botocore', level=logging.DEBUG)
+
     try:
+        # Initialize session
         session = boto3.Session(profile_name=args.profile, region_name=args.region)
         sts_client = session.client('sts')
 
         # 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:...)
+        # Dynamically extract partition (e.g., 'aws', 'aws-us-gov', 'aws-cn')
         partition = source_arn.split(':')[1]
 
-        print("="*60)
-        print("PRE-FLIGHT CONFIGURATION")
-        print("="*60)
-        print(f"  PARTITION DETECTED: {partition}")
-        print(f"  SOURCE IDENTITY:    {source_arn}")
-        print(f"  TARGET ROLE NAME:   {args.role_name}")
-        print("="*60 + "\n")
+        print("-" * 65)
+        print(f"AWS ORG CONNECTIVITY TEST - Version {__version__}")
+        print("-" * 65)
+        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("-" * 65)
 
         org_client = session.client('organizations')
         paginator = org_client.get_paginator('list_accounts')
@@ -40,25 +64,43 @@ def test_connectivity(args):
 
         for page in paginator.paginate():
             for account in page['Accounts']:
-                if account['Status'] != 'ACTIVE': continue
+                if account['Status'] != 'ACTIVE':
+                    continue
 
                 acc_id = account['Id']
-                # 2. Dynamically construct the ARN using the detected partition
+                acc_name = account['Name']
+
+                # 2. Construct ARN using the detected partition
                 role_arn = f"arn:{partition}:iam::{acc_id}:role/{args.role_name}"
 
                 try:
+                    # Attempt AssumeRole
                     assumed = sts_client.assume_role(
                         RoleArn=role_arn,
-                        RoleSessionName="MultiPartitionTest"
+                        RoleSessionName=f"ConnectivityTest-v{__version__.replace('.', '-')}"
                     )
-                    print(f"{account['Name'][:25]:<25} | {acc_id:<15} | ✅ Success")
+
+                    # Verify by creating a client 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=session.region_name
+                    )
+                    member_sts.get_caller_identity()
+
+                    print(f"{acc_name[:25]:<25} | {acc_id:<15} | SUCCESS")
+
                 except ClientError as e:
-                    print(f"{account['Name'][:25]:<25} | {acc_id:<15} | ❌ {e.response['Error']['Code']}")
+                    error_code = e.response['Error']['Code']
+                    print(f"{acc_name[:25]:<25} | {acc_id:<15} | FAILED ({error_code})")
 
     except ProfileNotFound:
-        print(f"Error: Profile '{args.profile}' not found.")
+        print(f"ERROR: Profile '{args.profile}' not found.")
     except Exception as e:
-        print(f"Error: {e}")
+        print(f"CRITICAL ERROR: {e}")
 
 if __name__ == "__main__":
     test_connectivity(get_args())