Testando Segurança de APIs WebSocket

Quando Usar

• Avaliando APIs de comunicação em tempo real que usam os protocolos WebSocket (ws://) ou Secure WebSocket (wss://)
• Testando para Cross-Site WebSocket Hijacking (CSWSH), onde a página de um atacante se conecta a um servidor WebSocket legítimo
• Avaliando a aplicação de autenticação e autorização em conexões e mensagens WebSocket
• Testando a validação de entrada em payloads de mensagens WebSocket para vulnerabilidades de injeção
• Avaliando implementações WebSocket para negação de serviço através de inundação de mensagens ou frames de tamanho excessivo

Não use sem autorização por escrito. Testes em WebSocket podem interromper serviços em tempo real e afetar outros usuários conectados.

Pré-requisitos

• Autorização por escrito especificando o endpoint WebSocket e o escopo do teste
• Burp Suite Professional com capacidade de interceptação de WebSocket
• Python 3.10+ com as bibliotecas websockets e asyncio
• Ferramentas de desenvolvedor do navegador para observar handshakes e frames de WebSocket
• Ferramenta de linha de comando wscat para interação manual com WebSocket: npm install -g wscat
• Conhecimento do subprotocolo WebSocket em uso (JSON-RPC, STOMP, customizado)

Fluxo de Trabalho

Passo 1: Descoberta do Endpoint WebSocket e Análise do Handshake

import asyncio
import websockets
import json
import ssl
import time

WS_URL = "wss://target-api.example.com/ws"
AUTH_TOKEN = "Bearer <token>"

# Capture and analyze the WebSocket handshake
async def analyze_handshake():
    """Analyze WebSocket upgrade request and response headers."""
    try:
        async with websockets.connect(
            WS_URL,
            extra_headers={"Authorization": AUTH_TOKEN},
            ssl=ssl.create_default_context()
        ) as ws:
            print(f"Connected to: {WS_URL}")
            print(f"Protocol: {ws.subprotocol}")
            print(f"Extensions: {ws.extensions}")

            # Send a test message
            test_msg = json.dumps({"type": "ping"})
            await ws.send(test_msg)
            response = await asyncio.wait_for(ws.recv(), timeout=5)
            print(f"Server response: {response}")

            return True
    except websockets.exceptions.InvalidStatusCode as e:
        print(f"Connection rejected: {e.status_code}")
        return False
    except Exception as e:
        print(f"Connection error: {e}")
        return False

asyncio.run(analyze_handshake())

Passo 2: Teste de Autenticação e Autorização

async def test_ws_authentication():
    """Testa se o WebSocket requer autenticação."""
    results = []

    # Teste 1: Conectar sem nenhuma autenticação
    try:
        async with websockets.connect(WS_URL) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({
                "test": "Sem autenticação",
                "status": "VULNERÁVEL",
                "response": resp[:200]
            })
            print(f"[VULN] WebSocket acessível sem autenticação")
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "Sem autenticação", "status": "SEGURO"})
    except Exception as e:
        results.append({"test": "Sem autenticação", "status": f"ERRO: {e}"})

    # Teste 2: Conectar com token inválido
    try:
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": "Bearer invalid_token"}) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({
                "test": "Token inválido",
                "status": "VULNERÁVEL",
                "response": resp[:200]
            })
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "Token inválido", "status": "SEGURO"})
    except Exception as e:
        results.append({"test": "Token inválido", "status": f"ERRO: {e}"})

    # Teste 3: Conectar com token expirado
    expired_token = "Bearer eyJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MDAwMDAwMDB9.expired"
    try:
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": expired_token}) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({"test": "Token expirado", "status": "VULNERÁVEL"})
    except (websockets.exceptions.InvalidStatusCode, Exception):

…