Automação de Coleta via APIs

Nota do capítulo 02 (Fontes de Dados e Coleta), que detalha a camada de clientes HTTP executada pelos coletores do pipeline.
Cobre os padrões de integração (polling, webhooks, streaming), autenticação, rate limits, caching, tratamento de erros e exemplo Python com requests.
Executa na prática o que o pipeline da nota 08 arquiteta e alimenta as métricas de saúde de fonte da nota 11.

1. Contexto e Posicionamento na Biblioteca

Praticamente toda fonte moderna de CTI expõe interface programática: feeds entregues por HTTP, coleções TAXII 2.1, APIs REST de plataformas como MISP e VirusTotal, webhooks de sandboxes e serviços de monitoramento, streams contínuos de eventos. A qualidade da automação que consome essas interfaces determina, na prática, a confiabilidade de toda a coleta: um coletor sem tratamento de rate limit derruba a própria credencial, um cliente sem idempotência duplica indicadores, um parser sem tolerância a variação quebra na primeira mudança da fonte. Automatizar coleta não é escrever um requests.get; é projetar um cliente resiliente.

Esta nota opera dentro do estágio de coletores do pipeline (08), que apenas baixa e publica a mensagem bruta na fila, e fornece os detalhes de implementação dessa camada: padrões de integração por tipo de fonte, autenticação segura, respeito a quotas, cache local e tratamento de erro que preserva dados. As métricas emitidas por esses clientes alimentam o monitoramento de saúde da nota 11, fechando o laço entre o que o código faz e o que o time consegue observar. Os exemplos usam Python com requests por ser o padrão de fato em ferramentas de CTI open source, e o raciocínio é transferível para qualquer linguagem.

2. Padrões de Integração: Polling, Webhooks e Streaming

A escolha do padrão de integração decorre de duas propriedades da fonte: o que ela oferece tecnicamente e quão rápido o dado precisa estar disponível. A tabela abaixo compara os três padrões dominantes, com critérios objetivos de escolha e riscos principais, servindo de guia para o desenho de cada coletor novo.

No polling, duas decisões definem a qualidade do coletor: o parâmetro incremental e a paginação. Toda consulta deve ser incremental por timestamp ou marcador persistido (cursor), evitando reprocessar a janela inteira a cada ciclo, e toda paginação deve seguir até o fim, tratando o caso de item novo surgir durante a varredura. O intervalo entre consultas deve respeitar o SLO de atualidade da fonte (definido na nota 11) e a quota disponível, com retrocesso automático quando o volume retornado cresce, pois intervalo adaptativo simples mantém o coletor estável diante de picos sem sacrificar a latência em regime normal.

Em webhooks, a segurança vem antes da conveniência: o endpoint receptor deve validar assinatura HMAC por segredo compartilhado, restringir IPs de origem quando a fonte os publica, responder rápido (2xx imediato, processamento assíncrono) e ser idempotente, pois reentrega é comportamento padrão de qualquer plataforma séria. Em streaming, o requisito central é a gestão da reconexão: backoff exponencial com teto, registro do último evento confirmado e política explícita para o gap de reconexão, seja rebuscando da última marcação na API, seja declarando a janela perdida nas métricas. Consumidor de stream que assume conexão perpétua acumula perda silenciosa de eventos, o pior tipo de falha em coleta.

3. Autenticação: Chaves de API, OAuth 2.0 e Segurança de Credenciais

As duas formas dominantes de autenticação em APIs de CTI são a chave estática (header dedicado, como x-apikey do VirusTotal, ou Authorization no MISP) e o OAuth 2.0 com fluxo client credentials, comum em serviços corporativos e nuvens. A regra de projeto é tratar credencial como segredo operacional: nunca em código, nunca em repositório, sempre em variável de ambiente, cofre (Vault, secret manager) ou arquivo protegido com permissão restrita. Chaves em query string devem ser evitadas mesmo quando a API permite, porque URL acaba em log de proxy, histórico e telemetria.

Para OAuth 2.0 com client credentials, o cliente deve gerenciar o ciclo de vida do token de acesso: armazenar em memória, renovar ao expirar, sincronizar renovação entre instâncias (lock simples ou cache compartilhado) e nunca chamar o endpoint de token por requisição, o que geraria latência e possível bloqueio do provedor de identidade. O esqueleto abaixo mostra a forma correta de obter e usar o token:

import os
import requests

def obter_token(idp_url, client_id, escopo):
    resposta = requests.post(
        f"{idp_url}/oauth/token",
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": os.environ["INTEGRACAO_CLIENT_SECRET"],
            "scope": escopo,
        },
        timeout=10,
    )
    resposta.raise_for_status()
    return resposta.json()["access_token"]

TOKEN = None

def cabecalhos(idp_url, client_id, escopo):
    global TOKEN
    if TOKEN is None:
        TOKEN = obter_token(idp_url, client_id, escopo)
    return {"Authorization": f"Bearer {TOKEN}"}

