Raccogliere i log della piattaforma Swimlane

Supportato in:

Questo documento spiega come importare i log della piattaforma Swimlane in Google Security Operations utilizzando Amazon S3.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Accesso con privilegi a Swimlane (amministratore dell'account in grado di generare un token di accesso personale)
  • Accesso con privilegi ad AWS (S3, IAM, Lambda, EventBridge)

Raccogli i prerequisiti della piattaforma Swimlane (ID, chiavi API, ID organizzazione, token)

  1. Accedi alla piattaforma Swimlane come amministratore account.
  2. Vai a Opzioni profilo.
  3. Fai clic su Profilo per aprire l'editor del profilo.
  4. Vai alla sezione Token di accesso personale.
  5. Fai clic su Genera token per creare un nuovo token di accesso personale.
  6. Copia immediatamente il token e conservalo in modo sicuro (non verrà mostrato di nuovo).
  7. Registra i seguenti dettagli per l'integrazione:
    • Token di accesso personale (PAT): utilizzato nell'intestazione Private-Token per le chiamate API.
    • ID account: obbligatorio per il percorso dell'API Audit Log /api/public/audit/account/{ACCOUNT_ID}/auditlogs. Se non conosci il tuo ID account, contatta l'amministratore di Swimlane.
    • URL di base: il tuo dominio Swimlane (ad esempio, https://eu.swimlane.app, https://us.swimlane.app).

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, swimlane-audit).
  3. Crea un utente seguendo questa guida: 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 un tag di 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 un utilizzo successivo.
  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.

Configura il ruolo e il criterio IAM per i caricamenti S3

  1. Nella console AWS, vai a IAM > Policy > Crea policy > scheda JSON.
  2. Inserisci la seguente policy:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutSwimlaneAuditObjects",
          "Effect": "Allow",
          "Action": ["s3:PutObject"],
          "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/*"
        },
        {
          "Sid": "AllowStateReadWrite",
          "Effect": "Allow",
          "Action": ["s3:GetObject", "s3:PutObject"],
          "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/state.json"
        }
      ]
    }
    
    • Sostituisci swimlane-audit se hai inserito un nome bucket diverso.
  3. Fai clic su Avanti > Crea criterio.

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

  5. Allega la policy appena creata e la policy gestita da AWS:

    • La norma personalizzata creata sopra
    • service-role/AWSLambdaBasicExecutionRole (CloudWatch Logs)
  6. Assegna al ruolo il nome WriteSwimlaneAuditToS3Role e fai clic su Crea ruolo.

