Raccogliere i log della piattaforma Swimlane
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)
- Accedi alla piattaforma Swimlane come amministratore account.
- Vai a Opzioni profilo.
- Fai clic su Profilo per aprire l'editor del profilo.
- Vai alla sezione Token di accesso personale.
- Fai clic su Genera token per creare un nuovo token di accesso personale.
- Copia immediatamente il token e conservalo in modo sicuro (non verrà mostrato di nuovo).
- 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
).
- Token di accesso personale (PAT): utilizzato nell'intestazione
Configura il bucket AWS S3 e IAM per Google SecOps
- Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket
- Salva il nome e la regione del bucket per riferimento futuro (ad esempio,
swimlane-audit
). - Crea un utente seguendo questa guida: Creazione di un utente IAM.
- Seleziona l'utente creato.
- Seleziona la scheda Credenziali di sicurezza.
- Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
- Seleziona Servizio di terze parti come Caso d'uso.
- Fai clic su Avanti.
- (Facoltativo) Aggiungi un tag di descrizione.
- Fai clic su Crea chiave di accesso.
- Fai clic su Scarica file CSV per salvare la chiave di accesso e la chiave di accesso segreta per un utilizzo successivo.
- Fai clic su Fine.
- Seleziona la scheda Autorizzazioni.
- Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
- Seleziona Aggiungi autorizzazioni.
- Seleziona Allega direttamente i criteri.
- Cerca e seleziona il criterio AmazonS3FullAccess.
- Fai clic su Avanti.
- Fai clic su Aggiungi autorizzazioni.
Configura il ruolo e il criterio IAM per i caricamenti S3
- Nella console AWS, vai a IAM > Policy > Crea policy > scheda JSON.
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.
- Sostituisci
Fai clic su Avanti > Crea criterio.
Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.
Allega la policy appena creata e la policy gestita da AWS:
- La norma personalizzata creata sopra
service-role/AWSLambdaBasicExecutionRole
(CloudWatch Logs)
Assegna al ruolo il nome
WriteSwimlaneAuditToS3Role
e fai clic su Crea ruolo.
Crea la funzione Lambda
- Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
- Fai clic su Crea autore da zero.
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
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())
Vai a Configurazione > Variabili di ambiente.
Fai clic su Modifica > Aggiungi nuova variabile di ambiente.
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
Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).
Seleziona la scheda Configurazione.
Nel riquadro Configurazione generale, fai clic su Modifica.
Modifica Timeout impostando 5 minuti (300 secondi) e fai clic su Salva.
Creare una pianificazione EventBridge
- Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
- 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
- Programma ricorrente: Tariffa (
- Fai clic su Crea pianificazione.
(Facoltativo) Crea chiavi e utenti IAM di sola lettura per Google SecOps
- Nella console AWS, vai a IAM > Utenti > Aggiungi utenti.
- Fai clic su Add users (Aggiungi utenti).
- Fornisci i seguenti dettagli di configurazione:
- Utente:
secops-reader
- Tipo di accesso: Chiave di accesso - Accesso programmatico
- Utente:
- Fai clic su Crea utente.
- Collega la criterio per la lettura minima (personalizzata): Utenti > secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Collega le norme direttamente > Crea norma.
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" } ] }
Imposta il nome su
secops-reader-policy
.Vai a Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.
Vai a Credenziali di sicurezza > Chiavi di accesso > Crea chiave di accesso.
Scarica il file CSV (questi valori vengono inseriti nel feed).
Configura un feed in Google SecOps per importare i log della piattaforma Swimlane
- Vai a Impostazioni SIEM > Feed.
- Fai clic su + Aggiungi nuovo feed.
- Nel campo Nome feed, inserisci un nome per il feed (ad esempio,
Swimlane Platform logs
). - Seleziona Amazon S3 V2 come Tipo di origine.
- Seleziona Swimlane Platform come Tipo di log.
- Fai clic su Avanti.
- 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.
- URI S3:
- Fai clic su Avanti.
- 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.