Coletar registros do 1Password

Compatível com:

Este documento explica como ingerir registros do 1Password no Google Security Operations usando o Google Cloud Storage V2.

O 1Password é uma plataforma de gerenciamento de senhas que ajuda as equipes a armazenar, compartilhar e gerenciar credenciais, secrets e informações sensíveis com segurança. A API 1Password Events fornece acesso a dados de tentativa de login e uso de itens da sua conta do 1Password Business.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

  • Uma instância do Google SecOps
  • Um projeto do GCP com a API Cloud Storage ativada
  • Permissões para criar e gerenciar buckets do GCS
  • Permissões para gerenciar políticas do IAM em buckets do GCS
  • Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Uma conta do 1Password Business
  • Uma função de proprietário ou administrador na sua conta do 1Password

Criar um bucket do Google Cloud Storage

  1. Acesse o Console do Google Cloud.
  2. Selecione seu projeto ou crie um novo.
  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 exclusivo globalmente, por exemplo, onepassword-secops-logs.
    Tipo de local Escolha com base nas 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 do 1Password

Configurar a integração de relatórios de eventos

  1. Faça login na sua conta em 1Password.com.
  2. Selecione Integrações na barra lateral.
  3. Se você já tiver configurado outras integrações, selecione Diretório na página Integrações.
  4. Na seção Relatórios de eventos, selecione Outro.
  5. No campo Nome, insira um nome para a integração (por exemplo, Google SecOps Integration).
  6. Clique em Adicionar integração.

Criar um token do portador

  1. Na página de detalhes da integração, clique em Adicionar um token de portador.
  2. Informe os seguintes detalhes de configuração:
    • Nome do token: insira um nome descritivo (por exemplo, SecOps GCS Collector - SignIn and ItemUsage).
    • Expira após: selecione o período de expiração de acordo com sua preferência. Selecione Nunca para um token que não expira ou escolha 30 dias, 90 dias ou 180 dias.
    • Eventos a serem denunciados: marque as seguintes caixas de seleção:
      • Tentativas de login
      • Eventos de uso de itens
  3. Clique em Emitir token.

  4. Na página Salvar seu token, clique em Salvar no 1Password ou copie o token e salve em um local seguro.

  5. Clique em Ver detalhes da integração para confirmar se ela está ativa.

Determinar o URL base da API Events

O URL base depende do servidor que hospeda sua conta do 1Password:

Se a conta estiver hospedada em O URL de base da API Events é
1password.com https://events.1password.com
ent.1password.com https://events.ent.1password.com
1password.ca https://events.1password.ca
1password.eu https://events.1password.eu

Testar o acesso à API

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

    # Replace with your actual bearer token and base URL
BEARER_TOKEN="<your-bearer-token>"
API_BASE="https://events.1password.com"

# Test API access using the introspect endpoint
curl -v \
    -H "Authorization: Bearer $BEARER_TOKEN" \
    "$API_BASE/api/v2/auth/introspect"

Uma resposta bem-sucedida retorna um objeto JSON com o campo features listando os tipos de eventos que seu token pode acessar (por exemplo, ["signinattempts", "itemusages"]).

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 no bucket do GCS e ser invocada pelo Pub/Sub.

Criar 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 onepassword-logs-collector-sa
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect 1Password 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 do Cloud Functions.
  6. Clique em Continuar.
  7. Clique em Concluído.

Esses papéis são necessários para:

  • Administrador de objetos do Storage: grava registros no bucket do GCS 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 GCS

