Coletar registros da plataforma MuleSoft Anypoint

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Observação: como o MuleSoft é um tipo de registro sem um analisador predefinido, este guia aborda apenas a ingestão de registros. Você precisa configurar um analisador personalizado para converter os registros do MuleSoft em UDM.

Este documento explica como ingerir eventos de trilha de auditoria dos registros da plataforma MuleSoft Anypoint no Google Security Operations usando o Google Cloud Storage.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

• Uma instância do Google SecOps
• Um projeto do GCP com a API Cloud Storage ativada
• Permissões para criar e gerenciar buckets do GCS
• Permissões para gerenciar políticas do IAM em buckets do GCS
• Permissões para criar funções do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
• Permissões para criar contas de serviço
• Acesso privilegiado à plataforma MuleSoft Anypoint

Encontrar o ID da organização do MuleSoft

1. Faça login na Anypoint Platform.
2. Acesse Gerenciamento de acesso > Organizações.
3. Na tabela Grupos empresariais, clique no nome da sua organização.
4. Copie o ID da organização (por exemplo, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Ou acesse Grupos empresariais do MuleSoft e copie o ID do URL.

Criar o app conectado do MuleSoft

1. Faça login na Anypoint Platform.
2. Acesse Gerenciamento de acesso > Apps conectados > Criar app.
3. Informe os seguintes detalhes de configuração:
   • Nome do app: insira um nome exclusivo (por exemplo, Google SecOps export).
   • Selecione O app age em seu próprio nome (credenciais do cliente).
4. Clique em Adicionar escopos > Visualizador de registros de auditoria > Próxima.
5. Selecione todos os grupos de empresas cujos registros você precisa.
6. Clique em Próxima > Adicionar escopos.

7. Clique em Salvar e copie o ID do cliente e a chave secreta do cliente.

   Observação: outros escopos podem ser necessários com base nas permissões da organização. Se você encontrar erros de permissão, verifique se o app conectado recebeu acesso a todos os grupos de empresas necessários.

Criar um bucket do Google Cloud Storage

1. Acesse o Console do Google Cloud.
2. Selecione seu projeto ou crie um novo.
3. No menu de navegação, acesse Cloud Storage > Buckets.
4. Clique em Criar bucket.

5. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nomeie seu bucket	Insira um nome exclusivo globalmente, por exemplo, mulesoft-audit-logs.
   Tipo de local	Escolha com base nas suas necessidades (região, birregional, multirregional)
   Local	Selecione o local (por exemplo, us-central1).
   Classe de armazenamento	Padrão (recomendado para registros acessados com frequência)
   Controle de acesso	Uniforme (recomendado)
   Ferramentas de proteção	Opcional: ativar o controle de versões de objetos ou a política de retenção

6. Clique em Criar.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS.

Criar conta de serviço

1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
2. Clique em Criar conta de serviço.
3. Informe os seguintes detalhes de configuração:
   • Nome da conta de serviço: insira mulesoft-logs-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect MuleSoft Anypoint logs.
4. Clique em Criar e continuar.
5. Na seção Conceda a essa conta de serviço acesso ao projeto:
   1. Clique em Selecionar papel.
   2. Pesquise e selecione Administrador de objetos do Storage.
   3. Clique em + Adicionar outro papel.
   4. Pesquise e selecione Invocador do Cloud Run.
   5. Clique em + Adicionar outro papel.
   6. Pesquise e selecione Invocador do Cloud Functions.
6. Clique em Continuar.
7. Clique em Concluído.

Esses papéis são necessários para:

• Administrador de objetos do Storage: grava registros em um bucket do GCS e gerencia arquivos de estado.
• Invocador do Cloud Run: permite que o Pub/Sub invoque a função
• Invocador do Cloud Functions: permite a invocação de funções

Conceder permissões do IAM no bucket do GCS

Conceda permissões de gravação à conta de serviço no bucket do GCS:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, mulesoft-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
   • Atribuir papéis: selecione Administrador de objetos do Storage.
6. Clique em Salvar.

Criar tópico Pub/Sub

Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

1. No Console do GCP, acesse Pub/Sub > Tópicos.
2. Selecione Criar tópico.
3. Informe os seguintes detalhes de configuração:
   • ID do tópico: insira mulesoft-audit-trigger.
   • Não altere as outras configurações.
4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API MuleSoft Anypoint e gravá-los no GCS.

1. No console do GCP, acesse o Cloud Run.
2. Clique em Criar serviço.
3. Selecione Função (use um editor in-line para criar uma função).

4. Na seção Configurar, forneça os seguintes detalhes de configuração:

   Configuração	Valor
   Nome do serviço	mulesoft-audit-collector
   Região	Selecione a região que corresponde ao seu bucket do GCS (por exemplo, us-central1).
   Ambiente de execução	Selecione Python 3.12 ou uma versão mais recente.

5. Na seção Acionador (opcional):

   1. Clique em + Adicionar gatilho.
   2. Selecione Cloud Pub/Sub.
   3. Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico mulesoft-audit-trigger.
   4. Clique em Salvar.

6. Na seção Autenticação:

   1. Selecione Exigir autenticação.
   2. Confira o Identity and Access Management (IAM).
   Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.

7. Role a tela para baixo e abra Contêineres, rede, segurança.

8. Acesse a guia Segurança:

   • Conta de serviço: selecione a conta de serviço mulesoft-logs-collector-sa.

9. Acesse a guia Contêineres:

   1. Clique em Variáveis e secrets.
   2. Clique em + Adicionar variável para cada variável de ambiente:

   Nome da variável	Valor de exemplo
   MULE_ORG_ID	your_org_id
   CLIENT_ID	your_client_id
   CLIENT_SECRET	your_client_secret
   GCS_BUCKET	mulesoft-audit-logs

10. Role a tela para baixo na guia Variáveis e secrets até Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos).

