How to Enforce Per-User Spending Limits on Amazon Bedrock

AWS does not natively offer per-user spend caps for Amazon Bedrock — AWS Budgets operates at account level only. This guide provides a production-ready architecture for tracking and enforcing per-user monthly budget thresholds on Bedrock API access for federated (SSO) users.

Note: This guide focuses on Identity Center (SSO) users but the aws:userId condition technique works for any IAM assumed-role session.

Since April 2026, AWS natively tracks per-user Bedrock costs via the line_item_iam_principal column in CUR 2.0 (documentation). However, the native feature only provides visibility — no alerts, thresholds, or enforcement. This guide builds that missing layer.

---

Choose Your Tracking Mode

Both options use the same enforcement mechanism: an IAM deny policy on the SSO Permission Set role with aws:userId conditions targeting specific user sessions.

	Option A: Standard Mode	Option B: Strict Mode
Cost source	CUR 2.0 via Athena (actual billing)	CloudWatch invocation logs (token estimate)
Accuracy	Exact (AWS-calculated)	Estimated (token × stored price)
Latency	~1 hour (CUR refresh)	Near real-time (seconds)
Pricing maintenance	Zero — AWS handles it	Manual (Parameter Store)
Monthly cost	~$3.50/mo	~$15.50/mo
Complexity	4 services, 1 Lambda	7 services, 2 Lambdas
Best for	$100+ budgets	Low budgets ($10–25), strict compliance
Prerequisite	CUR 2.0 with IAM principal data	Bedrock invocation logging

Recommendation: Start with Option A. It's simpler, cheaper, uses exact billing data, and the ~1 hour delay is acceptable for most budgets.

---

Architecture Overview

Option A: Standard Mode (CUR-based)

Option B: Strict Mode (Invocation Logs)

---

How Enforcement Works (Both Options)

Identity Center (SSO) users assume roles — they are not IAM users. You cannot attach policies to federated sessions directly. Instead, we add a Deny statement to the SSO Permission Set role with a Condition on aws:userId that matches the blocked user's session name.

When an Identity Center user assumes a role, the session looks like:

arn:aws:sts::123456789012:assumed-role/AWSReservedSSO_RoleName_abc123/user@domain.com

The aws:userId context key resolves to: AROA...:user@domain.com

So StringLike: {"aws:userId": "*:user@domain.com"} matches that specific user.

Deny policy applied to the SSO role:

{
  "Version": "2012-10-17",
  "Statement": [{
    "Sid": "BedrockBudgetDeny",
    "Effect": "Deny",
    "Action": [
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream",
      "bedrock:Converse",
      "bedrock:ConverseStream"
    ],
    "Resource": "*",
    "Condition": {
      "StringLike": {
        "aws:userId": ["*:user1@domain.com", "*:user2@domain.com"]
      }
    }
  }]
}

Verifying aws:userId for Your Environment

Run aws sts get-caller-identity as a federated user. You should see:

{
  "UserId": "AROAEXAMPLEID:jane.doe@company.com",
  "Account": "123456789012",
  "Arn": "arn:aws:sts::123456789012:assumed-role/AWSReservedSSO_PowerUser_abc123/jane.doe@company.com"
}

The UserId field format is <ROLE_ID>:<SESSION_NAME>. Our deny policy uses StringLike: {"aws:userId": "*:jane.doe@company.com"} — the *: prefix wildcards the role ID portion and matches only on the session name suffix.

If your session name is NOT the user's email (e.g., it's a username like jsmith or an opaque ID), adjust the userId extraction logic in the Lambda to match whatever your Identity Center passes as the session name.

---

Option A: Standard Mode — Implementation

Prerequisites

1. Enable CUR 2.0 with IAM principal data:

   • Billing Console → Data Exports → Create Standard Export (CUR 2.0)
   • Check: "Include caller identity (IAM principal) allocation data"
   • Destination: S3 bucket
   • Granularity: Hourly

