Raccogliere gli audit log di Google Cloud Looker

Supportato in:

Questo documento spiega come importare gli audit log di Google Cloud Looker in Google Security Operations utilizzando Google Cloud Storage o Amazon S3.

Looker è una piattaforma di business intelligence e analisi dei dati che consente alle organizzazioni di esplorare, analizzare e condividere approfondimenti aziendali in tempo reale. Il modello Attività di sistema di Looker monitora gli eventi di autenticazione degli utenti, la cronologia di esecuzione delle query, l'accesso a dashboard e Look, la creazione e la modifica dei contenuti, gli eventi di download, le chiamate API, gli eventi di distribuzione pianificata e le modifiche alle autorizzazioni.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un'istanza di Looker con accesso amministrativo o l'autorizzazione see_system_activity
  • Credenziali API Looker (ID client e client secret)
  • Per il percorso GCS: un progetto GCP con le API Cloud Storage, Cloud Run, Pub/Sub e Cloud Scheduler abilitate
  • Per il percorso S3: accesso privilegiato ad AWS (S3, IAM)

Configurare le credenziali dell'API Looker

Per consentire a Google SecOps di recuperare i dati di controllo dell'attività di sistema, devi creare le credenziali API in Looker e configurare un account di servizio con le autorizzazioni richieste.

Crea un account di servizio Looker per l'accesso API

  1. Accedi alla tua istanza di Looker come amministratore.
  2. Vai ad Amministrazione > Utenti.
  3. Fai clic su Aggiungi utenti.
  4. Nel campo Email, inserisci un'email del account di servizio (ad esempio, chronicle-integration@yourcompany.com).

  5. Nella sezione Ruoli, seleziona un ruolo che includa l'autorizzazione see_system_activity.

  6. Fai clic su Salva.

Genera credenziali API3

  1. Vai ad Amministrazione > Utenti.
  2. Trova l'utente del account di servizio che hai creato e fai clic su Modifica.
  3. Scorri verso il basso fino alla sezione Chiavi API.
  4. Fai clic su Nuova chiave API.

  5. Copia e salva i seguenti valori in una posizione sicura:

    • ID client: l'identificatore pubblico per l'autenticazione API
    • Client secret: la chiave privata per l'autenticazione API

Identificare l'URL di base dell'API Looker

  • L'URL di base dell'API Looker segue questo formato:

    https://<instance_name>.cloud.looker.com

Per le istanze di Looker ospitate su Google Cloud, Microsoft Azure e le istanze ospitate su AWS create il 07/07/2020 o in data successiva, l'API utilizza la porta 443 (HTTPS predefinito). Per le istanze ospitate su AWS meno recenti, l'API potrebbe utilizzare la porta 19999.

Puoi trovare l'URL dell'host dell'API andando ad Amministrazione > API nell'istanza di Looker.

Testare l'accesso API

  • Testa le tue credenziali prima di procedere con l'integrazione:

    # Replace with your actual credentials
LOOKER_BASE_URL="https://your-instance.cloud.looker.com"
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"

