Collecter les journaux d'authentification Duo

Ce document explique comment ingérer les journaux d'authentification Duo dans Google Security Operations à l'aide d'Amazon S3. L'analyseur extrait les journaux des messages au format JSON. Il transforme les données brutes des journaux en modèle de données unifié (UDM, Unified Data Model), en mappant des champs tels que les détails de l'utilisateur, de l'appareil, de l'application, de l'emplacement et de l'authentification. Il gère également divers facteurs et résultats d'authentification pour catégoriser les événements de sécurité. L'analyseur effectue également le nettoyage des données, la conversion des types et la gestion des exceptions pour garantir la qualité et la cohérence des données.

Avant de commencer

• Instance Google SecOps
• Accès privilégié au locataire Duo (application de l'API Admin)
• Accès privilégié à AWS (S3, IAM, Lambda, EventBridge)

Configurer l'application Duo Admin API

1. Connectez-vous au panneau d'administration Duo.
2. Accédez à Applications > Protéger une application.
3. Ajoutez l'application API Admin.
4. Copiez et enregistrez les valeurs suivantes dans un emplacement sécurisé :
   • Clé d'intégration (ikey)
   • Clé secrète (skey)
   • Nom d'hôte de l'API (par exemple, api-XXXXXXXX.duosecurity.com)
5. Dans Autorisations, activez Accorder l'accès en lecture aux journaux (pour lire les journaux d'authentification).
6. Enregistrez l'application.

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, duo-auth-logs).
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. Accédez à la console AWS> IAM> Policies> Create policy> onglet JSON.

2. Saisissez la règle suivante :

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutDuoAuthObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::duo-auth-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::duo-auth-logs/duo/auth/state.json"
       }
     ]
   }
   
   

   • Remplacez duo-auth-logs 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 règle que vous venez de créer.

6. Nommez le rôle WriteDuoAuthToS3Role, 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	duo_auth_to_s3
   Durée d'exécution	Python 3.13
   Architecture	x86_64
   Rôle d'exécution	WriteDuoAuthToS3Role

