Collecter les journaux de la plate-forme Swimlane

Compatible avec :

Ce document explique comment ingérer les journaux de la plate-forme Swimlane dans Google Security Operations à l'aide d'Amazon S3.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps
  • Accès privilégié à Swimlane (administrateur de compte capable de générer un jeton d'accès personnel)
  • Accès privilégié à AWS (S3, IAM, Lambda, EventBridge)

Collecter les conditions préalables de la plate-forme Swimlane (ID, clés API, ID d'organisation, jetons)

  1. Connectez-vous à la plate-forme Swimlane en tant qu'administrateur de compte.
  2. Accédez à Options du profil.
  3. Cliquez sur Profil pour ouvrir l'éditeur de profil.
  4. Accédez à la section Personal Access Token (Jeton d'accès personnel).
  5. Cliquez sur Générer un jeton pour créer un jeton d'accès personnel.
  6. Copiez immédiatement le jeton et stockez-le en lieu sûr (il ne s'affichera plus).
  7. Notez les informations suivantes concernant l'intégration :
    • Jeton d'accès personnel (PAT) : utilisé dans l'en-tête Private-Token pour les appels d'API.
    • ID du compte : obligatoire pour le chemin d'accès à l'API Audit Log /api/public/audit/account/{ACCOUNT_ID}/auditlogs. Contactez votre administrateur Swimlane si vous ne connaissez pas votre ID de compte.
    • URL de base : votre domaine Swimlane (par exemple, https://eu.swimlane.app, https://us.swimlane.app).

Configurer un bucket AWS S3 et IAM pour Google SecOps

  1. Créez un bucket Amazon S3 en suivant ce guide de l'utilisateur : Créer un bucket.
  2. Enregistrez le Nom et la Région du bucket pour référence ultérieure (par exemple, swimlane-audit).
  3. Créez un utilisateur en suivant ce guide : Créer un utilisateur IAM.
  4. Sélectionnez l'utilisateur créé.
  5. Sélectionnez l'onglet Informations d'identification de sécurité.
  6. Cliquez sur Créer une clé d'accès dans la section Clés d'accès.
  7. Sélectionnez Service tiers comme Cas d'utilisation.
  8. Cliquez sur Suivant.
  9. Facultatif : ajoutez un tag de description.
  10. Cliquez sur Créer une clé d'accès.
  11. Cliquez sur Télécharger le fichier CSV pour enregistrer la clé d'accès et la clé d'accès secrète pour une utilisation ultérieure.
  12. Cliquez sur OK.
  13. Sélectionnez l'onglet Autorisations.
  14. Cliquez sur Ajouter des autorisations dans la section Règles d'autorisation.
  15. Sélectionnez Ajouter des autorisations.
  16. Sélectionnez Joindre directement des règles.
  17. Recherchez et sélectionnez la règle AmazonS3FullAccess.
  18. Cliquez sur Suivant.
  19. Cliquez sur Ajouter des autorisations.

Configurer la stratégie et le rôle IAM pour les importations S3

  1. Dans la console AWS, accédez à IAM > Stratégies > Créer une stratégie > Onglet JSON.
  2. Saisissez la règle suivante :

    {
      "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"
        }
      ]
    }
    
    • Remplacez swimlane-audit si vous avez saisi un autre nom de bucket.
  3. Cliquez sur Suivant > Créer une règle.

  4. Accédez à IAM > Rôles > Créer un rôle > Service AWS > Lambda.

  5. Associez la stratégie nouvellement créée et la stratégie gérée par AWS :

    • Règle personnalisée créée ci-dessus
    • service-role/AWSLambdaBasicExecutionRole (journaux CloudWatch)
  6. Nommez le rôle WriteSwimlaneAuditToS3Role, puis cliquez sur Créer un rôle.

Créer la fonction Lambda

  1. Dans la console AWS, accédez à Lambda > Fonctions > Créer une fonction.
  2. Cliquez sur Créer à partir de zéro.
  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom swimlane_audit_to_s3
    Durée d'exécution Python 3.13
    Architecture x86_64
    Rôle d'exécution WriteSwimlaneAuditToS3Role
  4. Une fois la fonction créée, ouvrez l'onglet Code, supprimez le stub et saisissez le code suivant (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. Accédez à Configuration > Variables d'environnement.

  6. Cliquez sur Modifier > Ajouter une variable d'environnement.

  7. Saisissez les variables d'environnement suivantes en remplaçant par vos valeurs.

    Clé Exemple de valeur
    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 (facultatif)
    INCLUDE_ACCOUNT true
    PAGE_SIZE 100
    WINDOW_MINUTES 15
    SWIMLANE_PAT_TOKEN <your-personal-access-token>
    TIMEOUT 30
  8. Une fois la fonction créée, restez sur sa page (ou ouvrez Lambda > Fonctions > votre-fonction).

  9. Accédez à l'onglet Configuration.

  10. Dans le panneau Configuration générale, cliquez sur Modifier.

  11. Définissez le délai avant expiration sur 5 minutes (300 secondes), puis cliquez sur Enregistrer.

Créer une programmation EventBridge

  1. Accédez à Amazon EventBridge> Scheduler> Create schedule.
  2. Fournissez les informations de configuration suivantes :
    • Calendrier récurrent : Taux (15 min)
    • Cible : votre fonction Lambda swimlane_audit_to_s3
    • Nom : swimlane-audit-schedule-15min
  3. Cliquez sur Créer la programmation.

Facultatif : Créez un utilisateur et des clés IAM en lecture seule pour Google SecOps

  1. Dans la console AWS, accédez à IAM > Utilisateurs > Ajouter des utilisateurs.
  2. Cliquez sur Add users (Ajouter des utilisateurs).
  3. Fournissez les informations de configuration suivantes :
    • Utilisateur : secops-reader
    • Type d'accès : Clé d'accès – Accès programmatique
  4. Cliquez sur Créer un utilisateur.
  5. Associez une stratégie de lecture minimale (personnalisée) : Utilisateurs > secops-reader > Autorisations > Ajouter des autorisations > Associer des stratégies directement > Créer une stratégie.
  6. Dans l'éditeur JSON, saisissez la stratégie suivante :

    {
      "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. Définissez le nom sur secops-reader-policy.

  8. Accédez à Créer une règle > recherchez/sélectionnez > Suivant > Ajouter des autorisations.

  9. Accédez à Identifiants de sécurité > Clés d'accès > Créer une clé d'accès.

  10. Téléchargez le CSV (ces valeurs sont saisies dans le flux).

Configurer un flux dans Google SecOps pour ingérer les journaux de la plate-forme Swimlane

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur + Ajouter un flux.
  3. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Swimlane Platform logs).
  4. Sélectionnez Amazon S3 V2 comme type de source.
  5. Sélectionnez Swimlane Platform comme Type de journal.
  6. Cliquez sur Suivant.
  7. Spécifiez les valeurs des paramètres d'entrée suivants :
    • URI S3 : s3://swimlane-audit/swimlane/audit/
    • Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
    • Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours. 180 jours par défaut.
    • ID de clé d'accès : clé d'accès utilisateur ayant accès au bucket S3.
    • Clé d'accès secrète : clé secrète de l'utilisateur ayant accès au bucket S3.
    • Espace de noms de l'élément : espace de noms de l'élément.
    • Libellés d'ingestion : libellé appliqué aux événements de ce flux.
  8. Cliquez sur Suivant.
  9. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.