Além da mecânica, três práticas de segurança valem registro: rotação periódica de chave com janela de sobreposição planejada (nunca revogar antes do coletor novo estar em produção), credenciais distintas por finalidade (leitura de feed não usa a mesma chave da escrita no TIP) e princípio do menor privilégio no escopo solicitado. Incidentes de exposição de credencial em repositório são banais, e a mitigação barata é o scanner de segredos no pipeline de CI combinado à rotação testada, pois rotação nunca exercida não é capacidade, é esperança.

4. Rate Limits, Cache e Respeito à Fonte

Toda API séria publica limites de taxa, e o cliente deve tratá-los como parte do contrato. O caso mais conhecido da área é a API pública do VirusTotal: quatro requisições por minuto e quinhentas por dia, com respostas 429 quando o limite é rompido; planos pagos ampliam as quotas e acrescentam limites mensais. O cliente correto limita-se preventivamente (token bucket local) e, ainda assim, trata 429 e o header Retry-After como fluxo normal, nunca como exceção excepcional. Um limitador mínimo e correto cabe em poucas linhas:

import time

class Limitador:
    """Token bucket por janela de um minuto, thread-safe o bastante para coletores."""
    def __init__(self, max_por_minuto):
        self.max = max_por_minuto
        self.marcas = []

    def aguardar(self):
        agora = time.monotonic()
        self.marcas = [m for m in self.marcas if agora - m < 60]
        if len(self.marcas) >= self.max:
            time.sleep(max(60 - (agora - self.marcas[0]), 0.1))
        self.marcas.append(time.monotonic())

O cache é o outro lado da mesma moeda: a chamada mais barata é a que não acontece. Respostas de enriquecimento são naturalmente cacheáveis com TTL por classe de dado (anos para hash de arquivo, dias para IP), e respostas HTTP de listagem suportam cache condicional por ETag ou If-Modified-Since quando a fonte oferece, reduzindo tráfego e custo de processamento. No desenho do pipeline (08), o cache fica no estágio de enriquecimento; nos coletores, o cache de listagens evita rebaixar o mesmo arquivo inalterado a cada ciclo. Cuidado operacional: cache sem métrica de hit ratio é invisível, e um cache com hit ratio perto de zero está apenas ocupando memória e adiando o problema real de desenho.

O respeito à fonte também é postura de comunidade. Coletores identificáveis (User-Agent com nome e contato), consultas em horários e volumes civilizados, backoff ao primeiro sinal de pressão e adesão às regras de uso de cada serviço mantêm a credencial viva e a reputação do time limpa. Fonte bloqueada por abuso de coleta é perda de cobertura que nenhum slide de projeto compensa, e comunidades de compartilhamento (ISACs, grupos de confiança) tratam abuso de quota como problema de conduta, com efeito sobre futuras parcerias.

5. Tratamento de Erros, Retries e Resiliência

O tratamento de erro correto começa por classificar a falha, porque classes distintas pedem respostas distintas. Erros transitórios (timeout, 502, 503, 429) pedem nova tentativa com backoff exponencial e jitter, limitada em número de tentativas; erros permanentes (401, 403, 404) pedem parada e alerta, pois tentar de novo apenas atrasa a descoberta; erros de conteúdo (200 com corpo inesperado) pedem contagem e amostra, pois indicam mudança de contrato na fonte. A regra de ouro é nunca reaproveitar o mesmo bloco de retry para todas as classes, erro que produz tanto tempestade de requisições inúteis quanto cegueira diante de credencial revogada.

import requests

def baixar_com_resiliencia(url, cabecalhos, tentativas=4):
    for n in range(1, tentativas + 1):
        try:
            r = requests.get(url, headers=cabecalhos, timeout=(5, 30))
            if r.status_code == 429:
                esperar = int(r.headers.get("Retry-After", 2 ** n * 5))
                time.sleep(esperar)
                continue
            if r.status_code in (502, 503, 504):
                time.sleep(2 ** n * 5 + random.uniform(0, 3))  # backoff com jitter
                continue
            r.raise_for_status()
            return r
        except requests.exceptions.Timeout:
            if n == tentativas:
                raise
            time.sleep(2 ** n * 5)
    raise RuntimeError(f"fonte instavel apos {tentativas} tentativas: {url}")

A resiliência completa inclui três elementos fora do retry. O timeout sempre explícito, com timeout de conexão separado do de leitura, pois coletores presos em conexão zumbi são falha silenciosa clássica. O circuito de saúde: após falhas consecutivas acima do limiar, o coletor entra em pausa e emite evento para o canal de ingestão, evitando marteladas contra fonte em decomposição. E o compromisso de não perder dados: quando o coletor falha, o ciclo seguinte deve alcançar a janela perdida por meio do marcador incremental, que só avança após publicação confirmada na fila, nunca após o download. Marcador que avança antes da fila é a receita da lacuna silenciosa de coleta, detectável apenas semanas depois, se houver sorte.

6. Exemplo Integrado: Feed, VirusTotal e MISP