# Obtain access token
TOKEN=$(curl -s -X POST "${LOOKER_BASE_URL}/api/4.0/login" \
    -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" \
    | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

# Test System Activity access
curl -s -H "Authorization: token ${TOKEN}" \
    "${LOOKER_BASE_URL}/api/4.0/queries/run/json" \
    -X POST \
    -H "Content-Type: application/json" \
    -d '{"model":"system__activity","view":"event","fields":["event.name","event.created_time"],"limit":"5","sorts":["event.created_time desc"]}' \
    | python3 -m json.tool

Una risposta positiva restituisce un array JSON degli eventi Looker recenti.

Opzione A: configura l'importazione utilizzando Google Cloud Storage

Questa opzione utilizza una funzione Cloud Run per eseguire il polling dell'API Looker per gli eventi di controllo dell'attività di sistema e scriverli in un bucket GCS per l'importazione da parte di Google SecOps.

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 looker-audit-logs-gcs).
    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 degli accessi Uniforme (consigliato)
    Strumenti di protezione (Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Crea un account di servizio per la funzione Cloud Run

  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 looker-audit-collector-sa
    • Descrizione service account: inserisci Service account for Cloud Run function to collect Looker audit logs
  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 Cloud Functions Invoker.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Concedi autorizzazioni IAM sul bucket GCS

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (looker-audit-logs-gcs).
  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 (looker-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea argomento Pub/Sub

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

Crea una funzione Cloud Run per raccogliere i log

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

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

    Impostazione Valore
    Nome servizio looker-audit-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 looker-audit-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 looker-audit-collector-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 Descrizione
    GCS_BUCKET looker-audit-logs-gcs Nome bucket GCS
    GCS_PREFIX looker-audit Prefisso per i file di log
    STATE_KEY looker-audit/state.json Percorso file di stato
    LOOKER_BASE_URL https://your-instance.cloud.looker.com URL di base dell'API Looker
    LOOKER_CLIENT_ID your-client-id ID client API Looker
    LOOKER_CLIENT_SECRET your-client-secret Client secret dell'API Looker
    LOOKBACK_HOURS 24 Periodo di riferimento iniziale
    PAGE_SIZE 5000 Record per pagina API
    MAX_PAGES 20 Numero massimo di pagine per query

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

    • Timeout richiesta: inserisci 600 secondi (10 minuti)

  11. Vai alla scheda Impostazioni:

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

  12. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0
    • Numero massimo di istanze: inserisci 100

  13. Fai clic su Crea.

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

  15. Dopo aver creato il servizio, si aprirà automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

  1. Inserisci main nel campo Entry point (Punto di ingresso).
  2. Nell'editor di codice incorporato, crea due file:

  • main.py:

    import functions_framework
from google.cloud import storage
import json
import os
import urllib3
import urllib.parse
from datetime import datetime, timezone, timedelta

http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=10.0, read=60.0),
    retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'looker-audit')
STATE_KEY = os.environ.get('STATE_KEY', 'looker-audit/state.json')
LOOKER_BASE_URL = os.environ.get('LOOKER_BASE_URL', '').rstrip('/')
CLIENT_ID = os.environ.get('LOOKER_CLIENT_ID')
CLIENT_SECRET = os.environ.get('LOOKER_CLIENT_SECRET')
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '5000'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '20'))

@functions_framework.cloud_event
def main(cloud_event):
    if not all([GCS_BUCKET, LOOKER_BASE_URL, CLIENT_ID, CLIENT_SECRET]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket)
        now = datetime.now(timezone.utc)

        if isinstance(state, dict) and state.get('last_event_time'):
            try:
                last_val = state['last_event_time']
                if last_val.endswith('Z'):
                    last_val = last_val[:-1] + '+00:00'
                last_time = datetime.fromisoformat(last_val)
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")
                last_time = now - timedelta(hours=LOOKBACK_HOURS)
        else:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching events from {last_time.isoformat()} to {now.isoformat()}")

        token = get_access_token()

        events = fetch_system_activity(
            token, 'event',
            [
                'event.id', 'event.name', 'event.category',
                'event.created_time', 'event.is_api_call',
                'event.is_admin', 'event.is_looker_employee',
                'user.id', 'user.name', 'user.email'
            ],
            'event.created_time', last_time, now
        )

        history = fetch_system_activity(
            token, 'history',
            [
                'history.id', 'history.created_time',
                'history.completed_time', 'history.status',
                'history.source', 'history.issuer_source',
                'history.runtime', 'history.message',
                'query.id', 'query.model', 'query.view',
                'user.id', 'user.name', 'user.email',
                'dashboard.id', 'dashboard.title',
                'look.id', 'look.title'
            ],
            'history.created_time', last_time, now
        )

        all_records = []
        for e in events:
            e['_looker_record_type'] = 'event'
            all_records.append(e)
        for h in history:
            h['_looker_record_type'] = 'history'
            all_records.append(h)

        if not all_records:
            print("No new records found.")
            save_state(bucket, now.isoformat())
            return

        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/looker_audit_{timestamp}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join(
            [json.dumps(r, ensure_ascii=False, default=str) for r in all_records]
        ) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

        print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}")

        newest = find_newest_time(events, history)
        save_state(bucket, newest if newest else now.isoformat())

        print(f"Successfully processed {len(all_records)} records "
              f"(events: {len(events)}, history: {len(history)})")

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

