Collecter les journaux d'activité des utilisateurs Workday

Compatible avec :

Ce document explique comment ingérer les journaux d'activité des utilisateurs Workday dans Google Security Operations à l'aide de Cloud Storage V2.

Workday est une plate-forme dans le cloud de gestion du capital humain (HCM) et de gestion financière qui génère des journaux d'activité de l'utilisateur, d'authentification et d'audit de sécurité. L'API REST de journalisation de l'activité utilisateur permet d'accéder à ces journaux de manière programmatique.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps
  • Un projet Google Cloud avec {storage_api} activé
  • Autorisations pour créer et gérer des buckets Cloud Storage
  • Autorisations permettant de gérer les stratégies Identity and Access Management (IAM) sur les buckets Cloud Storage
  • Autorisations permettant de créer des services Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler
  • Accès privilégié à Workday avec des autorisations d'administrateur de la sécurité ou équivalentes
  • La journalisation de l'activité des utilisateurs est activée dans votre locataire Workday.

Créer un bucket Cloud Storage

  1. Accédez à la consoleGoogle Cloud .
  2. Sélectionnez votre projet ou créez-en un.
  3. Dans le menu de navigation, accédez à Cloud Storage > Buckets.
  4. Cliquez sur Créer un bucket.

  5. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Attribuez un nom au bucket Saisissez un nom unique (par exemple, workday-activity-logs).
    Type d'emplacement Choisissez en fonction de vos besoins (région, birégion, multirégion).
    Emplacement Sélectionnez l'emplacement (par exemple, us-central1).
    Classe de stockage Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
    Access control (Contrôle des accès) Uniforme (recommandé)
    Outils de protection Facultatif : Activez la gestion des versions des objets ou la règle de conservation.

  6. Cliquez sur Créer.

Collecter les identifiants de l'API Workday

Activer la journalisation de l'activité des utilisateurs

  1. Connectez-vous à votre locataire Workday en tant qu'administrateur de la sécurité.
  2. Dans la barre de recherche Workday, recherchez et ouvrez la tâche Edit Tenant Setup - System (Modifier la configuration du locataire – Système).
  3. Dans la section Sécurité, cochez la case Activer la journalisation de l'activité des utilisateurs.
  4. Cliquez sur OK pour enregistrer les modifications.

Créer un utilisateur du système d'intégration

  1. Dans la barre de recherche Workday, recherchez Create Integration System User.
  2. Saisissez un nom d'utilisateur (par exemple, ISU_SIEM_Export) et un mot de passe sécurisé.
  3. Décochez la case Exiger un nouveau mot de passe à la prochaine connexion.
  4. Cliquez sur OK.
  5. Recherchez Create Security Group (Créer un groupe de sécurité), puis créez un Integration System Security Group (Unconstrained) (Groupe de sécurité du système d'intégration (sans restriction)).
  6. Ajoutez l'unité d'identité et de signature (ISU_SIEM_Export) à ce groupe de sécurité.
  7. Recherchez Afficher le domaine et localisez le domaine Journalisation de l'activité des utilisateurs. Accordez l'accès Obtenir au groupe de sécurité ISU pour ce domaine.
  8. Recherchez Activer les modifications en attente de la stratégie de sécurité, puis activez les modifications.

Enregistrer un client API pour OAuth

  1. Dans la barre de recherche Workday, recherchez Register API Client for Integrations (Enregistrer le client API pour les intégrations).
  2. Saisissez un nom de client (par exemple, SIEM_OAuth_Client).
  3. Sélectionnez l'option Jetons d'actualisation sans expiration.
  4. Ajoutez le champ d'application : Système (ou le champ d'application requis pour l'API User Activity Logging).
  5. Cliquez sur OK.
  6. Copiez et enregistrez les informations suivantes dans un emplacement sécurisé :
    • ID client : identifiant du client API.
    • Code secret du client : code secret du client API.
  7. Recherchez Gérer les jetons d'actualisation pour les intégrations.
  8. Sélectionnez le compte ISU (ISU_SIEM_Export).
  9. Générez un jeton d'actualisation pour le client API.
  10. Copiez et enregistrez le jeton d'actualisation.

Déterminer l'URL du locataire

Le point de terminaison de l'API dépend de votre locataire Workday. Il a le format suivant :

Composant Valeur
Point de terminaison du jeton https://HOST/ccx/oauth2/TENANT/token
API Activity Logging https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

Vérifier les autorisations

Pour vérifier que l'ISU dispose des autorisations requises :

  1. Connectez-vous à Workday en tant qu'administrateur de la sécurité.
  2. Recherchez Afficher l'utilisateur du système d'intégration et sélectionnez l'utilisateur du système d'intégration (ISU_SIEM_Export).
  3. Vérifiez que l'utilisateur est membre du groupe de sécurité avec un accès Get au domaine User Activity Logging (Journalisation de l'activité des utilisateurs).
  4. Si l'accès est manquant, mettez à jour le groupe de sécurité et activez les modifications en attente de la stratégie de sécurité.