Conceda permissões de gravação à conta de serviço no bucket do GCS:

  1. Acesse Cloud Storage > Buckets.
  2. Clique no bucket (onepassword-secops-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar principais: insira onepassword-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com.
    • Atribuir papéis: selecione Administrador de objetos do Storage.
  6. Clique em Salvar.

Criar tópico Pub/Sub

Crie um 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 onepassword-logs-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 tentativas de login e registros de uso de itens da API 1Password Events e gravá-los no GCS.

  1. No console do GCP, acesse o Cloud Run.
  2. Clique em Criar serviço.
  3. Selecione Função (use um editor in-line para criar uma função).

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

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

  5. Na seção Acionador (opcional):

    1. Clique em + Adicionar gatilho.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, escolha onepassword-logs-trigger.
    4. Clique em Salvar.

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

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

  7. Role a tela para baixo e abra Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione onepassword-logs-collector-sa

  9. 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 onepassword-secops-logs Nome do bucket do GCS
      GCS_PREFIX onepassword Prefixo para arquivos de registro
      STATE_KEY onepassword/state.json Caminho do arquivo de estado
      OP_API_BASE https://events.1password.com URL base da API 1Password Events
      OP_BEARER_TOKEN <your-bearer-token> Token de autenticação da API 1Password Events
      MAX_RECORDS 10000 Número máximo de registros por endpoint por execução
      PAGE_SIZE 1000 Registros por página (máximo de 1.000)
      LOOKBACK_HOURS 24 Período inicial de retorno em horas

  10. Na seção Variáveis e secrets, role a tela até Solicitações:

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

  11. Acesse a guia Configurações:

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

  12. 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.

  13. Clique em Criar.

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

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

Adicionar código da função

  1. Insira main no campo Ponto de entrada.

  2. No editor de código em linha, 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

# 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', 'onepassword')
STATE_KEY = os.environ.get('STATE_KEY', 'onepassword/state.json')
API_BASE = os.environ.get('OP_API_BASE')
BEARER_TOKEN = os.environ.get('OP_BEARER_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

# 1Password Events API v2 endpoints for sign-in and item usage
ENDPOINTS = {
  'signinattempts': '/api/v2/signinattempts',
  'itemusages': '/api/v2/itemusages',
}

def parse_datetime(value: str) -> datetime:
  """Parse RFC 3339 datetime string to datetime object."""
  if value.endswith('Z'):
    value = value[:-1] + '+00:00'
  return datetime.fromisoformat(value)

@functions_framework.cloud_event
def main(cloud_event):
  """
  Cloud Run function triggered by Pub/Sub to fetch 1Password
  sign-in attempt and item usage logs and write them to GCS
  as NDJSON.
  """
  if not all([GCS_BUCKET, API_BASE, BEARER_TOKEN]):
    print('Error: Missing required environment variables '
        '(GCS_BUCKET, OP_API_BASE, OP_BEARER_TOKEN)')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)

    # Load state (stores cursors per endpoint)
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)
    total_records = 0

    for event_type, path in ENDPOINTS.items():
      print(f'--- Processing endpoint: {event_type} ---')

      saved_cursor = state.get(f'cursor_{event_type}')

      records, last_cursor = fetch_endpoint(
        api_base=API_BASE,
        path=path,
        bearer_token=BEARER_TOKEN,
        saved_cursor=saved_cursor,
        lookback_hours=LOOKBACK_HOURS,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
      )

      if records:
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = (f'{GCS_PREFIX}/'
                f'{event_type}_{timestamp}.ndjson')
        blob = bucket.blob(object_key)

        ndjson = '\n'.join(
          json.dumps(r, ensure_ascii=False) for r in records
        ) + '\n'
        blob.upload_from_string(
          ndjson, content_type='application/x-ndjson'
        )

        print(f'Wrote {len(records)} {event_type} records '
            f'to gs://{GCS_BUCKET}/{object_key}')

      # Save cursor even if no records (for next poll)
      if last_cursor:
        state[f'cursor_{event_type}'] = last_cursor

      total_records += len(records)

    # Save state with all cursors
    state['last_run'] = now.isoformat()
    save_state(bucket, STATE_KEY, state)

    print(f'Successfully processed {total_records} total 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}')
  return {}

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