11. Acesse a guia Configurações em Contêineres:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
    • Clique em Concluído.

12. Role até Ambiente de execução:

    • Selecione Padrão (recomendado).

13. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 ou ajuste com base na carga esperada.

14. Clique em Criar.

15. Aguarde a criação do serviço (1 a 2 minutos).

16. Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

1. Insira main em Ponto de entrada da função.

2. No editor de código em linha, crie dois arquivos:

   • Primeiro arquivo: main.py::

   import functions_framework
   from google.cloud import storage
   import json
   import os
   import urllib3
   from datetime import datetime, timedelta, timezone
   import uuid
   import time
   
   # Initialize HTTP client with timeouts
   http = urllib3.PoolManager(
       timeout=urllib3.Timeout(connect=5.0, read=30.0),
       retries=False,
   )
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   # MuleSoft API endpoints
   TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch MuleSoft audit logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       # Get environment variables
       org_id = os.environ.get('MULE_ORG_ID')
       client_id = os.environ.get('CLIENT_ID')
       client_secret = os.environ.get('CLIENT_SECRET')
       bucket_name = os.environ.get('GCS_BUCKET')
   
       if not all([org_id, client_id, client_secret, bucket_name]):
           print('Error: Missing required environment variables')
           return
   
       query_url = f"https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query"
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(bucket_name)
   
           # Get OAuth token
           token = get_token(client_id, client_secret)
   
           # Calculate time range (last 24 hours)
           now = datetime.now(timezone.utc).replace(microsecond=0)
           start = now - timedelta(days=1)
   
           print(f'Fetching audit logs from {start.isoformat()} to {now.isoformat()}')
   
           # Fetch audit logs
           events = list(fetch_audit(query_url, token, start, now))
   
           # Upload to GCS
           if events:
               upload_to_gcs(bucket, events, start)
               print(f'Uploaded {len(events)} events')
           else:
               print('No events in the last 24 hours')
   
       except Exception as e:
           print(f'Error processing logs: {str(e)}')
           raise
   
   def get_token(client_id, client_secret):
       """Get OAuth 2.0 access token from MuleSoft."""
       data = {
           'grant_type': 'client_credentials',
           'client_id': client_id,
           'client_secret': client_secret
       }
   
       encoded_data = urllib3.request.urlencode(data).encode('utf-8')
   
       backoff = 1.0
       max_retries = 3
   
       for attempt in range(max_retries):
           try:
               response = http.request(
                   'POST',
                   TOKEN_URL,
                   body=encoded_data,
                   headers={'Content-Type': 'application/x-www-form-urlencoded'}
               )
   
               if response.status == 429:
                   retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                   print(f'Rate limited (429) on token request. Retrying after {retry_after}s...')
                   time.sleep(retry_after)
                   backoff = min(backoff * 2, 30.0)
                   continue
   
               if response.status != 200:
                   raise Exception(f'Failed to get token: {response.status} - {response.data.decode()}')
   
               token_data = json.loads(response.data.decode('utf-8'))
               return token_data['access_token']
   
           except Exception as e:
               if attempt == max_retries - 1:
                   raise
               print(f'Token request failed (attempt {attempt + 1}/{max_retries}): {e}')
               time.sleep(backoff)
               backoff = min(backoff * 2, 30.0)
   
       raise Exception('Failed to get token after maximum retries')
   
   def fetch_audit(query_url, token, start, end):
       """Fetch audit logs from MuleSoft API with pagination."""
       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
       }
   
       backoff = 1.0
   
       while True:
           try:
               response = http.request(
                   'POST',
                   query_url,
                   body=json.dumps(body).encode('utf-8'),
                   headers=headers
               )
   
               # Handle rate limiting with exponential backoff
               if response.status == 429:
                   retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                   print(f'Rate limited (429). Retrying after {retry_after}s...')
                   time.sleep(retry_after)
                   backoff = min(backoff * 2, 30.0)
                   continue
   
               backoff = 1.0
   
               if response.status != 200:
                   print(f'HTTP Error: {response.status}')
                   response_text = response.data.decode('utf-8')
                   print(f'Response body: {response_text}')
                   break
   
               data = json.loads(response.data.decode('utf-8'))
   
               if not data.get('data'):
                   break
   
               yield from data['data']
               body['offset'] += body['limit']
   
           except Exception as e:
               print(f'Error fetching audit logs: {e}')
               break
   
   def upload_to_gcs(bucket, events, timestamp):
       """Upload events to GCS as compressed JSON."""
       import gzip
       import io
   
       # Create blob name with timestamp and UUID
       blob_name = f"{timestamp.strftime('%Y/%m/%d')}/mulesoft-audit-{uuid.uuid4()}.json.gz"
   
       # Compress events
       buf = io.BytesIO()
       with gzip.GzipFile(fileobj=buf, mode='w') as gz:
           for event in events:
               gz.write((json.dumps(event) + '\n').encode('utf-8'))
   
       buf.seek(0)
   
       # Upload to GCS
       blob = bucket.blob(blob_name)
       blob.upload_from_file(buf, content_type='application/gzip')
   
       print(f'Uploaded to gs://{bucket.name}/{blob_name}')
   

   • Segundo arquivo: requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. Clique em Implantar para salvar e implantar a função.

