Multi-Account Security Report — Assume Roles Across All Org Accounts

# CloudFormation StackSet to deploy SecurityAuditRole to all accounts
# (run from the Organizations management account)

aws cloudformation create-stack-set \
  --stack-set-name SecurityAuditRole \
  --template-body file://security-audit-role.yaml \
  --capabilities CAPABILITY_NAMED_IAM \
  --permission-model SERVICE_MANAGED \
  --auto-deployment Enabled=true,RetainStacksOnAccountRemoval=false

# security-audit-role.yaml
Resources:
  SecurityAuditRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: SecurityAuditRole
      AssumeRolePolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              AWS: "arn:aws:iam::MANAGEMENT_ACCOUNT_ID:root"
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/SecurityAudit
        - arn:aws:iam::aws:policy/ReadOnlyAccess

import boto3
import json
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")


# ── Step 1: Get all active org accounts ──────────────────────────
def get_all_org_accounts() -> list[dict]:
    """
    list_accounts() returns all accounts in the AWS Organization.
    We must call this from the MANAGEMENT (master) account.
    
    Only include ACTIVE accounts — SUSPENDED accounts cannot have
    roles assumed in them and would cause AssumeRole to fail.
    
    paginator handles the pagination automatically (max 20 per page).
    """
    org = boto3.client("organizations")
    accounts = []
    paginator = org.get_paginator("list_accounts")
    for page in paginator.paginate():
        accounts.extend(
            acct for acct in page["Accounts"] if acct["Status"] == "ACTIVE"
        )
    return accounts


# ── Step 2: Assume role in target account ────────────────────────
def assume_role(account_id: str, role_name: str = "SecurityAuditRole") -> dict | None:
    """
    assume_role() exchanges your current credentials for temporary
    credentials in the target account. The target account must have
    an IAM role that trusts your source account's identity.

    RoleArn format: arn:aws:iam::ACCOUNT_ID:role/ROLE_NAME
    RoleSessionName: appears in CloudTrail logs in the target account.
    DurationSeconds: max 3600s (1 hour) for chained role assumption.

    Returns a dict of credentials or None if assumption fails.
    The credentials dict has four keys:
      AccessKeyId, SecretAccessKey, SessionToken, Expiration
    """
    sts = boto3.client("sts")
    role_arn = f"arn:aws:iam::{account_id}:role/{role_name}"

    try:
        response = sts.assume_role(
            RoleArn=role_arn,
            RoleSessionName="SecurityAuditSession",
            DurationSeconds=3600,
        )
        creds = response["Credentials"]
        return {
            "aws_access_key_id":     creds["AccessKeyId"],
            "aws_secret_access_key": creds["SecretAccessKey"],
            "aws_session_token":     creds["SessionToken"],
        }
    except Exception as e:
        logger.warning(f"Cannot assume role in {account_id}: {e}")
        return None


# ── Step 3: Audit a single account ───────────────────────────────
def audit_account(account: dict) -> dict:
    """
    Run 5 security checks against a single AWS account using
    temporary credentials from assume_role().

    Each check is a separate boto3 client call in the target account.
    We use kwargs unpacking (**creds) to pass the temporary credentials.
    """
    account_id   = account["Id"]
    account_name = account["Name"]
    findings: dict = {
        "account_id":   account_id,
        "account_name": account_name,
        "checks":       {},
        "error":        None,
    }

    creds = assume_role(account_id)
    if creds is None:
        findings["error"] = "Cannot assume SecurityAuditRole"
        return findings

    try:
        # ── Check 1: Root MFA ──────────────────────────────────────
        # get_account_summary() returns aggregate IAM statistics.
        # AccountMFAEnabled=1 means MFA is on for the root account.
        # AccountMFAEnabled=0 means root has NO MFA — critical finding!
        iam = boto3.client("iam", **creds)
        summary = iam.get_account_summary()["SummaryMap"]
        findings["checks"]["root_mfa_enabled"] = bool(
            summary.get("AccountMFAEnabled", 0)
        )

        # ── Check 2: CloudTrail (multi-region) ────────────────────
        # describe_trails(includeShadowTrails=False) returns only trails
        # created in THIS region (not replicated shadow trails).
        # A multi-region trail captures API calls from all regions —
        # required for complete audit coverage.
        ct = boto3.client("cloudtrail", **creds)
        trails = ct.describe_trails(includeShadowTrails=False)["trailList"]
        findings["checks"]["cloudtrail_multi_region"] = any(
            t.get("IsMultiRegionTrail") for t in trails
        )

        # ── Check 3: GuardDuty ────────────────────────────────────
        # list_detectors() returns GuardDuty detector IDs in this region.
        # A detector is the GuardDuty "engine" — you need one per region.
        # Empty list = GuardDuty not enabled in this region.
        gd = boto3.client("guardduty", **creds)
        detectors = gd.list_detectors()["DetectorIds"]
        findings["checks"]["guardduty_enabled"] = len(detectors) > 0

        # ── Check 4: AWS Config ───────────────────────────────────
        # describe_configuration_recorders() returns the Config recorders.
        # You need at least one recording to track resource configurations.
        config = boto3.client("config", **creds)
        recorders = config.describe_configuration_recorders()[
            "ConfigurationRecorders"
        ]
        findings["checks"]["aws_config_enabled"] = len(recorders) > 0

        # ── Check 5: Password Policy ──────────────────────────────
        # get_account_password_policy() raises NoSuchEntityException
        # if no password policy is set — we treat that as non-compliant.
        try:
            policy = iam.get_account_password_policy()["PasswordPolicy"]
            findings["checks"]["password_min_length"] = policy.get(
                "MinimumPasswordLength", 0
            )
            findings["checks"]["password_requires_symbols"] = policy.get(
                "RequireSymbols", False
            )
            findings["checks"]["password_max_age_days"] = policy.get(
                "MaxPasswordAge", 9999
            )
        except iam.exceptions.NoSuchEntityException:
            findings["checks"]["password_policy"] = "NOT_CONFIGURED"

    except Exception as e:
        findings["error"] = str(e)
        logger.error(f"Audit failed for {account_id}: {e}")

    return findings