def fetch_endpoint(api_base, path, bearer_token, saved_cursor,
          lookback_hours, page_size, max_records):
  """
  Fetch events from a single 1Password Events API v2 endpoint.

  The 1Password Events API uses cursor-based pagination with POST
  requests. The first request sends a ResetCursor object with
  optional start_time and limit. Subsequent requests send the
  cursor returned from the previous response.

  Args:
    api_base: Events API base URL
    path: Endpoint path
    bearer_token: JWT bearer token
    saved_cursor: Cursor from previous run (or None)
    lookback_hours: Hours to look back on first run
    page_size: Max events per page (1-1000)
    max_records: Max total events per run

  Returns:
    Tuple of (records list, last_cursor string)
  """
  url = f'{api_base.rstrip("/")}{path}'

  headers = {
    'Authorization': f'Bearer {bearer_token}',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-1PasswordCollector/1.0',
  }

  records = []
  cursor = saved_cursor
  page_num = 0
  backoff = 1.0

  while True:
    page_num += 1

    if len(records) >= max_records:
      print(f'Reached max_records limit ({max_records})')
      break

    # Build request body
    if cursor:
      # Continuing cursor: resume from last position
      body = json.dumps({'cursor': cursor})
    else:
      # ResetCursor: first request or no saved state
      start_time = (
        datetime.now(timezone.utc)
        - timedelta(hours=lookback_hours)
      )
      body = json.dumps({
        'limit': page_size,
        'start_time': start_time.strftime(
          '%Y-%m-%dT%H:%M:%SZ'
        ),
      })

    try:
      response = http.request(
        'POST', url, body=body, headers=headers,
      )

      # Handle rate limiting (600 req/min, 30000 req/hour)
      if response.status == 429:
        retry_after = int(
          response.headers.get(
            'Retry-After', str(int(backoff))
          )
        )
        print(f'Rate limited (429). Retrying after '
            f'{retry_after}s...')
        time.sleep(retry_after)
        backoff = min(backoff * 2, 60.0)
        continue

      backoff = 1.0

      if response.status != 200:
        response_text = response.data.decode('utf-8')
        print(f'HTTP Error {response.status}: '
            f'{response_text}')
        break

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

      page_items = data.get('items', [])
      cursor = data.get('cursor')
      has_more = data.get('has_more', False)

      if page_items:
        print(f'Page {page_num}: Retrieved '
            f'{len(page_items)} events')
        records.extend(page_items)

      if not has_more:
        print(f'No more pages (has_more=false)')
        break

      if not cursor:
        print(f'No cursor returned, stopping')
        break

    except urllib3.exceptions.HTTPError as e:
      print(f'HTTP error: {str(e)}')
      break
    except Exception as e:
      print(f'Error fetching events: {str(e)}')
      break

  print(f'Retrieved {len(records)} total records '
      f'from {page_num} pages')
  return records, cursor

    • 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 o 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 onepassword-logs-collector-hourly
    Região Selecione a mesma região da função do Cloud Run
    Frequência 0 * * * * (a cada hora, na hora)
    Fuso horário Selecione o fuso horário (UTC recomendado)
    Tipo de destino Pub/Sub
    Tópico Selecionar onepassword-logs-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
Diariamente 0 0 * * * Coleta de dados históricos

Testar a integração

  1. No console do Cloud Scheduler, encontre onepassword-logs-collector-hourly.
  2. Clique em Executar à força para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique em onepassword-logs-collector.
  6. Clique na guia Registros.

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

    --- Processing endpoint: signinattempts ---
Page 1: Retrieved X events
Wrote X signinattempts records to gs://onepassword-secops-logs/onepassword/signinattempts_YYYYMMDD_HHMMSS.ndjson
--- Processing endpoint: itemusages ---
Page 1: Retrieved X events
Wrote X itemusages records to gs://onepassword-secops-logs/onepassword/itemusages_YYYYMMDD_HHMMSS.ndjson
Successfully processed X total records

  8. Acesse Cloud Storage > Buckets.

  9. Clique em onepassword-secops-logs.

  10. Navegue até a pasta onepassword/.

  11. Verifique se os novos arquivos .ndjson foram criados com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

  • HTTP 401: verifique o token de autenticação na variável de ambiente OP_BEARER_TOKEN. O token pode ter expirado.
  • HTTP 429: limitação de taxa. A função tenta novamente de forma automática com espera. A API 1Password Events permite 600 solicitações por minuto e 30.000 solicitações por hora.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas na configuração da função do Cloud Run.
  • Nenhum registro retornado: verifique se o token de portador tem acesso aos tipos de evento solicitados usando o endpoint de introspecção.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

  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, 1Password Sign-In and Item Usage Logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione 1Password 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 esse endereço de e-mail para usar 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 GCS com o caminho do prefixo:

      gs://onepassword-secops-logs/onepassword/

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

      • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
      • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.

      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.

    • Idade máxima do arquivo: inclui arquivos modificados nos últimos 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 do papel de Leitor de objetos do Storage no seu bucket do GCS.

  1. Acesse Cloud Storage > Buckets.
  2. Clique em onepassword-secops-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 Leitor de objetos do Storage.

  6. Clique em Salvar.

