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
Na Plataforma Axur, vá para configuração de chaves de API.
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.
Endpoint URL:
https://YOUR_ELASTIC_CLUSTER/axur-feed-input/_doc?pipeline=axur-feed-input-pipelineCabeçalhos a adicionar:
Authorization: ApiKey <YOUR_ENCODED_ELASTIC_API_KEY>(copie o formatoEncodedda UI de API Keys do Elastic)
Secret Key: configure uma
secretsegura no feed; a Axur assina as requisições com HMAC. Opcionalmente, verifique no lado do Elastic.
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
pipelineno caminho.
Você terminou: onde encontrar os dados e um exemplo de alerta
Tickets são indexados em:
axur-feed-ticket_splittedCredenciais (
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_splittedConsulta 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:
Vá para Stack Management → Security → API Keys → Create API key.
Nome: axur-feed-ingest.
Privilégios:
Privilégios de índice: adicione axur-feed-* (ou os índices específicos que você criou) com write e create_index.
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] 😊