Cette page a été traduite par l'API Cloud Translation.

Collecter les journaux de Duo Téléphonie

Compatible avec :

Google SecOps SIEM

Ce document explique comment ingérer des journaux Duo Telephony dans Google Security Operations à l'aide d'Amazon S3. L'analyseur extrait les champs des journaux, les transforme et les mappe au modèle de données unifié (UDM). Il gère différents formats de journaux Duo, convertit les codes temporels, extrait les informations utilisateur, les détails du réseau et les résultats de sécurité, et structure enfin la sortie au format UDM standardisé.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

Instance Google SecOps.
Accès privilégié au panneau d'administration Duo avec le rôle Propriétaire.
Accès privilégié à AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

Collecter les prérequis Duo (identifiants API)

Connectez-vous au panneau d'administration Duo en tant qu'administrateur disposant du rôle Propriétaire.
Accédez à Applications > Catalogue d'applications.
Recherchez l'entrée API Admin dans le catalogue.
Cliquez sur + Ajouter pour créer l'application.
Copiez et enregistrez les informations suivantes dans un emplacement sécurisé :
- Clé d'intégration
- Clé secrète
- Nom d'hôte de l'API (par exemple, api-yyyyyyyy.duosecurity.com)
Dans la section Autorisations, décochez toutes les options d'autorisation, à l'exception de Accorder l'accès en lecture aux journaux.
Cliquez sur Enregistrer les modifications.

Configurer un bucket AWS S3 et IAM pour Google SecOps

Créez un bucket Amazon S3 en suivant ce guide de l'utilisateur : Créer un bucket.
Enregistrez le Nom et la Région du bucket pour référence ultérieure (par exemple, duo-telephony-logs).
Créez un utilisateur en suivant ce guide de l'utilisateur : Créer un utilisateur IAM.
Sélectionnez l'utilisateur créé.
Sélectionnez l'onglet Informations d'identification de sécurité.
Cliquez sur Créer une clé d'accès dans la section Clés d'accès.
Sélectionnez Service tiers comme Cas d'utilisation.
Cliquez sur Suivant.
Facultatif : Ajoutez un tag de description.
Cliquez sur Créer une clé d'accès.
Cliquez sur Télécharger le fichier CSV pour enregistrer la clé d'accès et la clé d'accès secrète pour référence ultérieure.
Cliquez sur OK.
Sélectionnez l'onglet Autorisations.
Cliquez sur Ajouter des autorisations dans la section Règles relatives aux autorisations.
Sélectionnez Ajouter des autorisations.
Sélectionnez Joindre directement des règles.
Recherchez la règle AmazonS3FullAccess.
Sélectionnez la règle.
Cliquez sur Suivant.
Cliquez sur Ajouter des autorisations.

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

Dans la console AWS, accédez à IAM > Stratégies.
Cliquez sur Créer une règle > onglet JSON.
Copiez et collez le règlement suivant.

JSON de la règle (remplacez duo-telephony-logs si vous avez saisi un autre nom de bucket) :

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

Cliquez sur Suivant > Créer une règle.
Accédez à IAM > Rôles > Créer un rôle > Service AWS > Lambda.
Associez la règle que vous venez de créer.
Nommez le rôle duo-telephony-lambda-role, puis cliquez sur Créer un rôle.

Créer la fonction Lambda

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

Paramètre Valeur

Nom duo-telephony-logs-collector

Durée d'exécution Python 3.13

Architecture x86_64

Rôle d'exécution duo-telephony-lambda-role

Paramètre	Valeur
Nom	`duo-telephony-logs-collector`
Durée d'exécution	Python 3.13
Architecture	x86_64
Rôle d'exécution	`duo-telephony-lambda-role`

Une fois la fonction créée, ouvrez l'onglet Code, supprimez le stub et collez le code suivant (duo-telephony-logs-collector.py).

