Raccogliere i log di contesto dell'entità Duo

Versione del parser: 2.0

Supportato in:

Questo documento spiega come importare i dati di contesto delle entità Duo in Google Security Operations utilizzando Google Cloud Storage. Il parser trasforma i log JSON in un modello dei dati UDM (Unified Data Model) estraendo prima i campi dal JSON non elaborato e poi mappandoli agli attributi UDM. Gestisce vari scenari di dati, tra cui informazioni su utenti e asset, dettagli del software ed etichette di sicurezza, garantendo una rappresentazione completa all'interno dello schema UDM.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Accesso con privilegi al tenant Duo (applicazione API Admin con privilegi amministrativi sufficienti per gestire le applicazioni)
  • Un progetto GCP con l'API Storage Cloud 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

Configurare l'applicazione API Admin di Duo

  1. Accedi al pannello di amministrazione di Duo.
  2. Vai ad Applicazioni > Proteggi un'applicazione.
  3. Cerca API Admin e fai clic su Proteggi.
  4. Registra i seguenti valori:
    • Chiave di integrazione (ikey)
    • Chiave segreta (skey)
    • Nome host API (ad esempio, api-XXXXXXXX.duosecurity.com)
  5. In Autorizzazioni, attiva Concedi risorsa - Lettura (per leggere utenti, gruppi, telefoni, endpoint, token e credenziali WebAuthn).

  6. Fai clic su Salva.

Crea un bucket Google Cloud Storage

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

  5. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio duo-context).
    Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
    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

  6. Fai clic su Crea.

  7. Salva il nome del bucket e la regione per riferimento futuro.

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni per scrivere nel bucket GCS e per essere richiamato da Pub/Sub.

Crea service account

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service account.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome del service account: inserisci duo-entity-context-sa.
    • Descrizione service account: inserisci Service account for Cloud Run function to collect Duo entity context data.
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
    1. Fai clic su Seleziona un ruolo.
    2. Cerca e seleziona Storage Object Admin.
    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.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono necessari per:

  • Storage Object Admin: scrivi i log nel bucket GCS
  • 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 account di servizio le autorizzazioni di scrittura sul bucket GCS:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'email del account di servizio (ad esempio, duo-entity-context-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. 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à.

  1. Nella console GCP, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci duo-entity-context-trigger.
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i dati di contesto delle entità

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i dati di contesto delle entità dall'API Duo Admin e li scrive in GCS.

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

  4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio duo-entity-context-collector
    Regione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive

  5. 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-entity-context-trigger).
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Controlla Identity and Access Management (IAM).

  7. Scorri verso il basso ed espandi Container, networking, sicurezza.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio (duo-entity-context-sa).

  9. Vai alla scheda Container:

    1. Fai clic su Variabili e secret.
    2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
    Nome variabile Valore di esempio
    GCS_BUCKET duo-context
    GCS_PREFIX duo/context/
    DUO_IKEY DIXYZ...
    DUO_SKEY ****************
    DUO_API_HOSTNAME api-XXXXXXXX.duosecurity.com
    LIMIT 100
    RESOURCES users,groups,phones,endpoints,tokens,webauthncredentials

  10. Nella sezione Variabili e secret, scorri fino a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti).

  11. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
    • Fai clic su Fine.

  12. Scorri fino a Ambiente di esecuzione:

    • Seleziona Predefinito (opzione consigliata).

  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0.
    • Numero massimo di istanze: inserisci 100 (o modifica in base al carico previsto).

  14. Fai clic su Crea.

  15. Attendi la creazione del servizio (1-2 minuti).

  16. Dopo aver creato il servizio, si apre automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

  1. Inserisci main in Entry point della funzione

  2. 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 time
import hmac
import hashlib
import base64
import email.utils
import urllib.parse
from urllib.request import Request, urlopen

# Environment variables
DUO_IKEY = os.environ["DUO_IKEY"]
DUO_SKEY = os.environ["DUO_SKEY"]
DUO_API_HOSTNAME = os.environ["DUO_API_HOSTNAME"].strip()
GCS_BUCKET = os.environ["GCS_BUCKET"]
GCS_PREFIX = os.environ.get("GCS_PREFIX", "duo/context/")

