Raccogli i log delle attività utente di Workday

Supportata in:

Questo documento spiega come importare i log delle attività utente di Workday in Google Security Operations utilizzando Cloud Storage V2.

Workday è una piattaforma basata su cloud di gestione del capitale umano (HCM) e finanziaria che genera log di attività utente, autenticazione e log di controllo di sicurezza. L'API REST User Activity Logging fornisce l'accesso programmatico a questi log.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto Google Cloud con {storage_api} abilitata
  • Autorizzazioni per creare e gestire i bucket Cloud Storage
  • Autorizzazioni per gestire le policy di Identity and Access Management (IAM) sui bucket Cloud Storage
  • Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso con privilegi a Workday con autorizzazioni di amministratore della sicurezza o equivalenti
  • Logging delle attività utente abilitato nel tuo tenant Workday

Crea un bucket Cloud Storage

  1. Vai alla consoleGoogle 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 globalmente univoco (ad esempio workday-activity-logs)
    Tipo di località Scegli in base alle tue esigenze (Regione, a due regioni, multiregionale)
    Località Seleziona la località (ad esempio us-central1)
    Classe di archiviazione Standard (consigliato per i log ad accesso 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.

Raccogli le credenziali API di Workday

Abilita il logging delle attività utente

  1. Accedi al tuo tenant Workday come amministratore della sicurezza.
  2. Nella barra di ricerca di Workday, cerca e apri l'attività Edit Tenant Setup - System (Modifica configurazione tenant - Sistema).
  3. Nella sezione Sicurezza, seleziona la casella di controllo Abilita il logging delle attività utente.
  4. Fai clic su Ok per salvare le modifiche.

Crea un utente del sistema di integrazione (ISU)

  1. Nella barra di ricerca di Workday, cerca Create Integration System User (Crea utente di sistema di integrazione).
  2. Inserisci un nome utente (ad esempio ISU_SIEM_Export) e una password efficace.
  3. Deseleziona la casella di controllo Richiedi nuova password al prossimo accesso.
  4. Fai clic su OK.
  5. Cerca Create Security Group (Crea gruppo di sicurezza) e crea un Integration System Security Group (Unconstrained) (Gruppo di sicurezza del sistema di integrazione (senza vincoli)).
  6. Aggiungi l'ISU (ISU_SIEM_Export) a questo gruppo di sicurezza.
  7. Cerca Visualizza dominio e individua il dominio Log delle attività utente. Concedi l'accesso Get al gruppo di sicurezza ISU per questo dominio.
  8. Cerca Activate Pending Security Policy Changes (Attiva modifiche alle policy di sicurezza in attesa) e attiva le modifiche.

Registra un client API per OAuth

  1. Nella barra di ricerca di Workday, cerca Register API Client for Integrations (Registra client API per le integrazioni).
  2. Inserisci un Nome client (ad esempio SIEM_OAuth_Client).
  3. Seleziona l'opzione Non-Expiring Refresh Tokens (Token di aggiornamento senza scadenza).
  4. Aggiungi l'ambito: Sistema (o l'ambito richiesto per l'API User Activity Logging).
  5. Fai clic su OK.
  6. Copia e salva i seguenti dettagli in una posizione sicura:
    • ID client: l'identificatore del client API.
    • Client secret: il client secret dell'API.
  7. Cerca Manage Refresh Tokens for Integrations (Gestisci token di aggiornamento per le integrazioni).
  8. Seleziona l'account ISU (ISU_SIEM_Export).
  9. Genera un nuovo token di aggiornamento per il client API.
  10. Copia e salva il valore di Refresh Token (Token di aggiornamento).

Determina l'URL del tenant

L'endpoint API si basa sul tuo tenant Workday. Il formato è:

Componente Valore
Endpoint token https://HOST/ccx/oauth2/TENANT/token
API Log delle attività https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

Verifica le autorizzazioni

Per verificare che l'ISU disponga delle autorizzazioni richieste:

  1. Accedi a Workday come amministratore della sicurezza.
  2. Cerca View Integration System User (Visualizza utente del sistema di integrazione) e seleziona l'ISU (ISU_SIEM_Export).
  3. Verifica che l'utente sia un membro del gruppo di sicurezza con accesso Get al dominio User Activity Logging.
  4. Se l'accesso non è presente, aggiorna il gruppo di sicurezza e attiva le modifiche alla policy di sicurezza in attesa.

Testa l'accesso all'API

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

    ```bash
    # Replace with your actual credentials
    CLIENT_ID="your-client-id"
    CLIENT_SECRET="your-client-secret"
    REFRESH_TOKEN="your-refresh-token"
    HOST="your-workday-host"
    TENANT="your-tenant"
    
    # Get OAuth access token
    TOKEN=$(curl -s -X POST "https://${HOST}/ccx/oauth2/${TENANT}/token" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "grant_type=refresh_token&client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}&refresh_token=${REFRESH_TOKEN}" \
        | jq -r '.access_token')
    
    # Test API access
    curl -v -H "Authorization: Bearer ${TOKEN}" \
        "https://${HOST}/ccx/api/privacy/v1/${TENANT}/activityLogging?from=2024-01-01T00:00:00Z&to=2024-01-01T01:00:00Z"
    ```
    

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run ha bisogno di un account di servizio con le autorizzazioni per scrivere in un bucket Cloud Storage ed essere richiamata da Pub/Sub.

Crea il account di servizio

  1. Nella console Google Cloud, vai a IAM e amministrazione > Account di servizio.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome service account: inserisci workday-activity-collector-sa.
    • Descrizione service account: inserisci Service account for Cloud Run function to collect Workday User Activity 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.

Questi ruoli sono obbligatori per:

  • Storage Object Admin: scrive i log nel bucket Cloud Storage e gestisce i file di stato
  • Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
  • Cloud Functions Invoker: consente la chiamata della funzione

Concedi le autorizzazioni IAM sul bucket Cloud Storage

Concedi al account di servizio le autorizzazioni in scrittura sul bucket Cloud Storage nel seguente modo:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (ad esempio workday-activity-logs).
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'indirizzo email del account di servizio (ad esempio workday-activity-collector-sa@your-project.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea un argomento Pub/Sub

Crea l'argomento Pub/Sub in cui Cloud Scheduler pubblicherà i messaggi e a cui la funzione Cloud Run si iscriverà.

  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 workday-activity-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 Workday User Activity Logging e scriverli in Cloud Storage.

  1. Nella console Google Cloud, vai a Cloud Run.
  2. Fai clic su Scrivi una funzione.
  3. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio workday-activity-collector
    Regione Seleziona la regione corrispondente al tuo bucket Cloud Storage (ad esempio us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive
  4. Nella sezione Trigger, fai clic su Aggiungi trigger e seleziona Trigger Pub/Sub.

  5. Nel riquadro Trigger Eventarc, fornisci i seguenti dettagli di configurazione:

    • Nome trigger: mantieni il nome generato o inserisci un nome per il trigger.
    • Tipo di trigger: seleziona Origini Google.
    • Provider di eventi: seleziona Pub/Sub.
    • Tipo di evento: seleziona google.cloud.pubsub.topic.v1.messagePublished.
    • Seleziona un argomento Cloud Pub/Sub: scegli l'argomento workday-activity-trigger.
    • Regione: seleziona la stessa regione della funzione.
    • Service account: seleziona il account di servizio workday-activity-collector-sa.
  6. Fai clic su Salva trigger.

  7. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Seleziona Identity and Access Management (IAM).
  8. Vai a Container, networking, sicurezza ed espandi la sezione.

  9. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio workday-activity-collector-sa.
  10. 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 workday-activity-logs Nome bucket Cloud Storage
    GCS_PREFIX workday Prefisso per i file di log
    STATE_KEY workday-state.json Percorso dello stato, al di fuori del prefisso del log
    WORKDAY_HOST your-workday-host Nome host Workday
    WORKDAY_TENANT your-tenant Nome tenant Workday
    CLIENT_ID your-client-id ID client OAuth
    CLIENT_SECRET your-client-secret Client secret OAuth
    REFRESH_TOKEN your-refresh-token Token di aggiornamento OAuth
    MAX_RECORDS 10000 Numero massimo di record per esecuzione
    PAGE_SIZE 100 Record per pagina
    LOOKBACK_HOURS 24 Periodo di riferimento iniziale
  11. Nella scheda Variabili e secret, vai a Richieste:

    • Timeout della richiesta: inserisci 600 secondi (10 minuti).
  12. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0.
    • Numero massimo di istanze: inserisci 100 (o regola 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, l'editor di codice incorporato si aprirà automaticamente.

Aggiungi il 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
      import hashlib
      
      # 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', 'workday')
      STATE_KEY = os.environ.get('STATE_KEY', 'workday-state.json')
      WORKDAY_HOST = os.environ.get('WORKDAY_HOST', '')
      WORKDAY_TENANT = os.environ.get('WORKDAY_TENANT', '')
      CLIENT_ID = os.environ.get('CLIENT_ID', '')
      CLIENT_SECRET = os.environ.get('CLIENT_SECRET', '')
      REFRESH_TOKEN = os.environ.get('REFRESH_TOKEN', '')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      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)
      
      # Activity Logging entries carry no per-event identifier (taskId and
      # sessionId are reused), so the content hash is the identity.
      ID_FIELDS = ()
      
      def event_keys(event):
          """Return every identity this event can be recognised by."""
          keys = {'sha256:' + hashlib.sha256(
              json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
          ).hexdigest()}
          for field in ID_FIELDS:
              value = event.get(field)
              if value:
                  keys.add(f'{field}:{value}')
          return keys
      
      def get_token():
          """Get OAuth 2.0 access token using refresh token flow."""
          token_url = f"https://{WORKDAY_HOST}/ccx/oauth2/{WORKDAY_TENANT}/token"
      
          body = f"grant_type=refresh_token&client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}&refresh_token={REFRESH_TOKEN}"
      
          headers = {
              'Content-Type': 'application/x-www-form-urlencoded',
              'Accept': 'application/json'
          }
      
          backoff = 1.0
          max_retries = 3
      
          for attempt in range(max_retries):
              response = http.request('POST', token_url, body=body.encode('utf-8'), headers=headers)
      
              if response.status == 429:
                  retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                  print(f"Rate limited (429) on token request. Retrying after {retry_after}s...")
                  time.sleep(retry_after)
                  backoff = min(backoff * 2, 30.0)
                  continue
      
              if response.status != 200:
                  raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}")
      
              data = json.loads(response.data.decode('utf-8'))
              return data['access_token']
      
          raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")
      
      @functions_framework.cloud_event
      def main(cloud_event):
          """
          Cloud Run function triggered by Pub/Sub to fetch Workday User Activity logs and write to GCS.
      
          Args:
              cloud_event: CloudEvent object containing Pub/Sub message
          """
      
          if not all([GCS_BUCKET, WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN]):
              print('Error: Missing required environment variables')
              return
      
          try:
              bucket = storage_client.bucket(GCS_BUCKET)
      
              # Load state
              state = load_state(bucket, STATE_KEY)
              seen_keys = load_seen(bucket, STATE_KEY)
      
              # Determine time window
              now = datetime.now(timezone.utc)
              last_time = None
      
              if isinstance(state, dict) and state.get("last_event_time"):
                  try:
                      last_time = parse_datetime(state["last_event_time"])
                      last_time = last_time - timedelta(minutes=2)
                  except Exception as e:
                      print(f"Warning: Could not parse last_event_time: {e}")
      
              if last_time is None:
                  last_time = now - timedelta(hours=LOOKBACK_HOURS)
      
              print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
      
              # Get access token
              token = get_token()
      
              # Fetch logs
              records, newest_event_time = fetch_logs(
                  token=token,
                  start_time=last_time,
                  end_time=now,
                  page_size=PAGE_SIZE,
                  max_records=MAX_RECORDS,
              )
      
              # Drop the events the overlap window re-read from the previous run
              fresh = [r for r in records if not (event_keys(r) & seen_keys)]
              all_keys = set()
              for record in records:
                  all_keys |= event_keys(record)
      
              if not fresh:
                  print("No new log records found.")
                  return
      
              # Write to GCS as NDJSON
              timestamp = now.strftime('%Y%m%d_%H%M%S')
              object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
              blob = bucket.blob(object_key)
      
              ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in fresh]) + '\n'
              blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
              print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")
      
              # Remember every fetched record so the next overlap window skips it
              save_seen(bucket, STATE_KEY, all_keys)
      
              if newest_event_time:
                  save_state(bucket, STATE_KEY, newest_event_time)
              else:
                  save_state(bucket, STATE_KEY, now.isoformat())
      
              print(f"Successfully processed {len(fresh)} records")
      
          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}")
              raise
      
          return {}
      
      def save_state(bucket, key, last_event_time_iso: str):
          """Save the last event timestamp to GCS state file."""
          try:
              state = {'last_event_time': last_event_time_iso}
              blob = bucket.blob(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}")
              raise
      
      def load_seen(bucket, key):
          """Load the event keys written by the previous run (sidecar object)."""
          try:
              blob = bucket.blob(key + '.seen')
              if blob.exists():
                  return set(json.loads(blob.download_as_text()))
          except Exception as e:
              print(f"Warning: Could not load seen keys: {e}")
              raise
      
          return set()
      
      def save_seen(bucket, key, keys):
          """Save the event keys of this run to the sidecar object."""
          try:
              blob = bucket.blob(key + '.seen')
              blob.upload_from_string(
                  json.dumps(sorted(keys)),
                  content_type='application/json'
              )
              print(f"Saved {len(keys)} event keys for deduplication")
          except Exception as e:
              print(f"Warning: Could not save seen keys: {e}")
              raise
      
      def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
          """
          Fetch user activity logs from Workday Activity Logging API with pagination and rate limiting.
      
          Args:
              token: OAuth 2.0 access token
              start_time: Start time for log query
              end_time: End time for log query
              page_size: Number of records per page
              max_records: Maximum total records to fetch
      
          Returns:
              Tuple of (records list, newest_event_time ISO string)
          """
          base_url = f"https://{WORKDAY_HOST}/ccx/api/privacy/v1/{WORKDAY_TENANT}/activityLogging"
      
          headers = {
              'Authorization': f'Bearer {token}',
              'Accept': 'application/json',
              'User-Agent': 'GoogleSecOps-WorkdayCollector/1.0'
          }
      
          records = []
          newest_time = None
          page_num = 0
          backoff = 1.0
          offset = 0
      
          from_iso = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
          to_iso = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
      
          while True:
              page_num += 1
      
              if len(records) >= max_records:
                  print(f"Reached max_records limit ({max_records})")
                  break
      
              current_limit = min(page_size, max_records - len(records))
              url = f"{base_url}?from={from_iso}&to={to_iso}&limit={current_limit}&offset={offset}"
      
              try:
                  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
      
                  backoff = 1.0
      
                  if response.status != 200:
                      print(f"HTTP Error: {response.status}")
                      response_text = response.data.decode('utf-8')
                      print(f"Response body: {response_text}")
                      return [], None
      
                  data = json.loads(response.data.decode('utf-8'))
      
                  page_results = data.get('data', [])
      
                  if not page_results:
                      print(f"No more results (empty page)")
                      break
      
                  print(f"Page {page_num}: Retrieved {len(page_results)} events")
                  records.extend(page_results)
      
                  # Track newest event time
                  for event in page_results:
                      try:
                          event_time = event.get('requestTime') or event.get('systemAccount', {}).get('signOnDateTime')
                          if event_time:
                              if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                                  newest_time = event_time
                      except Exception as e:
                          print(f"Warning: Could not parse event time: {e}")
      
                  # Check pagination
                  total = data.get('total', 0)
                  if offset + len(page_results) >= total:
                      print(f"Reached end of results (offset={offset + len(page_results)}, total={total})")
                      break
      
                  offset += len(page_results)
      
              except Exception as e:
                  print(f"Error fetching logs: {e}")
                  return [], None
      
          print(f"Retrieved {len(records)} total records from {page_num} pages")
          return records, newest_time
      

    • 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 un job Cloud Scheduler

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

  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 workday-activity-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 (consigliato UTC)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento workday-activity-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 (consigliato)
