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())