Raccogli i log di SpyCloud

Supportato in:

Questo documento spiega come importare i log di SpyCloud in Google Security Operations utilizzando Google Cloud Storage V2.

SpyCloud è una piattaforma di prevenzione del takeover dell'account che fornisce informazioni su violazioni e credenziali rubate. Fornisce record di violazione, avvisi di watchlist e report sulle credenziali compromesse tramite un'API REST.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • 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
  • Un account SpyCloud con accesso API e una chiave API valida

Crea un bucket 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 spycloud-logs).
    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) Attivare il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Raccogli le credenziali API di SpyCloud

Ottenere la chiave API

  1. Accedi al portale SpyCloud come amministratore.
  2. Vai a Impostazioni > API.
  3. Copia e salva la chiave API in una posizione sicura.

Verifica le autorizzazioni

Per verificare che la chiave API disponga dell'accesso richiesto:

  1. Accedi al portale SpyCloud.
  2. Vai a Impostazioni > API.
  3. Verifica che la chiave API sia attiva e abbia accesso agli endpoint richiesti (dati sulle violazioni, watchlist, bussola).
  4. Se l'accesso è limitato, contatta l'amministratore di SpyCloud.

Testare l'accesso API

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

    # Replace with your actual API key
API_KEY="your-api-key"

# Test API access - fetch watchlist data
curl -v -H "X-API-Key: ${API_KEY}" \
  "https://api.spycloud.io/enterprise-v2/breach/data/watchlist?since=2024-01-01&until=2024-01-02"

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 spycloud-collector-sa.
    • Descrizione service account: inserisci Service account for Cloud Run function to collect SpyCloud 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 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.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono necessari per:

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

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (ad esempio spycloud-logs).
  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, spycloud-collector-sa@your-project.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 spycloud-trigger.
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run verrà attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API SpyCloud e scriverli 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 spycloud-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 spycloud-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 spycloud-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 spycloud-logs Nome del bucket GCS
    GCS_PREFIX spycloud Prefisso per i file di log
    STATE_KEY spycloud/state.json Percorso file di stato
    API_KEY your-api-key Chiave API SpyCloud
    API_BASE https://api.spycloud.io URL di base dell'API
    MAX_RECORDS 10000 Numero massimo di record per esecuzione
    LOOKBACK_DAYS 7 Periodo di riferimento iniziale in giorni
    STREAMS watchlist,catalog Stream di dati separati da virgole

  10. Scorri verso il basso nella scheda Variabili e secret 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.

  12. Nella sezione Scalabilità della revisione:

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

  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 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 urllib3
from datetime import datetime, timezone, timedelta
import time

# 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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'spycloud')
STATE_KEY = os.environ.get('STATE_KEY', 'spycloud/state.json')
API_KEY = os.environ.get('API_KEY', '')
API_BASE = os.environ.get('API_BASE', 'https://api.spycloud.io').rstrip('/')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
LOOKBACK_DAYS = int(os.environ.get('LOOKBACK_DAYS', '7'))
STREAMS = [s.strip() for s in os.environ.get('STREAMS', 'watchlist').split(',') if s.strip()]

def parse_datetime(value: str) -> datetime:
  """Parse ISO datetime string to datetime object."""
  if value.endswith("Z"):
    value = value[:-1] + "+00:00"
  return datetime.fromisoformat(value)

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

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

  if not all([GCS_BUCKET, API_KEY]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)

    # Load state
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)

    # Determine date range
    if isinstance(state, dict) and state.get("last_date"):
      since_date = state["last_date"]
    else:
      since_date = (now - timedelta(days=LOOKBACK_DAYS)).strftime('%Y-%m-%d')

    until_date = now.strftime('%Y-%m-%d')

    print(f"Fetching data from {since_date} to {until_date}")

    report = {}

    if 'watchlist' in STREAMS:
      print("Fetching watchlist breach data...")
      count = pull_watchlist(bucket, since_date, until_date)
      report['watchlist_records'] = count

    if 'catalog' in STREAMS:
      print("Fetching breach catalog...")
      count = pull_catalog(bucket, since_date, until_date)
      report['catalog_records'] = count

    if 'compass' in STREAMS:
      print("Fetching compass data...")
      count = pull_compass(bucket, since_date, until_date)
      report['compass_records'] = count

    # Update state
    save_state(bucket, STATE_KEY, until_date)
    print(f"Successfully processed: {json.dumps(report)}")

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

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: {e}")

  return {}

