Coletar registros de atividade do usuário do Workday

Compatível com:

Este documento explica como ingerir registros de atividade do usuário do Workday no Google Security Operations usando o Cloud Storage V2.

O Workday é uma plataforma de gerenciamento de capital humano (HCM) e finanças baseada na nuvem que gera registros de atividade do usuário, autenticação e auditoria de segurança. A API REST de registro de atividade do usuário oferece acesso programático a esses registros.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Uma instância do Google SecOps
  • Um projeto do Google Cloud com {storage_api} ativado
  • Permissões para criar e gerenciar buckets do Cloud Storage
  • Permissões para gerenciar políticas do Identity and Access Management (IAM) em buckets do Cloud Storage
  • Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Acesso privilegiado ao Workday com permissões de administrador de segurança ou equivalentes
  • O registro de atividade do usuário está ativado no seu locatário do Workday

Criar um bucket do Cloud Storage

  1. Acesse o console doGoogle Cloud .
  2. Selecione ou crie um projeto.
  3. No menu de navegação, acesse Cloud Storage > Buckets.
  4. Clique em Criar bucket.

  5. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nomeie seu bucket Insira um nome globalmente exclusivo, por exemplo, workday-activity-logs.
    Tipo de local Escolha de acordo com suas necessidades (região, birregional, multirregional)
    Local Selecione o local (por exemplo, us-central1).
    Classe de armazenamento Padrão (recomendado para registros acessados com frequência)
    Controle de acesso Uniforme (recomendado)
    Ferramentas de proteção Opcional: ativar o controle de versões de objetos ou a política de retenção

  6. Clique em Criar.

Coletar credenciais da API Workday

Ativar a geração de registros da atividade dos usuários

  1. Faça login no seu locatário do Workday como administrador de segurança.
  2. Na barra de pesquisa do Workday, procure e abra a tarefa Editar configuração do locatário - Sistema.
  3. Na seção Segurança, marque a caixa de seleção Ativar geração de registros de atividade do usuário.
  4. Clique em OK para salvar as alterações.

Criar um usuário do sistema de integração (ISU)

  1. Na barra de pesquisa do Workday, procure Criar usuário do sistema de integração.
  2. Insira um Nome de usuário (por exemplo, ISU_SIEM_Export) e uma Senha forte.
  3. Desmarque a caixa de seleção Exigir nova senha no próximo login.
  4. Clique em OK.
  5. Pesquise Criar grupo de segurança e crie um Grupo de segurança do sistema de integração (sem restrições).
  6. Adicione a ISU (ISU_SIEM_Export) a esse grupo de segurança.
  7. Pesquise Ver domínio e encontre o domínio Registro de atividade do usuário. Conceda acesso de Get ao grupo de segurança da ISU para esse domínio.
  8. Pesquise Ativar mudanças pendentes na política de segurança e ative as mudanças.

Registrar um cliente de API para OAuth

  1. Na barra de pesquisa do Workday, pesquise Registrar cliente da API para integrações.
  2. Insira um Nome do cliente (por exemplo, SIEM_OAuth_Client).
  3. Selecione a opção Tokens de atualização não expirados.
  4. Adicione o escopo: System (ou o escopo necessário para a API User Activity Logging).
  5. Clique em OK.
  6. Copie e salve os seguintes detalhes em um local seguro:
    • ID do cliente: o identificador do cliente da API.
    • Chave secreta do cliente: a chave secreta do cliente da API.
  7. Pesquise Gerenciar tokens de atualização para integrações.
  8. Selecione a conta de ISU (ISU_SIEM_Export).
  9. Gere um novo token de atualização para o cliente da API.
  10. Copie e salve o Refresh Token.

Determinar o URL do locatário

O endpoint de API é baseado no seu locatário do Workday. O formato é:

Componente Valor
Endpoint do token https://HOST/ccx/oauth2/TENANT/token
API Activity Logging https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

Verifique as permissões

Para verificar se a ISU tem as permissões necessárias:

  1. Faça login no Workday como administrador de segurança.
  2. Pesquise Ver usuário do sistema de integração e selecione o ISU (ISU_SIEM_Export).
  3. Verifique se o usuário é membro do grupo de segurança com acesso Get ao domínio Registro de atividade do usuário.
  4. Se o acesso estiver faltando, atualize o grupo de segurança e ative as mudanças pendentes na política de segurança.

