Coletar registros de atividade do Rippling

Compatível com:

Este documento explica como ingerir registros de atividades do Rippling no Google Security Operations usando o Amazon S3.

Antes de começar

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

  • Uma instância do Google SecOps.
  • Acesso privilegiado ao Rippling (token de API com acesso à Atividade da empresa).
  • Acesso privilegiado à AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

Conferir os pré-requisitos do Rippling

  1. Faça login no Rippling Admin.
  2. Abra Pesquisar > Tokens de API.
    Caminho alternativo: Configurações > Configurações da empresa > Tokens de API.
  3. Clique em Criar um token de API.
  4. Informe os seguintes detalhes de configuração:
    • Nome: insira um nome exclusivo e significativo (por exemplo, Google SecOps S3 Export).
    • Versão da API: API básica (v1)
    • Escopos/permissões: ative company:activity:read (necessário para Atividade da empresa).
  5. Clique em Criar e salve o valor do token em um local seguro. Ele será usado como um token do portador.

Configurar o bucket do AWS S3 e o IAM para o Google SecOps

  1. Crie um bucket do Amazon S3 seguindo este guia do usuário: Como criar um bucket
  2. Salve o Nome e a Região do bucket para referência futura (por exemplo, rippling-activity-logs).
  3. Crie um usuário seguindo este guia: Como criar um usuário do IAM.
  4. Selecione o usuário criado.
  5. Selecione a guia Credenciais de segurança.
  6. Clique em Criar chave de acesso na seção Chaves de acesso.
  7. Selecione Serviço de terceiros como Caso de uso.
  8. Clique em Próxima.
  9. Opcional: adicione uma tag de descrição.
  10. Clique em Criar chave de acesso.
  11. Clique em Fazer o download do arquivo CSV para salvar a chave de acesso e a chave de acesso secreta para referência futura.
  12. Clique em Concluído.
  13. Selecione a guia Permissões.
  14. Clique em Adicionar permissões na seção Políticas de permissões.
  15. Selecione Adicionar permissões.
  16. Selecione Anexar políticas diretamente.
  17. Pesquise a política AmazonS3FullAccess.
  18. Selecione a política.
  19. Clique em Próxima.
  20. Clique em Adicionar permissões

Configurar a política e o papel do IAM para uploads do S3

  1. No console da AWS, acesse IAM > Políticas.
  2. Clique em Criar política > guia JSON.
  3. Copie e cole a política a seguir.

  4. JSON da política (substitua os valores se você inseriu um bucket ou prefixo diferente):

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::rippling-activity-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::rippling-activity-logs/rippling/activity/state.json"
    }
  ]
}
````

  5. Clique em Próxima > Criar política.

  6. Acesse IAM > Funções > Criar função > Serviço da AWS > Lambda.

  7. Anexe a política recém-criada.

  8. Nomeie a função como WriteRipplingToS3Role e clique em Criar função.

Criar a função Lambda

  1. No console da AWS, acesse Lambda > Functions > Create function.
  2. Clique em Criar do zero.

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

    Configuração Valor
    Nome rippling_activity_to_s3
    Ambiente de execução Python 3.13
    Arquitetura x86_64
    Função de execução WriteRipplingToS3Role

  4. Depois que a função for criada, abra a guia Código, exclua o stub e cole o código a seguir (rippling_activity_to_s3.py).

    #!/usr/bin/env python3
# Lambda: Pull Rippling Company Activity logs to S3 (raw JSON, no transforms)

import os, json, time, urllib.parse
from urllib.request import Request, urlopen
from datetime import datetime, timezone, timedelta
import boto3

API_TOKEN = os.environ["RIPPLING_API_TOKEN"]
ACTIVITY_URL = os.environ.get("RIPPLING_ACTIVITY_URL", "https://api.rippling.com/platform/api/company_activity")
S3_BUCKET = os.environ["S3_BUCKET"]
S3_PREFIX = os.environ.get("S3_PREFIX", "rippling/activity/")
STATE_KEY = os.environ.get("STATE_KEY", "rippling/activity/state.json")

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

s3 = boto3.client("s3")

def _headers():
    return {"Authorization": f"Bearer {API_TOKEN}", "Accept": "application/json"}

def _get_state():
    try:
        obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        j = json.loads(obj["Body"].read())
        return {"since": j.get("since"), "next": j.get("next")}
    except Exception:
        return {"since": None, "next": None}

def _put_state(since_iso, next_cursor):
    body = json.dumps({"since": since_iso, "next": next_cursor}, separators=(",", ":")).encode("utf-8")
    s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=body)

def _get(url):
    req = Request(url, method="GET")
    for k, v in _headers().items():
        req.add_header(k, v)
    with urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode("utf-8"))

def _build_url(base, params):
    qs = urllib.parse.urlencode(params)
    return f"{base}?{qs}" if qs else base

