Passar para o conteúdo principal

Integração com Elastic Security

Este guia explica como ingerir feeds da Plataforma Axur no Elastic Security usando entrega Push (Webhook 2.0) sem infraestrutura externa. Utilizaremos apenas recursos nativos do Elastic Stack: índices, pipelines de ingestão e Watcher para dividir os dados em documentos em nível de item.

Importante: Esta é apenas uma abordagem de integração possível. O design do índice e as transformações de dados podem ser personalizados para corresponder ao seu ambiente e preferências. A abordagem abaixo foi projetada para ingerir dados diretamente do Feed da Axur e dividi-los em índices específicos por entidade (por exemplo, tickets e credentials).

Nota: Este tutorial assume que você já tem acesso à Plataforma Axur e à chave de API necessária.

Pré-requisitos

  • Elastic Stack (Elasticsearch + Kibana) versão 8.x

  • Acesso administrativo ao Kibana

  • Acesso de rede da infraestrutura da Axur ao seu cluster Elastic

  • Chave de API do Elastic com permissões de write para os índices de destino

Considerações sobre custo e faturamento

  • Importante: Esta integração usa a REST API e o Watcher embutidos do Elastic. Não há cobrança por parte da Axur para esta integração.

  • Principais fatores de custo:

    • Licenciamento/recursos do Elastic utilizados (Watcher requer licença apropriada)

    • Armazenamento para eventos ingeridos e índices derivados

    • Saída/entrada de rede dependendo da sua hospedagem

  • Dicas para controlar o custo:

    • Comece com filtros de feed mais restritivos para limitar o volume inicialmente

    • Aplique políticas de ILM aos índices criados aqui

    • Monitore a ingestão e ajuste os filtros do feed conforme necessário

Dica: Um glossário de termos do Elastic utilizados neste tutorial está disponível no final do documento.

1) Crie a chave de API da Axur

Crie uma chave de API. Mantenha o valor à mão para as próximas etapas.

Gerar uma chave de API

  1. Na Plataforma Axur, vá para configuração de chaves de API.

  2. Crie uma nova chave de API e copie o valor da chave de API com segurança.

Nota: O feed retorna apenas os dados que o usuário da chave de API tem permissão para acessar.

Capturas de tela:

A partir deste ponto (Etapas 2–6), toda a configuração é realizada no console Elastic/Kibana.

2) No Elastic: Crie o índice de entrada

Crie um índice para receber os dados da Axur diretamente via HTTP POST. Use flattened para lidar com campos dinâmicos.

PUT axur-feed-input
{
 "mappings": {
 "dynamic": true,
 "dynamic_templates": [
 {
 "collectionData_flattened": {
 "path_match": "collectionData.*",
 "mapping": { "type": "flattened" }
 }
 }
 ]
 }
}

Capturas de tela:

3) No Elastic: Crie o pipeline de ingestão

Adicione um carimbo de data/hora de ingestão a cada documento recebido no índice de entrada.

PUT _ingest/pipeline/axur-feed-input-pipeline
{
 "processors": [
 { "set": { "field": "ingested_at", "value": "{{_ingest.timestamp}}" } }
 ]
}

Importante: A URL Push do Feed da Axur deve incluir ?pipeline=axur-feed-input-pipeline para anexar este pipeline.

Capturas de tela:

4) No Elastic: Crie os índices de saída

Estes índices armazenarão documentos em nível de item, já divididos por tipo.

Índice de Tickets:

PUT axur-feed-ticket_splitted
{
 "mappings": {
 "dynamic": true,
 "properties": {
 "item": { "type": "flattened" },
 "feedData": { "type": "flattened" }
 }
 }
}

Índice de Credenciais (detections):

PUT axur-feed-credential_splitted
{
 "mappings": {
 "dynamic": true,
 "properties": {
 "item": { "type": "flattened" },
 "feedData": { "type": "flattened" }
 }
 }
}

Capturas de tela:

5) No Elastic: Crie o índice de checkpoint