Tester l'accès à l'API

  • Testez vos identifiants avant de procéder à l'intégration :

    ```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"
```

Créer un compte de service pour la fonction Cloud Run

La fonction Cloud Run a besoin d'un compte de service disposant des autorisations nécessaires pour écrire dans un bucket Cloud Storage et être appelée par Pub/Sub.

Créer le compte de service

  1. Dans la console GCP, accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Nom du compte de service : saisissez workday-activity-collector-sa.
    • Description du compte de service : saisissez Service account for Cloud Run function to collect Workday User Activity logs.
  4. Cliquez sur Créer et continuer.
  5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
    1. Cliquez sur Sélectionner un rôle.
    2. Recherchez et sélectionnez Administrateur des objets Storage.
    3. Cliquez sur + Ajouter un autre rôle.
    4. Recherchez et sélectionnez Demandeur Cloud Run.
    5. Cliquez sur + Ajouter un autre rôle.
    6. Recherchez et sélectionnez Demandeur Cloud Functions.
  6. Cliquez sur Continuer.
  7. Cliquez sur OK.

Ces rôles sont requis pour :

  • Administrateur des objets Storage : écrire des journaux dans le bucket Cloud Storage et gérer les fichiers d'état
  • Demandeur Cloud Run : autorise Pub/Sub à appeler la fonction
  • Demandeur Cloud Functions : autorise l'appel de fonctions

Accorder des autorisations IAM sur le bucket Cloud Storage