Ogni 6 ore 0 */6 * * * Basso volume, elaborazione batch
Ogni giorno 0 0 * * * Raccolta dei dati storici

Testa l'integrazione

  1. Nella console Cloud Scheduler, individua il tuo job (workday-activity-collector-hourly).
  2. Fai clic su Forza esecuzione per attivare manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > workday-activity-collector > Log.
  4. Verifica che la funzione sia stata eseguita correttamente. Cerca:

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X events
    Wrote X records to gs://workday-activity-logs/workday/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  5. Controlla il bucket Cloud Storage (workday-activity-logs) per verificare che i log siano stati scritti.

Se visualizzi errori nei log:

  • HTTP 401: controlla le credenziali OAuth nelle variabili di ambiente
  • HTTP 403: verifica che l'ISU disponga delle autorizzazioni necessarie per il dominio di Log delle attività utente
  • HTTP 429: limitazione di frequenza. La funzione riproverà automaticamente con backoff
  • Impossibile ottenere il token di accesso: verifica che WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET e REFRESH_TOKEN siano corretti

Configura un feed in Google SecOps per importare i log delle attività utente di Workday

  1. Vai a Impostazioni > Gestione dei dati > Feed. Se utilizzi una piattaforma solo SIEM, vai a Impostazioni > 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 es. Workday User Activity Logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Attività utente Workday come Tipo di log.
  7. Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email univoco del account di servizio, 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 Cloud Storage con il percorso del prefisso:

      gs://workday-activity-logs/workday/
      
      • Sostituisci:
        • workday-activity-logs: il nome del tuo bucket Cloud Storage.
        • workday: prefisso o percorso facoltativo in cui vengono archiviati i log (lascia vuoto per la directory principale).
    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Non eliminare mai i file: non eliminare mai i file dall'origine (opzione consigliata per i test).
      • Elimina i file trasferiti e le directory vuote: elimina i file e le directory vuote dall'origine dopo il completamento di un recupero riuscito.
    • Età massima del file: includi i file modificati negli ultimi 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 di notizie 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 richiede due ruoli nel bucket Cloud Storage: Storage Object Viewer per leggere gli oggetti di log e un ruolo a livello di bucket per leggere i metadati del bucket.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (workday-activity-logs).
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'indirizzo email del account di servizio Google SecOps.
    • Assegna i ruoli: seleziona entrambe le seguenti opzioni:
      • Storage Object Viewer: legge gli oggetti di log.
      • Storage Legacy Bucket Reader: legge i metadati del bucket. Se hai selezionato l'opzione di eliminazione Elimina i file trasferiti e le directory vuote, seleziona Storage Legacy Bucket Writer, che concede anche l'autorizzazione di eliminazione.
  6. Fai clic su Salva.

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
Account_Locked__Disabled_or_Expired_label additional.fields Unita
Authentication_Type_label additional.fields Unita
Created_Moment_label additional.fields Unita
Current_OMS_Version_label additional.fields Unita
Device_is_Trusted_label additional.fields Unita
Elapsed_Time__Minutes_label additional.fields Unita
Failed_Sign_On_label additional.fields Unita
Hire_Date_for_Signon_Worker_label additional.fields Unita
Job_Family_for_Signon_Worker_label additional.fields Unita
Operating_System_label additional.fields Unita
Pay_Rate_type_label additional.fields Unita
Request_Originator_label additional.fields Unita
SAML_Identity_Provider_label additional.fields Unita
Sign-on_Time_label additional.fields Unita
Signoff_Time_label additional.fields Unita
Signon_Day_of_the_Week_label additional.fields Unita
Signon_Hour_of_Day_label additional.fields Unita
Tenant_Access_Read_Only_label additional.fields Unita
UI_Client_Type_label additional.fields Unita
Worker_Active_for_Signon_Worker_label additional.fields Unita
Worker_Type_for_Signon_Worker_label additional.fields Unita
awaiting_details_label additional.fields Unita
data_label additional.fields Unita
evp_nme_label additional.fields Unita
evpminus_label additional.fields Unita
signon_ip_address_label additional.fields Unita
@requestTime metadata.event_timestamp Analizzato come ISO8601
Sign-on_Time metadata.event_timestamp Analizzato come yyyy-MM-ddTHH:mm:ssZZ
event_type metadata.event_type Mappatura diretta
status_code metadata.product_event_type Mappatura diretta
tenant metadata.product_event_type Mappatura diretta
taskId metadata.product_log_id Mappatura diretta
User_Agent network.http.parsed_user_agent Rinominato/mappato
Browser_Type network.http.user_agent Mappatura diretta
User_Agent network.http.user_agent Mappatura diretta
userAgent network.http.user_agent Mappatura diretta
Session_ID network.session_id Mappatura diretta
sessionId network.session_id Mappatura diretta
Client_TLS_Version network.tls.version Rinominato/mappato
Worker principal.asset.hostname Mappatura diretta
ip principal.asset.ip Unita
ipaddress principal.asset.ip Unita
Worker principal.hostname Mappatura diretta
manager_name principal.hostname Mappatura diretta
ip principal.ip Unita
ipaddress principal.ip Unita
Signon_Worker_label principal.user.attribute.labels Unita
Supervisory_Organization_of_Signon_Worker_label principal.user.attribute.labels Unita
System_Account_label principal.user.attribute.labels Unita
Workday_Account_Sign_On_label principal.user.attribute.labels Unita
user_credentials_label principal.user.attribute.labels Unita
User_Name principal.user.email_addresses Unita
Role principal.user.user_display_name Mappatura diretta
manager_id principal.user.userid Mappatura diretta
Active_Session_label security_result.detection_fields Unita
Authentication_Failure_Message_label security_result.detection_fields Unita
Forgotten_Password_Reset_Request_label security_result.detection_fields Unita
Invalid_Credentials_label security_result.detection_fields Unita
Is_Device_Managed_label security_result.detection_fields Unita
MFA_Exempt_label security_result.detection_fields Unita
Password_Changed_label security_result.detection_fields Unita
Position_label security_result.detection_fields Unita
taskDisplayName security_result.summary Mappatura diretta
hostname target.hostname Mappatura diretta
CITY_NME target.location.city Mappatura diretta
Location target.location.city Mappatura diretta
COUNTRY_NME target.location.country_or_region Mappatura diretta
Location_Country_for_Signon_Worker target.location.country_or_region Mappatura diretta
Location_State target.location.state Mappatura diretta
STATE_NME target.location.state Mappatura diretta
Exempt_labels target.resource.attribute.labels Unita
browser_type_label target.resource.attribute.labels Unita
class_of_instance_label target.resource.attribute.labels Unita
device target.resource.resource_subtype Mappatura diretta
device1 target.resource.resource_subtype Mappatura diretta
device target.resource.type Mappatura: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
device1 target.resource.type Mappatura: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
BUSINESSGROUP_NME target.user.department Unita
email_name target.user.email_addresses Unita
EMPLOYEE_ID target.user.employee_id Mappatura diretta
Employee_ID target.user.employee_id Mappatura diretta
FIRST_NME target.user.first_name Mappatura diretta
EMPLOYEE_TYPE target.user.group_identifiers Unita
LAST_NME target.user.last_name Mappatura diretta
JOBTITLE_NME target.user.title Mappatura diretta
title target.user.title Mappatura diretta
target.id target.user.userid Mappatura diretta
N/D metadata.product_name Costante: WORKDAY_USER_ACTIVITY
N/D metadata.vendor_name Costante: WORKDAY_USER_ACTIVITY
N/D target.resource.type Costante: DEVICE

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.