Recolha registos do MuleSoft Anypoint

Compatível com:

Este documento explica como carregar eventos de registos de trilhos de auditoria da plataforma MuleSoft Anypoint para o Google Security Operations através do AWS S3.

Antes de começar

Certifique-se de que tem os seguintes pré-requisitos:

  • Instância do Google SecOps
  • Acesso privilegiado ao MuleSoft
  • Acesso privilegiado ao AWS

Obtenha o ID da organização do MuleSoft

  1. Inicie sessão na Anypoint Platform.
  2. Aceda a Menu > Gestão de acesso.
  3. Na tabela Grupos de empresas, clique no nome da sua organização.
  4. Copie o ID da organização (por exemplo, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Configure o contentor do AWS S3 e o IAM para o Google SecOps

  1. Crie um contentor do Amazon S3 seguindo este manual do utilizador: Criar um contentor.
  2. Guarde o nome e a região do contentor para referência futura (por exemplo, mulesoft-audit-logs).
  3. Crie um utilizador seguindo este guia do utilizador: criar um utilizador do IAM.
  4. Selecione o utilizador criado.
  5. Selecione o separador Credenciais de segurança.
  6. Clique em Criar chave de acesso na secção Chaves de acesso.
  7. Selecione Serviço de terceiros como Exemplo de utilização.
  8. Clicar em Seguinte.
  9. Opcional: adicione a etiqueta de descrição.
  10. Clique em Criar chave de acesso.
  11. Clique em Transferir ficheiro CSV para guardar a chave de acesso e a chave de acesso secreta para referência futura.
  12. Clique em Concluído.
  13. Selecione o separador Autorizações.
  14. Clique em Adicionar autorizações na secção Políticas de autorizações.
  15. Selecione Adicionar autorizações.
  16. Selecione Anexar políticas diretamente.
  17. Pesquise e selecione a política AmazonS3FullAccess.
  18. Clicar em Seguinte.
  19. Clique em Adicionar autorizações.

Crie a app associada do MuleSoft

  1. Inicie sessão na Anypoint Platform.
  2. Aceda a Gestão de acesso > Apps ligadas > Criar app.
  3. Indique os seguintes detalhes de configuração:
    • Nome da app: introduza um nome exclusivo (por exemplo, Google SecOps export).
    • Selecione A app age em seu próprio nome (credenciais do cliente).
    • Clique em Adicionar âmbitos → Visualizador do registo de auditoria → Seguinte.
    • Selecione todos os grupos de empresas cujos registos necessita.
    • Clique em Seguinte > Adicionar âmbitos.
  4. Clique em Guardar e copie o ID de cliente e o segredo do cliente.

Configure a política e a função do IAM para carregamentos do S3

  1. JSON da política (substitua mulesoft-audit-logs pelo nome do seu contentor):

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutAuditObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::mulesoft-audit-logs/*"
    }
  ]
}

  2. Aceda a AWS console > IAM > Policies > Create policy > separador JSON.

  3. Copie e cole a política.

  4. Clique em Seguinte > Criar política.

  5. Aceda a IAM > Funções > Criar função > Serviço AWS > Lambda.

  6. Anexe a política criada recentemente.

  7. Dê o nome WriteMulesoftToS3Role à função e clique em Criar função.

Crie a função Lambda

Definição Valor
Nome mulesoft_audit_to_s3
Runtime Python 3.13
Arquitetura x86_64
Função de execução Use o > existente WriteMulesoftToS3Role

  1. Depois de criar a função, abra o separador Código, elimine o fragmento de código e introduza o seguinte código (mulesoft_audit_to_s3.py).

    #!/usr/bin/env python3

import os, json, gzip, io, uuid, datetime as dt, urllib.request, urllib.error, urllib.parse
import boto3

ORG_ID        = os.environ["MULE_ORG_ID"]
CLIENT_ID     = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
S3_BUCKET     = os.environ["S3_BUCKET_NAME"]

TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
QUERY_URL = f"https://anypoint.mulesoft.com/audit/v2/organizations/{ORG_ID}/query"

def http_post(url, data, headers=None):
    raw = json.dumps(data).encode() if headers else urllib.parse.urlencode(data).encode()
    req = urllib.request.Request(url, raw, headers or {})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.loads(r.read())
    except urllib.error.HTTPError as e:
        print("MuleSoft error body →", e.read().decode())
        raise

def get_token():
    return http_post(TOKEN_URL, {
        "grant_type": "client_credentials",
        "client_id":  CLIENT_ID,
        "client_secret": CLIENT_SECRET
    })["access_token"]

def fetch_audit(token, start, end):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type":  "application/json"
    }
    body = {
        "startDate": f"{start.isoformat(timespec='milliseconds')}Z",
        "endDate":   f"{end.isoformat(timespec='milliseconds')}Z",
        "limit": 200,
        "offset": 0,
        "ascending": False
    }
    while True:
        data = http_post(QUERY_URL, body, headers)
        if not data.get("data"):
            break
        yield from data["data"]
        body["offset"] += body["limit"]

def upload(events, ts):
    key = f"{ts:%Y/%m/%d}/mulesoft-audit-{uuid.uuid4()}.json.gz"
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="w") as gz:
        for ev in events:
            gz.write((json.dumps(ev) + "\n").encode())
    buf.seek(0)
    boto3.client("s3").upload_fileobj(buf, S3_BUCKET, key)

def lambda_handler(event=None, context=None):
    now   = dt.datetime.utcnow().replace(microsecond=0)
    start = now - dt.timedelta(days=1)

    token  = get_token()
    events = list(fetch_audit(token, start, now))

    if events:
        upload(events, start)
        print(f"Uploaded {len(events)} events")
    else:
        print("No events in the last 24 h")

# For local testing
if __name__ == "__main__":
    lambda_handler()

  2. Aceda a Configuração > Variáveis de ambiente > Editar > Adicionar nova variável de ambiente.

  3. Introduza as seguintes variáveis de ambiente fornecidas, substituindo-as pelo seu valor.

    Chave Valor de exemplo
    MULE_ORG_ID your_org_id
    CLIENT_ID your_client_id
    CLIENT_SECRET your_client_secret
    S3_BUCKET_NAME mulesoft-audit-logs

Agende a função Lambda (EventBridge Scheduler)

  1. Aceda a Configuration > Triggers > Add trigger > EventBridge Scheduler > Create rule.
  2. Indique os seguintes detalhes de configuração:
    • Nome: daily-mulesoft-audit export.
    • Padrão de agendamento: expressão cron.
    • Expressão: 0 2 * * * (executado diariamente às 02:00 UTC).
  3. Deixe o resto como predefinição e clique em Criar.

Configure um feed no Google SecOps para carregar os registos do MuleSoft

  1. Aceda a Definições do SIEM > Feeds.
  2. Clique em Adicionar novo.
  3. No campo Nome do feed, introduza um nome para o feed (por exemplo, MuleSoft Logs).
  4. Selecione Amazon S3 V2 como o Tipo de origem.
  5. Selecione Mulesoft como o Tipo de registo.
  6. Clicar em Seguinte.

  7. Especifique valores para os seguintes parâmetros de entrada:

    • URI do S3: o URI do contentor
      • s3://mulesoft-audit-logs/
        • Substitua mulesoft-audit-logs pelo nome real do contentor.

    • Opções de eliminação da origem: selecione a opção de eliminação de acordo com a sua preferência.

    • Idade máxima do ficheiro: inclua ficheiros modificados no último número de dias. A predefinição é 180 dias.

    • ID da chave de acesso: chave de acesso do utilizador com acesso ao contentor do S3.

    • Chave de acesso secreta: chave secreta do utilizador com acesso ao contentor do S3.

    • Espaço de nomes do recurso: o espaço de nomes do recurso.

    • Etiquetas de carregamento: a etiqueta a aplicar aos eventos deste feed.

  8. Clicar em Seguinte.

  9. Reveja a nova configuração do feed no ecrã Finalizar e, de seguida, clique em Enviar.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais da Google SecOps.