# Default resources can be adjusted via ENV
RESOURCES = [r.strip() for r in os.environ.get("RESOURCES", "users,groups,phones,endpoints,tokens,webauthncredentials,desktop_authenticators").split(",") if r.strip()]

# Duo paging: default 100; max varies by endpoint
LIMIT = int(os.environ.get("LIMIT", "100"))

# Initialize Storage client
storage_client = storage.Client()

def _canon_params(params: dict) -> str:
    """RFC 3986 encoding with '~' unescaped, keys sorted lexicographically."""
    if not params:
        return ""
    parts = []
    for k in sorted(params.keys()):
        v = params[k]
        if v is None:
            continue
        ks = urllib.parse.quote(str(k), safe="~")
        vs = urllib.parse.quote(str(v), safe="~")
        parts.append(f"{ks}={vs}")
    return "&".join(parts)

def _sign(method: str, host: str, path: str, params: dict) -> dict:
    """Construct Duo Admin API Authorization + Date headers (HMAC-SHA1)."""
    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("utf-8")).decode("utf-8")
    return {
        "Date": now,
        "Authorization": f"Basic {auth}"
    }

def _call(method: str, path: str, params: dict) -> dict:
    host = DUO_API_HOSTNAME
    assert host.startswith("api-") and host.endswith(".duosecurity.com"), \
        "DUO_API_HOSTNAME must be e.g. api-XXXXXXXX.duosecurity.com"

    qs = _canon_params(params)
    url = f"https://{host}{path}" + (f"?{qs}" if method.upper() == "GET" and qs else "")

    req = Request(url, method=method.upper())
    for k, v in _sign(method, host, path, params).items():
        req.add_header(k, v)

    with urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode("utf-8"))

def _write_json(obj: dict, when: float, resource: str, page: int) -> str:
    bucket = storage_client.bucket(GCS_BUCKET)
    prefix = GCS_PREFIX.strip("/") + "/" if GCS_PREFIX else ""
    key = f"{prefix}{time.strftime('%Y/%m/%d', time.gmtime(when))}/duo-{resource}-{page:05d}.json"

    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(obj, separators=(",", ":")),
        content_type="application/json"
    )
    return key