Testar o acesso à API

  • Teste suas credenciais antes de prosseguir com a integração:

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

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar em um bucket do Cloud Storage e ser invocada pelo Pub/Sub.

Criar a conta de serviço

  1. No console do GCP, acesse IAM e administrador > Contas de serviço.
  2. Clique em Criar conta de serviço.
  3. Informe os seguintes detalhes de configuração:
    • Nome da conta de serviço: insira workday-activity-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Workday User Activity logs.
  4. Clique em Criar e continuar.
  5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
    1. Clique em Selecionar papel.
    2. Pesquise e selecione Administrador de objetos do Storage.
    3. Clique em + Adicionar outro papel.
    4. Pesquise e selecione Invocador do Cloud Run.
    5. Clique em + Adicionar outro papel.
    6. Pesquise e selecione Invocador de funções do Cloud.
  6. Clique em Continuar.
  7. Clique em Concluído.

Essas funções são necessárias para:

  • Administrador de objetos do Storage: grava registros no bucket do Cloud Storage e gerencia arquivos de estado.
  • Invocador do Cloud Run: permite que o Pub/Sub invoque a função.
  • Invocador do Cloud Functions: permite a invocação de funções.

Conceder permissões do IAM no bucket do Cloud Storage

Conceda à conta de serviço permissões de gravação no bucket do Cloud Storage da seguinte maneira:

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (por exemplo, workday-activity-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, workday-activity-collector-sa@your-project.iam.gserviceaccount.com).
    • Atribuir papéis: selecione Administrador de objetos do Storage.
  6. Clique em Salvar.

Criar um tópico do Pub/Sub

Crie o tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

  1. No Console do GCP, acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.
  3. Informe os seguintes detalhes de configuração:
    • ID do tópico: insira workday-activity-trigger.
    • Não mude as outras configurações.
  4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API Workday User Activity Logging e gravá-los no Cloud Storage.

  1. No console do GCP, acesse o Cloud Run.
  2. Clique em Escrever uma função.

  3. Na seção Configurar, forneça os seguintes detalhes de configuração:

    Configuração Valor
    Nome do serviço workday-activity-collector
    Região Selecione a região que corresponde ao seu bucket do Cloud Storage (por exemplo, us-central1).
    Ambiente de execução Selecione Python 3.12 ou uma versão mais recente.

  4. Na seção Gatilho, clique em Adicionar gatilho e selecione Gatilho do Pub/Sub.

  5. No painel Gatilho do Eventarc, forneça os seguintes detalhes de configuração:

    • Nome do acionador: mantenha o nome gerado ou insira um nome para o acionador.
    • Tipo de gatilho: selecione Fontes do Google.
    • Provedor de eventos: selecione Pub/Sub.
    • Tipo de evento: selecione google.cloud.pubsub.topic.v1.messagePublished.
    • Selecionar um tópico do Cloud Pub/Sub: escolha o tópico workday-activity-trigger.
    • Região: selecione a mesma região da função.
    • Conta de serviço: selecione a conta de serviço workday-activity-collector-sa.

  6. Clique em Salvar acionador.

  7. Na seção Autenticação:

    1. Selecione Exigir autenticação.
    2. Selecione Identity and Access Management (IAM).

  8. Acesse e abra Contêineres, rede, segurança.

  9. Acesse a guia Segurança:

    • Conta de serviço: selecione a conta de serviço workday-activity-collector-sa.

  10. Acesse a guia Contêineres:

    1. Clique em Variáveis e secrets.
    2. Clique em + Adicionar variável para cada variável de ambiente:
    Nome da variável Valor de exemplo Descrição
    GCS_BUCKET workday-activity-logs Nome do bucket do Cloud Storage
    GCS_PREFIX workday Prefixo para arquivos de registro
    STATE_KEY workday-state.json Caminho do estado, fora do prefixo do registro
    WORKDAY_HOST your-workday-host Nome do host do Workday
    WORKDAY_TENANT your-tenant Nome do locatário do Workday
    CLIENT_ID your-client-id ID do cliente OAuth
    CLIENT_SECRET your-client-secret Chave secreta do cliente OAuth
    REFRESH_TOKEN your-refresh-token Token de atualização do OAuth
    MAX_RECORDS 10000 Máximo de registros por execução
    PAGE_SIZE 100 Registros por página
    LOOKBACK_HOURS 24 Período de lookback inicial

  11. Na guia Variáveis e secrets, acesse Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos).

  12. Acesse a guia Configurações em Contêineres:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.

  13. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 ou ajuste com base na carga esperada.

  14. Clique em Criar.

  15. Aguarde a criação do serviço (1 a 2 minutos).

  16. Depois que o serviço for criado, o editor de código inline será aberto automaticamente.

