Cross Site Request Forgery in Flask with Dynamodb

Cross Site Request Forgery (CSRF) in a Flask application that uses DynamoDB arises when state-changing requests rely on implicit trust of the request origin rather than anti-CSRF tokens. In Flask, routes that perform write or sensitive read operations—such as updating user settings, changing email, or modifying IAM-related attributes stored in DynamoDB—can be invoked from malicious sites if the browser automatically includes session cookies or authentication tokens. The combination of Flask’s common use of cookie-based sessions (e.g., Flask-Session or session cookies) and DynamoDB as the backend data store means that an attacker who tricks a logged-in user into visiting a crafted page can cause unauthorized DynamoDB operations on behalf of that user.

DynamoDB itself does not introduce CSRF; the risk is in how the Flask application invokes DynamoDB APIs. If the application uses long-lived AWS credentials or IAM roles associated with the server, compromised client-side requests might lead to privilege escalation patterns such as BFLA (Broken Function Level Authorization) when those credentials have broad permissions. For example, a Flask route that calls dynamodb.update_item to modify a user record may inadvertently allow parameter tampering if input validation and authorization checks are insufficient. Attack vectors include forged POST requests that change billing preferences or administrative flags stored in DynamoDB, especially when rate limiting and authentication are not enforced at the API level.

Another specific concern is unauthenticated LLM endpoint exposure when Flask applications expose endpoints that interact with DynamoDB and also provide AI features. If an endpoint accepts user input that is later used to query or update DynamoDB, and that endpoint is accessible without proper authentication, an attacker may leverage CSRF to induce unintended LLM interactions or data changes. This ties into broader OWASP API Top 10 risks such as Broken Object Level Authorization (BOLA/IDOR) when object identifiers in DynamoDB are predictable and not properly validated per user context. Therefore, securing CSRF in this stack requires strict separation of authentication, robust input validation, and strong authorization before any DynamoDB call originating from Flask routes.

Remediation focuses on ensuring every DynamoDB operation triggered by Flask is preceded by proper authentication checks and anti-CSRF defenses. Below are concrete, working examples for a Flask route that updates a user profile in DynamoDB securely.

1. Use anti-CSRF tokens for state-changing requests

Generate and validate per-session CSRF tokens for any POST/PUT/DELETE that affects DynamoDB. Store the token in the session and require it for state-changing operations.

from flask import Flask, session, request, jsonify
import secrets
import boto3
from botocore.exceptions import ClientError

app = Flask(__name__)
app.secret_key = 'super-secret-key'  # set a strong secret in production

def get_csrf_token():
    if '_csrf' not in session:
        session['_csrf'] = secrets.token_hex(16)
    return session['_csrf']

@app.before_request
def require_csrf_for_state_changes():
    if request.method in ['POST', 'PUT', 'DELETE', 'PATCH']:
        token = session.get('_csrf')
        if not token or token != request.form.get('_csrf'):
            return jsonify({'error': 'invalid csrf token'}), 403

# Secure DynamoDB update example
@app.route('/profile/update', methods=['POST'])
def update_profile():
    csrf = request.form.get('_csrf')
    if csrf != session.get('_csrf'):
        return jsonify({'error': 'invalid csrf token'}), 403

    user_id = session.get('user_id')  # ensure user is authenticated
    if not user_id:
        return jsonify({'error': 'unauthorized'}), 401

    new_email = request.form.get('email')
    if not new_email:
        return jsonify({'error': 'email required'}), 400

    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.Table('users')
    try:
        response = table.update_item(
            Key={'user_id': user_id},
            UpdateExpression='SET email = :val',
            ExpressionAttributeValues={':val': new_email},
            ReturnValues='UPDATED_NEW'
        )
        return jsonify({'status': 'ok', 'attributes': response.get('Attributes')})
    except ClientError as e:
        return jsonify({'error': str(e)}), 500

2. Apply fine-grained authorization before DynamoDB operations

Ensure the requesting user is authorized to modify the specific DynamoDB item. Avoid relying on client-supplied IDs alone; derive the resource owner from the authenticated session.

@app.route('/settings/change-email', methods=['POST'])
def change_email():
    user_id = session.get('user_id')
    if not user_id:
        return jsonify({'error': 'unauthorized'}), 401

    requested_user_id = request.form.get('user_id')
    # Prevent BOLA: ensure the requested user matches the authenticated user
    if requested_user_id != user_id:
        return jsonify({'error': 'forbidden'}), 403

    new_email = request.form.get('email')
    if not new_email:
        return jsonify({'error': 'email required'}), 400

    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('users')
    try:
        table.update_item(
            Key={'user_id': user_id},
            UpdateExpression='SET email = :email',
            ExpressionAttributeValues={':email': new_email},
            ConditionExpression='attribute_exists(user_id)'
        )
        return jsonify({'message': 'email updated'})
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            return jsonify({'error': 'item not found or precondition failed'}), 409
        return jsonify({'error': str(e)}), 500

3. Use short-term credentials and avoid broad IAM policies

Where possible, avoid long-lived AWS credentials in Flask configuration. Use temporary tokens via AWS Cognito or an assumed role with least privilege scoped to the specific DynamoDB table and actions. This limits the impact if a session is compromised via CSRF.

# Example using Cognito identity to get temporary credentials
import boto3
from botocore.session import get_session

def get_temp_dynamodb_client():
    cognito_identity = 'us-east-1:xxxx-xxxx-xxxx-xxxx'
    credentials = boto3.client('cognito-identity', region_name='us-east-1').get_credentials_for_identity(
        IdentityId=cognito_identity
    )
    return boto3.client(
        'dynamodb',
        aws_access_key_id=credentials['Credentials']['AccessKeyId'],
        aws_secret_access_key=credentials['Credentials']['SecretKey'],
        aws_session_token=credentials['Credentials']['SessionToken'],
        region_name='us-east-1'
    )

4. Validate and sanitize all inputs before DynamoDB calls

Treat all user input as untrusted. Validate email formats, length constraints, and ensure only permitted fields are updated to avoid accidental mass assignment.

import re

def is_valid_email(email: str) -> bool:
    pattern = r'^[^\s@]+@[^\s@]+\.[^\s@]+$'
    return re.match(pattern, email) is not None

@app.route('/settings/update', methods=['POST'])
def safe_update():
    user_id = session.get('user_id')
    if not user_id:
        return jsonify({'error': 'unauthorized'}), 401

    email = request.form.get('email')
    if not email or not is_valid_email(email):
        return jsonify({'error': 'invalid email'}), 400

    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('users')
    try:
        table.update_item(
            Key={'user_id': user_id},
            UpdateExpression='SET email = :email',
            ExpressionAttributeValues={':email': email}
        )
        return jsonify({'status': 'updated'})
    except ClientError as e:
        return jsonify({'error': str(e)}), 500