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