Adicionar o código da função

  1. Insira main em Ponto de entrada da função.

  2. No editor de código inline, crie dois arquivos:

    • Primeiro arquivo: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 arquivo: requirements.txt:

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

  3. Clique em Implantar para salvar e implantar a função.

  4. Aguarde a conclusão da implantação (2 a 3 minutos).

Criar um job do Cloud Scheduler

O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

  1. No console do GCP, acesse o Cloud Scheduler.
  2. Clique em Criar job.

  3. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nome workday-activity-collector-hourly
    Região Selecione a mesma região da função do Cloud Run
    Frequência 0 * * * * (de hora em hora)
    Fuso horário Selecionar fuso horário (UTC recomendado)
    Tipo de destino Pub/Sub
    Tópico Selecione o tópico workday-activity-trigger.
    Corpo da mensagem {} (objeto JSON vazio)

  4. Clique em Criar.

Opções de frequência de programação

Escolha a frequência com base no volume de registros e nos requisitos de latência:

Frequência Expressão Cron Caso de uso
A cada 5 minutos */5 * * * * Alto volume e baixa latência
A cada 15 minutos */15 * * * * Volume médio
A cada hora 0 * * * * Padrão (recomendado)
A cada 6 horas 0 */6 * * * Baixo volume, processamento em lote
Diário 0 0 * * * Coleta de dados históricos

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job (workday-activity-collector-hourly).
  2. Clique em Forçar execução para acionar manualmente.
  3. Aguarde alguns segundos e acesse Cloud Run > Serviços > workday-activity-collector > Registros.

  4. Verifique se a função foi executada com sucesso. Procure:

    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. Verifique o bucket do Cloud Storage (workday-activity-logs) para confirmar se os registros foram gravados.

Se você encontrar erros nos registros:

  • HTTP 401: verifique as credenciais do OAuth nas variáveis de ambiente
  • HTTP 403: verifique se o ISU tem as permissões necessárias para o domínio de registro de atividades do usuário.
  • HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera.
  • Falha ao receber o token de acesso: verifique se WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET e REFRESH_TOKEN estão corretos.

Configurar um feed no Google SecOps para ingerir registros de atividade do usuário do Workday

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em Adicionar novo feed.
  3. Clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Workday User Activity Logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione Atividade do usuário do Workday como o Tipo de registro.

  7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:

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

  8. Copie o endereço de e-mail. Ela será usada na próxima etapa.

  9. Clique em Próxima.

  10. Especifique valores para os seguintes parâmetros de entrada:

    • URL do bucket de armazenamento: insira o URI do bucket do Cloud Storage com o caminho do prefixo:

      gs://workday-activity-logs/workday/
      • Substitua:
        • workday-activity-logs: o nome do bucket do Cloud Storage.
        • workday: prefixo ou caminho opcional em que os registros são armazenados (deixe em branco para a raiz).

    • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

      • Nunca excluir arquivos: nunca exclua arquivos da origem (recomendado para testes).
      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios da origem após a conclusão de uma busca bem-sucedida.

    • Idade máxima do arquivo: inclua arquivos modificados no último número de dias (o padrão é 180 dias).

    • Namespace do recurso: o namespace do recurso.

    • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

  11. Clique em Próxima.

  12. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa de duas funções no seu bucket do Cloud Storage: Leitor de objetos do Storage para ler os objetos de registro e uma função no nível do bucket para ler os metadados dele.

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (workday-activity-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
    • Atribuir papéis: selecione as duas opções a seguir:
      • Leitor de objetos do Storage: lê os objetos de registro.
      • Leitor de bucket legado do Storage: lê os metadados do bucket. Se você selecionou a opção de exclusão Excluir arquivos transferidos e diretórios vazios, escolha Gravador de bucket legados do Storage, que também concede a permissão de exclusão.

  6. Clique em Salvar.

Tabela de mapeamento de UDM

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

Ver o registro de alterações deste analisador

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.