def get_access_token():
    url = f"{LOOKER_BASE_URL}/api/4.0/login"
    encoded_body = urllib.parse.urlencode({
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET
    }).encode('utf-8')

    response = http.request(
        'POST', url,
        body=encoded_body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )

    if response.status != 200:
        raise Exception(
            f"Looker login failed: {response.status} - "
            f"{response.data.decode('utf-8')}"
        )

    data = json.loads(response.data.decode('utf-8'))
    token = data.get('access_token')
    if not token:
        raise Exception("No access_token in login response")

    print("Successfully obtained Looker API access token")
    return token

def fetch_system_activity(token, view, fields, time_field, start_time, end_time):
    start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
    end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')

    all_records = []
    offset = 0

    for page in range(MAX_PAGES):
        query_body = {
            "model": "system__activity",
            "view": view,
            "fields": fields,
            "filters": {
                time_field: f"{start_str} to {end_str}"
            },
            "sorts": [f"{time_field} asc"],
            "limit": str(PAGE_SIZE),
            "offset": str(offset)
        }

        url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json"
        response = http.request(
            'POST', url,
            body=json.dumps(query_body).encode('utf-8'),
            headers={
                'Authorization': f'token {token}',
                'Content-Type': 'application/json'
            }
        )

        if response.status == 429:
            print(f"Rate limited on {view} query. Stopping pagination.")
            break

        if response.status != 200:
            print(f"{view} query failed: {response.status} - "
                  f"{response.data.decode('utf-8')}")
            break

        page_results = json.loads(response.data.decode('utf-8'))

        if not page_results:
            break

        all_records.extend(page_results)
        print(f"{view} page {page + 1}: {len(page_results)} records "
              f"(total: {len(all_records)})")

        if len(page_results) < PAGE_SIZE:
            break

        offset += PAGE_SIZE

    print(f"Total {view} records fetched: {len(all_records)}")
    return all_records

def find_newest_time(events, history):
    newest = None
    for e in events:
        t = e.get('event.created_time')
        if t and (newest is None or t > newest):
            newest = t
    for h in history:
        t = h.get('history.created_time')
        if t and (newest is None or t > newest):
            newest = t
    return newest

def load_state(bucket):
    try:
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
            return json.loads(blob.download_as_text())
    except Exception as e:
        print(f"Warning: Could not load state: {e}")
    return {}

def save_state(bucket, last_event_time_iso):
    try:
        state = {
            'last_event_time': last_event_time_iso,
            'last_run': datetime.now(timezone.utc).isoformat()
        }
        blob = bucket.blob(STATE_KEY)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: last_event_time={last_event_time_iso}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

  • requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

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

