Raccogliere i log di Duo Telephony

Supportato in:

Google secops SIEM

Questo documento spiega come importare i log di Duo Telephony in Google Security Operations utilizzando Google Cloud Storage. Il parser estrae i campi dai log, li trasforma e li mappa al modello Unified Data Model (UDM). Gestisce vari formati di log di Duo, convertendo i timestamp, estraendo le informazioni utente, i dettagli di rete e i risultati di sicurezza e infine strutturando l'output nel formato UDM standardizzato.

Prima di iniziare

Assicurati di disporre dei seguenti prerequisiti:

Un'istanza Google SecOps
Un progetto GCP con l'API Cloud Storage abilitata
Autorizzazioni per creare e gestire bucket GCS
Autorizzazioni per gestire le policy IAM nei bucket GCS
Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
Accesso privilegiato al pannello amministrativo di Duo con il ruolo Proprietario

Raccogli i prerequisiti di Duo (credenziali API)

Accedi al pannello amministrativo di Duo come amministratore con il ruolo di proprietario.
Vai ad Applicazioni > Catalogo applicazioni.
Individua la voce relativa all'API Admin nel catalogo.
Fai clic su + Aggiungi per creare l'applicazione.
Copia e salva in una posizione sicura i seguenti dettagli:
- Chiave di integrazione
- Chiave segreta
- Nome host API (ad esempio, api-yyyyyyyy.duosecurity.com)
Nella sezione Autorizzazioni, seleziona solo la casella di controllo dell'autorizzazione Lettura telefono e deseleziona tutte le altre autorizzazioni.
Fai clic su Salva modifiche.

Verifica le autorizzazioni

Per verificare che l'applicazione API Admin disponga delle autorizzazioni richieste:

Accedi al pannello di amministrazione di Duo.
Vai ad Applicazioni > Proteggi un'applicazione.
Individua l'applicazione API Admin.
Fai clic sul nome dell'applicazione per visualizzarne i dettagli.
Nella sezione Autorizzazioni, verifica che sia selezionata solo l'opzione Lettura telefono.
Se sono selezionate altre autorizzazioni o se Leggi telefono non è selezionata, aggiorna le autorizzazioni e fai clic su Salva modifiche.

Testare l'accesso API

Verifica le tue credenziali prima di procedere con l'integrazione:

# Replace with your actual credentials
DUO_IKEY="your-integration-key"
DUO_SKEY="your-secret-key"
DUO_HOST="api-yyyyyyyy.duosecurity.com"

# Test API access (requires signing - use Duo's API test tool or Python script)
# Visit https://duo.com/docs/adminapi#testing to test your credentials

Creazione di un bucket Google Cloud Storage

Vai alla console Google Cloud.
Seleziona il tuo progetto o creane uno nuovo.
Nel menu di navigazione, vai a Cloud Storage > Bucket.
Fai clic su Crea bucket.

Fornisci i seguenti dettagli di configurazione:

Impostazione	Valore
Assegna un nome al bucket	Inserisci un nome univoco globale (ad esempio `duo-telephony-logs`).
Tipo di località	Scegli in base alle tue esigenze (regione singola, doppia regione, più regioni)
Località	Seleziona la posizione (ad esempio, `us-central1`).
Classe di archiviazione	Standard (consigliato per i log a cui si accede di frequente)
Controllo dell'accesso	Uniforme (consigliato)
Strumenti di protezione	(Facoltativo) Attiva il controllo delle versioni degli oggetti o la policy di conservazione

Fai clic su Crea.

Crea un service account per la funzione Cloud Run

La funzione Cloud Run richiede un service account con autorizzazioni di scrittura nel bucket GCS e di invocazione da parte di Pub/Sub.

Crea service account

Nella console Google Cloud, vai a IAM e amministrazione > Service Accounts.
Fai clic su Crea service account.
Fornisci i seguenti dettagli di configurazione:
- Nome del service account: inserisci duo-telephony-collector-sa.
- Descrizione service account: inserisci Service account for Cloud Run function to collect Duo Telephony logs.
Fai clic su Crea e continua.
Nella sezione Concedi a questo service account l'accesso al progetto, aggiungi i seguenti ruoli:
1. Fai clic su Seleziona un ruolo.
2. Cerca e seleziona Amministratore oggetti di archiviazione.
3. Fai clic su + Aggiungi un altro ruolo.
4. Cerca e seleziona Cloud Run Invoker.
5. Fai clic su + Aggiungi un altro ruolo.
6. Cerca e seleziona Invoker di Cloud Functions.
Fai clic su Continua.
Fai clic su Fine.

