AWS: Bulk Update Alternate Contacts Across Organization

python3 UpdateAWSAccountContacts.py

#!/usr/bin/env python3
import os
import sys
import subprocess
import venv
import platform

def create_venv():
    """Create virtual environment and return the paths"""
    venv_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'venv')
    
    if platform.system() == 'Windows':
        venv_python = os.path.join(venv_dir, 'Scripts', 'python.exe')
        venv_pip = os.path.join(venv_dir, 'Scripts', 'pip.exe')
    else:
        venv_python = os.path.join(venv_dir, 'bin', 'python')
        venv_pip = os.path.join(venv_dir, 'bin', 'pip')

    if not os.path.exists(venv_dir):
        venv.create(venv_dir, with_pip=True)
    
    return venv_python, venv_pip, venv_dir

def install_requirements(venv_pip):
    """Install required packages in virtual environment"""
    requirements = ['boto3', 'setuptools']
    for package in requirements:
        subprocess.check_call([venv_pip, 'install', package], 
                            stdout=subprocess.DEVNULL, 
                            stderr=subprocess.DEVNULL)

def run_main_script(venv_python):
    """Run the main script content"""
    main_script = '''
import boto3
import time
import logging
import random
from concurrent.futures import ThreadPoolExecutor, as_completed
from botocore.exceptions import ClientError

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('contact_updates.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def get_management_account_id():
    """Get the management account ID using Organizations API"""
    try:
        org_client = boto3.client('organizations')
        response = org_client.describe_organization()
        management_account_id = response['Organization']['MasterAccountId']
        logger.info(f"Identified management account ID: {management_account_id}")
        return management_account_id
    except Exception as e:
        logger.error(f"Failed to get management account ID: {e}")
        raise
        
def check_aws_credentials():
    """Verify AWS credentials and get management account ID"""
    try:
        sts = boto3.client('sts')
        identity = sts.get_caller_identity()
        current_account = identity['Account']
        logger.info(f"Using AWS Account: {current_account}")
        
        # Get the actual management account ID
        management_account_id = get_management_account_id()
        logger.info(f"Organization management account: {management_account_id}")
        
        return True, management_account_id
    except Exception as e:
        print("\\n✗ AWS credentials not properly configured")
        print(f"Error: {str(e)}")
        print("\\nPlease configure AWS credentials using 'aws configure'")
        return False, None

class ContactUpdater:
    def __init__(self, management_account_id):
        self.org_client = boto3.client('organizations')
        self.account_client = boto3.client('account')
        self.management_account_id = management_account_id
        logger.info(f"Initialized with management account: {management_account_id}")
        
        # Verify management account ID
        try:
            org_info = self.org_client.describe_organization()
            if org_info['Organization']['MasterAccountId'] != management_account_id:
                raise ValueError("Management account ID mismatch")
        except Exception as e:
            logger.error(f"Failed to verify management account: {e}")
            raise

    def get_organization_accounts(self):
        """Get all active accounts in the organization except management account"""
        accounts = []
        try:
            paginator = self.org_client.get_paginator('list_accounts')
            for page in paginator.paginate():
                # Explicitly filter out the management account
                member_accounts = [
                    acct for acct in page['Accounts'] 
                    if (acct['Status'] == 'ACTIVE' and 
                        acct['Id'] != self.management_account_id)
                ]
                accounts.extend(member_accounts)
            
            logger.info(f"Found {len(accounts)} member accounts (excluding management account {self.management_account_id})")
            
            # Log all account IDs for verification
            for account in accounts:
                logger.info(f"Including account for processing: {account['Id']} ({account['Name']})")
            
            return accounts
        except Exception as e:
            logger.error(f"Failed to list organization accounts: {e}")
            return []

    def update_alternate_contact(self, account_id, contact_type, contact_info):
        """Update alternate contact for an account with exponential backoff retry"""
        # Skip if management account
        if account_id == self.management_account_id:
            logger.info(f"Skipping management account {account_id} for contact update")
            return False

        max_retries = 5
        base_delay = 2  # Base delay in seconds
        
        for attempt in range(max_retries):
            try:
                self.account_client.put_alternate_contact(
                    AlternateContactType=contact_type,
                    AccountId=account_id,
                    EmailAddress=contact_info['email'],
                    Name=contact_info['name'],
                    PhoneNumber=contact_info['phone'],
                    Title=contact_info['title']
                )
                logger.info(f"Successfully updated {contact_type} contact for account {account_id}")
                return True
            except ClientError as e:
                error_code = e.response.get('Error', {}).get('Code')
                
                # Handle TooManyRequestsException with exponential backoff
                if error_code == 'TooManyRequestsException':
                    if attempt < max_retries - 1:  # Don't sleep on the last attempt
                        # Calculate backoff with jitter
                        delay = (base_delay ** attempt) + random.uniform(0, 1)
                        logger.warning(f"Rate limit exceeded for account {account_id}. Retrying in {delay:.2f} seconds (attempt {attempt+1}/{max_retries})")
                        time.sleep(delay)
                    else:
                        logger.error(f"Failed to update {contact_type} contact for account {account_id} after {max_retries} attempts: {e}")
                        return False
                else:
                    # For other errors, log and return immediately
                    logger.error(f"Failed to update {contact_type} contact for account {account_id}: {e}")
                    return False
            except Exception as e:
                logger.error(f"Unexpected error updating {contact_type} contact for account {account_id}: {e}")
                return False
        
        return False  # If we get here, all retries failed

    def process_account(self, account, contact_type, contact_info):
        """Process a single account update"""
        account_id = account['Id']
        account_name = account['Name']
        
        # Additional check to prevent processing management account
        if account_id == self.management_account_id:
            logger.info(f"Skipping management account: {account_id}")
            return account_id, "Skipped (Management Account)"
        
        logger.info(f"Processing member account: {account_id} ({account_name})")
        
        success = self.update_alternate_contact(
            account_id,
            contact_type,
            contact_info
        )

        return account_id, "Success" if success else "Failed"

def get_contact_info():
    """Collect contact information interactively"""
    print("\\nPlease provide the following contact information:")
    contact_info = {
        'name': input("Full Name: ").strip(),
        'email': input("Email Address: ").strip(),
        'phone': input("Phone Number (E.164 format, e.g., +1-555-0123): ").strip(),
        'title': input("Job Title: ").strip()
    }
    
    if not all(contact_info.values()):
        raise ValueError("All fields are required")
    
    print("\\nPlease confirm the following information:")
    for key, value in contact_info.items():
        print(f"{key.capitalize()}: {value}")
    
    if input("\\nIs this information correct? (yes/no): ").lower() != 'yes':
        raise ValueError("Information not confirmed")
    
    return contact_info

def display_menu():
    """Display the main menu and get user choice"""
    print("\\n=== AWS Account Contact Update Tool ===")
    print("1. Update Billing Contacts")
    print("2. Update Operations Contacts")
    print("3. Update Security Contacts")
    print("4. Exit")
    
    while True:
        try:
            choice = int(input("\\nEnter your choice (1-4): "))
            if 1 <= choice <= 4:
                return choice
            print("Please enter a number between 1 and 4")
        except ValueError:
            print("Please enter a valid number")

def main():
    # Check AWS credentials and get management account ID
    credentials_valid, management_account_id = check_aws_credentials()
    if not credentials_valid:
        return

    print("\\nInitializing AWS Account Contact Update Tool...")
    print(f"Management Account ID: {management_account_id} (will be skipped)")
    updater = ContactUpdater(management_account_id)
    
    while True:
        choice = display_menu()
        
        if choice == 4:
            print("Exiting program...")
            break
            
        contact_types = {
            1: 'BILLING',
            2: 'OPERATIONS',
            3: 'SECURITY'
        }
        
        contact_type = contact_types[choice]
        
        try:
            print(f"\\nCollecting {contact_type.lower()} contact information...")
            contact_info = get_contact_info()

            # Get accounts and verify management account is excluded
            accounts = updater.get_organization_accounts()
            if not accounts:
                print("No member accounts found or error accessing organization")
                continue
            
            # Verify management account is not in the list
            accounts = [acct for acct in accounts if acct['Id'] != management_account_id]
            
            print(f"\\nThis will update {contact_type.lower()} contacts for {len(accounts)} member accounts.")
            print(f"Management account ({management_account_id}) will be skipped.")
            
            if input("Do you want to proceed? (yes/no): ").lower() != 'yes':
                print("Operation cancelled")
                continue
            
            # Set fixed concurrency level for large organizations
            max_workers = 5  # Fixed value optimized for large organizations
            print(f"Using concurrency level: {max_workers} (optimized for large organizations)")
            
            results = {}
            processed_accounts = set()  # Track processed accounts
            
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                future_to_account = {
                    executor.submit(updater.process_account, account, contact_type, contact_info): account
                    for account in accounts
                }
                
                for future in as_completed(future_to_account):
                    try:
                        account_id, result = future.result()
                        results[account_id] = result
                        processed_accounts.add(account_id)
                    except Exception as e:
                        logger.error(f"Error processing account: {e}")
            
            # Count results
            success_count = sum(1 for result in results.values() if result == "Success")
            failed_count = sum(1 for result in results.values() if result == "Failed")
            
            # Count skipped (management account)
            skipped_management = 1  # We always skip the management account
            
            print("\\nUpdate Summary:")
            print(f"Successfully updated: {success_count}")
            print(f"Failed: {failed_count}")
            print(f"Skipped (Management Account): {skipped_management}")
            print(f"Total member accounts processed: {len(processed_accounts)}")
            
            if input("\\nWould you like to see detailed results? (yes/no): ").lower() == 'yes':
                print("\\nDetailed Results:")
                print(f"Management Account {management_account_id}: Skipped")
                for account_id, result in results.items():
                    if account_id != management_account_id:
                        print(f"Account {account_id}: {result}")
        
        except Exception as e:
            logger.error(f"Error during execution: {e}")
            print(f"\\nError: {str(e)}")
            
        input("\\nPress Enter to continue...")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\\nProgram terminated by user")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        print(f"\\nAn unexpected error occurred: {e}")
'''

    # Write the main script to a temporary file
    script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'contact_updater.py')
    with open(script_path, 'w') as f:
        f.write(main_script)

    # Run the script using the virtual environment's Python
    subprocess.check_call([venv_python, script_path])

def main():
    try:
        print("Checking prerequisites...")
        # Create virtual environment
        venv_python, venv_pip, venv_dir = create_venv()

        # Install requirements silently
        install_requirements(venv_pip)

        print("✓ All prerequisites verified\\n")
        # Run the main script
        run_main_script(venv_python)

    except Exception as e:
        print(f"Error: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()