2. Set up Athena for CUR queries:

   • Create Glue crawler or use CUR auto-integration with Athena
   • Verify: SELECT * FROM cur_database.cur_table LIMIT 10;
   • Note the Glue database and table names created by the integration (e.g., athenacurcfn_my_export). These become your ATHENA_DATABASE and ATHENA_TABLE environment variables.

Lambda: bedrock-budget-sync

Single Lambda that runs hourly. Queries Athena for actual costs, checks limits from Parameter Store, sends SNS alerts, and directly applies/removes the IAM deny policy.

import json, os, time
from datetime import datetime, timezone
from decimal import Decimal
import boto3
from botocore.exceptions import ClientError

# --- Configuration ---
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']
DEFAULT_LIMIT_PARAM = os.environ.get('DEFAULT_LIMIT_PARAM', '/bedrock/budget/default-limit')
ATHENA_DATABASE = os.environ['ATHENA_DATABASE']
ATHENA_TABLE = os.environ['ATHENA_TABLE']
ATHENA_OUTPUT = os.environ['ATHENA_OUTPUT_BUCKET']
STATE_BUCKET = os.environ['STATE_BUCKET']
SSO_ROLE_NAME = os.environ['SSO_ROLE_NAME']
POLICY_NAME = 'BedrockBudgetDeny'
MAX_POLICY_USERS = 200
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')

athena = boto3.client('athena')
ssm = boto3.client('ssm')
sns = boto3.client('sns')
s3 = boto3.client('s3')
iam = boto3.client('iam')

def get_default_limit():
    try:
        r = ssm.get_parameter(Name=DEFAULT_LIMIT_PARAM)
        return Decimal(r['Parameter']['Value'])
    except ClientError:
        return Decimal('100')

def get_alert_state(period):
    """Load alert state from S3."""
    key = f"bedrock-budget-state/{period}.json"
    try:
        obj = s3.get_object(Bucket=STATE_BUCKET, Key=key)
        return json.loads(obj['Body'].read())
    except ClientError:
        return {}

def save_alert_state(period, state):
    """Save alert state to S3."""
    key = f"bedrock-budget-state/{period}.json"
    s3.put_object(Bucket=STATE_BUCKET, Key=key, Body=json.dumps(state),
                  ContentType='application/json')

def run_athena_query(query):
    r = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={'Database': ATHENA_DATABASE},
        ResultConfiguration={'OutputLocation': ATHENA_OUTPUT}
    )
    return r['QueryExecutionId']

def wait_for_query(eid, max_wait=60):
    for _ in range(max_wait):
        r = athena.get_query_execution(QueryExecutionId=eid)
        state = r['QueryExecution']['Status']['State']
        if state == 'SUCCEEDED': return True
        if state in ('FAILED', 'CANCELLED'):
            raise Exception(f"Query {state}: {r['QueryExecution']['Status'].get('StateChangeReason')}")
        time.sleep(1)
    raise Exception("Query timed out")

def get_per_user_costs():
    month_start = datetime.now(timezone.utc).strftime('%Y-%m-01')
    query = f"""
    SELECT line_item_iam_principal as user_arn,
           SUM(line_item_unblended_cost) as total_cost
    FROM {ATHENA_TABLE}
    WHERE product_product_name = 'Amazon Bedrock'
      AND line_item_iam_principal != ''
      AND line_item_usage_start_date >= DATE('{month_start}')
    GROUP BY line_item_iam_principal
    HAVING SUM(line_item_unblended_cost) > 0
    """
    eid = run_athena_query(query)
    wait_for_query(eid)
    results = []
    paginator = athena.get_paginator('get_query_results')
    for page in paginator.paginate(QueryExecutionId=eid):
        for row in page['ResultSet']['Rows'][1:]:
            arn = row['Data'][0].get('VarCharValue', '')
            cost = Decimal(row['Data'][1].get('VarCharValue', '0'))
            if ':assumed-role/' in arn:
                uid = arn.split('/')[-1]
            elif ':user/' in arn:
                uid = arn.split(':user/')[-1]
            else:
                uid = arn
            results.append({'userId': uid, 'cost': cost})
    return results