Questi ruoli sono necessari per:

Amministratore oggetti Storage: scrive i log nel bucket GCS e gestisce i file di stato
Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
Cloud Functions Invoker: consente la chiamata di funzioni

Concedi autorizzazioni IAM sul bucket GCS

Concedi al service account le autorizzazioni di scrittura sul bucket GCS:

Vai a Cloud Storage > Bucket.
Fai clic sul nome del bucket (ad esempio duo-telephony-logs).
Vai alla scheda Autorizzazioni.
Fai clic su Concedi l'accesso.
Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: inserisci l'email del service account (ad es. duo-telephony-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
- Assegna i ruoli: seleziona Storage Object Admin.
Fai clic su Salva.

Crea argomento Pub/Sub

Crea un argomento Pub/Sub a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

Nella console GCP, vai a Pub/Sub > Argomenti.
Fai clic su Crea argomento.
Fornisci i seguenti dettagli di configurazione:
- ID argomento: inserisci duo-telephony-trigger.
- Lascia le altre impostazioni sui valori predefiniti.
Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API Duo Telephony e li scrive in GCS.

Nella console GCP, vai a Cloud Run.
Fai clic su Crea servizio.
Seleziona Funzione (usa un editor in linea per creare una funzione).

Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

Impostazione	Valore
Nome servizio	`duo-telephony-logs-collector`
Regione	Seleziona la regione corrispondente al tuo bucket GCS (ad esempio `us-central1`)
Runtime	Seleziona Python 3.12 o versioni successive

Nella sezione Trigger (facoltativo):
1. Fai clic su + Aggiungi trigger.
2. Seleziona Cloud Pub/Sub.
3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento Pub/Sub (duo-telephony-trigger).
4. Fai clic su Salva.
Nella sezione Autenticazione:
1. Seleziona Richiedi autenticazione.
2. Controlla Identity and Access Management (IAM).
Nota: Pub/Sub gestisce automaticamente l'autenticazione quando richiama la funzione.
Scorri verso il basso ed espandi Container, networking, sicurezza.
Vai alla scheda Sicurezza:
- Service account: seleziona il service account (duo-telephony-collector-sa).

Vai alla scheda Container:

Fai clic su Variabili e secret.
Fai clic su + Aggiungi variabile per ogni variabile di ambiente:

Nome variabile	Valore di esempio
`GCS_BUCKET`	`duo-telephony-logs`
`GCS_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`

Scorri verso il basso nella scheda Variabili e secret fino a Richieste:
- Timeout richiesta: inserisci 600 secondi (10 minuti).
Vai alla scheda Impostazioni in Container:
- Nella sezione Risorse:
  - Memoria: seleziona 512 MiB o un valore superiore.
  - CPU: seleziona 1.
- Fai clic su Fine.
Scorri fino a Ambiente di esecuzione:
- Seleziona Predefinito (opzione consigliata).
Nella sezione Scalabilità della revisione:
- Numero minimo di istanze: inserisci 0.
- Numero massimo di istanze: inserisci 100 (o modifica in base al carico previsto).
Fai clic su Crea.
Attendi la creazione del servizio (1-2 minuti).
Dopo aver creato il servizio, si apre automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

Inserisci main nel campo Entry point (Punto di ingresso).

Nell'editor di codice incorporato, crea due file:

Primo file: main.py:

