check_bedrock

check_bedrock_accounts.py

"""
Audit AWS Organization accounts:
1. List all active accounts (including management account)
2. Check Bedrock Claude usage via CloudWatch metrics in the past 15 months
3. Retrieve contact information for each account
4. Classify by region (Mainland China, Hong Kong, Others) and generate report

Prerequisites:
- Run this script from the AWS Organizations MANAGEMENT account
- Python 3.12+ with boto3 installed (pip install boto3)
- AWS credentials configured (via environment variables, ~/.aws/credentials, or IAM role)

Required IAM permissions on the management account:
- organizations:ListAccounts
- sts:AssumeRole (to assume role in member accounts)
- account:GetContactInformation (with resource scope for all member accounts)

Required setup in each member account:
- A cross-account IAM role (default: "OrganizationAccountAccessRole") that:
  - Trusts the management account (allows sts:AssumeRole)
  - Has the following permissions:
    - cloudwatch:GetMetricStatistics

Configuration:
- Edit ROLE_NAME below if your cross-account role has a different name
- Edit REGIONS_TO_CHECK to include all regions where Bedrock may have been used
- Adjust MAX_WORKERS for concurrency (higher = faster but more API calls)

Output:
- Console summary with classification statistics
- CSV file (bedrock_account_report.csv) with full details

Post-check (manual verification of billing address):
- The contact information retrieved by this script is the account's PRIMARY contact,
  which may differ from the actual billing/payment address.
- The billing address can ONLY be verified through the AWS Console. No API is available.
- For each account flagged with Bedrock Claude usage in the report, log in to that
  account's AWS Console and navigate to:
    Billing and Cost Management > Payment Preferences
    (or directly: https://us-east-1.console.aws.amazon.com/billing/home#/paymentpreferences)
  The billing address is shown under the payment method details.
"""

import boto3
import csv
from datetime import datetime, timedelta, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed

# ============ Configuration ============
ROLE_NAME = "OrganizationAccountAccessRole"
REGIONS_TO_CHECK = ["us-east-1", "us-west-1", "us-west-2", "ap-southeast-1", "ap-northeast-1", "eu-west-1", "eu-central-1"]
MAX_WORKERS = 10
OUTPUT_CSV = "bedrock_account_report.csv"
# =======================================

# Global: management account ID (set in main)
MGMT_ACCOUNT_ID = None


def get_all_accounts():
    """List all active accounts in the organization, including the management account.

    Falls back to auditing only the current account when the account is not part
    of an AWS Organization (AWSOrganizationsNotInUseException) or lacks access to
    organizations:ListAccounts (AccessDeniedException). This lets the script work
    for a single standalone account without an Organization.
    """
    org = boto3.client("organizations")
    accounts = []
    try:
        paginator = org.get_paginator("list_accounts")
        for page in paginator.paginate():
            for acct in page["Accounts"]:
                if acct["Status"] == "ACTIVE":
                    accounts.append({"Id": acct["Id"], "Name": acct["Name"], "Email": acct["Email"]})
    except Exception as e:
        print(f"  Organizations unavailable ({type(e).__name__}); auditing current account only.")
        accounts = [{"Id": MGMT_ACCOUNT_ID, "Name": "(current account)", "Email": ""}]
    print(f"Found {len(accounts)} active accounts (management account: {MGMT_ACCOUNT_ID})")
    return accounts


def get_session(account_id, region="us-east-1"):
    """Get a boto3 session for the target account. Use current creds for management account."""
    if account_id == MGMT_ACCOUNT_ID:
        return boto3.Session(region_name=region)
    try:
        sts = boto3.client("sts")
        resp = sts.assume_role(
            RoleArn=f"arn:aws:iam::{account_id}:role/{ROLE_NAME}",
            RoleSessionName="bedrock-audit"
        )
        creds = resp["Credentials"]
        return boto3.Session(
            aws_access_key_id=creds["AccessKeyId"],
            aws_secret_access_key=creds["SecretAccessKey"],
            aws_session_token=creds["SessionToken"],
            region_name=region
        )
    except Exception:
        return None