# ── Step 4: Run all audits in parallel ───────────────────────────
def generate_consolidated_report(
    audit_role: str = "SecurityAuditRole",
    max_workers: int = 10,
    output_file: str = "org_security_report.json",
) -> list[dict]:
    """
    ThreadPoolExecutor runs audit_account() concurrently for all accounts.
    max_workers=10 means up to 10 accounts are audited simultaneously.
    More workers = faster, but watch STS assume_role rate limits.

    as_completed() yields futures as they finish (not in submission order).
    This means the report might not be in account ID order — we sort later.
    """
    accounts = get_all_org_accounts()
    logger.info(f"Auditing {len(accounts)} active accounts with {max_workers} workers")

    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(audit_account, acct): acct
            for acct in accounts
        }
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
            status = "✅" if not result.get("error") else "❌"
            checks_pass = all(
                bool(v) for v in result.get("checks", {}).values()
                if isinstance(v, bool)
            )
            print(
                f"{status} {result['account_id']} ({result['account_name']}) "
                f"{'✓ All checks passed' if checks_pass else '⚠ Issues found'}"
            )

    # Sort by account name for readability
    results.sort(key=lambda x: x["account_name"])

    # ── Print summary ──────────────────────────────────────────────
    print(f"\n{'='*70}")
    print("CONSOLIDATED SECURITY REPORT")
    print(f"Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}")
    print(f"{'='*70}")

    failed_checks = 0
    for r in results:
        print(f"\n  Account: {r['account_name']} ({r['account_id']})")
        if r.get("error"):
            print(f"    ⚠️  Error: {r['error']}")
            continue
        for check, val in r.get("checks", {}).items():
            if isinstance(val, bool):
                icon = "✅" if val else "❌"
                if not val:
                    failed_checks += 1
            else:
                icon = "ℹ️"
            print(f"    {icon} {check}: {val}")

    print(f"\n{'='*70}")
    print(f"Total accounts:   {len(results)}")
    print(f"Failed checks:    {failed_checks}")

    # Save full report
    with open(output_file, "w") as f:
        json.dump(results, f, indent=2, default=str)
    print(f"Full report saved to {output_file}")

    return results


if __name__ == "__main__":
    generate_consolidated_report(
        audit_role="SecurityAuditRole",
        max_workers=10,
    )

org = boto3.client("organizations")
paginator = org.get_paginator("list_accounts")
for page in paginator.paginate():
    accounts.extend(
        acct for acct in page["Accounts"] if acct["Status"] == "ACTIVE"
    )

sts = boto3.client("sts")
role_arn = f"arn:aws:iam::{account_id}:role/{role_name}"
response = sts.assume_role(
    RoleArn=role_arn,
    RoleSessionName="SecurityAuditSession",
    DurationSeconds=3600,
)
creds = response["Credentials"]
return {
    "aws_access_key_id":     creds["AccessKeyId"],
    "aws_secret_access_key": creds["SecretAccessKey"],
    "aws_session_token":     creds["SessionToken"],
}

iam = boto3.client("iam", **creds)
summary = iam.get_account_summary()["SummaryMap"]
findings["checks"]["root_mfa_enabled"] = bool(summary.get("AccountMFAEnabled", 0))

ct = boto3.client("cloudtrail", **creds)
trails = ct.describe_trails(includeShadowTrails=False)["trailList"]
findings["checks"]["cloudtrail_multi_region"] = any(t.get("IsMultiRegionTrail") for t in trails)

gd = boto3.client("guardduty", **creds)
detectors = gd.list_detectors()["DetectorIds"]
findings["checks"]["guardduty_enabled"] = len(detectors) > 0

config = boto3.client("config", **creds)
recorders = config.describe_configuration_recorders()["ConfigurationRecorders"]
findings["checks"]["aws_config_enabled"] = len(recorders) > 0

try:
    policy = iam.get_account_password_policy()["PasswordPolicy"]
    findings["checks"]["password_min_length"] = policy.get("MinimumPasswordLength", 0)
except iam.exceptions.NoSuchEntityException:
    findings["checks"]["password_policy"] = "NOT_CONFIGURED"

with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = {
        executor.submit(audit_account, acct): acct
        for acct in accounts
    }
    for future in as_completed(futures):
        result = future.result()
        results.append(result)

json.dump(results, f, indent=2, default=str)