From c86a34030a29cf4b702418792e654bdfa1827fa2 Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:17:31 -0500 Subject: [PATCH 01/10] v1.0.1 --- .../inf-org-crossaccount/test-cross-org.py | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100755 infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py diff --git a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py new file mode 100755 index 00000000..f9586017 --- /dev/null +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -0,0 +1,56 @@ +import boto3 +from botocore.exceptions import ClientError + +# --- SET THESE TO MATCH YOUR TERRAFORM --- +CROSS_ACCOUNT_ROLE_NAME = "OrganizationAdminAccessRole" +TARGET_REGION = "us-east-1" + +def test_connectivity(): + # 1. Initialize the Organizations client from Management Account + org_client = boto3.client('organizations') + sts_client = boto3.client('sts') + + print(f"Starting connectivity test using role: {CROSS_ACCOUNT_ROLE_NAME}\n") + print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") + print("-" * 55) + + # 2. Page through all accounts in the Org + paginator = org_client.get_paginator('list_accounts') + for page in paginator.paginate(): + for account in page['Accounts']: + # Skip accounts that aren't active (suspended/pending) + if account['State'] != 'ACTIVE': + continue + + acc_id = account['Id'] + acc_name = account['Name'] + role_arn = f"arn:aws:iam::{acc_id}:role/{CROSS_ACCOUNT_ROLE_NAME}" + + try: + # 3. Attempt to assume the role + response = sts_client.assume_role( + RoleArn=role_arn, + RoleSessionName="ConnectivityTestSession" + ) + + # 4. Use temporary credentials to create a target session + creds = response['Credentials'] + target_session = boto3.Session( + aws_access_key_id=creds['AccessKeyId'], + aws_secret_access_key=creds['SecretAccessKey'], + aws_session_token=creds['SessionToken'] + ) + + # 5. Perform a "whoami" call in the target account + target_sts = target_session.client('sts') + identity = target_sts.get_caller_identity() + + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success") + + except ClientError as e: + # Catch failures (e.g., role hasn't finished deploying or trust policy issue) + error_code = e.response['Error']['Code'] + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ Failed ({error_code})") + +if __name__ == "__main__": + test_connectivity() From 8939ec2d9257cad1674ea5ea4ffabf0e05b4d158 Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:17:39 -0500 Subject: [PATCH 02/10] v1.0.2 --- .../inf-org-crossaccount/test-cross-org.py | 121 ++++++++++-------- 1 file changed, 68 insertions(+), 53 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 index f9586017..ec9f088f 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -1,56 +1,71 @@ import boto3 -from botocore.exceptions import ClientError - -# --- SET THESE TO MATCH YOUR TERRAFORM --- -CROSS_ACCOUNT_ROLE_NAME = "OrganizationAdminAccessRole" -TARGET_REGION = "us-east-1" - -def test_connectivity(): - # 1. Initialize the Organizations client from Management Account - org_client = boto3.client('organizations') - sts_client = boto3.client('sts') - - print(f"Starting connectivity test using role: {CROSS_ACCOUNT_ROLE_NAME}\n") - print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") - print("-" * 55) - - # 2. Page through all accounts in the Org - paginator = org_client.get_paginator('list_accounts') - for page in paginator.paginate(): - for account in page['Accounts']: - # Skip accounts that aren't active (suspended/pending) - if account['State'] != 'ACTIVE': - continue - - acc_id = account['Id'] - acc_name = account['Name'] - role_arn = f"arn:aws:iam::{acc_id}:role/{CROSS_ACCOUNT_ROLE_NAME}" - - try: - # 3. Attempt to assume the role - response = sts_client.assume_role( - RoleArn=role_arn, - RoleSessionName="ConnectivityTestSession" - ) - - # 4. Use temporary credentials to create a target session - creds = response['Credentials'] - target_session = boto3.Session( - aws_access_key_id=creds['AccessKeyId'], - aws_secret_access_key=creds['SecretAccessKey'], - aws_session_token=creds['SessionToken'] - ) - - # 5. Perform a "whoami" call in the target account - target_sts = target_session.client('sts') - identity = target_sts.get_caller_identity() - - print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success") - - except ClientError as e: - # Catch failures (e.g., role hasn't finished deploying or trust policy issue) - error_code = e.response['Error']['Code'] - print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ Failed ({error_code})") +import argparse +import sys +from botocore.exceptions import ClientError, ProfileNotFound + +def get_args(): + parser = argparse.ArgumentParser(description="Test AWS Org Connectivity with assuming roles.") + parser.add_argument("--profile", required=True, help="AWS CLI profile to use (e.g., your-sso-profile)") + parser.add_argument("--region", default="us-east-1", help="Default region for the session (default: us-east-1)") + parser.add_argument("--role-name", required=True, help="The name of the role to assume in member accounts") + return parser.parse_args() + +def test_connectivity(args): + try: + # Create the base session using your local profile + session = boto3.Session(profile_name=args.profile, region_name=args.region) + org_client = session.client('organizations') + sts_client = session.client('sts') + + # Verify who you are in the Management Account first + caller = sts_client.get_caller_identity() + print(f"Authenticated as: {caller['Arn']}") + print(f"Targeting role: {args.role_name}\n") + + print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") + print("-" * 60) + + paginator = org_client.get_paginator('list_accounts') + 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:aws:iam::{acc_id}:role/{args.role_name}" + + try: + # Assume role in the member account + assumed_role = sts_client.assume_role( + RoleArn=role_arn, + RoleSessionName="OrgTestSession" + ) + + # Prove it worked by creating a client with the temp credentials + creds = assumed_role['Credentials'] + member_session = boto3.Session( + aws_access_key_id=creds['AccessKeyId'], + aws_secret_access_key=creds['SecretAccessKey'], + aws_session_token=creds['SessionToken'], + region_name=args.region + ) + + # Call target account identity + member_id = member_session.client('sts').get_caller_identity()['Account'] + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success (Verified: {member_id})") + + except ClientError as e: + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ Failed ({e.response['Error']['Code']})") + + except ProfileNotFound: + print(f"Error: The profile '{args.profile}' was not found in your AWS config.") + sys.exit(1) + except Exception as e: + print(f"An unexpected error occurred: {e}") + sys.exit(1) if __name__ == "__main__": - test_connectivity() + cli_args = get_args() + test_connectivity(cli_args) + From a3115b6ddf82212d6ae85b47240208ba79301e72 Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:19 -0500 Subject: [PATCH 03/10] v1.0.4 --- .../inf-org-crossaccount/test-cross-org.py | 74 +++++++++++-------- 1 file changed, 43 insertions(+), 31 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 index ec9f088f..22927268 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -1,29 +1,46 @@ import boto3 import argparse +import os import sys from botocore.exceptions import ClientError, ProfileNotFound def get_args(): - parser = argparse.ArgumentParser(description="Test AWS Org Connectivity with assuming roles.") - parser.add_argument("--profile", required=True, help="AWS CLI profile to use (e.g., your-sso-profile)") - parser.add_argument("--region", default="us-east-1", help="Default region for the session (default: us-east-1)") - parser.add_argument("--role-name", required=True, help="The name of the role to assume in member accounts") + parser = argparse.ArgumentParser(description="Test AWS Org Connectivity by assuming roles.") + + # Defaults are pulled from environment variables if the flags are omitted + parser.add_argument("--profile", + default=os.environ.get("AWS_PROFILE"), + help="AWS CLI profile (default: AWS_PROFILE env var)") + + 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)") + + parser.add_argument("--role-name", + required=True, + help="The name of the role to assume in member accounts") + return parser.parse_args() def test_connectivity(args): try: - # Create the base session using your local profile + # Initialize session. Boto3 naturally handles None for profile_name + # by falling back to 'default' or environment credentials. session = boto3.Session(profile_name=args.profile, region_name=args.region) - org_client = session.client('organizations') + sts_client = session.client('sts') + org_client = session.client('organizations') - # Verify who you are in the Management Account first + # Verify Management Account Identity caller = sts_client.get_caller_identity() - print(f"Authenticated as: {caller['Arn']}") - print(f"Targeting role: {args.role_name}\n") + 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"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") - print("-" * 60) + print("-" * 65) paginator = org_client.get_paginator('list_accounts') for page in paginator.paginate(): @@ -36,36 +53,31 @@ def test_connectivity(args): role_arn = f"arn:aws:iam::{acc_id}:role/{args.role_name}" try: - # Assume role in the member account - assumed_role = sts_client.assume_role( + # Assume role using the Management Account's credentials + assumed = sts_client.assume_role( RoleArn=role_arn, - RoleSessionName="OrgTestSession" + RoleSessionName="OrgTest" ) - # Prove it worked by creating a client with the temp credentials - creds = assumed_role['Credentials'] - member_session = boto3.Session( - aws_access_key_id=creds['AccessKeyId'], - aws_secret_access_key=creds['SecretAccessKey'], - aws_session_token=creds['SessionToken'], - region_name=args.region + # Verify by calling STS inside the member account + temp = assumed['Credentials'] + member_sts = boto3.client( + 'sts', + aws_access_key_id=temp['AccessKeyId'], + aws_secret_access_key=temp['SecretAccessKey'], + aws_session_token=temp['SessionToken'] ) + member_id = member_sts.get_caller_identity()['Account'] - # Call target account identity - member_id = member_session.client('sts').get_caller_identity()['Account'] - print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success (Verified: {member_id})") + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success") except ClientError as e: - print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ Failed ({e.response['Error']['Code']})") + print(f"{acc_name[:25]:<25} | {acc_id:<15} | ❌ {e.response['Error']['Code']}") except ProfileNotFound: - print(f"Error: The profile '{args.profile}' was not found in your AWS config.") - sys.exit(1) + print(f"Error: Profile '{args.profile}' not found.") except Exception as e: - print(f"An unexpected error occurred: {e}") - sys.exit(1) + print(f"Error: {e}") if __name__ == "__main__": - cli_args = get_args() - test_connectivity(cli_args) - + test_connectivity(get_args()) From ef75628c4ca9832c728dd2f846ac9e092ef9e5cd Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:23 -0500 Subject: [PATCH 04/10] v1.0.5 --- .../inf-org-crossaccount/test-cross-org.py | 60 +++++++++++-------- 1 file changed, 34 insertions(+), 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 index 22927268..84a6215d 100755 --- 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()) From c43335b31152bd69feb34b0198f080bac0e62341 Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:28 -0500 Subject: [PATCH 05/10] v1.0.6 --- .../inf-org-crossaccount/test-cross-org.py | 22 ++++++++++++++----- 1 file changed, 17 insertions(+), 5 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 index 84a6215d..b403c6d4 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -2,6 +2,7 @@ import argparse import os import sys +import logging from botocore.exceptions import ClientError, ProfileNotFound def get_args(): @@ -19,15 +20,26 @@ def get_args(): 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") 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) - # 1. Print Pre-flight Check (The "Where am I?" info) + # 2. Print Configuration Summary current_profile = args.profile or "Default/Environment" current_region = session.region_name @@ -36,14 +48,15 @@ def test_connectivity(args): 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) - # 2. Get Management Account Identity + # 3. Get Management Account Identity sts_client = session.client('sts') caller = sts_client.get_caller_identity() print(f"Sourcing identity from: {caller['Arn']}\n") - # 3. Start Organization loop + # 4. Start Organization loop org_client = session.client('organizations') print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") print("-" * 60) @@ -65,7 +78,7 @@ def test_connectivity(args): RoleSessionName="ConnectivityTest" ) - # Prove success by verifying identity inside the member account + # Verify identity inside the member account creds = assumed['Credentials'] member_sts = boto3.client( 'sts', @@ -75,7 +88,6 @@ def test_connectivity(args): region_name=current_region ) - # If this succeeds, connectivity is confirmed member_sts.get_caller_identity() print(f"{acc_name[:25]:<25} | {acc_id:<15} | ✅ Success") From e025c9937360d910404864482d26a04302ad511d Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:34 -0500 Subject: [PATCH 06/10] v1.0.7 --- .../inf-org-crossaccount/test-cross-org.py | 100 +++++------------- 1 file changed, 29 insertions(+), 71 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 index b403c6d4..b8cf699a 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/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()) From aea550e6572b4888f1e914458917a47278a2467f Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:39 -0500 Subject: [PATCH 07/10] v1.0.8 --- .../inf-org-crossaccount/test-cross-org.py | 45 ++++++++++--------- 1 file changed, 24 insertions(+), 21 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 index b8cf699a..886aeca0 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/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()) From f0b36297dd1c1e40f6b4dc6c369d8fe2c4024bff Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:47 -0500 Subject: [PATCH 08/10] v1.0.9 --- .../inf-org-crossaccount/test-cross-org.py | 82 ++++++++++++++----- 1 file changed, 62 insertions(+), 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 index 886aeca0..419f3060 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -2,18 +2,39 @@ 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') @@ -21,16 +42,19 @@ def test_connectivity(args): 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()) From d056f97ef452ecdb80aa6d11e5b600e50ccdb983 Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:53 -0500 Subject: [PATCH 09/10] v1.0.10 --- .../inf-org-crossaccount/test-cross-org.py | 40 ++++++++++--------- 1 file changed, 21 insertions(+), 19 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 index 419f3060..bc7f912a 100755 --- a/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py +++ b/infrastructure/global/stacksets/inf-org-crossaccount/test-cross-org.py @@ -6,7 +6,7 @@ from botocore.exceptions import ClientError, ProfileNotFound # --- VERSIONING --- -__version__ = "1.0.9" +__version__ = "1.0.10" def get_args(): parser = argparse.ArgumentParser(description=f"Multi-Partition AWS Org Connectivity Test v{__version__}") @@ -34,33 +34,32 @@ def test_connectivity(args): boto3.set_stream_logger(name='botocore', level=logging.DEBUG) try: - # Initialize session + # Initialize management session session = boto3.Session(profile_name=args.profile, region_name=args.region) sts_client = session.client('sts') - # 1. Identify Partition and Identity + # Identify Partition and Identity caller = sts_client.get_caller_identity() source_arn = caller['Arn'] - - # Dynamically extract partition (e.g., 'aws', 'aws-us-gov', 'aws-cn') partition = source_arn.split(':')[1] - print("-" * 65) + print("-" * 100) print(f"AWS ORG CONNECTIVITY TEST - Version {__version__}") - print("-" * 65) + 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("-" * 65) + print("-" * 100) org_client = session.client('organizations') paginator = org_client.get_paginator('list_accounts') - print(f"{'Account Name':<25} | {'Account ID':<15} | {'Status'}") - print("-" * 65) + # Table Header + print(f"{'Account Name':<25} | {'Alias':<25} | {'Account ID':<15} | {'Status'}") + print("-" * 100) for page in paginator.paginate(): for account in page['Accounts']: @@ -69,33 +68,36 @@ def test_connectivity(args): acc_id = account['Id'] acc_name = account['Name'] - - # 2. Construct ARN using the detected partition role_arn = f"arn:{partition}:iam::{acc_id}:role/{args.role_name}" + alias = "N/A" try: - # Attempt AssumeRole + # Assume Role assumed = sts_client.assume_role( RoleArn=role_arn, RoleSessionName=f"ConnectivityTest-v{__version__.replace('.', '-')}" ) - # Verify by creating a client inside the member account + # Create temporary session for member account creds = assumed['Credentials'] - member_sts = boto3.client( - 'sts', + 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 ) - member_sts.get_caller_identity() - print(f"{acc_name[:25]:<25} | {acc_id:<15} | SUCCESS") + # 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} | {acc_id:<15} | FAILED ({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.") From 5d3754fc4de71e0f9da6af47c342b6dba590c1e8 Mon Sep 17 00:00:00 2001 From: badra001 Date: Wed, 31 Dec 2025 08:03:45 -0500 Subject: [PATCH 10/10] update --- .../inf-org-crossaccount/test-cross-org.py | 193 +++++++++++------- 1 file changed, 115 insertions(+), 78 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 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()) +