CLAUDE_MODEL_IDS = [
    # Claude 2.x
    "anthropic.claude-v2", "anthropic.claude-v2:1",
    "anthropic.claude-instant-v1",
    # Claude 3
    "anthropic.claude-3-haiku-20240307-v1:0",
    "anthropic.claude-3-sonnet-20240229-v1:0",
    "anthropic.claude-3-opus-20240229-v1:0",
    # Claude 3.5
    "anthropic.claude-3-5-sonnet-20240620-v1:0",
    "anthropic.claude-3-5-sonnet-20241022-v2:0",
    "anthropic.claude-3-5-haiku-20241022-v1:0",
    # Claude 3.7
    "anthropic.claude-3-7-sonnet-20250219-v1:0",
    # Claude 4
    "anthropic.claude-sonnet-4-20250514-v1:0",
    "anthropic.claude-opus-4-20250514-v1:0",
    # Claude 4.5
    "anthropic.claude-sonnet-4-5-20250929-v1:0",
    "anthropic.claude-haiku-4-5-20251001-v1:0",
    "anthropic.claude-opus-4-5-20251101-v1:0",
    # Claude 4.6
    "anthropic.claude-sonnet-4-6",
    "anthropic.claude-opus-4-6-v1",
    # Claude 4.7
    "anthropic.claude-opus-4-7",
    # Claude 4.8
    "anthropic.claude-opus-4-8",
]


def check_bedrock_usage(account_id):
    """Check if the account has invoked Bedrock Claude models in the past 15 months.

    Uses CloudWatch GetMetricData to query all model metrics in a single (batched)
    call per region, instead of one GetMetricStatistics call per model. CloudWatch
    is regional, so each region still needs its own call, but all models within a
    region are fetched together.
    """
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=456)

    # Map a unique CloudWatch query Id to each model. Query Ids must match
    # ^[a-z][a-zA-Z0-9_]*$, so we use "m{index}" and map results back via that Id.
    id_to_model = {f"m{i}": model_id for i, model_id in enumerate(CLAUDE_MODEL_IDS)}
    all_queries = [
        {
            "Id": qid,
            "MetricStat": {
                "Metric": {
                    "Namespace": "AWS/Bedrock",
                    "MetricName": "Invocations",
                    "Dimensions": [{"Name": "ModelId", "Value": model_id}],
                },
                "Period": 86400,
                "Stat": "Sum",
            },
            "ReturnData": True,
        }
        for qid, model_id in id_to_model.items()
    ]

    for region in REGIONS_TO_CHECK:
        session = get_session(account_id, region)
        if not session:
            continue
        cw = session.client("cloudwatch")
        try:
            sums = {}
            # GetMetricData accepts up to 500 queries per call; chunk to stay within
            # the limit (future-proof as the model list grows). The paginator handles
            # NextToken if the datapoint volume spans multiple pages.
            paginator = cw.get_paginator("get_metric_data")
            for i in range(0, len(all_queries), 500):
                chunk = all_queries[i:i + 500]
                for page in paginator.paginate(
                    MetricDataQueries=chunk,
                    StartTime=start_time,
                    EndTime=end_time,
                ):
                    for r in page["MetricDataResults"]:
                        sums[r["Id"]] = sums.get(r["Id"], 0) + sum(r["Values"])

            matched_models = []
            total = 0
            for qid, model_id in id_to_model.items():
                s = sums.get(qid, 0)
                if s > 0:
                    matched_models.append(model_id)
                    total += int(s)

            if matched_models:
                return {"used": True, "region": region, "models": matched_models, "total_invocations": total}
        except Exception:
            continue

    return {"used": False}


def get_contact_info(account_id):
    """Retrieve contact information for the account."""
    # Try from management account first (delegated access)
    try:
        client = boto3.client("account")
        params = {} if account_id == MGMT_ACCOUNT_ID else {"AccountId": account_id}
        resp = client.get_contact_information(**params)
        c = resp["ContactInformation"]
        return {
            "country": c.get("CountryCode", ""),
            "state": c.get("StateOrRegion", ""),
            "city": c.get("City", ""),
            "company": c.get("CompanyName", ""),
            "address": c.get("AddressLine1", ""),
            "phone": c.get("PhoneNumber", ""),
        }
    except Exception:
        pass

    # Fallback: assume role into the account
    session = get_session(account_id)
    if not session:
        return {"country": "UNKNOWN"}
    try:
        client = session.client("account")
        resp = client.get_contact_information()
        c = resp["ContactInformation"]
        return {
            "country": c.get("CountryCode", ""),
            "state": c.get("StateOrRegion", ""),
            "city": c.get("City", ""),
            "company": c.get("CompanyName", ""),
            "address": c.get("AddressLine1", ""),
            "phone": c.get("PhoneNumber", ""),
        }
    except Exception:
        return {"country": "UNKNOWN"}