4. Une fois la fonction créée, ouvrez l'onglet Code, supprimez le stub et saisissez le code suivant (duo_auth_to_s3.py) :

   #!/usr/bin/env python3
   # Lambda: Pull Duo Admin API v2 Authentication Logs to S3 (raw JSON pages)
   # Notes:
   # - Duo v2 requires mintime/maxtime in *milliseconds* (13-digit epoch).
   # - Pagination via metadata.next_offset ("<millis>,<txid>").
   # - We save state (mintime_ms) in ms to resume next run without gaps.
   
   import os, json, time, hmac, hashlib, base64, email.utils, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError, URLError
   import boto3
   
   DUO_IKEY = os.environ["DUO_IKEY"]
   DUO_SKEY = os.environ["DUO_SKEY"]
   DUO_API_HOSTNAME = os.environ["DUO_API_HOSTNAME"].strip()
   S3_BUCKET = os.environ["S3_BUCKET"]
   S3_PREFIX = os.environ.get("S3_PREFIX", "duo/auth/").strip("/")
   STATE_KEY = os.environ.get("STATE_KEY", "duo/auth/state.json")
   LIMIT = min(int(os.environ.get("LIMIT", "500")), 1000)  # default 100, max 1000
   
   s3 = boto3.client("s3")
   
   def _canon_params(params: dict) -> str:
       parts = []
       for k in sorted(params.keys()):
           v = params[k]
           if v is None:
               continue
           parts.append(f"{urllib.parse.quote(str(k), '~')}={urllib.parse.quote(str(v), '~')}")
       return "&".join(parts)
   
   def _sign(method: str, host: str, path: str, params: dict) -> dict:
       now = email.utils.formatdate()
       canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)])
       sig = hmac.new(DUO_SKEY.encode("utf-8"), canon.encode("utf-8"), hashlib.sha1).hexdigest()
       auth = base64.b64encode(f"{DUO_IKEY}:{sig}".encode()).decode()
       return {"Date": now, "Authorization": f"Basic {auth}"}
   
   def _http(method: str, path: str, params: dict, timeout: int = 60, max_retries: int = 5) -> dict:
       host = DUO_API_HOSTNAME
       assert host.startswith("api-") and host.endswith(".duosecurity.com"), \
           "DUO_API_HOSTNAME must be like api-XXXXXXXX.duosecurity.com"
       qs = _canon_params(params)
       url = f"https://{host}{path}" + (f"?{qs}" if qs else "")
   
       attempt, backoff = 0, 1.0
       while True:
           req = Request(url, method=method.upper())
           req.add_header("Accept", "application/json")
           for k, v in _sign(method, host, path, params).items():
               req.add_header(k, v)
           try:
               with urlopen(req, timeout=timeout) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
           except URLError:
               if attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
   
   def _read_state_ms() -> int | None:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           val = json.loads(obj["Body"].read()).get("mintime")
           if val is None:
               return None
           # Backward safety: if seconds were stored, convert to ms
           return int(val) * 1000 if len(str(int(val))) <= 10 else int(val)
       except Exception:
           return None
   
   def _write_state_ms(mintime_ms: int):
       body = json.dumps({"mintime": int(mintime_ms)}).encode("utf-8")
       s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")
   
   def _write_page(payload: dict, when_epoch_s: int, page: int) -> str:
       key = f"{S3_PREFIX}/{time.strftime('%Y/%m/%d', time.gmtime(when_epoch_s))}/duo-auth-{page:05d}.json"
       s3.put_object(
           Bucket=S3_BUCKET,
           Key=key,
           Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
       return key
   
   def fetch_and_store():
       now_s = int(time.time())
       # Duo recommends a ~2-minute delay buffer; use maxtime = now - 120 seconds (in ms)
       maxtime_ms = (now_s - 120) * 1000
       mintime_ms = _read_state_ms() or (maxtime_ms - 3600 * 1000)  # 1 hour on first run
   
       page = 0
       total = 0
       next_offset = None
   
       while True:
           params = {"mintime": mintime_ms, "maxtime": maxtime_ms, "limit": LIMIT}
           if next_offset:
               params["next_offset"] = next_offset
   
           data = _http("GET", "/admin/v2/logs/authentication", params)
           _write_page(data, maxtime_ms // 1000, page)
           page += 1
   
           resp = data.get("response")
           items = resp if isinstance(resp, list) else []
           total += len(items)
   
           meta = data.get("metadata") or {}
           next_offset = meta.get("next_offset")
           if not next_offset:
               break
   
       # Advance window to maxtime_ms for next run
       _write_state_ms(maxtime_ms)
       return {"ok": True, "pages": page, "events": total, "next_mintime_ms": maxtime_ms}
   
   def lambda_handler(event=None, context=None):
       return fetch_and_store()
   
   if __name__ == "__main__":
       print(lambda_handler())
   
   

5. Accédez à Configuration> Variables d'environnement> Modifier> Ajouter une variable d'environnement.

6. Saisissez les variables d'environnement fournies ci-dessous, en les remplaçant par vos valeurs :

   Clé	Exemple
   S3_BUCKET	duo-auth-logs
   S3_PREFIX	duo/auth/
   STATE_KEY	duo/auth/state.json
   DUO_IKEY	DIXYZ...
   DUO_SKEY	****************
   DUO_API_HOSTNAME	api-XXXXXXXX.duosecurity.com
   LIMIT	500

7. Une fois la fonction créée, restez sur sa page (ou ouvrez Lambda > Fonctions > votre‑fonction).

8. Accédez à l'onglet Configuration.

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

10. 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 (Créer une programmation).
2. Fournissez les informations de configuration suivantes :
   • Planning récurrent : Tarif (1 hour).
   • Cible : votre fonction Lambda.
   • Nom : duo-auth-1h.
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, puis cliquez sur Ajouter des utilisateurs.
2. Fournissez les informations de configuration suivantes :
   • Utilisateur : saisissez un nom unique (par exemple, secops-reader).
   • Type d'accès : sélectionnez Clé d'accès – Accès programmatique.
   • Cliquez sur Créer un utilisateur.
3. Associez une règle de lecture minimale (personnalisée) : Utilisateurs> sélectionnez secops-reader > Autorisations> Ajouter des autorisations> Associer des règles directement> Créer une règle

4. Dans l'éditeur JSON, saisissez la stratégie suivante :

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::<your-bucket>/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::<your-bucket>"
       }
     ]
   }
   

5. Définissez le nom sur secops-reader-policy.

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

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

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

Configurer un flux dans Google SecOps pour ingérer les journaux d'authentification Duo

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, Duo Authentication Logs).
4. Sélectionnez Amazon S3 V2 comme type de source.
5. Sélectionnez Duo Auth comme Type de journal.
6. Cliquez sur Suivant.
7. Spécifiez les valeurs des paramètres d'entrée suivants :
   • URI S3 : s3://duo-auth-logs/duo/auth/
   • Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
   • Âge maximal des fichiers : 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.

Table de mappage UDM

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