def _parse_iso(ts):
    if ts.endswith("Z"):
        ts = ts[:-1] + "+00:00"
    return datetime.fromisoformat(ts)

def _iso_from_epoch(sec):
    return datetime.fromtimestamp(sec, tz=timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")

def _write(payload, run_ts_iso, page_index, source="company_activity"):
    day_path = _parse_iso(run_ts_iso).strftime("%Y/%m/%d")
    key = f"{S3_PREFIX.strip('/')}/{day_path}/{run_ts_iso.replace(':','').replace('-','')}-page{page_index:05d}-{source}.json"
    s3.put_object(Bucket=S3_BUCKET, Key=key, Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"))
    return key

def lambda_handler(event=None, context=None):
    state = _get_state()
    run_end = datetime.now(timezone.utc) - timedelta(seconds=END_LAG_SECONDS)
    end_iso = run_end.replace(microsecond=0).isoformat().replace("+00:00", "Z")

    since_iso = state["since"]
    next_cursor = state["next"]

    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)

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

    while pages < MAX_PAGES:
        params = {"limit": str(LIMIT)}
        if next_cursor:
            params["next"] = next_cursor
        else:
            params["startDate"] = since_iso
            params["endDate"] = end_iso

        url = _build_url(ACTIVITY_URL, params)
        data = _get(url)
        _write(data, run_ts_iso, pages)

        events = data.get("events") or data.get("items") or data.get("data") or []
        total += len(events) if isinstance(events, list) else 0

        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

        nxt = data.get("next") or data.get("next_cursor") or None
        pages += 1

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

    new_since_iso = (newest_ts or run_end).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    _put_state(new_since_iso, pending_next)

    return {"ok": True, "pages": pages, "events": total, "since": new_since_iso, "next": pending_next}

  5. Acesse Configuração > Variáveis de ambiente.

  6. Clique em Editar > Adicionar nova variável de ambiente.

  7. Insira as variáveis de ambiente fornecidas na tabela a seguir, substituindo os valores de exemplo pelos seus.

    Variáveis de ambiente

    Chave Valor de exemplo
    S3_BUCKET rippling-activity-logs
    S3_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

  8. Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > sua-função.

  9. Selecione a guia Configuração.

  10. No painel Configuração geral, clique em Editar.

  11. Mude Tempo limite para 5 minutos (300 segundos) e clique em Salvar.

Criar uma programação do EventBridge

  1. Acesse Amazon EventBridge > Scheduler > Criar programação.
  2. Informe os seguintes detalhes de configuração:
    • Programação recorrente: Taxa (1 hour).
    • Destino: sua função Lambda rippling_activity_to_s3.
    • Nome: rippling-activity-logs-1h.
  3. Clique em Criar programação.

(Opcional) Criar um usuário e chaves do IAM somente leitura para o Google SecOps

  1. No console da AWS, acesse IAM > Usuários > Adicionar usuários.
  2. Clique em Add users.
  3. Informe os seguintes detalhes de configuração:
    • Usuário: insira secops-reader.
    • Tipo de acesso: selecione Chave de acesso — Acesso programático.
  4. Clique em Criar usuário.
  5. Anexe a política de leitura mínima (personalizada): Usuários > secops-reader > Permissões > Adicionar permissões > Anexar políticas diretamente > Criar política.

  6. JSON:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::rippling-activity-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::rippling-activity-logs"
    }
  ]
}

  7. Name = secops-reader-policy.

  8. Clique em Criar política > pesquisar/selecionar > Próxima > Adicionar permissões.

  9. Crie uma chave de acesso para secops-reader: Credenciais de segurança > Chaves de acesso.

  10. Clique em Criar chave de acesso.

  11. Faça o download do .CSV. Cole esses valores no feed.

Configurar um feed no Google SecOps para ingerir registros de atividades do Rippling

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em + Adicionar novo feed.
  3. No campo Nome do feed, insira um nome para o feed (por exemplo, Rippling Activity Logs).
  4. Selecione Amazon S3 V2 como o Tipo de origem.
  5. Selecione Registros de atividade do Rippling como o Tipo de registro.
  6. Clique em Próxima.
  7. Especifique valores para os seguintes parâmetros de entrada:
    • URI do S3: s3://rippling-activity-logs/rippling/activity/
    • Opções de exclusão de fontes: selecione a opção de exclusão de acordo com sua preferência.
    • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
    • ID da chave de acesso: chave de acesso do usuário com acesso ao bucket do S3.
    • Chave de acesso secreta: chave secreta do usuário com acesso ao bucket do S3.
    • Namespace do recurso: rippling.activity
    • Opcional: Rótulos de ingestão: adicione o rótulo de ingestão.
  8. Clique em Próxima.
  9. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

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