def classify_by_phone(phone):
    """Classify region by the phone number's international dialing prefix.

    Returns one of "CN", "HK", "Others", or "UNKNOWN".
    "UNKNOWN" covers an empty phone number.
    """
    if not phone:
        return "UNKNOWN"
    # Normalize: strip spaces, dashes, parentheses
    p = phone.strip()
    for ch in (" ", "-", "(", ")"):
        p = p.replace(ch, "")
    digits = "".join(ch for ch in p if ch.isdigit())
    # No digits at all (e.g. whitespace/punctuation only) -> cannot determine region
    if not digits:
        return "UNKNOWN"

    # Explicit international prefixes (+ or 00): always trusted, no length guard.
    # Order matters: check the longer HK prefix before the CN one.
    if p.startswith("+852") or p.startswith("00852"):
        return "HK"
    if p.startswith("+86") or p.startswith("0086"):
        return "CN"
    # Bare country code (user wrote "86..."/"852..." without "+"): require enough
    # total digits to look like a full international number, so a 10-digit US local
    # number with an 86x area code (e.g. 860/862/863/864/865) is not misread as CN.
    # A China number with country code 86 is at least 11 digits (mobile 13, landline 11-13).
    if p.startswith("852") and len(digits) >= 11:
        return "HK"
    if p.startswith("86") and len(digits) >= 11:
        return "CN"
    return "Others"


def classify_region(contact):
    """Classify account by country code, refined by phone number prefix.

    The country code is the primary signal. The phone number prefix is used
    to (a) resolve accounts whose country is missing/unknown, and (b) flag
    cases where the country code and phone prefix disagree.
    """
    country = (contact.get("country") or "").upper()
    phone_region = classify_by_phone(contact.get("phone", ""))

    # Primary: country code
    if country == "CN" or phone_region == "CN":
        country_region = "Mainland China"
    elif country == "HK" or phone_region == "HK":
        country_region = "Hong Kong"
    elif country in ("", "UNKNOWN") and phone_region == "UNKNOWN":
        country_region = "Unknown"
    else:
        country_region = "Others"

    return country_region


def process_account(account):
    """Process a single account: check Bedrock usage and contact info."""
    account_id = account["Id"]
    print(f"  Checking account: {account_id} ({account['Name']})")

    usage = check_bedrock_usage(account_id)
    contact = get_contact_info(account_id)

    return {
        "account_id": account_id,
        "account_name": account["Name"],
        "email": account["Email"],
        "bedrock_claude_used": usage["used"],
        "models": ", ".join(usage.get("models", [])),
        "region": usage.get("region", ""),
        "total_invocations": usage.get("total_invocations", 0),
        "country": contact.get("country", ""),
        "city": contact.get("city", ""),
        "company": contact.get("company", ""),
        "address": contact.get("address", ""),
        "phone": contact.get("phone", ""),
        "classification": classify_region(contact),
    }


def generate_report(results):
    """Generate summary report and CSV output."""
    bedrock_users = [r for r in results if r["bedrock_claude_used"]]

    stats = {"Mainland China": [], "Hong Kong": [], "Others": [], "Unknown": []}
    for r in bedrock_users:
        stats[r["classification"]].append(r)

    print("\n" + "=" * 70)
    print("Bedrock Claude Usage Audit Report")
    print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 70)
    print(f"\nTotal accounts scanned: {len(results)}")
    print(f"Accounts with Bedrock Claude usage: {len(bedrock_users)}")
    print(f"\nClassification by billing address:")
    print(f"  Mainland China (CN): {len(stats['Mainland China'])}")
    print(f"  Hong Kong (HK):      {len(stats['Hong Kong'])}")
    print(f"  Others:              {len(stats['Others'])}")
    print(f"  Unknown:             {len(stats['Unknown'])}")

    for category, accounts in stats.items():
        if accounts:
            print(f"\n--- {category} ({len(accounts)} accounts) ---")
            for a in accounts:
                print(f"  {a['account_id']} | {a['company']} | {a['city']} | {a['phone']} | Models: {a['models']}")

    # Write CSV
    with open(OUTPUT_CSV, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "account_id", "account_name", "email", "bedrock_claude_used",
            "models", "region", "total_invocations",
            "country", "city", "company", "address", "phone", "classification"
        ])
        writer.writeheader()
        writer.writerows(results)

    print(f"\nFull report exported to: {OUTPUT_CSV}")


def main():
    global MGMT_ACCOUNT_ID
    sts = boto3.client("sts")
    MGMT_ACCOUNT_ID = sts.get_caller_identity()["Account"]

    print("Starting audit...")
    accounts = get_all_accounts()

    results = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(process_account, acct): acct for acct in accounts}
        for future in as_completed(futures):
            acct = futures[future]
            try:
                results.append(future.result())
            except Exception as e:
                print(f"  !! Failed to process account {acct['Id']}: {e}")
                results.append({
                    "account_id": acct["Id"], "account_name": acct["Name"],
                    "email": acct["Email"], "bedrock_claude_used": False,
                    "models": "", "region": "", "total_invocations": 0,
                    "country": "ERROR", "city": "", "company": "",
                    "address": "", "phone": "", "classification": "Unknown"
                })

    generate_report(results)


if __name__ == "__main__":
    main()