import json
import boto3
import os
import hmac
import hashlib
import base64
import urllib.parse
import urllib.request
import email.utils
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional
from botocore.exceptions import ClientError

s3 = boto3.client('s3')

def lambda_handler(event, context):
    """
    Lambda function to fetch Duo telephony logs and store them in S3.
    """
    try:
        # Get configuration from environment variables
        bucket_name = os.environ['S3_BUCKET']
        s3_prefix = os.environ['S3_PREFIX'].rstrip('/')
        state_key = os.environ['STATE_KEY']
        integration_key = os.environ['DUO_IKEY']
        secret_key = os.environ['DUO_SKEY']
        api_hostname = os.environ['DUO_API_HOST']

        # Load state
        state = load_state(bucket_name, state_key)

        # Calculate time range
        now = datetime.now(timezone.utc)
        if state.get('last_offset'):
            # Continue from last offset
            next_offset = state['last_offset']
            logs = []
            has_more = True
        else:
            # Start from last timestamp or 24 hours ago
            mintime = state.get('last_timestamp_ms', 
                              int((now - timedelta(hours=24)).timestamp() * 1000))
            # Apply 2-minute delay as recommended by Duo
            maxtime = int((now - timedelta(minutes=2)).timestamp() * 1000)
            next_offset = None
            logs = []
            has_more = True

        # Fetch logs with pagination
        total_fetched = 0
        max_iterations = int(os.environ.get('MAX_ITERATIONS', '10'))

        while has_more and total_fetched < max_iterations:
            if next_offset:
                # Use offset for pagination
                params = {
                    'limit': '1000',
                    'next_offset': next_offset
                }
            else:
                # Initial request with time range
                params = {
                    'mintime': str(mintime),
                    'maxtime': str(maxtime),
                    'limit': '1000',
                    'sort': 'ts:asc'
                }

            # Make API request with retry logic
            response = duo_api_call_with_retry(
                'GET',
                api_hostname,
                '/admin/v2/logs/telephony',
                params,
                integration_key,
                secret_key
            )

            if 'items' in response:
                logs.extend(response['items'])
                total_fetched += 1

                # Check for more data
                if 'metadata' in response and 'next_offset' in response['metadata']:
                    next_offset = response['metadata']['next_offset']
                    state['last_offset'] = next_offset
                else:
                    has_more = False
                    state['last_offset'] = None

                    # Update timestamp for next run
                    if logs:
                        # Get the latest timestamp from logs
                        latest_ts = max([log.get('ts', '') for log in logs])
                        if latest_ts:
                            # Convert ISO timestamp to milliseconds
                            dt = datetime.fromisoformat(latest_ts.replace('Z', '+00:00'))
                            state['last_timestamp_ms'] = int(dt.timestamp() * 1000) + 1
            else:
                has_more = False

        # Save logs to S3 if any were fetched
        if logs:
            timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
            key = f"{s3_prefix}/telephony_{timestamp}.json"

            # Format logs as newline-delimited JSON
            log_data = '\n'.join(json.dumps(log) for log in logs)

            s3.put_object(
                Bucket=bucket_name,
                Key=key,
                Body=log_data.encode('utf-8'),
                ContentType='application/x-ndjson'
            )

            print(f"Saved {len(logs)} telephony logs to s3://{bucket_name}/{key}")
        else:
            print("No new telephony logs found")

        # Save state
        save_state(bucket_name, state_key, state)

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Successfully processed {len(logs)} telephony logs',
                'logs_count': len(logs)
            })
        }

    except Exception as e:
        print(f"Error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def duo_api_call_with_retry(method: str, host: str, path: str, params: Dict[str, str], 
                            ikey: str, skey: str, max_retries: int = 3) -> Dict[str, Any]:
    """
    Make an authenticated API call to Duo Admin API with retry logic.
    """
    for attempt in range(max_retries):
        try:
            return duo_api_call(method, host, path, params, ikey, skey)
        except Exception as e:
            if '429' in str(e) or '5' in str(e)[:1]:  # Rate limit or server error
                if attempt < max_retries - 1:
                    wait_time = (2 ** attempt) * 2  # Exponential backoff
                    print(f"Retrying after {wait_time} seconds...")
                    import time
                    time.sleep(wait_time)
                    continue
            raise

