Raccogliere i log delle attività di Rippling

Supportato in:

Questo documento spiega come importare i log attività di Rippling in Google Security Operations utilizzando Google Cloud Storage. Rippling è una piattaforma di gestione della forza lavoro che fornisce soluzioni per le RU, l'IT e la finanza, tra cui buste paga, benefit, onboarding dei dipendenti, gestione dei dispositivi e provisioning delle applicazioni. L'API Company Activity fornisce i log di controllo delle azioni amministrative e degli utenti sulla piattaforma Rippling.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto GCP con l'API Cloud Storage abilitata
  • Autorizzazioni per creare e gestire bucket GCS
  • Autorizzazioni per gestire le policy IAM nei bucket GCS
  • Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso con privilegi a Rippling (token API con accesso all'attività dell'azienda)

Crea un bucket Google Cloud Storage

  1. Vai alla console Google Cloud.
  2. Seleziona il tuo progetto o creane uno nuovo.
  3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
  4. Fai clic su Crea bucket.

  5. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio rippling-activity-logs).
    Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
    Località Seleziona la posizione (ad esempio, us-central1).
    Classe di archiviazione Standard (consigliato per i log a cui si accede di frequente)
    Controllo degli accessi Uniforme (consigliato)
    Strumenti di protezione (Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Raccogliere le credenziali API di Rippling

  1. Accedi a Rippling Admin.
  2. Vai a Search > Token API.
    • Percorso alternativo: Impostazioni > Impostazioni azienda > Token API.
  3. Fai clic su Genera token API.
  4. Fornisci i seguenti dettagli di configurazione:
    • Nome: inserisci un nome univoco e significativo (ad esempio, Google SecOps GCS Export).
    • Versione API: seleziona API di base (v1).
    • Ambiti/Autorizzazioni: attiva company:activity:read (obbligatorio per l'attività aziendale).
  5. Fai clic su Crea.

  6. Copia e salva il valore del token in un luogo sicuro. Lo utilizzerai come token di autenticazione.

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni di scrittura nel bucket GCS.

Crea service account

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service account.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome service account: inserisci rippling-logs-collector-sa.
    • Descrizione del service account: inserisci Service account for Cloud Run function to collect Rippling activity logs.
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo account di servizio l'accesso al progetto:
    1. Fai clic su Seleziona un ruolo.
    2. Cerca e seleziona Storage Object Admin.
    3. Fai clic su + Aggiungi un altro ruolo.
    4. Cerca e seleziona Cloud Run Invoker.
    5. Fai clic su + Aggiungi un altro ruolo.
    6. Cerca e seleziona Cloud Functions Invoker.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono necessari per:

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

Concedi autorizzazioni IAM sul bucket GCS

Concedi al account di servizio le autorizzazioni di scrittura sul bucket GCS:

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

Crea argomento Pub/Sub

Crea un argomento Pub/Sub a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

  1. Nella console Google Cloud, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci rippling-activity-trigger.
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API Rippling Company Activity e scriverli in GCS.

  1. Nella console Google Cloud, vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione (utilizza un editor in linea per creare una funzione).

  4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio rippling-activity-collector
    Regione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive.

  5. Nella sezione Trigger (facoltativo):

    1. Fai clic su + Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento rippling-activity-trigger.
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Controlla Identity and Access Management (IAM).

  7. Scorri verso il basso ed espandi Container, networking, sicurezza.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio rippling-logs-collector-sa.

  9. Vai alla scheda Container:

    1. Fai clic su Variabili e secret.
    2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
    Nome variabile Valore di esempio
    GCS_BUCKET rippling-activity-logs
    GCS_PREFIX rippling/activity/
    STATE_KEY rippling/activity/state.json
    RIPPLING_API_TOKEN your-api-token
    RIPPLING_ACTIVITY_URL https://api.rippling.com/platform/api/company_activity
    LIMIT 1000
    MAX_PAGES 10
    LOOKBACK_MINUTES 60
    END_LAG_SECONDS 120

  10. Scorri verso il basso nella scheda Variabili e secret fino a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti).

  11. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
    • Fai clic su Fine.

  12. Scorri fino a Ambiente di esecuzione:

    • Seleziona Predefinita (opzione consigliata).

  13. Nella sezione Scalabilità della revisione:

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

  14. Fai clic su Crea.

  15. Attendi la creazione del servizio (1-2 minuti).

  16. Dopo aver creato il servizio, si apre automaticamente l'editor di codice incorporato.

Aggiungi codice per la funzione

  1. Inserisci main in Entry point della funzione

  2. Nell'editor di codice incorporato, crea due file:

    • Primo file: main.py:
    import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time

# Initialize HTTP client
http = urllib3.PoolManager(timeout=urllib3.Timeout(connect=5.0, read=60.0), retries=False)

# Initialize Storage client
storage_client = storage.Client()

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

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

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'rippling/activity/')
    state_key = os.environ.get('STATE_KEY', 'rippling/activity/state.json')

    # Rippling API configuration
    api_token = os.environ.get('RIPPLING_API_TOKEN')
    activity_url = os.environ.get('RIPPLING_ACTIVITY_URL', 'https://api.rippling.com/platform/api/company_activity')
    limit = int(os.environ.get('LIMIT', '1000'))
    max_pages = int(os.environ.get('MAX_PAGES', '10'))
    lookback_minutes = int(os.environ.get('LOOKBACK_MINUTES', '60'))
    end_lag_seconds = int(os.environ.get('END_LAG_SECONDS', '120'))

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

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        # Load state (last processed timestamp and cursor)
        state = load_state(bucket, state_key)
        since_iso = state.get('since')
        next_cursor = state.get('next')

        # Calculate time window
        run_end = datetime.now(timezone.utc) - timedelta(seconds=end_lag_seconds)
        end_iso = run_end.replace(microsecond=0).isoformat().replace('+00:00', 'Z')

        if since_iso is None:
            since_iso = iso_from_epoch(time.time() - lookback_minutes * 60)
        else:
            try:
                since_iso = (parse_iso(since_iso) + timedelta(seconds=1)).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
            except Exception:
                since_iso = iso_from_epoch(time.time() - lookback_minutes * 60)

        print(f'Processing logs from {since_iso} to {end_iso}')

        run_ts_iso = end_iso
        pages = 0
        total = 0
        newest_ts = None
        pending_next = None

        # Fetch logs with pagination
        while pages < max_pages:
            params = {'limit': str(limit)}

            if next_cursor:
                params['next'] = next_cursor
            else:
                params['startDate'] = since_iso
                params['endDate'] = end_iso

            # Build URL with query parameters
            url = build_url(activity_url, params)

            # Fetch data from Rippling API
            headers = {
                'Authorization': f'Bearer {api_token}',
                'Accept': 'application/json'
            }

            # Implement exponential backoff for rate limiting
            backoff = 1.0
            max_retries = 3
            retry_count = 0

            while retry_count < max_retries:
                response = http.request('GET', url, headers=headers, timeout=60.0)

                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)
                    retry_count += 1
                    continue

                break

            if response.status != 200:
                print(f'Error: API returned status {response.status}')
                break

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

            # Write page to GCS
            write_to_gcs(bucket, prefix, data, run_ts_iso, pages)

            # Extract events
            events = data.get('events') or []
            total += len(events) if isinstance(events, list) else 0

            # Track newest timestamp
            if isinstance(events, list):
                for ev in events:
                    t = ev.get('timestamp') or ev.get('time') or ev.get('event_time')
                    if isinstance(t, str):
                        try:
                            dt_ts = parse_iso(t)
                            if newest_ts is None or dt_ts > newest_ts:
                                newest_ts = dt_ts
                        except Exception:
                            pass

            # Check for next page
            nxt = data.get('next')
            pages += 1

            if nxt:
                next_cursor = nxt
                pending_next = nxt
                continue
            else:
                pending_next = None
                break

        # Update state
        new_since_iso = (newest_ts or run_end).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
        save_state(bucket, state_key, {'since': new_since_iso, 'next': pending_next})

        print(f'Successfully processed {total} events across {pages} pages')
        print(f'Updated state: since={new_since_iso}, next={pending_next}')

    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: {str(e)}')
    return {}

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, separators=(',', ':')),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

