v1.0.5

terraform · Dec 30, 2025 · ef75628 · ef75628
1 parent a3115b6
commit ef75628
Showing 1 changed file with 34 additions and 26 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
@@ -5,42 +5,48 @@
 from botocore.exceptions import ClientError, ProfileNotFound
 
 def get_args():
-    parser = argparse.ArgumentParser(description="Test AWS Org Connectivity by assuming roles.")
+    parser = argparse.ArgumentParser(description="Test AWS Org Connectivity with assuming roles.")
 
-    # Defaults are pulled from environment variables if the flags are omitted
+    # Defaults pull from environment variables (AWS_PROFILE / AWS_DEFAULT_REGION)
     parser.add_argument("--profile", 
                         default=os.environ.get("AWS_PROFILE"),
-                        help="AWS CLI profile (default: AWS_PROFILE env var)")
+                        help="AWS CLI profile name")
 
     parser.add_argument("--region", 
                         default=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
-                        help="AWS Region (default: AWS_DEFAULT_REGION or us-east-1)")
+                        help="AWS Region (default: us-east-1)")
 
     parser.add_argument("--role-name", 
                         required=True, 
-                        help="The name of the role to assume in member accounts")
+                        help="The IAM role name to assume in member accounts")
 
     return parser.parse_args()
 
 def test_connectivity(args):
     try:
-        # Initialize session. Boto3 naturally handles None for profile_name 
-        # by falling back to 'default' or environment credentials.
+        # Initialize the base session
         session = boto3.Session(profile_name=args.profile, region_name=args.region)
 
-        sts_client = session.client('sts')
-        org_client = session.client('organizations')
+        # 1. Print Pre-flight Check (The "Where am I?" info)
+        current_profile = args.profile or "Default/Environment"
+        current_region = session.region_name
 
-        # Verify Management Account Identity
+        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("-" * 60)
+
+        # 2. Get Management Account Identity
+        sts_client = session.client('sts')
         caller = sts_client.get_caller_identity()
-        print(f"--- Session Initiated ---")
-        print(f"Identity: {caller['Arn']}")
-        print(f"Profile:  {args.profile or 'Environment/Default'}")
-        print(f"Region:   {session.region_name}")
-        print(f"Role:     {args.role_name}\n")
-
+        print(f"Sourcing identity from: {caller['Arn']}\n")
+
+        # 3. Start Organization loop
+        org_client = session.client('organizations')
         print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}")
-        print("-" * 65)
+        print("-" * 60)
 
         paginator = org_client.get_paginator('list_accounts')
         for page in paginator.paginate():
@@ -53,22 +59,24 @@ def test_connectivity(args):
                 role_arn = f"arn:aws:iam::{acc_id}:role/{args.role_name}"
 
                 try:
-                    # Assume role using the Management Account's credentials
+                    # Attempting the AssumeRole
                     assumed = sts_client.assume_role(
                         RoleArn=role_arn,
-                        RoleSessionName="OrgTest"
+                        RoleSessionName="ConnectivityTest"
                     )
 
-                    # Verify by calling STS inside the member account
-                    temp = assumed['Credentials']
+                    # Prove success by verifying identity inside the member account
+                    creds = assumed['Credentials']
                     member_sts = boto3.client(
                         'sts',
-                        aws_access_key_id=temp['AccessKeyId'],
-                        aws_secret_access_key=temp['SecretAccessKey'],
-                        aws_session_token=temp['SessionToken']
+                        aws_access_key_id=creds['AccessKeyId'],
+                        aws_secret_access_key=creds['SecretAccessKey'],
+                        aws_session_token=creds['SessionToken'],
+                        region_name=current_region
                     )
-                    member_id = member_sts.get_caller_identity()['Account']
 
+                    # If this succeeds, connectivity is confirmed
+                    member_sts.get_caller_identity()
                     print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success")
 
                 except ClientError as e:
@@ -77,7 +85,7 @@ def test_connectivity(args):
     except ProfileNotFound:
         print(f"Error: Profile '{args.profile}' not found.")
     except Exception as e:
-        print(f"Error: {e}")
+        print(f"An unexpected error occurred: {e}")
 
 if __name__ == "__main__":
     test_connectivity(get_args())