From aea550e6572b4888f1e914458917a47278a2467f Mon Sep 17 00:00:00 2001 From: badra001 Date: Tue, 30 Dec 2025 16:18:39 -0500 Subject: [PATCH] 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())