Crea job Cloud Scheduler

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

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome looker-audit-collector-hourly
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza 0 * * * * (ogni ora, allo scoccare dell'ora)
    Fuso orario Seleziona il fuso orario (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona looker-audit-trigger
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Testare l'integrazione

  1. Nella console Cloud Scheduler, trova il tuo job (looker-audit-collector-hourly).
  2. Fai clic su Forza esecuzione per attivare il job manualmente.
  3. Attendi alcuni secondi.
  4. Vai a Cloud Run > Servizi.
  5. Fai clic su looker-audit-collector.
  6. Fai clic sulla scheda Log.

  7. Verifica che la funzione sia stata eseguita correttamente. Cerca:

    Fetching events from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Successfully obtained Looker API access token
event page 1: X records (total: X)
history page 1: X records (total: X)
Wrote X records to gs://looker-audit-logs-gcs/looker-audit/looker_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records (events: X, history: X)

  8. Vai a Cloud Storage > Bucket.

  9. Fai clic su looker-audit-logs-gcs.

  10. Vai alla cartella looker-audit/.

  11. Verifica che sia stato creato un nuovo file .ndjson con il timestamp corrente.

Se nei log vengono visualizzati errori:

  • HTTP 401: verifica che le variabili di ambiente LOOKER_CLIENT_ID e LOOKER_CLIENT_SECRET siano corrette
  • HTTP 403: verifica che l'utente Looker disponga dell'autorizzazione see_system_activity
  • HTTP 429: limitazione della frequenza: la funzione interromperà la paginazione e riprenderà alla successiva esecuzione pianificata
  • Variabili di ambiente mancanti: verifica che tutte le variabili richieste siano impostate nella configurazione di Cloud Run Functions

Recuperare il account di servizio Google SecOps

  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, Looker Audit Logs GCS).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Audit di Looker come Tipo di log.

  7. Fai clic su Ottieni service account. Verrà 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.

  9. Fai clic su Avanti.

  10. 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://looker-audit-logs-gcs/looker-audit/

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

    • 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

  11. Fai clic su Avanti.

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

Concedi le autorizzazioni IAM al account di servizio Google SecOps

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic su looker-audit-logs-gcs.
  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 i ruoli: seleziona Visualizzatore oggetti Storage.

  6. Fai clic su Salva.

Opzione B: configura l'importazione utilizzando Amazon S3

Questa opzione utilizza una funzione AWS Lambda per eseguire il polling dell'API Looker per gli eventi di controllo dell'attività di sistema e scriverli in un bucket S3 per l'importazione da parte di Google SecOps.

Configura il bucket AWS S3 e IAM per Google SecOps

  1. Crea un bucket Amazon S3 seguendo questa guida dell'utente: Creazione di un bucket
  2. Salva il nome e la regione del bucket per riferimento futuro (ad esempio, looker-audit-logs).
  3. Crea un utente seguendo questa guida utente: Creazione di un utente IAM.
  4. Seleziona l'utente creato.
  5. Seleziona la scheda Credenziali di sicurezza.
  6. Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
  7. Seleziona Servizio di terze parti come Caso d'uso.
  8. Fai clic su Avanti.
  9. (Facoltativo) Aggiungi il tag di descrizione.
  10. Fai clic su Crea chiave di accesso.
  11. Fai clic su Scarica file .csv per salvare la chiave di accesso e la chiave di accesso segreta per riferimento futuro.
  12. Fai clic su Fine.
  13. Seleziona la scheda Autorizzazioni.
  14. Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
  15. Seleziona Aggiungi autorizzazioni.
  16. Seleziona Allega direttamente le norme.
  17. Cerca la policy AmazonS3FullAccess.
  18. Seleziona la policy.
  19. Fai clic su Avanti.
  20. Fai clic su Aggiungi autorizzazioni.

Configura la policy e il ruolo IAM per i caricamenti S3

  1. Nella console AWS, vai a IAM > Policy > Crea policy > scheda JSON.
  2. Copia e incolla il criterio riportato di seguito.

  3. JSON della policy (sostituisci looker-audit-logs se hai inserito un nome del bucket diverso):

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

  4. Fai clic su Avanti > Crea policy.

  5. Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.

  6. Allega la policy appena creata.

  7. Assegna al ruolo il nome LookerAuditCollectorRole e fai clic su Crea ruolo.

Crea la funzione Lambda

  1. Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
  2. Fai clic su Crea autore da zero.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome LookerAuditCollector
    Tempo di esecuzione Python 3.13
    Architettura x86_64
    Ruolo di esecuzione LookerAuditCollectorRole

  4. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e incolla il codice riportato di seguito (LookerAuditCollector.py).

    import urllib3
import json
import boto3
import os
from datetime import datetime, timezone, timedelta
import logging
import urllib.parse

logger = logging.getLogger()
logger.setLevel(logging.INFO)

http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=10.0, read=60.0),
    retries=False,
)
s3 = boto3.client('s3')

BUCKET = os.environ['S3_BUCKET']
PREFIX = os.environ['S3_PREFIX']
STATE_KEY = os.environ['STATE_KEY']
LOOKER_BASE_URL = os.environ['LOOKER_BASE_URL'].rstrip('/')
CLIENT_ID = os.environ['LOOKER_CLIENT_ID']
CLIENT_SECRET = os.environ['LOOKER_CLIENT_SECRET']
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '5000'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '20'))

