Recopila registros de actividad del usuario de Workday

Compatible con:

En este documento, se explica cómo transferir registros de actividad del usuario de Workday a Google Security Operations con Cloud Storage V2.

Workday es una plataforma de administración de capital humano (HCM) y finanzas basada en la nube que genera registros de actividad de los usuarios, autenticación y auditoría de seguridad. La API de REST de registro de actividad del usuario proporciona acceso programático a estos registros.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Un proyecto de Google Cloud con la {storage_api} habilitada
  • Permisos para crear y administrar buckets de Cloud Storage
  • Permisos para administrar las políticas de Identity and Access Management (IAM) en los buckets de Cloud Storage
  • Permisos para crear servicios de Cloud Run, temas de Pub/Sub y trabajos de Cloud Scheduler
  • Acceso con privilegios a Workday con permisos de administrador de seguridad o equivalentes
  • Registro de actividad del usuario habilitado en tu usuario de Workday

Cree un bucket de Cloud Storage

  1. Ve a la consola deGoogle Cloud .
  2. Selecciona tu proyecto o crea uno nuevo.
  3. En el menú de navegación, ve a Cloud Storage > Buckets.
  4. Haz clic en Crear bucket.
  5. Proporciona los siguientes detalles de configuración:

    Parámetro de configuración Valor
    Asigna un nombre a tu bucket Ingresa un nombre único a nivel global (por ejemplo, workday-activity-logs).
    Tipo de ubicación Elige una en función de tus necesidades (región, birregional o multirregional).
    Ubicación Selecciona la ubicación (por ejemplo, us-central1).
    Clase de almacenamiento Estándar (recomendado para los registros a los que se accede con frecuencia)
    Control de acceso Uniforme (recomendado)
    Herramientas de protección Opcional: Habilita el control de versiones de objetos o la política de retención
  6. Haz clic en Crear.

Recopila las credenciales de la API de Workday

Habilitar registro de actividad del usuario

  1. Accede a tu usuario de Workday como administrador de seguridad.
  2. En la barra de búsqueda de Workday, busca y abre la tarea Edit Tenant Setup - System.
  3. En la sección Seguridad, selecciona la casilla de verificación Habilitar el registro de actividad del usuario.
  4. Haz clic en Aceptar para guardar los cambios.

Crea un usuario del sistema de integración (ISU)

  1. En la barra de búsqueda de Workday, busca Create Integration System User.
  2. Ingresa un Nombre de usuario (por ejemplo, ISU_SIEM_Export) y una Contraseña segura.
  3. Desmarca la casilla de verificación Solicitar una contraseña nueva en el próximo acceso.
  4. Haz clic en Aceptar.
  5. Busca Create Security Group y crea un Integration System Security Group (Unconstrained).
  6. Agrega el ISU (ISU_SIEM_Export) a este grupo de seguridad.
  7. Busca Ver dominio y localiza el dominio Registro de actividad del usuario. Otorga acceso de obtención al grupo de seguridad de ISU para este dominio.
  8. Busca Activate Pending Security Policy Changes y activa los cambios.

Registra un cliente de API para OAuth

  1. En la barra de búsqueda de Workday, busca Register API Client for Integrations.
  2. Ingresa un Nombre del cliente (por ejemplo, SIEM_OAuth_Client).
  3. Selecciona la opción Tokens de actualización no caducables.
  4. Agrega el permiso: System (o el permiso que se requiere para la API de User Activity Logging).
  5. Haz clic en Aceptar.
  6. Copia y guarda los siguientes datos en una ubicación segura:
    • ID de cliente: Es el identificador del cliente de API.
    • Secreto del cliente: Es el secreto del cliente de API.
  7. Busca Manage Refresh Tokens for Integrations.
  8. Selecciona la cuenta de ISU (ISU_SIEM_Export).
  9. Genera un nuevo token de actualización para el cliente de API.
  10. Copia y guarda el token de actualización.

Determina la URL del usuario

El extremo de API se basa en tu tenant de Workday. El formato es el siguiente:

Componente Valor
Extremo del token https://HOST/ccx/oauth2/TENANT/token
API de Activity Logging https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

Verifica los permisos