Armazene o último carimbo de data/hora processado para evitar reprocessamento.

PUT axur-feed-checkpoint
{
 "mappings": {
 "properties": {
 "last_ts": { "type": "date" }
 }
 }
}

Capturas de tela:

6) No Elastic: Crie o Watcher para dividir itens

Este Watcher processa apenas novos documentos de entrada e indexa itens divididos nos índices de saída, atualizando o checkpoint.

PUT _watcher/watch/axur_split_batch
{
 "trigger": { "schedule": { "interval": "1m" } },
 "input": {
 "chain": {
 "inputs": [
 {
 "cp": {
 "search": {
 "request": {
 "indices": ["axur-feed-checkpoint"],
 "body": {
 "size": 1,
 "query": { "ids": { "values": ["checkpoint"] } }
 }
 }
 }
 }
 },
 {
 "src": {
 "search": {
 "request": {
 "indices": ["axur-feed-input"],
 "body": {
 "size": 200,
 "query": {
 "range": {
 "ingested_at": {
 "gt": "{{#ctx.payload.cp.hits.hits}}{{ctx.payload.cp.hits.hits.0._source.last_ts}}{{/ctx.payload.cp.hits.hits}}{{^ctx.payload.cp.hits.hits}}1970-01-01T00:00:00Z{{/ctx.payload.cp.hits.hits}}"
 }
 }
 },
 "sort": [{ "ingested_at": "asc" }]
 }
 }
 }
 }
 }
 ]
 }
 },
 "condition": {
 "script": {
 "lang": "painless",
 "source": "def h = ctx?.payload?.src?.hits; if (h == null) return false; def t = h.total; if (t == null) return false; if (t instanceof Map) return (t.value != null && t.value > 0); return (t > 0);"
 }
 },
 "transform": {
 "script": {
 "lang": "painless",
 "source": """
 def tics = new ArrayList();
 def creds = new ArrayList();
 def maxTs = null;

 for (def hit : ctx.payload.src.hits.hits) {
 def src = hit._source;
 if (src == null) continue;

 def ts = src.containsKey('ingested_at') ? src.ingested_at : null;
 if (ts != null) {
 if (maxTs == null) { maxTs = ts; }
 else if (java.time.ZonedDateTime.parse(ts).isAfter(java.time.ZonedDateTime.parse(maxTs))) { maxTs = ts; }
 }

 def feed = src.containsKey('feedData') ? src.feedData : null;
 if (!src.containsKey('collectionData')) continue;
 def cd = src.collectionData;

 if (cd != null && cd.containsKey('tickets') && cd.tickets != null) {
 for (def it : cd.tickets) {
 if (it == null) continue;
 tics.add(['feedData': feed, 'item': it]);
 }
 }
 if (cd != null && cd.containsKey('detections') && cd.detections != null) {
 for (def it2 : cd.detections) {
 if (it2 == null) continue;
 creds.add(['feedData': feed, 'item': it2]);
 }
 }
 }

 return ['tics': tics, 'creds': creds, 'next_checkpoint': maxTs];
 """
 }
 },
 "actions": {
 "index_tics": {
 "foreach": "ctx.payload.tics",
 "max_iterations": 500,
 "transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
 "index": { "index": "axur-feed-ticket_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
 },
 "index_creds": {
 "foreach": "ctx.payload.creds",
 "max_iterations": 500,
 "transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
 "index": { "index": "axur-feed-credential_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
 },
 "update_checkpoint": {
 "condition": { "script": { "lang": "painless", "source": "return ctx.payload.next_checkpoint != null;" } },
 "transform": { "script": { "lang": "painless", "source": "return ['last_ts': ctx.payload.next_checkpoint];" } },
 "index": { "index": "axur-feed-checkpoint", "doc_id": "checkpoint", "execution_time_field": "updated_at", "refresh": "wait_for" }
 }
 }
}

Capturas de tela:

7) Configure o endpoint do Feed PUSH da Axur

Aponte o Feed da Axur para o seu índice de entrada do Elastic e anexe o pipeline.

Capturas de tela:

Validação

Após a configuração, envie um documento de teste para o índice de entrada (Kibana → Dev Tools):

POST axur-feed-input/_doc?pipeline=axur-feed-input-pipeline
{
 "feedData": { "feedType": "ticket", "customerKey": "TEST" },
 "collectionData": {
 "tickets": [ { "ticket": { "id": "TCK123", "type": "leak" } } ],
 "detections": [ { "credential": { "user": "alice" } } ]
 }
}

Aguarde ~1 minuto e verifique:

GET axur-feed-ticket_splitted/_search

GET axur-feed-credential_splitted/_search

GET axur-feed-checkpoint/_doc/checkpoint

Você deve ver documentos em nível de item em cada índice de saída e um last_ts atualizado no checkpoint. Execuções subsequentes não devem reprocessar os mesmos documentos de entrada.

Capturas de tela:

Solução de problemas

  • Nenhum documento nos índices de saída: certifique-se de que o Watcher esteja ativo e não esteja falhando; verifique Watcher > History.

  • Erros de indexação 401/403: valide as permissões da chave de API do Elastic (write, create_index).

  • Duplicatas: confirme se o documento de checkpoint é atualizado; valide a sincronização de tempo do cluster.

  • Nada ingerido: verifique a acessibilidade da URL Push da Axur e o parâmetro pipeline no caminho.

Você terminou: onde encontrar os dados e um exemplo de alerta

  • Tickets são indexados em: axur-feed-ticket_splitted

  • Credenciais (detections) são indexadas em: axur-feed-credential_splitted

Exemplos de consultas (Kibana → Discover):

GET axur-feed-ticket_splitted/_search
{
 "query": { "term": { "item.detection.status": "open" } },
 "size": 20,
 "sort": [{ "indexed_at": { "order": "desc" } }]
}
GET axur-feed-credential_splitted/_search
{
 "query": { "exists": { "field": "item.credential.user" } },
 "size": 20,
 "sort": [{ "indexed_at": { "order": "desc" } }]
}

Crie uma regra de detecção (Elastic Security → Rules → Create rule → Custom query):

  • Padrão de índice: axur-feed-credential_splitted

  • Consulta personalizada (KQL): item.credential.user:* and item.access.domain:"example.com"

  • Agendamento: A cada 5 minutos

  • Gravidade: Média

  • Ações: adicione suas notificações (e-mail, Slack, etc.)

Como gerar a chave de API do Elastic

Crie uma chave de API no Elastic para autenticar as requisições Push da Axur.

No Kibana:

  1. Vá para Stack Management → Security → API Keys → Create API key.

  2. Nome: axur-feed-ingest.

  3. Privilégios:

    1. Privilégios de índice: adicione axur-feed-* (ou os índices específicos que você criou) com write e create_index.

  4. Crie e copie a chave no formato Encoded. Mantenha-a segura.

Capturas de tela:

Glossário

  • Elasticsearch: Motor de busca e análise distribuído do Elastic Stack.

  • Kibana: Interface de usuário web para gerenciar e visualizar dados do Elasticsearch.

  • Index: Um espaço de nomes lógico para documentos no Elasticsearch.

  • Ingest Pipeline: Processadores executados antes de indexar um documento.

  • Watcher: Recurso de alerta/automação usado aqui para implementar a divisão periódica.

  • ILM (Index Lifecycle Management): Política para gerenciar a retenção de dados e o desempenho.

  • API Key: Mecanismo de autenticação para o Elasticsearch.

  • Push (Webhook 2.0): Modo de entrega da Axur que envia eventos diretamente para o seu endpoint.

Se ficar com qualquer dúvida, é só chamar a gente no [email protected] 😊

Artigos relacionados
Add-on "Axur Data Connector" no Splunk
Integração com Microsoft Sentinel
Integração IBM QRadar
Respondeu à sua pergunta?