def write_to_gcs(bucket, prefix, payload, run_ts_iso, page_index):
    """Write payload to GCS."""
    try:
        day_path = parse_iso(run_ts_iso).strftime('%Y/%m/%d')
        key = f"{prefix.strip('/')}/{day_path}/{run_ts_iso.replace(':', '').replace('-', '')}-page{page_index:05d}-company_activity.json"

        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(payload, separators=(',', ':')),
            content_type='application/json'
        )
        print(f'Wrote page {page_index} to {key}')
    except Exception as e:
        print(f'Error writing to GCS: {str(e)}')
        raise

def parse_iso(ts):
    """Parse ISO 8601 timestamp."""
    if ts.endswith('Z'):
        ts = ts[:-1] + '+00:00'
    return datetime.fromisoformat(ts)

def iso_from_epoch(sec):
    """Convert epoch seconds to ISO 8601 timestamp."""
    return datetime.fromtimestamp(sec, tz=timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z')

def build_url(base, params):
    """Build URL with query parameters."""
    if not params:
        return base
    query_string = '&'.join([f'{k}={v}' for k, v in params.items()])
    return f'{base}?{query_string}'
    • Secondo file: requirements.txt::
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

  4. Attendi il completamento del deployment (2-3 minuti).

Crea job Cloud Scheduler

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

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

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome rippling-activity-hourly
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza 0 * * * * (ogni ora, allo scoccare dell'ora)
    Fuso orario Seleziona il fuso orario (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento rippling-activity-trigger
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Opzioni di frequenza di pianificazione

  • Scegli la frequenza in base al volume dei log e ai requisiti di latenza:

    Frequenza Espressione cron Caso d'uso
    Ogni 5 minuti */5 * * * * Volume elevato, bassa latenza
    Ogni 15 minuti */15 * * * * Volume medio
    Ogni ora 0 * * * * Standard (consigliato)
    Ogni 6 ore 0 */6 * * * Volume basso, elaborazione batch
    Ogni giorno 0 0 * * * Raccolta dei dati storici

Testa il job di pianificazione

  1. Nella console Cloud Scheduler, trova il tuo job.
  2. Fai clic su Forza esecuzione per attivarlo manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > rippling-activity-collector > Log.
  4. Verifica che la funzione sia stata eseguita correttamente.
  5. Controlla il bucket GCS per verificare che i log siano stati scritti.

Recuperare il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal tuo bucket GCS. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'email del account di servizio

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Rippling Activity Logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Log delle attività di Rippling come Tipo di log.

  7. Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email del account di servizio univoco, ad esempio:

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

  8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel tuo bucket GCS.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'email del account di servizio Google SecOps.
    • Assegna ruoli: seleziona Visualizzatore oggetti Storage.

  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i log attività di Rippling

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Rippling Activity Logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Log delle attività di Rippling come Tipo di log.
  7. Fai clic su Avanti.

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

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

      gs://rippling-activity-logs/rippling/activity/

      • Sostituisci:

        • rippling-activity-logs: il nome del bucket GCS.
        • rippling/activity/: il percorso del prefisso/della cartella in cui sono archiviati i log (deve corrispondere alla variabile di ambiente GCS_PREFIX).

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
      • Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.

      • Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.

    • Età massima file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset (ad esempio, rippling.activity).

    • Etichette di importazione: etichetta facoltativa da applicare agli eventi di questo feed.

  9. Fai clic su Avanti.

  10. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Tabella di mappatura UDM

Campo log mappatura UDM Logica
accuracy_radius_label additional.fields Unita
company_label additional.fields Unita
finger_label additional.fields Unita
linked_events_key_label additional.fields Unita
owner_label additional.fields Unita
processing_state_label additional.fields Unita
spoke_label additional.fields Unita
subject_version_label additional.fields Unita
has_target extensions.auth.type Mappato: true → PHYSICAL
event_type metadata.event_type Mappato direttamente
name metadata.product_event_type Mappato direttamente
prop_id metadata.product_log_id Mappato direttamente
data.request_data.authentication_context.login_username network.email.mail_id Mappato direttamente
data.request_data.authentication_context.mfa_method network.http.method Mappato direttamente
data.request_data.authentication_context.login_agent network.http.user_agent Mappato direttamente
data.initiator.display_name principal.asset.hostname Mappato direttamente
data.search_string principal.asset.hostname Mappato direttamente
subjects.display_name principal.asset.hostname Mappato direttamente
data.initiator.display_name principal.hostname Mappato direttamente
data.search_string principal.hostname Mappato direttamente
subjects.display_name principal.hostname Mappato direttamente
data.initiator.icon principal.url Mappato direttamente
subjects.icon principal.url Mappato direttamente
data.initiator.email principal.user.email_addresses Unita
emailid principal.user.email_addresses Mappato: ^.+@.+$ → data.initiator.email
data._subject_group_name_cache principal.user.group_identifiers Unita
data._subject_role_name_cache principal.user.role_name Mappato direttamente
data.initiator.role principal.user.userid Mappato direttamente
data.subject_group principal.user.userid Mappato direttamente
subjects.instance principal.user.userid Mappato direttamente
data.event_reason.reason security_result.category_details Unita
data.event_reason.message security_result.summary Mappato direttamente
data.request_data.os src.platform_version Mappato direttamente
data.request_data.ip target.asset.ip Unita
data.request_data.ip target.ip Unita
data.request_data.city target.location.city Mappato direttamente
data.request_data.country target.location.country_or_region Mappato direttamente
geo_latitude target.location.region_latitude Rinominate/mappate
geo_longitude target.location.region_longitude Rinominate/mappate
data.request_data.state target.location.state Mappato direttamente
N/D extensions.auth.type Costante: PHYSICAL
N/D metadata.product_name Costante: RIPPLING_ACTIVITYLOGS
N/D metadata.vendor_name Costante: RIPPLING_ACTIVITYLOGS
N/D src.platform Costante: MAC
N/D target.resource.resource_type Costante: TASK

Log delle modifiche

Visualizza il log delle modifiche per questo parser

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