def _fetch_resource(resource: str) -> dict:
    """Fetch all pages for a list endpoint using limit/offset + metadata.next_offset."""
    path = f"/admin/v1/{resource}"
    offset = 0
    page = 0
    now = time.time()
    total_items = 0

    while True:
        params = {"limit": LIMIT, "offset": offset}
        data = _call("GET", path, params)
        _write_json(data, now, resource, page)
        page += 1

        resp = data.get("response")
        # most endpoints return a list; if not a list, count as 1 object page
        if isinstance(resp, list):
            total_items += len(resp)
        elif resp is not None:
            total_items += 1

        meta = data.get("metadata") or {}
        next_offset = meta.get("next_offset")
        if next_offset is None:
            break

        # Duo returns next_offset as int
        try:
            offset = int(next_offset)
        except Exception:
            break

    return {
        "resource": resource,
        "pages": page,
        "objects": total_items
    }

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

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """
    results = []
    for res in RESOURCES:
        print(f"Fetching resource: {res}")
        result = _fetch_resource(res)
        results.append(result)
        print(f"Completed {res}: {result['pages']} pages, {result['objects']} objects")

    print(f"All resources fetched successfully: {results}")
    • Secondo file: requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*

  3. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

  4. Attendi il completamento del deployment (2-3 minuti).

Crea job Cloud Scheduler

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

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

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome duo-entity-context-hourly
    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-entity-context-trigger)
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Opzioni di frequenza di pianificazione

  • Scegli la frequenza in base ai requisiti di aggiornamento dei dati:

    Frequenza Espressione cron Caso d'uso
    Ogni ora 0 * * * * Standard (consigliato)
    Ogni 2 ore 0 */2 * * * Moderatamente attuali
    Ogni 6 ore 0 */6 * * * Aggiornamenti a bassa frequenza
    Ogni giorno 0 0 * * * Aggiornamenti minimi

Testa il job di pianificazione

  1. Nella console Cloud Scheduler, trova il tuo job (duo-entity-context-hourly).
  2. Fai clic su Forza esecuzione per attivare manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > duo-entity-context-collector > Log.
  4. Verifica che la funzione sia stata eseguita correttamente.
  5. Controlla il bucket GCS per verificare che i dati di contesto dell'entità siano stati scritti.

Recuperare il account di servizio Google SecOps

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

Recupera l'email del account di servizio

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Duo Entity Context).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Duo Entity context data (Dati contestuali dell'entità Duo) come Tipo di log.

  7. Fai clic su Ottieni service account. Viene visualizzato un indirizzo email del account di servizio univoco, ad esempio:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

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

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'email del account di servizio Google SecOps.
    • Assegna ruoli: seleziona Visualizzatore oggetti Storage.

  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i dati del contesto delle entità di Duo

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Duo Entity Context).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Duo Entity context data (Dati contestuali dell'entità Duo) come Tipo di log.
  7. Fai clic su Avanti.

  8. 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-context/duo/context/

      • Sostituisci:

        • duo-context: il nome del bucket GCS.
        • duo/context/: il percorso del prefisso/della cartella in cui sono archiviati i log (deve corrispondere alla variabile di ambiente GCS_PREFIX).

    • 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 i 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.

    • 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.

  9. Fai clic su Avanti.

  10. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
attivato entity.asset.deployment_status Se "activated" è false, imposta "DECOMISSIONED", altrimenti "ACTIVE".
browsers.browser_family entity.asset.software.name Estratto dall'array "browser" nel log non elaborato.
browsers.browser_version entity.asset.software.version Estratto dall'array "browser" nel log non elaborato.
device_name entity.asset.hostname Mappato direttamente dal log non elaborato.
disk_encryption_status entity.asset.attribute.labels.key: "disk_encryption_status", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
email entity.user.email_addresses Mappato direttamente dal log non elaborato se contiene "@", altrimenti utilizza "username" o "username1" se contengono "@".
criptato entity.asset.attribute.labels.key: "Encrypted", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
epkey entity.asset.product_object_id Utilizzato come "product_object_id" se presente, altrimenti utilizza "phone_id" o "token_id".
impronta entity.asset.attribute.labels.key: "Finger Print", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
firewall_status entity.asset.attribute.labels.key: "firewall_status", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
hardware_uuid entity.asset.asset_id Utilizzato come "asset_id" se presente, altrimenti utilizza "user_id".
last_seen entity.asset.last_discover_time Analizzato come timestamp ISO8601 e mappato.
modello entity.asset.hardware.model Mappato direttamente dal log non elaborato.
numero entity.user.phone_numbers Mappato direttamente dal log non elaborato.
os_family entity.asset.platform_software.platform Mappato su "WINDOWS", "LINUX" o "MAC" in base al valore, senza distinzione tra maiuscole e minuscole.
os_version entity.asset.platform_software.platform_version Mappato direttamente dal log non elaborato.
password_status entity.asset.attribute.labels.key: "password_status", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
phone_id entity.asset.product_object_id Utilizzato come "product_object_id" se "epkey" non è presente, altrimenti utilizza "token_id".
security_agents.security_agent entity.asset.software.name Estratto dall'array "security_agents" nel log non elaborato.
security_agents.version entity.asset.software.version Estratto dall'array "security_agents" nel log non elaborato.
timestamp entity.metadata.collected_timestamp Compila il campo "collected_timestamp" all'interno dell'oggetto "metadata".
token_id entity.asset.product_object_id Utilizzato come "product_object_id" se "epkey" e "phone_id" non sono presenti.
trusted_endpoint entity.asset.attribute.labels.key: "trusted_endpoint", entity.asset.attribute.labels.value Mappato direttamente dal log non elaborato, convertito in minuscolo.
tipo entity.asset.type Se il "tipo" del log grezzo contiene "mobile" (senza distinzione tra maiuscole e minuscole), impostalo su "MOBILE", altrimenti su "LAPTOP".
user_id entity.asset.asset_id Utilizzato come "asset_id" se "hardware_uuid" non è presente.
users.email entity.user.email_addresses Utilizzato come "email_addresses" se è il primo utente nell'array "users" e contiene "@".
users.username entity.user.userid Nome utente estratto prima di "@" e utilizzato come "userid" se è il primo utente nell'array "users".
entity.metadata.vendor_name "Duo"
entity.metadata.product_name "Duo Entity Context Data" (Dati di contesto dell'entità Duo)
entity.metadata.entity_type RISORSA
entity.relations.entity_type UTENTE
entity.relations.relationship OWNS

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