def send_alert(user_id, pct, cost, limit):
    subject = f"Bedrock Budget: {user_id} at {pct}% (${cost:.2f}/${limit:.2f})"
    msg = f"User: {user_id}\nSpend: ${cost:.2f}\nLimit: ${limit:.2f}\nThreshold: {pct}%\n"
    msg += "ACTION: User BLOCKED." if pct >= 100 else "Warning: approaching limit."
    sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=subject[:100], Message=msg)

def apply_deny_policy(blocked_users):
    """Apply or remove IAM deny policy on SSO role."""
    if not blocked_users:
        try:
            iam.delete_role_policy(RoleName=SSO_ROLE_NAME, PolicyName=POLICY_NAME)
        except ClientError as e:
            if e.response['Error']['Code'] != 'NoSuchEntity': raise
        return
    patterns = [f"*:{uid}" for uid in blocked_users[:MAX_POLICY_USERS]]
    policy = json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "BedrockBudgetDeny",
            "Effect": "Deny",
            "Action": ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream",
                       "bedrock:Converse", "bedrock:ConverseStream"],
            "Resource": "*",
            "Condition": {"StringLike": {"aws:userId": patterns}}
        }]
    })
    iam.put_role_policy(RoleName=SSO_ROLE_NAME, PolicyName=POLICY_NAME, PolicyDocument=policy)

def lambda_handler(event, context):
    period = datetime.now(timezone.utc).strftime('%Y-%m')
    default_limit = get_default_limit()
    state = get_alert_state(period)
    user_costs = get_per_user_costs()
    blocked_users = []
    for u in user_costs:
        uid, cost = u['userId'], u['cost']
        # Per-user override: check Parameter Store, fall back to default
        try:
            r = ssm.get_parameter(Name=f'/bedrock/budget/user/{uid}')
            limit = Decimal(r['Parameter']['Value'])
        except ClientError:
            limit = default_limit
        pct = (cost / limit) * 100
        user_state = state.get(uid, {})
        if pct >= 75 and not user_state.get('notified75'):
            send_alert(uid, 75, cost, limit)
            user_state['notified75'] = True
        if pct >= 90 and not user_state.get('notified90'):
            send_alert(uid, 90, cost, limit)
            user_state['notified90'] = True
        if pct >= 100:
            if not user_state.get('blocked'):
                send_alert(uid, 100, cost, limit)
                user_state['blocked'] = True
            blocked_users.append(uid)
        state[uid] = user_state
    apply_deny_policy(blocked_users)
    save_alert_state(period, state)
    return {'statusCode': 200, 'blocked': len(blocked_users)}

Environment Variables — Option A

Variable	Description	Example
SNS_TOPIC_ARN	Budget alert SNS topic	arn:aws:sns:us-east-1:123456789012:bedrock-budget-alerts
SSO_ROLE_NAME	IAM role used by SSO Permission Set	AWSReservedSSO_PowerUser_abc123
ATHENA_DATABASE	Glue database for CUR queries	athenacurcfn_my_export
ATHENA_TABLE	Glue table for CUR data	my_cur_export
ATHENA_OUTPUT_BUCKET	S3 URI for Athena results	s3://my-bucket/athena-results/
STATE_BUCKET	S3 bucket for alert state JSON	my-budget-state-bucket
DEFAULT_LIMIT_PARAM	SSM parameter for default limit	/bedrock/budget/default-limit

IAM Execution Role Policy — Option A