def lambda_handler(event, context):
    try:
        state = load_state()
        now = datetime.now(timezone.utc)

        if state and state.get('last_event_time'):
            try:
                last_time = datetime.fromisoformat(
                    state['last_event_time'].replace('Z', '+00:00')
                )
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                logger.warning(f"Could not parse last_event_time: {e}")
                last_time = now - timedelta(hours=LOOKBACK_HOURS)
        else:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        logger.info(f"Fetching events from {last_time.isoformat()} to {now.isoformat()}")

        token = get_access_token()

        events = fetch_events(token, last_time, now)
        history = fetch_history(token, last_time, now)

        all_records = []
        for e in events:
            e['_looker_record_type'] = 'event'
            all_records.append(e)
        for h in history:
            h['_looker_record_type'] = 'history'
            all_records.append(h)

        if not all_records:
            logger.info("No new records found.")
            save_state(now.isoformat())
            return {'statusCode': 200, 'body': json.dumps({'events': 0})}

        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{PREFIX}/looker_audit_{timestamp}.ndjson"

        ndjson = '\n'.join(
            [json.dumps(r, ensure_ascii=False, default=str) for r in all_records]
        ) + '\n'

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

        logger.info(f"Wrote {len(all_records)} records to s3://{BUCKET}/{object_key}")

        newest_time = find_newest_time(events, history)
        save_state(newest_time if newest_time else now.isoformat())

        return {
            'statusCode': 200,
            'body': json.dumps({
                'events': len(events),
                'history': len(history),
                'total': len(all_records)
            })
        }

    except Exception as e:
        logger.error(f"Lambda execution failed: {str(e)}")
        raise

def get_access_token():
    url = f"{LOOKER_BASE_URL}/api/4.0/login"
    encoded_body = urllib.parse.urlencode({
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET
    }).encode('utf-8')

    response = http.request(
        'POST', url,
        body=encoded_body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )

    if response.status != 200:
        raise Exception(
            f"Login failed with status {response.status}: "
            f"{response.data.decode('utf-8')}"
        )

    data = json.loads(response.data.decode('utf-8'))
    token = data.get('access_token')
    if not token:
        raise Exception("No access_token in login response")

    logger.info("Successfully obtained Looker API access token")
    return token

def fetch_events(token, start_time, end_time):
    start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
    end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')

    all_events = []
    offset = 0

    for page in range(MAX_PAGES):
        query_body = {
            "model": "system__activity",
            "view": "event",
            "fields": [
                "event.id",
                "event.name",
                "event.category",
                "event.created_time",
                "event.is_api_call",
                "event.is_admin",
                "event.is_looker_employee",
                "user.id",
                "user.name",
                "user.email"
            ],
            "filters": {
                "event.created_time": f"{start_str} to {end_str}"
            },
            "sorts": ["event.created_time asc"],
            "limit": str(PAGE_SIZE),
            "offset": str(offset)
        }

        url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json"
        response = http.request(
            'POST', url,
            body=json.dumps(query_body).encode('utf-8'),
            headers={
                'Authorization': f'token {token}',
                'Content-Type': 'application/json'
            }
        )

        if response.status == 429:
            logger.warning("Rate limited on events query. Stopping pagination.")
            break

        if response.status != 200:
            logger.error(
                f"Events query failed: {response.status} - "
                f"{response.data.decode('utf-8')}"
            )
            break

        page_results = json.loads(response.data.decode('utf-8'))

        if not page_results:
            logger.info(f"Events: No more results at offset {offset}")
            break

        all_events.extend(page_results)
        logger.info(
            f"Events page {page + 1}: Retrieved {len(page_results)} records "
            f"(total: {len(all_events)})"
        )

        if len(page_results) < PAGE_SIZE:
            break

        offset += PAGE_SIZE

    logger.info(f"Total events fetched: {len(all_events)}")
    return all_events

