Amazon Bedrockを題材にした生成AIセキュリティワークショップに参加してきた #AWSreInvent

# Lambda function to handle CloudWatch alarms and update user status in DynamoDB
import boto3
import json
import logging

# Initialize logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDB table name
DYNAMODB_TABLE_NAME = "UserTable"

# Initialize DynamoDB resource
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
table = dynamodb.Table(DYNAMODB_TABLE_NAME)

def update_user_status(username, status):
    """
    Updates the user's status in DynamoDB.
    """
    try:
        response = table.update_item(
            Key={'name': username},
            UpdateExpression="SET #status = :status",
            ExpressionAttributeNames={"#status": "status"},
            ExpressionAttributeValues={":status": status},
            ReturnValues="UPDATED_NEW"
        )
        logger.info(f"User {username} status updated to {status}: {response}")
    except Exception as e:
        logger.error(f"Failed to update user {username} status to {status}: {e}")
        raise

def lambda_handler(event, context):
    """
    Lambda function to process CloudWatch alarm events and update user status in DynamoDB.
    """
    logger.info(f"Received event: {json.dumps(event)}")

    try:
        # Extract alarm details
        alarm_data = event.get('alarmData', {})
        if not alarm_data:
            raise ValueError("Missing 'alarmData' in event.")

        # Extract current alarm state and previous state
        state = alarm_data.get('state', {}).get('value')
        previous_state = alarm_data.get('previousState', {}).get('value')

        if not state or not previous_state:
            raise ValueError("Missing 'state' or 'previousState' in alarmData.")

        # Extract metrics configuration
        metrics = alarm_data.get('configuration', {}).get('metrics', [])
        if not metrics:
            raise ValueError("Missing 'metrics' in alarm configuration.")

        # Find username from dimensions
        for metric in metrics:
            dimensions = metric.get('metricStat', {}).get('metric', {}).get('dimensions', {})
            username = dimensions.get('username')
            if username:
                logger.info(f"Processing user: {username}")

                # Handle state transitions
                if previous_state == "OK" and state == "ALARM":
                    logger.info(f"Transition detected: OK -> ALARM for user {username}. Disabling user.")
                    update_user_status(username, "disabled")
                    return {"statusCode": 200, "body": f"User {username} disabled successfully."}
                elif previous_state == "ALARM" and state == "OK":
                    logger.info(f"Transition detected: ALARM -> OK for user {username}. Re-enabling user.")
                    update_user_status(username, "enabled")
                    return {"statusCode": 200, "body": f"User {username} re-enabled successfully."}
                else:
                    logger.info(f"No action required for transition {previous_state} -> {state}.")
                    return {"statusCode": 200, "body": f"No action required for {username}."}

        raise ValueError("No username found in alarm dimensions.")

    except Exception as e:
        logger.error(f"Error processing CloudWatch alarm: {e}")
        return {"statusCode": 500, "body": str(e)}