Raccogliere i log di MuleSoft Anypoint

Supportato in:

Questo documento spiega come importare gli eventi di audit trail dai log della piattaforma MuleSoft Anypoint in Google Security Operations utilizzando AWS S3.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Istanza Google SecOps
  • Accesso con privilegi a MuleSoft
  • Accesso privilegiato ad AWS

Ottenere l'ID organizzazione MuleSoft

  1. Accedi ad Anypoint Platform.
  2. Vai a Menu > Gestione accessi.
  3. Nella tabella Gruppi aziendali, fai clic sul nome della tua organizzazione.
  4. Copia l'ID organizzazione (ad esempio, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Configura il bucket AWS S3 e IAM per Google SecOps

  1. Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket.
  2. Salva il nome e la regione del bucket per riferimento futuro (ad esempio, mulesoft-audit-logs).
  3. Crea un utente seguendo questa guida utente: Creazione di un utente IAM.
  4. Seleziona l'utente creato.
  5. Seleziona la scheda Credenziali di sicurezza.
  6. Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
  7. Seleziona Servizio di terze parti come Caso d'uso.
  8. Fai clic su Avanti.
  9. (Facoltativo) Aggiungi il tag della descrizione.
  10. Fai clic su Crea chiave di accesso.
  11. Fai clic su Scarica file CSV per salvare la chiave di accesso e la chiave di accesso segreta per riferimento futuro.
  12. Fai clic su Fine.
  13. Seleziona la scheda Autorizzazioni.
  14. Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
  15. Seleziona Aggiungi autorizzazioni.
  16. Seleziona Allega direttamente i criteri.
  17. Cerca e seleziona il criterio AmazonS3FullAccess.
  18. Fai clic su Avanti.
  19. Fai clic su Aggiungi autorizzazioni.

Crea l'app connessa MuleSoft

  1. Accedi ad Anypoint Platform.
  2. Vai a Gestione accessi > App connesse > Crea app.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome app: inserisci un nome univoco (ad esempio, Google SecOps export).
    • Seleziona L'app agisce per proprio conto (credenziali client).
    • Fai clic su Aggiungi ambiti → Visualizzatore log di controllo → Avanti.
    • Seleziona tutti i gruppi di attività di cui hai bisogno dei log.
    • Fai clic su Avanti > Aggiungi ambiti.
  4. Fai clic su Salva e copia l'ID client e il client secret.

Configura il ruolo e il criterio IAM per i caricamenti S3

  1. JSON delle policy (sostituisci mulesoft-audit-logs con il nome del tuo bucket):

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

  2. Vai alla console AWS > IAM > Policy > Crea policy > scheda JSON.

  3. Copia e incolla le norme.

  4. Fai clic su Avanti > Crea criterio.

  5. Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.

  6. Allega il criterio appena creato.

  7. Assegna al ruolo il nome WriteMulesoftToS3Role e fai clic su Crea ruolo.

Crea la funzione Lambda

Impostazione Valore
Nome mulesoft_audit_to_s3
Tempo di esecuzione Python 3.13
Architettura x86_64
Ruolo di esecuzione Utilizza esistente > WriteMulesoftToS3Role

  1. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e inserisci il seguente codice (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. Vai a Configurazione > Variabili di ambiente > Modifica > Aggiungi nuova variabile di ambiente.

  3. Inserisci le seguenti variabili di ambiente fornite, sostituendole con il tuo valore.

    Chiave Valore di esempio
    MULE_ORG_ID your_org_id
    CLIENT_ID your_client_id
    CLIENT_SECRET your_client_secret
    S3_BUCKET_NAME mulesoft-audit-logs

Pianifica la funzione Lambda (EventBridge Scheduler)

  1. Vai a Configurazione > Trigger > Aggiungi trigger > EventBridge Scheduler > Crea regola.
  2. Fornisci i seguenti dettagli di configurazione:
    • Nome: daily-mulesoft-audit export
    • Pattern di pianificazione: espressione cron.
    • Espressione: 0 2 * * * (viene eseguita ogni giorno alle ore 02:00 UTC).
  3. Lascia invariate le altre impostazioni predefinite e fai clic su Crea.

Configura un feed in Google SecOps per importare i log MuleSoft

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo.
  3. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, MuleSoft Logs).
  4. Seleziona Amazon S3 V2 come Tipo di origine.
  5. Seleziona Mulesoft come tipo di log.
  6. Fai clic su Avanti.

  7. Specifica i valori per i seguenti parametri di input:

    • URI S3: l'URI del bucket
      • s3://mulesoft-audit-logs/
        • Sostituisci mulesoft-audit-logs con il nome effettivo del bucket.

    • Opzioni di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.

    • Età massima del file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.

    • ID chiave di accesso: chiave di accesso utente con accesso al bucket S3.

    • Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3.

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.

  8. Fai clic su Avanti.

  9. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.