def fetch_history(token, start_time, end_time):
    start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
    end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')

    all_history = []
    offset = 0

    for page in range(MAX_PAGES):
        query_body = {
            "model": "system__activity",
            "view": "history",
            "fields": [
                "history.id",
                "history.created_time",
                "history.completed_time",
                "history.status",
                "history.source",
                "history.issuer_source",
                "history.runtime",
                "history.message",
                "query.id",
                "query.model",
                "query.view",
                "user.id",
                "user.name",
                "user.email",
                "dashboard.id",
                "dashboard.title",
                "look.id",
                "look.title"
            ],
            "filters": {
                "history.created_time": f"{start_str} to {end_str}"
            },
            "sorts": ["history.created_time asc"],
            "limit": str(PAGE_SIZE),
            "offset": str(offset)
        }

        url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json"
        response = http.request(
            'POST', url,
            body=json.dumps(query_body).encode('utf-8'),
            headers={
                'Authorization': f'token {token}',
                'Content-Type': 'application/json'
            }
        )

        if response.status == 429:
            logger.warning("Rate limited on history query. Stopping pagination.")
            break

        if response.status != 200:
            logger.error(
                f"History query failed: {response.status} - "
                f"{response.data.decode('utf-8')}"
            )
            break

        page_results = json.loads(response.data.decode('utf-8'))

        if not page_results:
            logger.info(f"History: No more results at offset {offset}")
            break

        all_history.extend(page_results)
        logger.info(
            f"History page {page + 1}: Retrieved {len(page_results)} records "
            f"(total: {len(all_history)})"
        )

        if len(page_results) < PAGE_SIZE:
            break

        offset += PAGE_SIZE

    logger.info(f"Total history records fetched: {len(all_history)}")
    return all_history

def find_newest_time(events, history):
    newest = None
    for e in events:
        t = e.get('event.created_time')
        if t and (newest is None or t > newest):
            newest = t
    for h in history:
        t = h.get('history.created_time')
        if t and (newest is None or t > newest):
            newest = t
    return newest

def load_state():
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        return json.loads(obj['Body'].read().decode('utf-8'))
    except s3.exceptions.NoSuchKey:
        logger.info("No previous state found, starting fresh")
        return None
    except Exception as e:
        logger.warning(f"Could not load state: {e}")
        return None

def save_state(last_event_time):
    state = {
        'last_event_time': last_event_time,
        'last_run': datetime.now(timezone.utc).isoformat()
    }
    s3.put_object(
        Bucket=BUCKET,
        Key=STATE_KEY,
        Body=json.dumps(state, indent=2).encode('utf-8'),
        ContentType='application/json'
    )
    logger.info(f"Saved state: last_event_time={last_event_time}")

  5. Vai a Configurazione > Variabili di ambiente > Modifica > Aggiungi nuova variabile di ambiente.

  6. Inserisci le variabili di ambiente fornite di seguito, sostituendole con i tuoi valori.

Variabili di ambiente

Chiave Valore di esempio
S3_BUCKET looker-audit-logs
S3_PREFIX looker-audit/
STATE_KEY looker-audit/state.json
LOOKER_BASE_URL https://your-instance.cloud.looker.com
LOOKER_CLIENT_ID your-looker-client-id
LOOKER_CLIENT_SECRET your-looker-client-secret
LOOKBACK_HOURS 24
PAGE_SIZE 5000
MAX_PAGES 20
  1. Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).
  2. Seleziona la scheda Configurazione.
  3. Nel riquadro Configurazione generale, fai clic su Modifica.

  4. Modifica Timeout impostando 5 minuti (300 secondi) e fai clic su Salva.

Creare una pianificazione EventBridge

  1. Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
  2. Fornisci i seguenti dettagli di configurazione:
    • Pianificazione ricorrente: Tariffa (1 hour)
    • Target: la tua funzione Lambda LookerAuditCollector
    • Nome: LookerAuditCollector-1h
  3. Fai clic su Crea pianificazione.

Configura un feed in Google SecOps per importare i log di controllo di Looker

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Nella pagina successiva, fai clic su Configura un singolo feed.
  4. Inserisci un nome univoco per il nome feed.
  5. Seleziona Amazon S3 V2 come Tipo di origine.
  6. Seleziona Audit di Looker come Tipo di log.
  7. Fai clic su Avanti e poi su Invia.

  8. Specifica i valori per i seguenti campi:

    • URI S3: s3://looker-audit-logs/looker-audit/
    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.
    • Età massima del file: includi i file modificati nell'ultimo numero di giorni (il valore predefinito è 180 giorni)
    • ID chiave di accesso: chiave di accesso utente con accesso al bucket S3
    • Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3
    • 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 e poi su Invia.