Crea la funzione Lambda

  1. Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
  2. Fai clic su Crea autore da zero.
  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome swimlane_audit_to_s3
    Tempo di esecuzione Python 3.13
    Architettura x86_64
    Ruolo di esecuzione WriteSwimlaneAuditToS3Role
  4. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e inserisci il seguente codice (swimlane_audit_to_s3.py):

    #!/usr/bin/env python3
    import os, json, gzip, io, uuid, datetime as dt, urllib.parse, urllib.request
    import boto3
    
    # ---- Environment ----
    S3_BUCKET = os.environ["S3_BUCKET"]
    S3_PREFIX = os.environ.get("S3_PREFIX", "swimlane/audit/")
    STATE_KEY = os.environ.get("STATE_KEY", S3_PREFIX + "state.json")
    BASE_URL = os.environ["SWIMLANE_BASE_URL"].rstrip("/")  # e.g., https://eu.swimlane.app
    ACCOUNT_ID = os.environ["SWIMLANE_ACCOUNT_ID"]
    TENANT_LIST = os.environ.get("SWIMLANE_TENANT_LIST", "")  # comma-separated; optional
    INCLUDE_ACCOUNT = os.environ.get("INCLUDE_ACCOUNT", "true").lower() == "true"
    PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "100"))  # max 100
    WINDOW_MINUTES = int(os.environ.get("WINDOW_MINUTES", "15"))  # time range per run
    PAT_TOKEN = os.environ["SWIMLANE_PAT_TOKEN"]  # Personal Access Token
    TIMEOUT = int(os.environ.get("TIMEOUT", "30"))
    
    AUDIT_URL = f"{BASE_URL}/api/public/audit/account/{ACCOUNT_ID}/auditlogs"
    
    s3 = boto3.client("s3")
    
    # ---- Helpers ----
    
    def _http(req: urllib.request.Request):
        return urllib.request.urlopen(req, timeout=TIMEOUT)
    
    def _now():
        return dt.datetime.utcnow()
    
    def get_state() -> dict:
        try:
            obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
            return json.loads(obj["Body"].read())
        except Exception:
            return {}
    
    def put_state(state: dict) -> None:
        state["updated_at"] = _now().isoformat() + "Z"
        s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state).encode())
    
    def build_url(from_dt: dt.datetime, to_dt: dt.datetime, page: int) -> str:
        params = {
            "pageNumber": str(page),
            "pageSize": str(PAGE_SIZE),
            "includeAccount": str(INCLUDE_ACCOUNT).lower(),
            "fromdate": from_dt.replace(microsecond=0).isoformat() + "Z",
            "todate": to_dt.replace(microsecond=0).isoformat() + "Z",
        }
        if TENANT_LIST:
            params["tenantList"] = TENANT_LIST
        return AUDIT_URL + "?" + urllib.parse.urlencode(params)
    
    def fetch_page(url: str) -> dict:
        headers = {
            "Accept": "application/json",
            "Private-Token": PAT_TOKEN,
        }
        req = urllib.request.Request(url, headers=headers)
        with _http(req) as r:
            return json.loads(r.read())
    
    def write_chunk(items: list[dict], ts: dt.datetime) -> str:
        key = f"{S3_PREFIX}{ts:%Y/%m/%d}/swimlane-audit-{uuid.uuid4()}.json.gz"
        buf = io.BytesIO()
        with gzip.GzipFile(fileobj=buf, mode="w") as gz:
            for rec in items:
                gz.write((json.dumps(rec) + "n").encode())
        buf.seek(0)
        s3.upload_fileobj(buf, S3_BUCKET, key)
        return key
    
    def lambda_handler(event=None, context=None):
        state = get_state()
    
        # determine window
        to_dt = _now()
        from_dt = to_dt - dt.timedelta(minutes=WINDOW_MINUTES)
        if (prev := state.get("last_to_dt")):
            try:
                from_dt = dt.datetime.fromisoformat(prev.replace("Z", "+00:00"))
            except Exception:
                pass
    
        page = int(state.get("page", 1))
        total_written = 0
    
        while True:
            url = build_url(from_dt, to_dt, page)
            resp = fetch_page(url)
            items = resp.get("auditlogs", []) or []
            if items:
                write_chunk(items, _now())
                total_written += len(items)
            next_path = resp.get("next")
            if not next_path:
                break
            page += 1
            state["page"] = page
    
        # advance state window
        state["last_to_dt"] = to_dt.replace(microsecond=0).isoformat() + "Z"
        state["page"] = 1
        put_state(state)
    
        return {"ok": True, "written": total_written, "from": from_dt.isoformat() + "Z", "to": to_dt.isoformat() + "Z"}
    
    if __name__ == "__main__":
        print(lambda_handler())
    
  5. Vai a Configurazione > Variabili di ambiente.

  6. Fai clic su Modifica > Aggiungi nuova variabile di ambiente.

  7. Inserisci le seguenti variabili di ambiente, sostituendole con i tuoi valori.

    Chiave Valore di esempio
    S3_BUCKET swimlane-audit
    S3_PREFIX swimlane/audit/
    STATE_KEY swimlane/audit/state.json
    SWIMLANE_BASE_URL https://eu.swimlane.app
    SWIMLANE_ACCOUNT_ID xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
    SWIMLANE_TENANT_LIST tenantA,tenantB (facoltativo)
    INCLUDE_ACCOUNT true
    PAGE_SIZE 100
    WINDOW_MINUTES 15
    SWIMLANE_PAT_TOKEN <your-personal-access-token>
    TIMEOUT 30
  8. Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).

  9. Seleziona la scheda Configurazione.

  10. Nel riquadro Configurazione generale, fai clic su Modifica.

  11. Modifica Timeout impostando 5 minuti (300 secondi) e fai clic su Salva.

Creare una pianificazione EventBridge

  1. Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
  2. Fornisci i seguenti dettagli di configurazione:
    • Programma ricorrente: Tariffa (15 min)
    • Target: la tua funzione Lambda swimlane_audit_to_s3
    • Nome: swimlane-audit-schedule-15min
  3. Fai clic su Crea pianificazione.

(Facoltativo) Crea chiavi e utenti IAM di sola lettura per Google SecOps

  1. Nella console AWS, vai a IAM > Utenti > Aggiungi utenti.
  2. Fai clic su Add users (Aggiungi utenti).
  3. Fornisci i seguenti dettagli di configurazione:
    • Utente: secops-reader
    • Tipo di accesso: Chiave di accesso - Accesso programmatico
  4. Fai clic su Crea utente.
  5. Collega la criterio per la lettura minima (personalizzata): Utenti > secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Collega le norme direttamente > Crea norma.
  6. Nell'editor JSON, inserisci la seguente policy:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::swimlane-audit/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::swimlane-audit"
        }
      ]
    }
    
  7. Imposta il nome su secops-reader-policy.

  8. Vai a Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.

  9. Vai a Credenziali di sicurezza > Chiavi di accesso > Crea chiave di accesso.

  10. Scarica il file CSV (questi valori vengono inseriti nel feed).

Configura un feed in Google SecOps per importare i log della piattaforma Swimlane

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su + Aggiungi nuovo feed.
  3. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Swimlane Platform logs).
  4. Seleziona Amazon S3 V2 come Tipo di origine.
  5. Seleziona Swimlane Platform come Tipo di log.
  6. Fai clic su Avanti.
  7. Specifica i valori per i seguenti parametri di input:
    • URI S3: s3://swimlane-audit/swimlane/audit/
    • 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. 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 applicata 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.