The bedrock-budget-sync Lambda requires these permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AthenaAccess",
      "Effect": "Allow",
      "Action": ["athena:StartQueryExecution", "athena:GetQueryExecution", "athena:GetQueryResults"],
      "Resource": "arn:aws:athena:*:*:workgroup/primary"
    },
    {
      "Sid": "GlueAccess",
      "Effect": "Allow",
      "Action": ["glue:GetTable", "glue:GetPartitions", "glue:GetDatabase"],
      "Resource": "*"
    },
    {
      "Sid": "S3CURAndState",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:ListBucket", "s3:GetBucketLocation"],
      "Resource": ["arn:aws:s3:::<CUR_BUCKET>/*", "arn:aws:s3:::<STATE_BUCKET>/*", "arn:aws:s3:::<ATHENA_OUTPUT_BUCKET>/*"]
    },
    {
      "Sid": "ParameterStore",
      "Effect": "Allow",
      "Action": ["ssm:GetParameter"],
      "Resource": "arn:aws:ssm:*:*:parameter/bedrock/budget/*"
    },
    {
      "Sid": "SNSPublish",
      "Effect": "Allow",
      "Action": ["sns:Publish"],
      "Resource": "arn:aws:sns:*:*:bedrock-budget-alerts"
    },
    {
      "Sid": "IAMEnforcement",
      "Effect": "Allow",
      "Action": ["iam:PutRolePolicy", "iam:DeleteRolePolicy"],
      "Resource": "arn:aws:iam::*:role/AWSReservedSSO_*"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": "*"
    }
  ]
}

Operational Queries (Athena)

Top 10 Bedrock spenders this month:

-- Top 10 Bedrock spenders this month
-- Replace <ATHENA_DATABASE>.<ATHENA_TABLE> with your Glue catalog values
SELECT line_item_iam_principal,
       SUM(line_item_unblended_cost) as total_cost,
       COUNT(*) as api_calls
FROM <ATHENA_DATABASE>.<ATHENA_TABLE>
WHERE product_product_name = 'Amazon Bedrock'
  AND line_item_iam_principal != ''
  AND line_item_usage_start_date >= DATE_TRUNC('month', current_date)
GROUP BY line_item_iam_principal
ORDER BY total_cost DESC LIMIT 10;

Per-user cost by model:

-- Per-user cost by model
SELECT line_item_iam_principal,
       line_item_resource_id as model,
       SUM(line_item_unblended_cost) as cost
FROM <ATHENA_DATABASE>.<ATHENA_TABLE>
WHERE product_product_name = 'Amazon Bedrock'
  AND line_item_usage_start_date >= DATE_TRUNC('month', current_date)
GROUP BY line_item_iam_principal, line_item_resource_id
ORDER BY cost DESC;

---

Option B: Strict Mode — Implementation

Step 1: Enable Bedrock Invocation Logging

1. Navigate to Amazon Bedrock → Settings → Model invocation logging
2. Enable "Log model invocation inputs and outputs"
3. Select destination: CloudWatch Logs
4. Log group name: /aws/bedrock/invocations
5. Ensure the Bedrock service role has permission to write to CloudWatch Logs

Important: Invocation logging must be enabled in EACH region where users invoke Bedrock models.

Required log fields:

Field	Path in Log	Usage
User ARN	identity.arn	Extract userId
Model ID	modelId	Look up pricing
Input Tokens	input.inputTokenCount	Calculate input cost
Output Tokens	output.outputTokenCount	Calculate output cost

Step 2: DynamoDB Table

aws dynamodb create-table \
  --table-name BedrockUserCosts \
  --attribute-definitions \
    AttributeName=userId,AttributeType=S \
    AttributeName=period,AttributeType=S \
    AttributeName=blocked,AttributeType=S \
  --key-schema \
    AttributeName=userId,KeyType=HASH \
    AttributeName=period,KeyType=RANGE \
  --global-secondary-indexes \
    '[{"IndexName":"BlockedIndex","KeySchema":[{"AttributeName":"blocked","KeyType":"HASH"},{"AttributeName":"userId","KeyType":"RANGE"}],"Projection":{"ProjectionType":"ALL"}}]' \
  --billing-mode PAY_PER_REQUEST

Schema:

Attribute	Type	Key	Description
userId	String	PK	IAM user/role identifier
period	String	SK	Billing period (e.g., 2026-07)
totalCost	Number	—	Running total in USD
blocked	String	GSI PK	'true' or 'false'
userLimit	Number	—	Per-user override (optional)
notified75	Boolean	—	75% alert sent
notified90	Boolean	—	90% alert sent