def save_state(bucket, key, last_date: str):
  """Save the last query date to GCS state file."""
  try:
    state = {'last_date': last_date, 'last_event_time': datetime.now(timezone.utc).isoformat()}
    blob = bucket.blob(key)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_date={last_date}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")

def api_get(endpoint: str, params: dict = None):
  """Make authenticated GET request to SpyCloud API with rate limiting."""
  url = f"{API_BASE}{endpoint}"
  if params:
    query = '&'.join([f"{k}={v}" for k, v in params.items()])
    url = f"{url}?{query}"

  headers = {
    'X-API-Key': API_KEY,
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-SpyCloudCollector/1.0'
  }

  backoff = 1.0
  max_retries = 3

  for attempt in range(max_retries):
    response = http.request('GET', url, headers=headers)

    if response.status == 429:
      retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
      print(f"Rate limited (429). Retrying after {retry_after}s...")
      time.sleep(retry_after)
      backoff = min(backoff * 2, 30.0)
      continue

    if response.status != 200:
      print(f"HTTP Error: {response.status} - {response.data.decode('utf-8')}")
      return None

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

  print(f"Failed after {max_retries} retries due to rate limiting")
  return None

def write_ndjson(bucket, prefix: str, stream_name: str, records: list):
  """Write records to GCS as NDJSON."""
  if not records:
    return 0

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

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

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

def pull_watchlist(bucket, since_date: str, until_date: str):
  """Fetch watchlist breach data."""
  cursor = None
  all_records = []

  while True:
    params = {'since': since_date, 'until': until_date}
    if cursor:
      params['cursor'] = cursor

    data = api_get('/enterprise-v2/breach/data/watchlist', params)
    if not data:
      break

    results = data.get('results', [])
    if not results:
      break

    all_records.extend(results)

    if len(all_records) >= MAX_RECORDS:
      print(f"Reached max_records limit ({MAX_RECORDS})")
      break

    cursor = data.get('cursor')
    if not cursor:
      break

  return write_ndjson(bucket, GCS_PREFIX, 'watchlist', all_records)

def pull_catalog(bucket, since_date: str, until_date: str):
  """Fetch breach catalog."""
  params = {'since': since_date, 'until': until_date}
  data = api_get('/enterprise-v2/breach/catalog', params)
  if not data:
    return 0

  results = data.get('results', [])
  return write_ndjson(bucket, GCS_PREFIX, 'catalog', results)