Accordez au compte de service des autorisations d'écriture sur le bucket Cloud Storage comme suit :

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (par exemple, workday-activity-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (par exemple, workday-activity-collector-sa@your-project.iam.gserviceaccount.com).
    • Attribuer des rôles : sélectionnez Administrateur des objets Storage.
  6. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

Créez le sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.

  1. Dans la console GCP, accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. Fournissez les informations de configuration suivantes :
    • ID du sujet : saisissez workday-activity-trigger.
    • Conservez les valeurs par défaut des autres paramètres.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

La fonction Cloud Run sera déclenchée par les messages Pub/Sub de Cloud Scheduler pour extraire les journaux de l'API Workday User Activity Logging et les écrire dans Cloud Storage.

  1. Dans la console GCP, accédez à Cloud Run.
  2. Cliquez sur Écrire une fonction.

  3. Dans la section Configurer, fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom du service workday-activity-collector
    Région Sélectionnez la région correspondant à votre bucket Cloud Storage (par exemple, us-central1).
    Durée d'exécution Sélectionnez Python 3.12 ou version ultérieure.

  4. Dans la section Déclencheur, cliquez sur Ajouter un déclencheur, puis sélectionnez Déclencheur Pub/Sub.

  5. Dans le volet Déclencheur Eventarc, fournissez les informations de configuration suivantes :

    • Nom du déclencheur : conservez le nom généré ou saisissez-en un.
    • Type de déclencheur : sélectionnez Sources Google.
    • Fournisseur d'événements : sélectionnez Pub/Sub.
    • Type d'événement : sélectionnez google.cloud.pubsub.topic.v1.messagePublished.
    • Sélectionner un sujet Cloud Pub/Sub : choisissez le sujet workday-activity-trigger.
    • Région : sélectionnez la même région que la fonction.
    • Compte de service : sélectionnez le compte de service workday-activity-collector-sa.

  6. Cliquez sur Enregistrer le déclencheur.

  7. Dans la section Authentification :

    1. Sélectionnez Exiger l'authentification.
    2. Sélectionnez Identity and Access Management (IAM).

  8. Accédez à Conteneurs, mise en réseau, sécurité et développez cette section.

  9. Accédez à l'onglet Sécurité :

    • Compte de service : sélectionnez le compte de service workday-activity-collector-sa.

  10. Accédez à l'onglet Conteneurs :

    1. Cliquez sur Variables et secrets.
    2. Cliquez sur + Ajouter une variable pour chaque variable d'environnement :
    Nom de la variable Exemple de valeur Description
    GCS_BUCKET workday-activity-logs Nom du bucket Cloud Storage
    GCS_PREFIX workday Préfixe des fichiers journaux
    STATE_KEY workday-state.json Chemin d'état, en dehors du préfixe du journal
    WORKDAY_HOST your-workday-host Nom d'hôte Workday
    WORKDAY_TENANT your-tenant Nom du locataire Workday
    CLIENT_ID your-client-id ID client OAuth
    CLIENT_SECRET your-client-secret Code secret du client OAuth
    REFRESH_TOKEN your-refresh-token Jeton d'actualisation OAuth
    MAX_RECORDS 10000 Nombre maximal d'enregistrements par exécution
    PAGE_SIZE 100 Enregistrements par page
    LOOKBACK_HOURS 24 Période d'analyse initiale

  11. Dans l'onglet Variables et secrets, accédez à Demandes :

    • Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).

  12. Accédez à l'onglet Paramètres dans Conteneurs :

    • Dans la section Ressources :
      • Mémoire : sélectionnez 512 Mio ou plus.
      • CPU : sélectionnez 1.

  13. Dans la section Scaling de révision :

    • Nombre minimal d'instances : saisissez 0.
    • Nombre maximal d'instances : saisissez 100 (ou ajustez en fonction de la charge attendue).

  14. Cliquez sur Créer.

  15. Attendez que le service soit créé (une à deux minutes).

  16. Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter le code de la fonction

  1. Saisissez main dans Point d'entrée de la fonction.

  2. Dans l'éditeur de code intégré, créez deux fichiers :

    • Premier fichier : 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

    • Deuxième fichier : requirements.txt: :

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

  3. Cliquez sur Déployer pour enregistrer et déployer la fonction.

  4. Attendez la fin du déploiement (deux à trois minutes).

Créer une tâche Cloud Scheduler

