Unsafe Consumption of APIs

CWE: CWE-20 - Improper Input Validation
Problema: La aplicación confía ciegamente en APIs externas

---

¿Qué es?

La aplicación integra con APIs externas pero:

• No valida respuestas
• No verifica SSL/certificados
• No maneja errores adecuadamente
• No tiene timeout
• No valida origen de datos

---

Vectores de Ataque

1. MITM en API Externas

# Aplicación integra con API externa sin validar SSL
import requests

# ✗ VULNERABLE
response = requests.get(
    'https://payment-api.com/verify',
    verify=False  # ← No valida certificado
)

# Atacante intercepta traffic
# Respuesta modificada
{
  "payment_status": "verified",
  "amount": 100.00
}

# Atacante cambia a:
{
  "payment_status": "verified",
  "amount": 0.01
}

2. XXE en Respuestas XML

# API retorna XML
import xml.etree.ElementTree as ET

response = requests.get('https://external-api.com/data')
# Respuesta:
# <?xml version="1.0"?>
# <!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>
# <data>&xxe;</data>

# ✗ VULNERABLE
root = ET.fromstring(response.content)
# Lee /etc/passwd

# ✓ SEGURO - Deshabilitar XXE
from defusedxml import ElementTree as DET
root = DET.fromstring(response.content)

3. XML Bomb (Billion Laughs)

# API retorna XML con recursión
bomb_xml = '''<?xml version="1.0"?>
<!DOCTYPE lolz [
  <!ENTITY lol "lol">
  <!ENTITY lol2 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
  <!ENTITY lol3 "&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;">
]>
<lolz>&lol3;</lolz>
'''

# Parser intenta expandir - consume CPU/RAM
root = ET.fromstring(bomb_xml)
# Resultado: DoS

4. Inyección en Respuestas JSON

# Aplicación confía en JSON
response = requests.get('https://api.external.com/user/123')
data = response.json()

# ✗ VULNERABLE
user_name = data['name']
html = f"<h1>User: {user_name}</h1>"
# Si data['name'] contiene JS:
# <h1>User: <script>alert('xss')</script></h1>
# XSS en la aplicación

# ✓ SEGURO - Escapar HTML
from markupsafe import escape
user_name = escape(data['name'])

5. Redirection Chains

# Aplicación sigue redirects sin validar
response = requests.get('https://api.external.com/endpoint')

# ¿Redirect a?
# 1. Mismo dominio - OK
# 2. Trusted partner - OK
# 3. file:///etc/passwd - ✗ PELIGROSO
# 4. http://internal-server - ✗ PELIGROSO
# 5. 127.0.0.1:8080 - ✗ PELIGROSO

# ✓ SEGURO - Validar redirects
response = requests.get(
    'https://api.external.com/endpoint',
    allow_redirects=False  # O validar manualmente
)

6. Inyección de Dependencias vía API

# Aplicación integra con API de plugins
response = requests.get('https://plugin-api.com/get-plugins')
plugins = response.json()

for plugin in plugins:
    # ✗ VULNERABLE - Ejecutar código de API
    exec(plugin['code'])

# Atacante compromete plugin-api.com
# Inyecta código malicioso

---

Explotación Práctica

Caso 1: MITM en Pago

# Aplicación verifica pago llamando API
def verify_payment(transaction_id):
    response = requests.get(
        f'https://payment-gateway.com/verify/{transaction_id}',
        verify=False  # ← VULNERABLE
    )
    
    if response.json()['status'] == 'approved':
        complete_order()
        return True
    return False

# Atacante (MITM):
# 1. Intercepta solicitud
# 2. Cambia respuesta a {"status": "approved"}
# 3. Orden se completa sin pagar

Caso 2: XXE → RCE

# API de terceros retorna XML vulnerable a XXE
def get_user_data(user_id):
    response = requests.get(
        f'https://crm-api.com/users/{user_id}',
        timeout=5
    )
    
    # Parse XML sin protección
    root = ET.fromstring(response.content)
    return root.find('name').text

# Payload XXE:
# <?xml version="1.0"?>
# <!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/shadow">]>
# <user><name>&xxe;</name></user>

# Resultado: /etc/shadow leído

---

Mitigación

1. Validar Certificados SSL

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.ssl_ import create_urllib3_context

# ✗ NUNCA
response = requests.get(url, verify=False)

# ✓ SIEMPRE verificar
response = requests.get(url, verify=True)

# ✓ CON certificado personalizado
response = requests.get(
    url,
    verify='/path/to/ca-bundle.crt',
    cert=('/path/to/client.crt', '/path/to/client.key')
)

2. Deshabilitar XXE

