# CSRF via State-Changing GET Endpoint in Flask

Language: Python
Severity: Critical
CWE: CWE-352

## Source
10-11

## Flow
10-12, 11-12-14

## Sink
12, 14

## Vulnerable Code
```python
from flask import Flask, request, jsonify
import boto3

app = Flask(__name__)
s3_client = boto3.client('s3')

@app.route('/cloud/bucket/purge', methods=['GET'])
def purge_s3_bucket():
    bucket_name = request.args.get('bucket', 'user-uploads-prod')
    prefix = request.args.get('prefix', '')
    objects = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    if 'Contents' in objects:
        delete_keys = [{'Key': obj['Key']} for obj in objects['Contents']]
        s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': delete_keys})
    return jsonify({'status': 'success', 'deleted': len(delete_keys) if 'Contents' in objects else 0})
```

## Explanation

This endpoint performs destructive S3 operations (listing and deleting objects) via GET request without CSRF protection. An attacker can craft a malicious link or embed it in an image tag that, when visited by an authenticated administrator, will delete S3 bucket contents. The use of GET for state-changing operations violates HTTP semantics and enables CSRF attacks through simple GET requests.

## Remediation

The fix changes the endpoint from GET to POST to follow HTTP semantics for state-changing operations, adds CSRF token validation via a decorator that checks a session-stored token against the request header or form field, and adds input validation including an allowlist for bucket names and requiring a non-empty prefix to prevent accidental full bucket deletion.

## Secure Code
```python
from flask import Flask, request, jsonify, session, abort
import boto3
import secrets
import functools

app = Flask(__name__)
app.secret_key = secrets.token_hex(32)
s3_client = boto3.client('s3')

ALLOWED_BUCKETS = ['user-uploads-prod', 'user-uploads-staging']

def require_csrf_token(f):
    @functools.wraps(f)
    def decorated_function(*args, **kwargs):
        token = request.headers.get('X-CSRF-Token') or request.form.get('csrf_token')
        if not token or token != session.get('csrf_token'):
            abort(403, description='CSRF token missing or invalid')
        return f(*args, **kwargs)
    return decorated_function

@app.route('/cloud/bucket/csrf-token', methods=['GET'])
def get_csrf_token():
    if 'csrf_token' not in session:
        session['csrf_token'] = secrets.token_hex(32)
    return jsonify({'csrf_token': session['csrf_token']})

@app.route('/cloud/bucket/purge', methods=['POST'])
@require_csrf_token
def purge_s3_bucket():
    data = request.get_json() or request.form
    bucket_name = data.get('bucket', 'user-uploads-prod')
    prefix = data.get('prefix', '')

    if bucket_name not in ALLOWED_BUCKETS:
        return jsonify({'status': 'error', 'message': 'Bucket not allowed'}), 400

    if not prefix:
        return jsonify({'status': 'error', 'message': 'Prefix is required to prevent full bucket purge'}), 400

    deleted_count = 0
    objects = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    if 'Contents' in objects:
        delete_keys = [{'Key': obj['Key']} for obj in objects['Contents']]
        s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': delete_keys})
        deleted_count = len(delete_keys)

    return jsonify({'status': 'success', 'deleted': deleted_count})
```