4. Aguarde a conclusão da implantação (2 a 3 minutos).

Considerações importantes

Limitação de taxa:o endpoint de consulta de registros de auditoria aplica limites de taxa por IP nos três planos de controle. O plano de controle dos EUA permite 700 solicitações por minuto e IP, enquanto os planos de controle da UE e do governo permitem 40 solicitações por minuto e IP. A função implementa espera exponencial para lidar com a limitação de taxa automaticamente.

Expiração do token:os tokens de acesso geralmente expiram de 30 a 60 minutos após a emissão. A função solicita um novo token para cada execução. Para implantações de produção com execuções frequentes, considere implementar o armazenamento em cache de tokens com lógica de atualização.

Retenção de registros de auditoria:os registros de auditoria têm um período de retenção padrão de um ano. Se a organização foi criada antes de 10 de julho de 2023 e você não mudou manualmente o período de retenção, ele é de seis anos. Faça o download dos registros periodicamente se precisar mantê-los além do período de retenção configurado.

Criar o job do Cloud Scheduler

O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

1. No Console do GCP, acesse o Cloud Scheduler.
2. Clique em Criar job.

3. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nome	daily-mulesoft-audit-export
   Região	Selecione a mesma região da função do Cloud Run
   Frequência	0 2 * * * (executado diariamente às 2h UTC)
   Fuso horário	Selecione o fuso horário (UTC recomendado)
   Tipo de destino	Pub/Sub
   Tópico	Selecione o tópico mulesoft-audit-trigger.
   Corpo da mensagem	{} (objeto JSON vazio)

4. Clique em Criar.

Testar o job do programador

1. No console do Cloud Scheduler, encontre seu job.
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > mulesoft-audit-collector > Registros.
4. Verifique se a função foi executada com sucesso.
5. Verifique o bucket do GCS para confirmar se os registros foram gravados.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, MuleSoft Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Mulesoft como o Tipo de registro.

7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:

   [e-mail]
   

8. Copie esse endereço de e-mail para usar na próxima etapa.

   Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
   • Atribuir papéis: selecione Leitor de objetos do Storage.

6. Clique em Salvar.

   Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar um feed no Google SecOps para ingerir os registros do MuleSoft

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, MuleSoft Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Mulesoft como o Tipo de registro.
7. Clique em Próxima.

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

   • URL do bucket de armazenamento: insira o URI do bucket do GCS:

     gs://mulesoft-audit-logs/
     

     • Substitua mulesoft-audit-logs pelo nome real do bucket.
     Observação: sempre inclua a barra (/) no final do URI.

   • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

     • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
     • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.

     • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.

       Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.

   • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.

   • Namespace do recurso: o namespace do recurso.

   • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

9. Clique em Próxima.

10. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

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