Replay Attack in Flask with Dynamodb

A replay attack occurs when an attacker intercepts a valid request and retransmits it to reproduce the intended effect without authorization. In a Flask application that uses Amazon DynamoDB as its persistence layer, the combination of HTTP-based communication, predictable or missing nonces, and insufficient idempotency controls can enable this attack.

Flask does not provide built-in replay protection. If endpoints accept state-changing operations (such as payments, transfers, or resource creation) and rely only on transport-layer security (TLS), an adversary who captures a signed request—including headers, body, and timestamp—can replay it while the signature or token remains valid. This is especially relevant when DynamoDB is used as the backend datastore: a captured request that writes to a DynamoDB table (for example, using the put_item or update_item APIs) may create duplicate records or modify state when replayed.

DynamoDB itself does not prevent replays; it processes each request independently. Without application-level safeguards such as idempotency tokens, unique request identifiers, or strict timestamp windows, the same item write can be applied multiple times. For instance, a payment request that writes {"orderId": "123", "status": "paid", "amount": 100} to a DynamoDB table will produce the same result if replayed, potentially leading to double-charging or duplicated orders. The risk is compounded when primary keys are not designed to detect duplicates, or when conditional writes are not used to enforce uniqueness.

Attack surface in this stack includes missing or weak authentication on some endpoints, lack of per-request nonces, and insufficient validation of request timestamps. An attacker may also exploit unauthenticated or weakly authenticated endpoints to probe for valid request formats and then replay them against authenticated operations. Because DynamoDB stores the resulting state, replayed writes can persist and affect downstream business logic, making it critical to validate and de-duplicate at the application layer.

To mitigate replay attacks in a Flask application using DynamoDB, implement idempotency at the request and item level. Use unique client-generated idempotency tokens and enforce them in DynamoDB with conditional writes. Always include a timestamp or nonce in the request and validate freshness server-side.

Idempotency token with DynamoDB conditional write

Store idempotency tokens in a dedicated DynamoDB table and use a conditional write to ensure only the first request succeeds. Subsequent replays will fail the condition and can be safely ignored or logged.

import boto3
from flask import Flask, request, jsonify
import uuid
import time

app = Flask(__name__)
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
tokens_table = dynamodb.Table('IdempotencyTokens')

@app.route('/transfer', methods=['POST'])
def transfer_funds():
    data = request.get_json()
    token = request.headers.get('Idempotency-Token')
    if not token:
        token = str(uuid.uuid4())
    
    # Ensure token is recent (e.g., within 24 hours)
    now = int(time.time())
    
    try:
        tokens_table.put_item(
            Item={
                'idempotency_token': token,
                'created_at': now,
                'request_hash': hash(str(data)),  # application-level fingerprint
                'status': 'PENDING'
            },
            ConditionExpression='attribute_not_exists(idempotency_token)'
        )
    except Exception as e:
        # Conditional check failed — token already used; safely return previous result
        existing = tokens_table.get_item(Key={'idempotency_token': token})
        if 'Item' in existing:
            return jsonify({'status': 'already_processed', 'original_response': existing['Item'].get('response')}), 200
        # Token expired or invalid
        return jsonify({'error': 'invalid_or_expired_token'}), 400

    # Process the transfer (write to target table)
    target_table = dynamodb.Table('Accounts')
    # Example: debit source, credit destination with conditional balance checks
    try:
        target_table.update_item(
            Key={'account_id': data['from']},
            UpdateExpression='SET balance = balance - :val',
            ConditionExpression='balance >= :val',
            ExpressionAttributeValues={':val': data['amount']}
        )
        target_table.update_item(
            Key={'account_id': data['to']},
            UpdateExpression='SET balance = balance + :val',
            ExpressionAttributeValues={':val': data['amount']}
        )
        response = {'status': 'success', 'transaction_id': str(uuid.uuid4())}
    except Exception as transact_error:
        # Mark token as failed to prevent reuse
        tokens_table.update_item(
            Key={'idempotency_token': token},
            UpdateExpression='SET #s = :failed',
            ExpressionAttributeNames={'#s': 'status'},
            ExpressionAttributeValues={':failed': 'FAILED'}
        )
        return jsonify({'error': 'transfer_failed', 'details': str(transact_error)}), 400

    # Record successful response for replay handling
    tokens_table.update_item(
        Key={'idempotency_token': token},
        UpdateExpression='SET #s = :done, response = :resp, updated_at = :upd',
        ExpressionAttributeNames={'#s': 'status'},
        ExpressionAttributeValues={':done': 'COMPLETED', 'resp': response, 'upd': now}
    )
    return jsonify(response), 200

Timestamp and window validation

Reject requests with timestamps too far from server time to narrow the replay window. Combine this with HTTPS and strong authentication to reduce exposure.

from datetime import datetime, timezone
import time

def validate_request_timestamp(ts: int, skew_seconds: int = 300) -> bool:
    """Return True if request timestamp is within allowed skew."""
    now = int(time.time())
    return abs(now - ts) <= skew_seconds

@app.before_request
def enforce_timestamp():
    if request.method in ('POST', 'PUT', 'DELETE'):
        ts = request.headers.get('X-Request-Timestamp')
        if not ts or not validate_request_timestamp(int(ts)):
            return jsonify({'error': 'stale_request'}), 400

DynamoDB best practices to reduce replay impact

• Use conditional writes (e.g., attribute_not_exists or version checks) to enforce uniqueness on idempotency tokens and critical operations.
• Design primary keys to include idempotency tokens or include a deduplication attribute where appropriate.
• Enable DynamoDB Streams and process mutations to detect and alert on unexpected duplicate patterns, but remember this is detection, not prevention.

middleBrick can scan this Flask+DynamoDB setup and surface missing idempotency controls or weak timestamp validations as part of its 12 checks. If you want continuous monitoring, the Pro plan provides scheduled scans and alerts; the CLI (`middlebrick scan `) and GitHub Action can enforce a minimum security score in CI/CD.