V4 · API y Servicios Web#

from flask import Flask, jsonify, make_response

app = Flask(__name__)

# ❌ MAL: responder JSON sin declarar el Content-Type correcto
@app.route('/api/usuario')
def usuario_mal():
    return '{"id": 1, "nombre": "Ana"}'
    # El navegador puede interpretar esto como texto plano y abrirlo directamente,
    # lo que en ciertos contextos puede llevar a XSS

# ✅ BIEN: Content-Type explícito con charset
@app.route('/api/usuario')
def usuario_bien():
    data = {"id": 1, "nombre": "Ana"}
    response = make_response(jsonify(data))
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    return response

# Flask's jsonify ya pone application/json, pero verificar el charset
# es especialmente importante para APIs que retornan texto con caracteres especiales

from flask import Flask, request, jsonify, abort

app = Flask(__name__)

# ✅ Definir explícitamente qué métodos acepta cada endpoint
@app.route('/api/usuarios', methods=['GET', 'POST'])
def usuarios():
    if request.method == 'GET':
        return jsonify(obtener_usuarios())
    elif request.method == 'POST':
        return jsonify(crear_usuario(request.json)), 201

# Flask devolverá 405 Method Not Allowed automáticamente para otros métodos.
# Para APIs más estrictas, se puede agregar un handler global:
@app.errorhandler(405)
def metodo_no_permitido(e):
    return jsonify({
        'error': 'Método no permitido',
        'metodos_permitidos': e.valid_methods
    }), 405

# Escenario: el app confía en X-Real-IP para auditoría o rate limiting.
# Un atacante podría enviar su propia cabecera X-Real-IP para falsificar su IP.

# ❌ MAL: confiar directamente en el header enviado por el cliente
@app.route('/api/accion')
def accion_mal():
    ip_cliente = request.headers.get('X-Real-IP', request.remote_addr)
    registrar_accion(ip_cliente)  # El atacante puede falsificar su IP

# ✅ BIEN: el proxy/load balancer debe sobrescribir este header,
# y la app debe rechazar el header si viene directamente del cliente.
# En Nginx:
#   proxy_set_header X-Real-IP $remote_addr;  ← Nginx sobrescribe con la IP real

# En Flask, verificar que el request viene de un proxy confiable:
PROXIES_CONFIABLES = {'10.0.0.1', '10.0.0.2'}  # IPs de los load balancers

@app.route('/api/accion')
def accion_bien():
    if request.remote_addr not in PROXIES_CONFIABLES:
        # Request directo al app — no usar headers de proxy
        ip_real = request.remote_addr
    else:
        # Request de un proxy confiable — usar el header
        ip_real = request.headers.get('X-Real-IP', request.remote_addr)

    registrar_accion(ip_real)

import re

# HTTP/2 no permite \r o \n en valores de headers
# Si construyes headers dinámicamente, valida antes de enviarlos

def sanitizar_valor_header(valor: str) -> str:
    """Eliminar caracteres de control que podrían inyectar headers."""
    # Rechazar si contiene CR, LF o CRLF
    if re.search(r'[\r\n]', valor):
        raise ValueError(f"Valor de header inválido: contiene caracteres de control")
    return valor

# Ejemplo de uso al construir un header de respuesta dinámico:
@app.route('/api/exportar')
def exportar():
    nombre_archivo = request.args.get('nombre', 'export')

    # ❌ MAL: usar el input del usuario directamente en un header
    # response.headers['Content-Disposition'] = f'attachment; filename={nombre_archivo}'
    # Un atacante podría enviar nombre_archivo = "x\r\nX-Injected: evil"

    # ✅ BIEN: sanitizar antes de usar en header
    nombre_seguro = sanitizar_valor_header(nombre_archivo)
    # Además, solo permitir caracteres alfanuméricos en el nombre
    nombre_seguro = re.sub(r'[^a-zA-Z0-9_\-]', '_', nombre_seguro)

    response = make_response(generar_csv())
    response.headers['Content-Disposition'] = f'attachment; filename="{nombre_seguro}.csv"'
    return response

# Un atacante puede enviar una query profundamente anidada para agotar recursos:
# (query "bomb" — exponencial en tiempo de procesamiento)
{
  usuarios {
    amigos {
      amigos {
        amigos {
          amigos {
            amigos {
              nombre  # 6 niveles de profundidad = miles de resolvers ejecutados
            }
          }
        }
      }
    }
  }
}