Step 3: Lambda Cost Tracker

Triggered by CloudWatch Logs subscription filter. Calculates cost from token counts and Parameter Store pricing, updates DynamoDB, sends alerts.

import json, base64, gzip, os, time
from datetime import datetime, timezone
from decimal import Decimal
import boto3
from botocore.exceptions import ClientError

TABLE_NAME = os.environ['TABLE_NAME']
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']
PARAMETER_PREFIX = os.environ.get('PARAMETER_PREFIX', '/bedrock/pricing')
DEFAULT_LIMIT_PARAM = os.environ.get('DEFAULT_LIMIT_PARAM', '/bedrock/budget/default-limit')

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(TABLE_NAME)
ssm = boto3.client('ssm')
sns = boto3.client('sns')

# Per-instance in-memory cache; NOT shared across concurrent Lambda executions
_cache = {}
_cache_ts = 0

def get_pricing(model_id):
    global _cache, _cache_ts
    now = time.time()
    if now - _cache_ts < 300 and model_id in _cache:
        return _cache[model_id]
    try:
        r = ssm.get_parameter(Name=f"{PARAMETER_PREFIX}/{model_id}")
        p = json.loads(r['Parameter']['Value'])
        _cache[model_id] = p; _cache_ts = now; return p
    except:
        return {"input_per_1k": 0.01, "output_per_1k": 0.03}

def get_limit():
    try:
        return Decimal(ssm.get_parameter(Name=DEFAULT_LIMIT_PARAM)['Parameter']['Value'])
    except:
        return Decimal('100')

def lambda_handler(event, context):
    data = json.loads(gzip.decompress(base64.b64decode(event['awslogs']['data'])))
    if data.get('messageType') == 'CONTROL_MESSAGE':
        return {'statusCode': 200}
    period = datetime.now(timezone.utc).strftime('%Y-%m')
    for le in data.get('logEvents', []):
        try:
            msg = json.loads(le['message'])
            arn = msg.get('identity', {}).get('arn', '')
            mid = msg.get('modelId', '')
            it = msg.get('input', {}).get('inputTokenCount', 0)
            ot = msg.get('output', {}).get('outputTokenCount', 0)
            if not arn or not mid: continue
            uid = arn.split('/')[-1] if '/' in arn else arn
            pr = get_pricing(mid)
            cost = Decimal(str(it))/1000*Decimal(str(pr['input_per_1k'])) + \
                   Decimal(str(ot))/1000*Decimal(str(pr['output_per_1k']))
            r = table.update_item(
                Key={'userId': uid, 'period': period},
                UpdateExpression='ADD totalCost :c, invocationCount :o SET lastUpdated=:n, blocked=if_not_exists(blocked,:f), notified75=if_not_exists(notified75,:nf), notified90=if_not_exists(notified90,:nf)',
                ExpressionAttributeValues={':c': cost, ':o': 1, ':n': datetime.now(timezone.utc).isoformat(), ':f': 'false', ':nf': False},
                ReturnValues='ALL_NEW')
            item = r['Attributes']
            lim = item.get('userLimit', get_limit())
            pct = (item['totalCost'] / lim) * 100
            if pct >= 75 and not item.get('notified75'):
                sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=f"Budget 75%: {uid}", Message=f"User {uid} at {pct:.0f}%")
                table.update_item(Key={'userId': uid, 'period': period}, UpdateExpression='SET notified75=:t', ExpressionAttributeValues={':t': True})
            if pct >= 90 and not item.get('notified90'):
                sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=f"Budget 90%: {uid}", Message=f"User {uid} at {pct:.0f}%")
                table.update_item(Key={'userId': uid, 'period': period}, UpdateExpression='SET notified90=:t', ExpressionAttributeValues={':t': True})
            if pct >= 100 and item.get('blocked') != 'true':
                table.update_item(Key={'userId': uid, 'period': period}, UpdateExpression='SET blocked=:b', ExpressionAttributeValues={':b': 'true'})
                sns.publish(TopicArn=SNS_TOPIC_ARN, Subject=f"BLOCKED: {uid}", Message=f"User {uid} blocked at ${item['totalCost']:.4f}")
        except Exception as e:
            print(f"Error: {e}"); continue
    return {'statusCode': 200}