def duo_api_call(method: str, host: str, path: str, params: Dict[str, str], 
                ikey: str, skey: str) -> Dict[str, Any]:
    """
    Make an authenticated API call to Duo Admin API.
    """
    # Create canonical string for signing using RFC 2822 date format
    now = email.utils.formatdate()
    canon = [now, method.upper(), host.lower(), path]

    # Add parameters
    args = []
    for key in sorted(params.keys()):
        val = params[key]
        args.append(f"{urllib.parse.quote(key, '~')}={urllib.parse.quote(val, '~')}")
    canon.append('&'.join(args))
    canon_str = '\n'.join(canon)

    # Sign the request
    sig = hmac.new(
        skey.encode('utf-8'),
        canon_str.encode('utf-8'),
        hashlib.sha1
    ).hexdigest()

    # Create authorization header
    auth = base64.b64encode(f"{ikey}:{sig}".encode('utf-8')).decode('utf-8')

    # Build URL
    url = f"https://{host}{path}"
    if params:
        url += '?' + '&'.join(args)

    # Make request
    req = urllib.request.Request(url)
    req.add_header('Authorization', f'Basic {auth}')
    req.add_header('Date', now)
    req.add_header('Host', host)
    req.add_header('User-Agent', 'duo-telephony-s3-ingestor/1.0')

    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            data = json.loads(response.read().decode('utf-8'))
            if data.get('stat') == 'OK':
                return data.get('response', {})
            else:
                raise Exception(f"API error: {data.get('message', 'Unknown error')}")
    except urllib.error.HTTPError as e:
        error_body = e.read().decode('utf-8')
        raise Exception(f"HTTP error {e.code}: {error_body}")

def load_state(bucket: str, key: str) -> Dict[str, Any]:
    """Load state from S3."""
    try:
        response = s3.get_object(Bucket=bucket, Key=key)
        return json.loads(response['Body'].read().decode('utf-8'))
    except ClientError as e:
        if e.response.get('Error', {}).get('Code') in ('NoSuchKey', '404'):
            return {}
        print(f"Error loading state: {e}")
        return {}
    except Exception as e:
        print(f"Error loading state: {e}")
        return {}

def save_state(bucket: str, key: str, state: Dict[str, Any]):
    """Save state to S3."""
    try:
        s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=json.dumps(state).encode('utf-8'),
            ContentType='application/json'
        )
    except Exception as e:
        print(f"Error saving state: {e}")

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

Saisissez les variables d'environnement fournies dans le tableau suivant, en remplaçant les exemples de valeurs par les vôtres.

Variables d'environnement

Clé	Exemple de valeur
`S3_BUCKET`	`duo-telephony-logs`
`S3_PREFIX`	`duo-telephony/`
`STATE_KEY`	`duo-telephony/state.json`
`DUO_IKEY`	`<your-integration-key>`
`DUO_SKEY`	`<your-secret-key>`
`DUO_API_HOST`	`api-yyyyyyyy.duosecurity.com`
`MAX_ITERATIONS`	`10`

Une fois la fonction créée, restez sur sa page (ou ouvrez Lambda> Fonctions> duo-telephony-logs-collector).
Accédez à l'onglet Configuration.
Dans le panneau Configuration générale, cliquez sur Modifier.
Définissez le délai avant expiration sur 5 minutes (300 secondes), puis cliquez sur Enregistrer.