Tabela de mapeamento do UDM

Campo de registro Mapeamento do UDM Lógica
categoria security_result.category_details O valor é extraído do campo category no registro bruto.
client.app_name principal.application O valor é extraído do campo client.app_name no registro bruto.
client.app_version metadata.product_version O valor é extraído do campo client.app_version no registro bruto.
client.ip_address principal.ip O valor é extraído do campo client.ip_address no registro bruto.
client.os_name principal.platform O valor é extraído do campo client.os_name no registro bruto e mapeado para o valor correspondente da plataforma UDM (LINUX, WINDOWS, MAC).
client.os_version principal.platform_version O valor é extraído do campo client.os_version no registro bruto.
client.platform_name principal.resource.attribute.labels O valor é extraído do campo client.platform_name no registro bruto.
client.platform_version principal.asset.platform_software.platform_version O valor é extraído do campo client.platform_version no registro bruto.
país principal.location.country_or_region O valor é extraído do campo country no registro bruto se location.country não estiver presente.
item_uuid security_result.about.resource.attribute.labels O valor é extraído do campo item_uuid no registro bruto.
location.city principal.location.city O valor é extraído do campo location.city no registro bruto.
location.country principal.location.country_or_region O valor é extraído do campo location.country no registro bruto.
location.latitude principal.location.region_latitude O valor é extraído do campo location.latitude no registro bruto.
location.longitude principal.location.region_longitude O valor é extraído do campo location.longitude no registro bruto.
location.region principal.location.name O valor é extraído do campo location.region no registro bruto.
session.ip principal.ip O valor é extraído do campo session.ip no registro bruto.
session_uuid network.session_id O valor é extraído do campo session_uuid no registro bruto.
target_user.email target.user.email_addresses O valor é extraído do campo target_user.email no registro bruto.
target_user.uuid target.user.userid O valor é extraído do campo target_user.uuid no registro bruto.
timestamp metadata.event_timestamp O valor é extraído do campo timestamp no registro bruto e convertido em segundos e nanossegundos.
tipo additional.fields O valor é extraído do campo type no registro bruto.
user.email principal.user.email_addresses O valor é extraído do campo user.email no registro bruto.
user.name principal.user.user_display_name O valor é extraído do campo user.name no registro bruto.
used_version additional.fields O valor é extraído do campo used_version no registro bruto.
uuid principal.resource.attribute.labels O valor é extraído do campo uuid no registro bruto.
vault_uuid security_result.about.resource.attribute.labels O valor é extraído do campo vault_uuid no registro bruto.
N/A extensions.auth Um objeto vazio é criado para esse campo.
N/A metadata.event_type Defina como USER_LOGIN se a categoria for success ou firewall_reported_success, STATUS_UPDATE se não houver informações do usuário e USER_UNCATEGORIZED caso contrário.
N/A metadata.log_type Defina como ONEPASSWORD.
N/A metadata.product_name Defina como ONEPASSWORD.
N/A metadata.vendor_name Defina como ONEPASSWORD.
N/A security_result.action Defina como ALLOW se a categoria for success ou firewall_reported_success, BLOCK se a categoria for credentials_failed, mfa_failed, modern_version_failed ou firewall_failed. Caso contrário, deixe em branco.

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