Para verificar que la ISU tenga los permisos necesarios, sigue estos pasos:

  1. Accede a Workday como administrador de seguridad.
  2. Busca View Integration System User y selecciona el ISU (ISU_SIEM_Export).
  3. Verifica que el usuario sea miembro del grupo de seguridad con acceso de obtención al dominio de registro de actividad del usuario.
  4. Si falta el acceso, actualiza el grupo de seguridad y activa los cambios pendientes en la política de seguridad.

Prueba el acceso a la API

  • Prueba tus credenciales antes de continuar con la integración:

    ```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 una cuenta de servicio para la función de Cloud Run

La función de Cloud Run necesita una cuenta de servicio con permisos para escribir en un bucket de Cloud Storage y que Pub/Sub la invoque.

Crea la cuenta de servicio

  1. En Google Cloud Console, ve a IAM y administración > Cuentas de servicio.
  2. Haz clic en Crear cuenta de servicio.
  3. Proporciona los siguientes detalles de configuración:
    • Nombre de la cuenta de servicio: Ingresa workday-activity-collector-sa.
    • Descripción de la cuenta de servicio: Ingresa Service account for Cloud Run function to collect Workday User Activity logs.
  4. Haz clic en Crear y continuar.
  5. En la sección Otorga acceso al proyecto a esta cuenta de servicio, agrega los siguientes roles:
    1. Haz clic en Selecciona un rol.
    2. Busca y selecciona Administrador de objetos de Storage.
    3. Haz clic en + Agregar otro rol.
    4. Busca y selecciona Invocador de Cloud Run.
    5. Haz clic en + Agregar otro rol.
    6. Busca y selecciona Invocador de Cloud Functions.
  6. Haz clic en Continuar.
  7. Haz clic en Listo.

Estos roles son necesarios para lo siguiente:

  • Administrador de objetos de Storage: Escribe registros en el bucket de Cloud Storage y administra archivos de estado
  • Invocador de Cloud Run: Permite que Pub/Sub invoque la función
  • Invocador de Cloud Functions: Permite la invocación de funciones

Otorga permisos de IAM en el bucket de Cloud Storage

Otorga a la cuenta de servicio permisos de escritura en el bucket de Cloud Storage de la siguiente manera:

  1. Ve a Cloud Storage > Buckets.
  2. Haz clic en el nombre de tu bucket (por ejemplo, workday-activity-logs).
  3. Ve a la pestaña Permisos.
  4. Haz clic en Otorgar acceso.
  5. Proporciona los siguientes detalles de configuración:
    • Agregar entidades: Ingresa el correo electrónico de la cuenta de servicio (por ejemplo, workday-activity-collector-sa@your-project.iam.gserviceaccount.com).
    • Asignar roles: Selecciona Administrador de objetos de Storage.
  6. Haz clic en Guardar.

Crea un tema de Pub/Sub

Crea el tema de Pub/Sub en el que Cloud Scheduler publicará y al que se suscribirá la función de Cloud Run.

  1. En Google Cloud Console, ve a Pub/Sub > Temas.
  2. Haz clic en Crear tema.
  3. Proporciona los siguientes detalles de configuración:
    • ID del tema: Ingresa workday-activity-trigger.
    • Deja el resto de la configuración con sus valores predeterminados.
  4. Haz clic en Crear.

Crea una función de Cloud Run para recopilar registros

La función de Cloud Run se activará con mensajes de Pub/Sub de Cloud Scheduler para recuperar registros de la API de Logging de actividad del usuario de Workday y escribirlos en Cloud Storage.

  1. En Google Cloud Console, ve a Cloud Run.
  2. Haz clic en Escribe una función.
  3. En la sección Configurar, proporciona los siguientes detalles de configuración:

    Parámetro de configuración Valor
    Nombre del servicio workday-activity-collector
    Región Selecciona la región que coincida con tu bucket de Cloud Storage (por ejemplo, us-central1)
    Tiempo de ejecución Selecciona Python 3.12 o una versión posterior.
  4. En la sección Activador, haz clic en Agregar activador y selecciona Activador de Pub/Sub.

  5. En el panel Activador de Eventarc, proporciona los siguientes detalles de configuración:

    • Nombre del activador: Conserva el nombre generado o ingresa un nombre para el activador.
    • Tipo de activador: Selecciona Fuentes de Google.
    • Proveedor de eventos: Selecciona Pub/Sub.
    • Tipo de evento: Selecciona google.cloud.pubsub.topic.v1.messagePublished.
    • Selecciona un tema de Cloud Pub/Sub: Elige el tema workday-activity-trigger.
    • Región: Selecciona la misma región que la función.
    • Cuenta de servicio: Selecciona la cuenta de servicio workday-activity-collector-sa.
  6. Haz clic en Guardar activador.

  7. En la sección Autenticación, haz lo siguiente:

    1. Selecciona Necesita autenticación.
    2. Selecciona Identity and Access Management (IAM).
  8. Navega a Contenedores, redes y seguridad y expande la sección.

  9. Ve a la pestaña Seguridad:

    • Cuenta de servicio: Selecciona la cuenta de servicio workday-activity-collector-sa.
  10. Ve a la pestaña Contenedores:

    1. Haz clic en Variables y secretos.
    2. Haz clic en + Agregar variable para cada variable de entorno:
    Nombre de la variable Valor de ejemplo Descripción
    GCS_BUCKET workday-activity-logs Nombre del bucket de Cloud Storage
    GCS_PREFIX workday Prefijo para los archivos de registro
    STATE_KEY workday-state.json Ruta de acceso de estado, fuera del prefijo de registro
    WORKDAY_HOST your-workday-host Nombre de host de Workday
    WORKDAY_TENANT your-tenant Nombre del usuario de Workday
    CLIENT_ID your-client-id ID de cliente de OAuth
    CLIENT_SECRET your-client-secret Secreto del cliente de OAuth
    REFRESH_TOKEN your-refresh-token Token de actualización de OAuth
    MAX_RECORDS 10000 Cantidad máxima de registros por ejecución
    PAGE_SIZE 100 Registros por página
    LOOKBACK_HOURS 24 Período de visualización inicial
  11. En la pestaña Variables y secretos, ve a Solicitudes:

    • Tiempo de espera de la solicitud: Ingresa 600 segundos (10 minutos).
  12. Ve a la pestaña Configuración en Contenedores:

    • En la sección Recursos, haz lo siguiente:
      • Memoria: Selecciona 512 MiB o más.
      • CPU: Selecciona 1.
  13. En la sección Escalamiento de revisión, haz lo siguiente:

    • Cantidad mínima de instancias: Ingresa 0.
    • Cantidad máxima de instancias: Ingresa 100 (o ajusta según la carga esperada).
  14. Haz clic en Crear.

  15. Espera a que se cree el servicio (de 1 a 2 minutos).

  16. Después de crear el servicio, se abrirá automáticamente el editor de código intercalado.

Agrega el código de la función

  1. Ingresa main en Punto de entrada de la función.
  2. En el editor de código intercalado, crea dos archivos:

    • Primer archivo: 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
      

    • Segundo archivo: requirements.txt:

      functions-framework==3.*
      google-cloud-storage==2.*
      urllib3>=2.0.0
      
  3. Haz clic en Implementar para guardar y, luego, implementar la función.

  4. Espera a que se complete la implementación (de 2 a 3 minutos).

Crea un trabajo de Cloud Scheduler

Cloud Scheduler publica mensajes en el tema de Pub/Sub a intervalos regulares, lo que activa la función de Cloud Run.

  1. En Google Cloud Console, ve a Cloud Scheduler.
  2. Haz clic en Crear trabajo.
  3. Proporciona los siguientes detalles de configuración:

    Parámetro de configuración Valor
    Nombre workday-activity-collector-hourly
    Región Selecciona la misma región que la función de Cloud Run
    Frecuencia 0 * * * * (cada hora, en punto)
    Zona horaria Selecciona la zona horaria (se recomienda UTC)
    Tipo de objetivo Pub/Sub
    Tema Selecciona el tema workday-activity-trigger.
    Cuerpo del mensaje {} (objeto JSON vacío)
  4. Haz clic en Crear.

Opciones de frecuencia de programación

Elige la frecuencia según el volumen de registros y los requisitos de latencia:

Frecuencia Expresión cron Caso de uso
Cada 5 minutos */5 * * * * Alto volumen y baja latencia
Cada 15 minutos */15 * * * * Volumen medio
Cada 1 hora 0 * * * * Estándar (recomendado)
Cada 6 horas 0 */6 * * * Procesamiento por lotes de bajo volumen
Diario 0 0 * * * Recopilación de datos históricos

Prueba la integración

  1. En la consola de Cloud Scheduler, busca tu trabajo (workday-activity-collector-hourly).
  2. Haz clic en Forzar ejecución para activar la tarea de forma manual.
  3. Espera unos segundos y ve a Cloud Run > Servicios > workday-activity-collector > Registros.
  4. Verifica que la función se haya ejecutado correctamente. Busca lo siguiente:

    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. Verifica el bucket de Cloud Storage (workday-activity-logs) para confirmar que se hayan escrito los registros.

Si ves errores en los registros, haz lo siguiente:

  • HTTP 401: Verifica las credenciales de OAuth en las variables de entorno.
  • HTTP 403: Verifica que la ISU tenga los permisos necesarios para el dominio de registro de actividad del usuario
  • HTTP 429: Límite de frecuencia: La función volverá a intentarlo automáticamente con retirada
  • No se pudo obtener el token de acceso: Verifica que WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET y REFRESH_TOKEN sean correctos

Configura un feed en Google SecOps para transferir registros de actividad del usuario de Workday

  1. Ve a Configuración > Administración de datos > Feeds. Si estás en una plataforma solo de SIEM, ve a Configuración > Feeds.
  2. Haz clic en Agregar feed nuevo.
  3. Haz clic en Configura un feed único.
  4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Workday User Activity Logs).
  5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.
  6. Selecciona Actividad del usuario de Workday como el Tipo de registro.
  7. Haz clic en Obtener cuenta de servicio. Se mostrará un correo electrónico único de la cuenta de servicio, por ejemplo:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. Copia esta dirección de correo electrónico. La necesitarás en el próximo paso.

  9. Haz clic en Siguiente.

  10. Especifica valores para los siguientes parámetros de entrada:

    • URL del bucket de almacenamiento: Ingresa el URI del bucket de Cloud Storage con la ruta del prefijo:

      gs://workday-activity-logs/workday/
      
      • Reemplaza:
        • workday-activity-logs: Es el nombre de tu bucket de Cloud Storage.
        • workday: Es el prefijo o la ruta de acceso opcional en la que se almacenan los registros (deja este campo vacío para la raíz).
    • Opción de eliminación del código fuente: Selecciona la opción de eliminación según tu preferencia:

      • No borrar archivos nunca: Nunca borra archivos de la fuente (se recomienda para las pruebas).
      • Borrar los archivos transferidos y los directorios vacíos: Borra los archivos y los directorios vacíos de la fuente después de que se complete correctamente la recuperación.
    • Antigüedad máxima del archivo: Incluye los archivos modificados en la última cantidad de días (el valor predeterminado es 180 días).

    • Espacio de nombres del recurso: Es el espacio de nombres del recurso.

    • Etiquetas de transferencia: Es la etiqueta que se aplicará a los eventos de este feed.

  11. Haz clic en Siguiente.

  12. Revisa la configuración de tu feed de noticias nuevo en la pantalla Finalizar y, luego, haz clic en Enviar.

Otorga permisos de IAM a la cuenta de servicio de Google SecOps

La cuenta de servicio de Google SecOps necesita dos roles en tu bucket de Cloud Storage: Visualizador de objetos de Storage para leer los objetos de registro y un rol a nivel del bucket para leer los metadatos del bucket.

  1. Ve a Cloud Storage > Buckets.
  2. Haz clic en el nombre de tu bucket (workday-activity-logs).
  3. Ve a la pestaña Permisos.
  4. Haz clic en Otorgar acceso.
  5. Proporciona los siguientes detalles de configuración:
    • Agregar entidades: Pega el correo electrónico de la cuenta de servicio de Google SecOps.
    • Asignar roles: Selecciona las siguientes opciones:
      • Visualizador de objetos de Storage: Lee los objetos de registro.
      • Lector de buckets heredados de Storage: Lee los metadatos del bucket. Si seleccionaste la opción de borrado Borrar los archivos transferidos y los directorios vacíos, selecciona Escritor de buckets heredados de Storage, que también otorga el permiso de borrado.
  6. Haz clic en Guardar.

Tabla de asignación de UDM

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

Registro de cambios

Consulta el registro de cambios de este analizador

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.