Remarque : Le délai d'expiration est une limite supérieure stricte. Cela signifie que si la fonction est toujours en cours d'exécution lorsque le minuteur atteint cinq minutes, AWS Lambda l'arrête. Toute exécution plus courte se termine normalement. Vous n'êtes facturé que pour la durée d'exécution réelle.

Créer une programmation EventBridge

Accédez à Amazon EventBridge> Scheduler> Create schedule.
Fournissez les informations de configuration suivantes :
- Planning récurrent : Tarif (1 hour).
- Cible : votre fonction Lambda duo-telephony-logs-collector.
- Nom : duo-telephony-logs-1h.
Cliquez sur Créer la programmation.

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

Accédez à Console AWS > IAM > Utilisateurs.
Cliquez sur Add users (Ajouter des utilisateurs).
Fournissez les informations de configuration suivantes :
- Utilisateur : saisissez secops-reader.
- Type d'accès : sélectionnez Clé d'accès – Accès programmatique.
Cliquez sur Créer un utilisateur.
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.

JSON :

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::duo-telephony-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::duo-telephony-logs"
    }
  ]
}

Nom = secops-reader-policy.
Cliquez sur Créer une règle> recherchez/sélectionnez > Suivant> Ajouter des autorisations.
Créez une clé d'accès pour secops-reader : Identifiants de sécurité > Clés d'accès.
Cliquez sur Créer une clé d'accès.
Téléchargez le fichier .CSV. (Vous collerez ces valeurs dans le flux.)

Configurer un flux dans Google SecOps pour ingérer les journaux de téléphonie Duo

Accédez à Paramètres SIEM> Flux.
Cliquez sur + Ajouter un flux.
Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Duo Telephony logs).
Sélectionnez Amazon S3 V2 comme type de source.
Sélectionnez Journaux de téléphonie Duo comme Type de journal.
Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
- URI S3 : s3://duo-telephony-logs/duo-telephony/
- 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. La valeur par défaut est de 180 jours.
- 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.
Cliquez sur Suivant.
Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Table de mappage UDM

Champ du journal	Mappage UDM	Logique
`context`	`metadata.product_event_type`	Directement mappé à partir du champ `context` dans le journal brut.
`credits`	`security_result.detection_fields.value`	Directement mappé à partir du champ `credits` du journal brut, imbriqué sous un objet `detection_fields` avec la clé `credits` correspondante.
`eventtype`	`security_result.detection_fields.value`	Directement mappé à partir du champ `eventtype` du journal brut, imbriqué sous un objet `detection_fields` avec la clé `eventtype` correspondante.
`host`	`principal.hostname`	Directement mappé à partir du champ `host` dans le journal brut s'il ne s'agit pas d'une adresse IP. Définissez une valeur statique "ALLOW" dans l'analyseur. Définissez une valeur statique "MECHANISM_UNSPECIFIED" dans l'analyseur. Analysé à partir du champ `timestamp` du journal brut, qui représente les secondes depuis l'epoch. Définissez la valeur sur "USER_UNCATEGORIZED" si les champs `context` et `host` sont présents dans le journal brut. Définissez sur "STATUS_UPDATE" si seul `host` est présent. Sinon, définissez-le sur "GENERIC_EVENT". Directement extraite du champ `log_type` du journal brut. Définissez une valeur statique "Telephony" dans l'analyseur. Définissez une valeur statique "Duo" dans l'analyseur.
`phone`	`principal.user.phone_numbers`	Directement mappé à partir du champ `phone` dans le journal brut.
`phone`	`principal.user.userid`	Directement mappé à partir du champ `phone` dans le journal brut. Définissez une valeur statique "INFORMATIONAL" dans l'analyseur. Définissez une valeur statique "Duo Telephony" dans l'analyseur.
`timestamp`	`metadata.event_timestamp`	Analysé à partir du champ `timestamp` du journal brut, qui représente les secondes depuis l'epoch.
`type`	`security_result.summary`	Directement mappé à partir du champ `type` dans le journal brut.

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