# Usando la librería graphene con validaciones de seguridad

from graphene import ObjectType, String, List, Schema
from graphql import validate, parse
from graphql.validation import NoSchemaIntrospectionCustomRule

# --- Límite de profundidad ---
def calcular_profundidad(node, profundidad=0):
    """Calcular la profundidad máxima de una query GraphQL."""
    max_profundidad = profundidad
    for campo in getattr(node, 'selection_set', None) and \
                 node.selection_set.selections or []:
        max_profundidad = max(
            max_profundidad,
            calcular_profundidad(campo, profundidad + 1)
        )
    return max_profundidad

MAX_PROFUNDIDAD = 5
MAX_COMPLEJIDAD = 100

@app.route('/graphql', methods=['POST'])
def graphql_endpoint():
    body = request.get_json()
    query_str = body.get('query', '')

    # Parsear sin ejecutar
    try:
        documento = parse(query_str)
    except Exception:
        return jsonify({'error': 'Query inválida'}), 400

    # Verificar profundidad
    for definicion in documento.definitions:
        profundidad = calcular_profundidad(definicion)
        if profundidad > MAX_PROFUNDIDAD:
            return jsonify({
                'error': f'Query demasiado profunda: {profundidad} niveles (máx {MAX_PROFUNDIDAD})'
            }), 400

    # Ejecutar solo si pasa los controles
    resultado = schema.execute(query_str)
    return jsonify(resultado.data)


# --- Deshabilitar introspección en producción ---
import os

if os.environ.get('FLASK_ENV') == 'production':
    # Agregar regla de validación que bloquea queries de introspección
    extra_validaciones = [NoSchemaIntrospectionCustomRule]
else:
    extra_validaciones = []

# Esto evita que atacantes mapeen todo el esquema de datos de la app
# ejecutando queries como: { __schema { types { name fields { name } } } }

from flask import Flask, request, session
from flask_socketio import SocketIO, disconnect, emit

app = Flask(__name__)
app.secret_key = 'clave_secreta_segura'

# ✅ Usar solo WSS (WebSocket Secure) — configurar en el servidor web (Nginx/etc.)
# En producción: el servidor Nginx maneja el TLS, Flask recibe la conexión internamente
socketio = SocketIO(app, cors_allowed_origins=[
    'https://app.midominio.com'   # ← Allowlist de orígenes: solo nuestra app
])

# ❌ NUNCA usar: cors_allowed_origins='*' para WebSockets con sesiones

@socketio.on('connect')
def manejar_conexion():
    # Validar que el usuario tiene una sesión HTTP activa ANTES de aceptar WebSocket
    usuario_id = session.get('usuario_id')

    if not usuario_id:
        # Rechazar la conexión WebSocket si no hay sesión HTTP autenticada
        disconnect()
        return False

    # Generar token de sesión dedicado para esta conexión WebSocket
    token_ws = generar_token_websocket(usuario_id)
    session['ws_token'] = token_ws

    emit('autenticado', {'token': token_ws})
    print(f"WebSocket aceptado para usuario {usuario_id}")

@socketio.on('accion_sensible')
def manejar_accion(data):
    # Re-validar en cada mensaje sensible, no solo en la conexión
    token_recibido = data.get('ws_token')
    token_esperado = session.get('ws_token')

    if not token_recibido or token_recibido != token_esperado:
        emit('error', {'mensaje': 'Token de sesión inválido'})
        disconnect()
        return

    # Procesar acción solo si el token es válido
    procesar_accion(data)

# A diferencia de fetch() y XMLHttpRequest, el navegador no aplica
# CORS automáticamente a WebSocket. Cualquier sitio puede iniciar
# una conexión WebSocket a tu servidor si no verificas el Origin.

# Sin verificación, un sitio malicioso puede hacer:
# new WebSocket('wss://tu-api.com/ws')
# ...y si el usuario tiene una cookie de sesión, la conexión se autentica.

# Con Flask-SocketIO, el parámetro cors_allowed_origins maneja esto.
# Para implementación manual:

from flask import request

@app.route('/ws-upgrade')
def websocket_upgrade():
    origen = request.headers.get('Origin', '')
    origenes_permitidos = {'https://app.midominio.com', 'https://m.midominio.com'}

    if origen not in origenes_permitidos:
        return 'Origen no permitido', 403

    # Continuar con el upgrade a WebSocket...