Detecção de Endpoints de API Sombra

Visão Geral

APIs Sombra são endpoints de API que operam dentro do ambiente de uma organização sem serem rastreados, documentados ou protegidos. Elas surgem de ciclos de desenvolvimento rápidos, ambientes de teste esquecidos, versões de API descontinuadas que continuam em execução, integrações de terceiros ou projetos paralelos de desenvolvedores implantados sem governança. As APIs Sombra contornam os controles de autenticação e monitoramento, criando pontos de entrada ocultos para atacantes. Estudos mostram que até 30% dos endpoints de API em grandes organizações são indocumentados, tornando a detecção de APIs Sombra um componente crítico do gerenciamento da postura de segurança de API.

Quando Usar

• Ao investigar incidentes de segurança que exigem a detecção de endpoints de API sombra.
• Ao construir regras de detecção ou consultas de caça a ameaças para este domínio.
• Quando analistas de SOC precisam de procedimentos estruturados para este tipo de análise.
• Ao validar a cobertura de monitoramento de segurança para técnicas de ataque relacionadas.

Pré-requisitos

• Gateway de API ou proxy reverso com registro de tráfego (Kong, AWS API Gateway, Envoy).
• Capacidade de captura de tráfego de rede (broker de pacotes, espelhamento de porta).
• Acesso a repositórios de código-fonte e configurações de pipeline CI/CD.
• Acesso ao provedor de nuvem para escaneamento de configuração (AWS, GCP, Azure).
• Inventário de documentação de API (especificações OpenAPI, documentos Swagger).
• Python 3.8+ para ferramentas de descoberta personalizadas.

Métodos de Detecção

1. Análise e Comparação de Tráfego

Compare o tráfego de API ao vivo com as especificações OpenAPI documentadas para identificar endpoints indocumentados:

#!/usr/bin/env python3
"""Detector de Endpoints de API Sombra

Compara padrões de tráfego de API observados com especificações
OpenAPI documentadas para identificar endpoints indocumentados (sombra).
"""

import json
import re
import yaml
import sys
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Set, Tuple, Optional
from dataclasses import dataclass, field

@dataclass
class DiscoveredEndpoint:
    method: str
    path_pattern: str
    first_seen: str
    last_seen: str
    request_count: int
    source_ips: Set[str] = field(default_factory=set)
    status_codes: Set[int] = field(default_factory=set)
    has_auth_header: bool = False
    documented: bool = False

class ShadowAPIDetector:
    # Padrões comuns para segmentos de caminho parametrizados
    PARAM_PATTERNS = [
        (re.compile(r'/\d+'), '/{id}'),
        (re.compile(r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'), '/{uuid}'),
        (re.compile(r'/[a-zA-Z0-9]{20,40}'), '/{token}'),
    ]

    def __init__(self):
        self.documented_endpoints: Set[Tuple[str, str]] = set()
        self.discovered_endpoints: Dict[Tuple[str, str], DiscoveredEndpoint] = {}

    def load_openapi_spec(self, spec_path: str):
        """Carrega endpoints documentados da especificação OpenAPI."""
        with open(spec_path, 'r') as f:
            if spec_path.endswith('.json'):
                spec = json.load(f)
            else:
                spec = yaml.safe_load(f)

        paths = spec.get('paths', {})
        for path, methods in paths.items():
            # Normaliza parâmetros de caminho OpenAPI
            normalized_path = re.sub(r'\{[^}]+\}', '{id}', path)
            for method in methods:
                if method.upper() in ('GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'):
                    self.documented_endpoints.add((method.upper(), normalized_path))

        print(f"Carregados {len(self.documented_endpoints)} endpoints documentados de {spec_path}")

    def normalize_path(self, path: str) -> str:
        """Normaliza um caminho observado substituindo segmentos dinâmicos por placeholders."""
        # Remove a string de consulta
        path = path.split('?')[0]

        for pattern, replacement in self.PARAM_PATTERNS:
            path = pattern.sub(replacement, path)

        return path

    def process_access_log(self, log_file: str, log_format: str = "common"):
        """Processa logs de acesso de API para descobrir endpoints."""
        patterns = {
            "common": re.compile(
                r'(?P<ip>[\d.]+)\s+\S+\s+\S+\s+\[(?P<time>[^\]]+)\]\s+'
                r'"(?P<method>\w+)\s+(?P<path>\S+)\s+\S+"\s+(?P<status>\d+)'
            ),
            "json": None  # Lida com logs JSON separadamente
        }

        with open(log_file, 'r') as f:
            for line in f:
                if log_format == "json":
                    try:
                        entry = json.loads(line)

…