Environment Variables — Option B Cost Tracker

Variable	Description	Example
TABLE_NAME	DynamoDB table for cost tracking	BedrockUserCosts
SNS_TOPIC_ARN	Budget alert SNS topic	arn:aws:sns:us-east-1:123456789012:bedrock-budget-alerts
PARAMETER_PREFIX	SSM prefix for model pricing	/bedrock/pricing
DEFAULT_LIMIT_PARAM	SSM parameter for default limit	/bedrock/budget/default-limit

Step 4: Lambda Budget Enforcer

Runs every 5 minutes. Queries blocked users from DynamoDB GSI and applies the IAM deny policy.

import json, os
from datetime import datetime, timezone
import boto3
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key, Attr

TABLE_NAME = os.environ['TABLE_NAME']
SSO_ROLE_NAME = os.environ['SSO_ROLE_NAME']
POLICY_NAME = 'BedrockBudgetDeny'
MAX_POLICY_USERS = 200

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(TABLE_NAME)
iam = boto3.client('iam')

def lambda_handler(event, context):
    period = datetime.now(timezone.utc).strftime('%Y-%m')
    blocked = []
    r = table.query(IndexName='BlockedIndex',
                    KeyConditionExpression=Key('blocked').eq('true'),
                    FilterExpression=Attr('period').eq(period))
    blocked.extend(r.get('Items', []))
    while 'LastEvaluatedKey' in r:
        r = table.query(IndexName='BlockedIndex',
                        KeyConditionExpression=Key('blocked').eq('true'),
                        FilterExpression=Attr('period').eq(period),
                        ExclusiveStartKey=r['LastEvaluatedKey'])
        blocked.extend(r.get('Items', []))
    
    blocked_ids = [i['userId'] for i in blocked]
    
    if blocked_ids:
        user_patterns = [f"*:{uid}" for uid in blocked_ids[:MAX_POLICY_USERS]]
        policy = json.dumps({
            "Version": "2012-10-17",
            "Statement": [{
                "Sid": "BedrockBudgetDeny",
                "Effect": "Deny",
                "Action": ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream",
                           "bedrock:Converse", "bedrock:ConverseStream"],
                "Resource": "*",
                "Condition": {"StringLike": {"aws:userId": user_patterns}}
            }]
        })
        iam.put_role_policy(RoleName=SSO_ROLE_NAME, PolicyName=POLICY_NAME, PolicyDocument=policy)
    else:
        try:
            iam.delete_role_policy(RoleName=SSO_ROLE_NAME, PolicyName=POLICY_NAME)
        except: pass
    
    return {'statusCode': 200, 'blocked': len(blocked_ids)}

Environment Variables — Option B Enforcer

Variable	Description	Example
TABLE_NAME	DynamoDB table for cost tracking	BedrockUserCosts
SSO_ROLE_NAME	IAM role used by SSO Permission Set	AWSReservedSSO_PowerUser_abc123

IAM Execution Role Policies — Option B

bedrock-cost-tracker Lambda:

{
  "Version": "2012-10-17",
  "Statement": [
    {"Sid": "DynamoDB", "Effect": "Allow", "Action": ["dynamodb:UpdateItem", "dynamodb:GetItem"], "Resource": "arn:aws:dynamodb:*:*:table/BedrockUserCosts"},
    {"Sid": "ParameterStore", "Effect": "Allow", "Action": ["ssm:GetParameter"], "Resource": "arn:aws:ssm:*:*:parameter/bedrock/*"},
    {"Sid": "SNSPublish", "Effect": "Allow", "Action": ["sns:Publish"], "Resource": "arn:aws:sns:*:*:bedrock-budget-alerts"},
    {"Sid": "CloudWatchLogs", "Effect": "Allow", "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"], "Resource": "*"}
  ]
}

bedrock-budget-enforcer Lambda:

{
  "Version": "2012-10-17",
  "Statement": [
    {"Sid": "DynamoDBQuery", "Effect": "Allow", "Action": ["dynamodb:Query"], "Resource": ["arn:aws:dynamodb:*:*:table/BedrockUserCosts", "arn:aws:dynamodb:*:*:table/BedrockUserCosts/index/BlockedIndex"]},
    {"Sid": "IAMEnforcement", "Effect": "Allow", "Action": ["iam:PutRolePolicy", "iam:DeleteRolePolicy"], "Resource": "arn:aws:iam::*:role/AWSReservedSSO_*"},
    {"Sid": "CloudWatchLogs", "Effect": "Allow", "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"], "Resource": "*"}
  ]
}

---

Operational Runbook

Unblock a User

• Option A: Remove the user's entry from the S3 state JSON, or wait for the next hourly sync.
• Option B: Update DynamoDB: set blocked = 'false'. Enforcer removes the policy on next run (≤5 min).

Increase a User's Limit

• Option A: Add a Parameter Store parameter /bedrock/budget/user/email@company.com with the custom limit value (e.g., 200). The Lambda checks this on each run.
• Option B: Update the userLimit attribute on their DynamoDB item.

Monthly Reset

Both options auto-reset. New month = new billing period. Old blocked users don't appear in current-period queries, so the deny policy is automatically removed.

Edge case: If the Lambda fails at the start of a new month (before processing any users), the deny policy from the previous month stays attached. Set up a CloudWatch alarm on Lambda errors, and audit the IAM deny policy on the SSO role at the start of each billing period.

---

Limitations

Option A (Standard Mode)

• Enforcement delay: Up to ~1 hour (CUR refresh cycle)
• Unblock latency: Also ~1 hour
• CUR setup required: Payer account must enable CUR 2.0 with IAM principal data
• Athena cost: ~$5/TB scanned (typically <$2/mo for Bedrock-only queries)

Option B (Strict Mode)

• Enforcement delay: Up to 5 minutes (Enforcer Lambda schedule)
• Token estimation: Streaming responses may have delayed token counts
• Pricing drift: Parameter Store prices must be updated when AWS changes pricing

Both Options

• IAM propagation: Deny policies take up to 60 seconds to propagate
• Policy size limit: 10,240 characters (~200 blocked users). For larger deployments:
  • Option 1: Split across multiple inline policies on the same role (IAM allows up to 10 inline policies per role)
  • Option 2: Use Service Control Policies (SCPs) — requires AWS Organizations, applies at account/OU level with same aws:userId condition logic
• Session name dependency: Requires Identity Center to pass user email as session name (default behavior — verify with aws sts get-caller-identity)

---

Security Best Practices

• Least privilege: Lambda IAM roles scoped to specific resource ARNs (see IAM policies above). The iam:PutRolePolicy permission is restricted to arn:aws:iam::*:role/AWSReservedSSO_* — never use Resource: "*" for IAM write operations.
• CloudTrail auditing: All iam:PutRolePolicy and iam:DeleteRolePolicy calls are logged in CloudTrail. Set up a CloudTrail event rule to alert on any modifications to your SSO role outside of the enforcer Lambda.
• Input validation: User IDs extracted from CUR/CloudWatch Logs are inserted into IAM policy conditions. Validate that extracted userId values match expected patterns (e.g., email format) before including them in the deny policy to prevent policy injection.
• Encryption: DynamoDB uses AWS-managed encryption at rest. Parameter Store supports SecureString for sensitive values.
• Monitoring: All Lambda executions logged to CloudWatch. Set CloudWatch alarms on Lambda Errors metric (threshold: ≥1 in 5 minutes) to catch enforcement failures.

---

References

• IAM Principal Cost Allocation for Bedrock (CUR 2.0)
• Amazon Bedrock Model Invocation Logging
• IAM Condition Context Keys Reference

---

Replace placeholder values (S3 bucket names, SNS topic ARNs, Athena database/table names, SSO role name) with your environment-specific values before deployment.