Riferimento ai dati dell'attività del sistema di Looker

La tabella seguente descrive i dati chiave disponibili nelle esplorazioni dell'attività di sistema di Looker raccolti da questa integrazione:

Esplora Dati raccolti Conservazione
Evento Eventi di autenticazione utente, creazione e modifica di contenuti, modifiche alle autorizzazioni, chiamate API, eventi di pubblicazione pianificata, eventi di download 90 giorni (predefinito)
Cronologia Cronologia di esecuzione delle query, dashboard e accesso a Look, runtime e stato delle query, origine della query (UI, API, pianificazione) 90 giorni (predefinito)

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
Link di modifica di gruppo additional.fields.Group_Edit_Link_label.value.string_value Valore copiato direttamente
ID gruppo additional.fields.Group_ID_label.value.string_value Valore copiato direttamente
Durata dell'esecuzione più recente della cronologia in secondi additional.fields.History_Most_Recent_Run_Length_in_Seconds_label.value.string_value Valore copiato direttamente
Slug della cronologia additional.fields.History_Slug_label.value.string_value Valore copiato direttamente
Origine della cronologia additional.fields.History_Source_label.value.string_value Valore copiato direttamente
Stato cronologia additional.fields.History_Status_label.value.string_value Valore copiato direttamente
Link Look additional.fields.Look_Link_label.value.string_value Valore copiato direttamente
Titolo del look additional.fields.Look_Title_label.value.string_value Valore copiato direttamente
Link di modifica utente additional.fields.User_Edit_Link_label.value.string_value Valore copiato direttamente
Cartella Home utente additional.fields.User_Home_Folder_label.value.string_value Valore copiato direttamente
dashboard.link additional.fields.dashboard_link_label.value.string_value Valore copiato direttamente
dashboard.title additional.fields.dashboard_title_label.value.string_value Valore copiato direttamente
history.source additional.fields.history_source_label.value.string_value Valore copiato direttamente
history.status additional.fields.history_status_label.value.string_value Valore copiato direttamente
history.id additional.fields.id_label.value.string_value Convertito in stringa
history.connection_name additional.fields.name_label.value.string_value Valore copiato direttamente
query.model additional.fields.query_model_label.value.string_value Valore copiato direttamente
query.view additional.fields.query_view_label.value.string_value Valore copiato direttamente
sql_text.text additional.fields.sql_text_text_label.value.string_value Valore copiato direttamente
Data/ora creazione cronologia metadata.event_timestamp Analizzato utilizzando il formato ISO8601, RFC 3339 o aaaa-MM-gg HH:mm:ss
has_principal_user metadata.event_type Impostato su "NETWORK_CONNECTION" se has_principal e has_target sono true; "USER_UNCATEGORIZED" se has_principal_user è true; "STATUS_UPDATE" se has_principal è true; altrimenti "GENERIC_EVENT"
has_principal metadata.event_type
has_target metadata.event_type
Indirizzo email dell'utente principal.email Valore di Email utente se non è vuoto, altrimenti user.email
user.email principal.email
Nome gruppo principal.group.group_display_name Valore copiato direttamente
ID utente principal.user.product_object_id Valore di User-ID se non è vuoto, altrimenti user.id
user.id principal.user.product_object_id
Nome utente principal.user.userid Valore di User Name se non è vuoto, altrimenti user.name
nome.utente principal.user.userid
Kevin Liu security_result.category_details Unito a "Kevin_Liu_label" se Kevin Liu non è vuoto, "History_ID_label" se History ID non è vuoto, "History_Created_Date_label" se History Created Date non è vuoto
ID cronologia security_result.category_details
Data di creazione della cronologia security_result.category_details
Descrizione del look security_result.description Valore copiato direttamente
Nome utente ordinato target.hostname Valore copiato se il nome del ramo di sviluppo utente non è vuoto
user.name principal.user.user_display_name Mappato dal log delle modifiche

Log delle modifiche

Visualizza il log delle modifiche per questo parser

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