def pull_compass(bucket, since_date: str, until_date: str):
  """Fetch compass findings."""
  cursor = None
  all_records = []

  while True:
    params = {'since': since_date, 'until': until_date}
    if cursor:
      params['cursor'] = cursor

    data = api_get('/enterprise-v2/compass/data', params)
    if not data:
      break

    results = data.get('results', [])
    if not results:
      break

    all_records.extend(results)

    if len(all_records) >= MAX_RECORDS:
      print(f"Reached max_records limit ({MAX_RECORDS})")
      break

    cursor = data.get('cursor')
    if not cursor:
      break

  return write_ndjson(bucket, GCS_PREFIX, 'compass', all_records)

    • Secondo file: requirements.txt:

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

  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 pubblicherà messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

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

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome spycloud-collector-daily
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza 0 0 * * * (ogni giorno a mezzanotte)
    Fuso orario Seleziona il fuso orario (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento spycloud-trigger
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. 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
Ogni 6 ore 0 */6 * * * Volume basso, elaborazione batch
Ogni giorno 0 0 * * * Raccolta dei dati storici (consigliata per i dati sulle violazioni)

Testare l'integrazione

  1. Nella console Cloud Scheduler, trova il tuo job (spycloud-collector-daily).
  2. Fai clic su Forza esecuzione per attivare manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > spycloud-collector > Log.

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

    Fetching data from YYYY-MM-DD to YYYY-MM-DD
Fetching watchlist breach data...
Wrote X records to gs://spycloud-logs/spycloud/watchlist/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed: {"watchlist_records": X, "catalog_records": Y}

  5. Controlla il bucket GCS (spycloud-logs) per verificare che i log siano stati scritti.

Se visualizzi errori nei log:

  • HTTP 401: controlla la chiave API nelle variabili di ambiente
  • HTTP 403: verifica che la chiave API abbia accesso agli endpoint richiesti
  • HTTP 429: limitazione di frequenza: la funzione riproverà automaticamente con backoff
  • Variabili di ambiente mancanti: controlla che tutte le variabili richieste siano impostate

Configura un feed in Google SecOps per importare i log di SpyCloud

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

  7. Fai clic su Ottieni service account. Verrà visualizzata un'email dell'account di servizio univoca, ad esempio:

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

  8. Copia l'indirizzo email. Lo utilizzerai nel prossimo passaggio.

  9. Fai clic su Avanti.

  10. Specifica i valori per i seguenti parametri di input:

    • URL bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:

      gs://spycloud-logs/spycloud/
      • Sostituisci:
        • spycloud-logs: il nome del bucket GCS.
        • spycloud: (Facoltativo) prefisso/percorso della cartella in cui vengono archiviati i log (lascia vuoto per la radice).

    • Opzione di eliminazione della fonte: 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 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

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 (spycloud-logs).
  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.

Tabella di mappatura UDM

Campo log Mappatura UDM Logic
av_softwares, assets.av_softwares, assets.country, assets.country_code, assets.display_resolution, assets.email, assets.full_name, assets.infected_machine_id, assets.infected_path, assets.infected_time, assets.ip_addresses, assets.keyboard_languages, assets.password, assets.target_url, assets.username, assets.user_browser, assets.user_hostname, assets.user_os, assets.user_sys_registered_owner additional.fields Unite dalle etichette create da questi campi e da av_software_list
password extensions.auth.auth_details Valore copiato direttamente
quando metadata.event_timestamp Analizzato come timestamp ISO8601
metadata.event_type Impostato in base alle condizioni: NETWORK_CONNECTION se ha has_principal, has_target, has_network; USER_UNCATEGORIZED se ha has_principal e has_principal_userid; STATUS_UPDATE se ha has_principal e non has_principal_ip; altrimenti GENERIC_EVENT; o USER_UNCATEGORIZED se sono presenti user_hostname, ip o infected_machine_id
infected_time metadata.ingested_timestamp Analizzato come timestamp con i formati aaaa-MM-ggTHH:mm:ssZ, RFC3339, ISO8601
log_id, assets.log_id, uuid metadata.product_log_id Valore di log_id se non è vuoto, altrimenti assets.log_id, altrimenti uuid
user_os network.http.parsed_user_agent Convertito in user agent analizzato
user_os network.http.user_agent Valore copiato direttamente
cookie_domain principal.administrative_domain Valore copiato direttamente
country principal.asset.location.country_or_region Valore copiato direttamente
infected_machine_id principal.asset_id Concatenato come "id: " + infected_machine_id
infected_path principal.file.full_path Valore copiato direttamente
user_hostname, domain principal.hostname Impostato su user_hostname se non è vuoto, poi su dominio se non è vuoto
ip, ip_addresses principal.ip Unito dall'array ip e ip_addresses
country_code principal.resource.attribute.labels Unito da country_code_label creato da country_code
id principal.resource.id Valore copiato direttamente (convertito in stringa)
home page principal.url Valore copiato direttamente
email principal.user.email_addresses Valore copiato direttamente
full_name principal.user.user_display_name Valore copiato direttamente
user_sys_registered_owner, email_username principal.user.userid Impostato su user_sys_registered_owner, poi su email_username se non è vuoto
confidenza security_result.confidence_details Convertito in stringa
descrizione security_result.description Valore copiato direttamente
cookie_expiration, cookie_name, cookie_subdomain, cookie_value, day, document_id, locality_zone, source_id, spycloud_publishdate, spycloud_publish_date, user_browser, infected_time, timezone, password_type, password_plaintext, email_domain, api_token, account_status, breach_category, breach_main_category, consumer_category, malware_family, num_records, premium_flag, sensitive_source, short_title, site_description, title, tlp, type, display_resolution, keyboard_languages security_result.detection_fields Unite dalle etichette create da questi campi
gravità security_result.severity Imposta su LOW se "2", INFORMATIONAL se "5", HIGH se "20", CRITICAL se "25" o "26"
gravità security_result.severity_details Valore copiato direttamente
target_subdomain target.administrative_domain Valore copiato direttamente
target_domain target.asset.hostname Valore copiato direttamente
target_domain target.hostname Valore copiato direttamente
target_url target.url Valore copiato direttamente
nome utente target.user.userid Valore copiato direttamente
metadata.product_name Impostato su "SPYCLOUD"
metadata.vendor_name Impostato su "SpyCloud"

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