O exemplo abaixo consolida os padrões da nota em um coletor realista: baixa um feed incremental, enriquece hashes com a API v3 do VirusTotal respeitando o limite público de quatro requisições por minuto e publica atributos no MISP pela API (/attributes/add), com credenciais vindas do ambiente e cache em processo:

import os
import requests
from requests.adapters import HTTPAdapter

MISP_URL = os.environ["MISP_URL"]
MISP_KEY = os.environ["MISP_KEY"]
VT_KEY = os.environ["VT_API_KEY"]
limite_vt = Limitador(max_por_minuto=4)
cache_vt = {}

def buscar_vt(hash_sha256):
    if hash_sha256 in cache_vt:
        return cache_vt[hash_sha256]
    limite_vt.aguardar()
    r = requests.get(
        f"https://www.virustotal.com/api/v3/files/{hash_sha256}",
        headers={"x-apikey": VT_KEY},
        timeout=(5, 30),
    )
    if r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", 60)))
        return buscar_vt(hash_sha256)
    r.raise_for_status()
    dados = r.json()["data"]["attributes"]
    cache_vt[hash_sha256] = {
        "deteccoes": dados["last_analysis_stats"].get("malicious", 0),
        "familia": next(iter(dados.get("popular_threat_classification", {})
                             .get("suggested_threat_label", "").split(".")), None),
    }
    return cache_vt[hash_sha256]

def publicar_misp(atributos):
    r = requests.post(
        f"{MISP_URL}/attributes/add/{os.environ['MISP_EVENTO_ID']}",
        headers={"Authorization": MISP_KEY, "Content-Type": "application/json"},
        json={"attribute": atributos},
        timeout=(5, 30),
    )
    r.raise_for_status()
    return r.json()

Três detalhes desse exemplo merecem explicitação porque generalizam para qualquer integração real. A resposta do MISP à criação em lote deve ser verificada por item (a API aceita a requisição e rejeita itens individualmente), o que exige leitura do corpo e contagem de sucesso. O cache do VirusTotal em processo serve para um coletor; em produção, cache compartilhado (Redis) evita que múltiplos coletores queimem a mesma quota com as mesmas chaves. E o usuário do feed nunca deve receber a URL enriquecida sem desarme, mantendo a regra da nota 09 sobre manuseio seguro de observáveis.

7. Erros Comuns e Pontos de Atenção

O erro mais frequente é ignorar o Retry-After e tratar 429 com backoff genérico, prolongando o bloqueio que se tentava evitar; a segunda posição é o coletor sem timeout explícito, que estagna conexões e consome workers. Em sequência vêm: credencial em código ou repositório, descoberta tarde por vazamento; marcador incremental que avança antes da confirmação de publicação, gerando lacunas silenciosas; e parsing frágil por posição ou por chave única, que quebra na primeira mudança cosmética da fonte, quando o contrato versionado da nota 08 já teria evitado o susto.

Na automação de enriquecimento via API, os erros clássicos são: enriquecer tudo na ingestão (em vez do mínimo para triagem), queimar quota paga com dado que ninguém usará; cache sem TTL por classe, servindo dado velho como se fosse novo; e cachear erros, transformando uma falha pontual do provedor em falha persistente por horas. No streaming, o erro típico é a reconexão sem recuperação de janela, aceitando perda como custo de operação; em webhooks, é o receptor que processa antes de responder, acumulando timeouts e reentregas que multiplicam o processamento.

Um ponto de atenção final é a observabilidade do coletor como cidadão de primeira classe. Cada coletor deve emitir as métricas acordadas na nota 11 (volume por ciclo, taxa de rejeição, latência, falhas por classe) com rótulo de fonte, e seu log de execução deve ser suficientemente explícito para reconstruir o que aconteceu em qualquer janela sem ler código. Coletor invisível não é automação: é aposta, e apostas em coleta se pagam com lacuna de cobertura descoberta no pior momento possível.

8. Integração com o Restante da Biblioteca

Os coletores descritos aqui são o ponto de entrada do pipeline da nota 08, e todas as regras de contrato de mensagem, idempotência e filtragem inicial permanecem válidas para o código desta nota, que publica bruto e deixa decisões para os workers. As métricas emitidas por cliente alimentam diretamente o monitoramento da nota 11, que devolve SLO por fonte e orienta ajuste de intervalo, cache e quota. A nota de governança (13) define retenção do que os coletores armazenam, inclusive os brutos, e as restrições de dados pessoais que um coletor pode trazer de fontes externas para dentro do ambiente.

Com o restante da biblioteca, esta camada sustenta o controle CTI.4 da seção 15 (coleta de fontes técnicas) e os pontos de integração do PLT (MISP e OpenCTI), além de prover o fluxo de observáveis que os capítulos de hunting consomem como watchlists. A evolução de maturidade natural é tratar cada coletor como serviço com pacote, teste de contrato contra fonte real gravada (fixture), rollout versionado e rollback testado, porque coletores deixam de ser scripts quando passam a sustentar detecção e bloqueio; nessa fase, a pergunta de release deixa de ser funciona hoje e passa a ser quebra de forma detectável, com falha visível, recuperação automática e auditoria completa da janela afetada.