from defusedxml import ElementTree as DET
from defusedxml.minidom import parseString

# ✗ VULNERABLE
import xml.etree.ElementTree as ET
root = ET.fromstring(response.content)

# ✓ SEGURO
root = DET.fromstring(response.content)

# O manualmente deshabilitar
parser = ET.XMLParser()
parser.parser.SetParamEntityParsing(ET.XML_PARAM_ENTITY_PARSING_NEVER)
root = ET.fromstring(response.content, parser=parser)

3. Validar Redirects

import requests
from urllib.parse import urlparse

ALLOWED_DOMAINS = [
    'api.app.com',
    'partner1.com',
    'partner2.com'
]

def safe_request(url):
    """
    Request con validación de redirects
    """
    
    response = requests.get(
        url,
        allow_redirects=False  # No seguir automáticamente
    )
    
    # Si hay redirect, validar
    if response.status_code in [301, 302, 307, 308]:
        redirect_url = response.headers.get('Location')
        parsed = urlparse(redirect_url)
        
        # Whitelist de dominios
        if parsed.netloc not in ALLOWED_DOMAINS:
            raise Exception(f"Redirect to untrusted domain: {parsed.netloc}")
        
        # Seguir redirect manualmente
        return requests.get(redirect_url)
    
    return response

# Usar
response = safe_request('https://api.partner.com/endpoint')

4. Input Validation y Sanitization

from markupsafe import escape
import html
import json

def process_external_api_response(data):
    """
    Validar y sanitizar respuesta de API
    """
    
    # Si es JSON string, parsear
    if isinstance(data, str):
        data = json.loads(data)
    
    # Validar estructura
    if not isinstance(data, dict):
        raise ValueError("Invalid response format")
    
    # Validar y escapar cada campo
    result = {}
    
    for key, value in data.items():
        # Validar tipo
        if not isinstance(value, (str, int, float, bool)):
            continue  # Ignorar tipos inesperados
        
        # Escapar strings
        if isinstance(value, str):
            result[key] = escape(value)
        else:
            result[key] = value
    
    return result

# Uso
response = requests.get('https://external-api.com/data')
safe_data = process_external_api_response(response.json())

5. Timeout y Límite de Tamaño

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def requests_with_retry(max_retries=3, timeout=5, max_size=1024*1024):
    """
    Requests con timeout, retry y límite de tamaño
    """
    
    session = requests.Session()
    
    # Timeout global
    session.timeout = timeout
    
    # Retry strategy
    retry = Retry(
        total=max_retries,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503, 504]
    )
    
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    
    # Wrapper para limitar tamaño
    original_request = session.request
    
    def request_with_size_limit(*args, **kwargs):
        kwargs['stream'] = True
        response = original_request(*args, **kwargs)
        
        # Verificar tamaño
        size = int(response.headers.get('content-length', 0))
        if size > max_size:
            raise Exception(f"Response too large: {size} bytes")
        
        return response
    
    session.request = request_with_size_limit
    return session

# Usar
session = requests_with_retry(timeout=5, max_size=10*1024*1024)
response = session.get('https://external-api.com/data')

6. Implementar Circuit Breaker

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure = None
        self.state = 'closed'  # closed, open, half_open
    
    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if datetime.now() - self.last_failure > timedelta(seconds=self.timeout):
                self.state = 'half_open'
            else:
                raise Exception("Circuit is open")
        
        try:
            result = func(*args, **kwargs)
            self.failures = 0
            self.state = 'closed'
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = datetime.now()
            
            if self.failures >= self.failure_threshold:
                self.state = 'open'
            
            raise

# Usar
breaker = CircuitBreaker(failure_threshold=5)

def call_external_api():
    return requests.get('https://external-api.com/data')

try:
    response = breaker.call(call_external_api)
except Exception as e:
    print(f"API error: {e}")

---

Checklist

• [ ] Siempre verificar certificados SSL (verify=True)
• [ ] Usar defusedxml para parsear XML
• [ ] Deshabilitar XXE
• [ ] Validar y sanitizar todas las respuestas
• [ ] Timeout en todas las requests (máx 5-10 segundos)
• [ ] Límite de tamaño de respuesta
• [ ] Validar redirects (whitelist de dominios)
• [ ] Circuit breaker para APIs externas
• [ ] Logging de errores de integración
• [ ] Monitoreo de latencia de APIs externas
• [ ] Retry logic con backoff exponencial

---

Referencias

• OWASP: https://owasp.org/www-project-api-security/
• defusedxml: https://github.com/tiran/defusedxml
• Requests: https://requests.readthedocs.io/
• CWE-20: https://cwe.mitre.org/data/definitions/20.html