import functions_framework
from google.cloud import storage
import json
import os
import hmac
import hashlib
import base64
import urllib.parse
import urllib3
import email.utils
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Duo telephony logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'duo-telephony').rstrip('/')
    state_key = os.environ.get('STATE_KEY', 'duo-telephony/state.json')
    integration_key = os.environ.get('DUO_IKEY')
    secret_key = os.environ.get('DUO_SKEY')
    api_hostname = os.environ.get('DUO_API_HOST')

    if not all([bucket_name, integration_key, secret_key, api_hostname]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        # Load state
        state = load_state(bucket, 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:
            page_num = total_fetched + 1

            if next_offset:
                # Use offset for pagination
                params = {
                    'limit': '100',
                    'next_offset': next_offset
                }
            else:
                # Initial request with time range
                params = {
                    'mintime': str(mintime),
                    'maxtime': str(maxtime),
                    'limit': '100',
                    '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 (ISO 8601 format)
                    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

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

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

            blob = bucket.blob(blob_name)
            blob.upload_from_string(
                log_data,
                content_type='application/x-ndjson'
            )

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

        # Save state
        save_state(bucket, state_key, state)

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

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]:
                if attempt < max_retries - 1:
                    wait_time = (2 ** attempt) * 2
                    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
    headers = {
        'Authorization': f'Basic {auth}',
        'Date': now,
        'Host': host,
        'User-Agent': 'duo-telephony-gcs-ingestor/1.0'
    }

    try:
        response = http.request('GET', url, headers=headers)
        data = json.loads(response.data.decode('utf-8'))

        if data.get('stat') == 'OK':
            return data.get('response', {})
        else:
            raise Exception(f"API error: {data.get('message', 'Unknown error')}")
    except urllib3.exceptions.HTTPError as e:
        raise Exception(f"HTTP error: {str(e)}")

def load_state(bucket, key):
    """Load state from GCS."""
    try:
        blob = bucket.blob(key)
        if blob.exists():
            state_data = blob.download_as_text()
            return json.loads(state_data)
    except Exception as e:
        print(f'Warning: Could not load state: {str(e)}')
    return {}

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

Secondo file: requirements.txt::

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.
Attendi il completamento del deployment (2-3 minuti).

Nota :la configurazione del trigger Pub/Sub crea automaticamente le sottoscrizioni e le autorizzazioni necessarie. I limiti di frequenza dell'API Duo sono 5 richieste al secondo e 1000 richieste al minuto. La funzione utilizza un limite di 100 record per pagina (anziché il massimo di 1000) per migliorare il rendimento dell'API e la conformità ai consigli di Duo.

Crea job Cloud Scheduler

Cloud Scheduler pubblica messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

Nella console di GCP, vai a Cloud Scheduler.
Fai clic su Crea job.

Fornisci i seguenti dettagli di configurazione:

Impostazione	Valore
Nome	`duo-telephony-logs-1h`
Regione	Seleziona la stessa regione della funzione Cloud Run
Frequenza	`0 * * * *` (ogni ora, all'ora)
Fuso orario	Seleziona il fuso orario (UTC consigliato)
Tipo di target	Pub/Sub
Argomento	Seleziona l'argomento Pub/Sub (`duo-telephony-trigger`)
Corpo del messaggio	`{}` (oggetto JSON vuoto)

Fai clic su Crea.

Opzioni di frequenza di pianificazione

Scegli la frequenza in base al volume dei log e ai requisiti di latenza:

Frequenza	Espressione cron	Caso d'uso
Ogni 5 minuti	`/5 * * *`	Volume elevato, bassa latenza
Ogni 15 minuti	`/15 * * *`	Volume medio
Ogni ora	`0 * * * *`	Standard (consigliato)
Ogni 6 ore	`0 /6 * *`	Volume basso, elaborazione batch
Ogni giorno	`0 0 * * *`	Raccolta dei dati storici

Testa il job di pianificazione

Nella console Cloud Scheduler, trova il tuo job (duo-telephony-logs-1h).
Fai clic su Forza esecuzione per attivare manualmente.
Attendi qualche secondo e vai a Cloud Run > Servizi.
Fai clic sul nome della funzione (duo-telephony-logs-collector).
Fai clic sulla scheda Log.
Verifica che la funzione sia stata eseguita correttamente.
Vai a Cloud Storage > Bucket.
Fai clic sul nome del bucket (duo-telephony-logs).
Vai alla cartella del prefisso (duo-telephony/).
Verifica che sia stato creato un nuovo file .json con i log.

Se visualizzi errori nei log:

HTTP 401: controlla le credenziali API (DUO_IKEY, DUO_SKEY, DUO_API_HOST) nelle variabili di ambiente
HTTP 403: verifica che l'applicazione API Admin abbia l'autorizzazione Lettura telefonia attivata
HTTP 429: limitazione della frequenza: la funzione riproverà automaticamente con backoff esponenziale
Variabili di ambiente mancanti: verifica che tutte le variabili richieste siano impostate nella configurazione della funzione Cloud Run

Recuperare il service account Google SecOps

Google SecOps utilizza un service account univoco per leggere i dati dal tuo bucket GCS. Devi concedere a questo service account l'accesso al tuo bucket.

Recuperare l'email del service account

Vai a Impostazioni SIEM > Feed.
Fai clic su Aggiungi nuovo feed.
Fai clic su Configura un singolo feed.
Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Duo Telephony logs).
Seleziona Google Cloud Storage V2 come Tipo di origine.
Seleziona Log di telefonia Duo come Tipo di log.
Fai clic su Ottieni service account. Viene visualizzata un'email del service account univoca, ad esempio:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

Nota: ogni istanza di Google SecOps ha un service account univoco. Non utilizzare service account di altre documentazioni o esempi.

Concedi le autorizzazioni IAM al service account Google SecOps

Il service account Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel bucket GCS.

Vai a Cloud Storage > Bucket.
Fai clic sul nome del bucket (ad esempio duo-telephony-logs).
Vai alla scheda Autorizzazioni.
Fai clic su Concedi l'accesso.
Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: incolla l'email del service account Google SecOps.
- Assegna i ruoli: seleziona Visualizzatore oggetti Storage.
Fai clic su Salva.

Nota: se prevedi di utilizzare l'opzione di eliminazione "Elimina i file trasferiti" o "Elimina i file trasferiti e le directory vuote", assegna il ruolo Storage Object Admin anziché Storage Object Viewer.

Configurare un feed in Google SecOps per importare i log di Duo Telephony

Vai a Impostazioni SIEM > Feed.
Fai clic su Aggiungi nuovo feed.
Fai clic su Configura un singolo feed.
Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Duo Telephony logs).
Seleziona Google Cloud Storage V2 come Tipo di origine.
Seleziona Log di telefonia Duo come Tipo di log.
Fai clic su Avanti.
Specifica i valori per i seguenti parametri di input:
- URL del bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:
```
gs://duo-telephony-logs/duo-telephony/
```
  - Sostituisci:
    - duo-telephony-logs: il nome del bucket GCS.
    - duo-telephony: (Facoltativo) prefisso/percorso della cartella in cui vengono archiviati i log (lascia vuoto per la radice).
  - Esempi:
    - Bucket radice: gs://duo-telephony-logs/
    - Con prefisso: gs://duo-telephony-logs/duo-telephony/
  Nota: includi sempre la barra finale (/) alla fine dell'URI.
- Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:
  - Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
  - Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.
  - Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.
  Nota: se selezioni un'opzione di eliminazione, il service account deve disporre del ruolo Storage Object Admin anziché di Storage Object Viewer. Aggiorna le autorizzazioni IAM di conseguenza.
- Età massima del file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.
- Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.
- Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.
Fai clic su Avanti.
Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Tabella di mappatura UDM

Campo log	Mappatura UDM	Funzione logica
context	metadata.product_event_type	Mappato direttamente dal campo di contesto nel log non elaborato.
crediti	security_result.detection_fields.value	Mappato direttamente dal campo dei crediti nel log non elaborato, nidificato in un oggetto detection_fields con i crediti chiave corrispondenti.
eventtype	security_result.detection_fields.value	Mappato direttamente dal campo eventtype nel log non elaborato, nidificato in un oggetto detection_fields con l'eventtype chiave corrispondente.
host	principal.hostname	Mappato direttamente dal campo host nel log non elaborato se non è un indirizzo IP.
	security_result.action	Imposta un valore statico di "ALLOW" nel parser.
	security_result.detection_fields.value	Imposta un valore statico di "MECHANISM_UNSPECIFIED" nel parser.
	metadata.event_timestamp	Analizzato dal campo ts nel log non elaborato, che è una stringa di timestamp ISO 8601.
	metadata.event_type	Imposta "USER_UNCATEGORIZED" se nel log non elaborato sono presenti sia i campi contesto che host. Impostato su "STATUS_UPDATE" se è presente solo l'host. In caso contrario, imposta "GENERIC_EVENT".
	metadata.log_type	Estratto direttamente dal campo log_type del log non elaborato.
	metadata.product_name	Imposta un valore statico di "Telefonia" nel parser.
	metadata.vendor_name	Imposta un valore statico di "Duo" nel parser.
telefono	principal.user.phone_numbers	Mappato direttamente dal campo del telefono nel log non elaborato.
telefono	principal.user.userid	Mappato direttamente dal campo del telefono nel log non elaborato.
	security_result.severity	Impostato sul valore statico "INFORMATIONAL" nel parser.
	security_result.summary	Imposta un valore statico di "Duo Telephony" nel parser.
ts	metadata.event_timestamp	Analizzato dal campo ts nel log non elaborato, che è una stringa di timestamp ISO 8601.
tipo	security_result.summary	Mappato direttamente dal campo tipo nel log non elaborato.

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.