Cloud Scheduler publie des messages sur le sujet Pub/Sub à intervalles réguliers, ce qui déclenche la fonction Cloud Run.

  1. Dans la console GCP, accédez à Cloud Scheduler.
  2. Cliquez sur Créer une tâche.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom workday-activity-collector-hourly
    Région Sélectionnez la même région que la fonction Cloud Run.
    Fréquence 0 * * * * (toutes les heures, à l'heure pile)
    Fuseau horaire Sélectionnez un fuseau horaire (UTC recommandé).
    Type de cible Pub/Sub
    Sujet Sélectionnez le thème workday-activity-trigger.
    Corps du message {} (objet JSON vide)

  4. Cliquez sur Créer.

Options de fréquence de programmation

Choisissez la fréquence en fonction du volume de journaux et des exigences de latence :

Fréquence Expression Cron Cas d'utilisation
Toutes les 5 minutes */5 * * * * Volume élevé, faible latence
Toutes les 15 minutes */15 * * * * Volume moyen
Toutes les heures 0 * * * * Standard (recommandé)
Toutes les 6 heures 0 */6 * * * Traitement par lot à faible volume
Tous les jours 0 0 * * * Collecte de données historiques

Tester l'intégration

  1. Dans la console Cloud Scheduler, recherchez votre job (workday-activity-collector-hourly).
  2. Cliquez sur Exécuter de force pour déclencher manuellement.
  3. Patientez quelques secondes, puis accédez à Cloud Run > Services > workday-activity-collector > Journaux.

  4. Vérifiez que la fonction s'est exécutée correctement. Par exemple :

    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. Vérifiez le bucket Cloud Storage (workday-activity-logs) pour confirmer que les journaux ont été écrits.

Si des erreurs s'affichent dans les journaux :

  • HTTP 401 : vérifiez les identifiants OAuth dans les variables d'environnement.
  • HTTP 403 : vérifiez que l'ISU dispose des autorisations requises pour le domaine de journalisation de l'activité des utilisateurs.
  • HTTP 429 : limitation du débit. La fonction effectuera automatiquement une nouvelle tentative avec un intervalle entre les tentatives.
  • Échec de l'obtention du jeton d'accès : vérifiez que WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET et REFRESH_TOKEN sont corrects.

Configurer un flux dans Google SecOps pour ingérer les journaux d'activité des utilisateurs Workday

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Ajouter un flux.
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Workday User Activity Logs).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez Activité des utilisateurs Workday comme Type de journal.

  7. Cliquez sur Obtenir un compte de service. Une adresse e-mail de compte de service unique s'affiche, par exemple :

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

  8. Copiez cette adresse e-mail. Vous l'utiliserez à l'étape suivante.

  9. Cliquez sur Suivant.

  10. Spécifiez les valeurs des paramètres d'entrée suivants :

    • URL du bucket de stockage : saisissez l'URI du bucket Cloud Storage avec le chemin d'accès au préfixe :

      gs://workday-activity-logs/workday/
      • Remplacez :
        • workday-activity-logs : nom de votre bucket Cloud Storage.
        • workday : préfixe ou chemin d'accès facultatif où les journaux sont stockés (laissez vide pour la racine).

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Ne jamais supprimer les fichiers : ne supprimez jamais les fichiers de la source (recommandé pour les tests).
      • Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides de la source une fois l'extraction réussie.

    • Âge maximal du fichier : incluez les fichiers modifiés au cours des derniers jours (180 jours par défaut).

    • Espace de noms de l'élément : espace de noms de l'élément.

    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux.

  11. Cliquez sur Suivant.

  12. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps a besoin de deux rôles sur votre bucket Cloud Storage : Lecteur des objets Storage pour lire les objets journaux et un rôle au niveau du bucket pour lire les métadonnées du bucket.

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (workday-activity-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
    • Attribuer des rôles : sélectionnez les deux rôles suivants :
      • Lecteur des objets Storage : lit les objets journaux.
      • Lecteur des anciens buckets Storage : lit les métadonnées du bucket. Si vous avez sélectionné l'option de suppression Supprimer les fichiers transférés et les répertoires vides, sélectionnez plutôt Rédacteur de bucket Storage hérité, qui accorde également l'autorisation de suppression.

  6. Cliquez sur Enregistrer.

Table de mappage UDM

Champ de journal Mappage UDM Logique
Account_Locked__Disabled_or_Expired_label additional.fields Fusionné
Authentication_Type_label additional.fields Fusionné
Created_Moment_label additional.fields Fusionné
Current_OMS_Version_label additional.fields Fusionné
Device_is_Trusted_label additional.fields Fusionné
Elapsed_Time__Minutes_label additional.fields Fusionné
Failed_Sign_On_label additional.fields Fusionné
Hire_Date_for_Signon_Worker_label additional.fields Fusionné
Job_Family_for_Signon_Worker_label additional.fields Fusionné
Operating_System_label additional.fields Fusionné
Pay_Rate_type_label additional.fields Fusionné
Request_Originator_label additional.fields Fusionné
SAML_Identity_Provider_label additional.fields Fusionné
Sign-on_Time_label additional.fields Fusionné
Signoff_Time_label additional.fields Fusionné
Signon_Day_of_the_Week_label additional.fields Fusionné
Signon_Hour_of_Day_label additional.fields Fusionné
Tenant_Access_Read_Only_label additional.fields Fusionné
UI_Client_Type_label additional.fields Fusionné
Worker_Active_for_Signon_Worker_label additional.fields Fusionné
Worker_Type_for_Signon_Worker_label additional.fields Fusionné
awaiting_details_label additional.fields Fusionné
data_label additional.fields Fusionné
evp_nme_label additional.fields Fusionné
evpminus_label additional.fields Fusionné
signon_ip_address_label additional.fields Fusionné
@requestTime metadata.event_timestamp Analysé en tant que ISO8601
Sign-on_Time metadata.event_timestamp Analysé en tant que yyyy-MM-ddTHH:mm:ssZZ
event_type metadata.event_type Mappé directement
status_code metadata.product_event_type Mappé directement
tenant metadata.product_event_type Mappé directement
taskId metadata.product_log_id Mappé directement
User_Agent network.http.parsed_user_agent Renommé/Mappé
Browser_Type network.http.user_agent Mappé directement
User_Agent network.http.user_agent Mappé directement
userAgent network.http.user_agent Mappé directement
Session_ID network.session_id Mappé directement
sessionId network.session_id Mappé directement
Client_TLS_Version network.tls.version Renommé/Mappé
Worker principal.asset.hostname Mappé directement
ip principal.asset.ip Fusionné
ipaddress principal.asset.ip Fusionné
Worker principal.hostname Mappé directement
manager_name principal.hostname Mappé directement
ip principal.ip Fusionné
ipaddress principal.ip Fusionné
Signon_Worker_label principal.user.attribute.labels Fusionné
Supervisory_Organization_of_Signon_Worker_label principal.user.attribute.labels Fusionné
System_Account_label principal.user.attribute.labels Fusionné
Workday_Account_Sign_On_label principal.user.attribute.labels Fusionné
user_credentials_label principal.user.attribute.labels Fusionné
User_Name principal.user.email_addresses Fusionné
Role principal.user.user_display_name Mappé directement
manager_id principal.user.userid Mappé directement
Active_Session_label security_result.detection_fields Fusionné
Authentication_Failure_Message_label security_result.detection_fields Fusionné
Forgotten_Password_Reset_Request_label security_result.detection_fields Fusionné
Invalid_Credentials_label security_result.detection_fields Fusionné
Is_Device_Managed_label security_result.detection_fields Fusionné
MFA_Exempt_label security_result.detection_fields Fusionné
Password_Changed_label security_result.detection_fields Fusionné
Position_label security_result.detection_fields Fusionné
taskDisplayName security_result.summary Mappé directement
hostname target.hostname Mappé directement
CITY_NME target.location.city Mappé directement
Location target.location.city Mappé directement
COUNTRY_NME target.location.country_or_region Mappé directement
Location_Country_for_Signon_Worker target.location.country_or_region Mappé directement
Location_State target.location.state Mappé directement
STATE_NME target.location.state Mappé directement
Exempt_labels target.resource.attribute.labels Fusionné
browser_type_label target.resource.attribute.labels Fusionné
class_of_instance_label target.resource.attribute.labels Fusionné
device target.resource.resource_subtype Mappé directement
device1 target.resource.resource_subtype Mappé directement
device target.resource.type Mappé : (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
device1 target.resource.type Mappé : (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
BUSINESSGROUP_NME target.user.department Fusionné
email_name target.user.email_addresses Fusionné
EMPLOYEE_ID target.user.employee_id Mappé directement
Employee_ID target.user.employee_id Mappé directement
FIRST_NME target.user.first_name Mappé directement
EMPLOYEE_TYPE target.user.group_identifiers Fusionné
LAST_NME target.user.last_name Mappé directement
JOBTITLE_NME target.user.title Mappé directement
title target.user.title Mappé directement
target.id target.user.userid Mappé directement
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

Journal des modifications

Afficher le journal des modifications pour ce parseur

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.