Coletar registros de eventos do Bitwarden Enterprise

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros de eventos do Bitwarden Enterprise no Google Security Operations usando o Google Cloud Storage. O analisador transforma registros de eventos brutos formatados em JSON em um formato estruturado de acordo com o UDM do SecOps. Ele extrai campos relevantes, como detalhes do usuário, endereços IP e tipos de eventos, mapeando-os para os campos correspondentes do UDM para uma análise de segurança consistente.

Antes de começar

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

Uma instância do Google SecOps
Acesso privilegiado ao locatário do Bitwarden
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

Receber a chave de API e o URL do Bitwarden

No Admin Console do Bitwarden, acesse Configurações > Informações da organização > Ver chave de API.
Copie e salve os seguintes detalhes em um local seguro:
- Client-ID
- Client Secret
Determine seus endpoints do Bitwarden (com base na região):
- IDENTITY_URL = https://identity.bitwarden.com/connect/token (UE: https://identity.bitwarden.eu/connect/token)
- API_BASE = https://api.bitwarden.com (EU: https://api.bitwarden.eu)
Observação: as organizações empresariais do Bitwarden geralmente precisam ter o recurso "Eventos" ativado para que os eventos da organização estejam disponíveis. No Admin Console, acesse Configurações > Configurações da organização para verificar se os eventos estão ativados caso você não esteja recebendo eventos da API.
Observação: somente os proprietários da organização (e outras funções permitidas pelo Bitwarden) podem ver ou alternar a chave de API da organização. A chave de API da organização usada aqui precisa ser uma API Organization (escopo api.organization, client_id começando com "organization.") e não uma chave de API pessoal.

Criar um bucket do Google Cloud Storage

Acesse o Console do Google Cloud.
Selecione seu projeto ou crie um novo.
No menu de navegação, acesse Cloud Storage > Buckets.
Clique em Criar bucket.

Informe os seguintes detalhes de configuração:

Configuração	Valor
Nomeie seu bucket	Insira um nome exclusivo globalmente, por exemplo, `bitwarden-events`.
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

Clique em Criar.

Coletar os pré-requisitos da API Bitwarden

Você já coletou as credenciais da API Bitwarden na etapa anterior:

ID do cliente: ID do cliente da organização no Admin Console do Bitwarden.
Chave secreta do cliente: chave secreta do cliente da organização no Admin Console do Bitwarden
IDENTITY_URL: https://identity.bitwarden.com/connect/token (ou endpoint da UE)
API_BASE: https://api.bitwarden.com (ou endpoint da UE)

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

No Console do GCP, acesse IAM e administrador > Contas de serviço.
Clique em Criar conta de serviço.
Informe os seguintes detalhes de configuração:
- Nome da conta de serviço: insira bitwarden-events-collector-sa.
- Descrição da conta de serviço: insira Service account for Cloud Run function to collect Bitwarden Enterprise Event logs.
Clique em Criar e continuar.
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.
Clique em Continuar.
Clique em Concluído.

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

Administrador de objetos do Storage: grava registros em um 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:

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Acesse a guia Permissões.
Clique em Conceder acesso.
Informe os seguintes detalhes de configuração:
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo, bitwarden-events-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
- Atribuir papéis: selecione Administrador de objetos do Storage.
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.

No Console do GCP, acesse Pub/Sub > Tópicos.
Selecione Criar tópico.
Informe os seguintes detalhes de configuração:
- ID do tópico: insira bitwarden-events-trigger.
- Não altere as outras configurações.
Clique em Criar.

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

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API Bitwarden Events e gravá-los no GCS.

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

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

Configuração	Valor
Nome do serviço	`bitwarden-events-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.

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 o tópico do Pub/Sub (bitwarden-events-trigger).
4. Clique em Salvar.
Na seção Autenticação:
1. Selecione Exigir autenticação.
2. Confira o Identity and Access Management (IAM).
Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione a conta de serviço (bitwarden-events-collector-sa).

Acesse a guia Contêineres:

Clique em Variáveis e secrets.
Clique em + Adicionar variável para cada variável de ambiente:

Nome da variável	Valor de exemplo
`GCS_BUCKET`	`bitwarden-events`
`GCS_PREFIX`	`bitwarden/events`
`STATE_KEY`	`bitwarden/events/state.json`
`BW_CLIENT_ID`	`organization.your-client-id`
`BW_CLIENT_SECRET`	`your-client-secret`
`IDENTITY_URL`	`https://identity.bitwarden.com/connect/token`
`API_BASE`	`https://api.bitwarden.com`
`MAX_PAGES`	`10`

Na seção Variáveis e secrets, role a tela para baixo até Solicitações:
- Tempo limite da solicitação: insira 600 segundos (10 minutos).
Acesse a guia Configurações:
- Na seção Recursos:
  - Memória: selecione 512 MiB ou mais.
  - CPU: selecione 1.
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.
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

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

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
import base64

# 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()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch events from Bitwarden 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', 'bitwarden/events').strip('/')
    state_key = os.environ.get('STATE_KEY', 'bitwarden/events/state.json')

    # Bitwarden API credentials
    identity_url = os.environ.get('IDENTITY_URL', 'https://identity.bitwarden.com/connect/token')
    api_base = os.environ.get('API_BASE', 'https://api.bitwarden.com').rstrip('/')
    client_id = os.environ.get('BW_CLIENT_ID')
    client_secret = os.environ.get('BW_CLIENT_SECRET')
    max_pages = int(os.environ.get('MAX_PAGES', '10'))

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

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

        # Load state (continuation token)
        state = load_state(bucket, state_key)
        continuation_token = state.get('continuationToken')

        print(f'Processing events with continuation token: {continuation_token}')

        # Get OAuth token
        access_token = get_oauth_token(identity_url, client_id, client_secret)

        # Fetch events from Bitwarden API
        run_timestamp = int(datetime.now(timezone.utc).timestamp())
        pages = 0
        total_events = 0
        written_files = []

        while pages < max_pages:
            events_data = fetch_events(api_base, access_token, continuation_token)

            # Extract events array from API response
            events = events_data.get('data', [])

            # Only write file if there are events
            if events:
                gcs_key = write_events_jsonl(bucket, events, prefix, run_timestamp, pages)
                if gcs_key:
                    written_files.append(gcs_key)
                total_events += len(events)

            pages += 1

            # Check for next page token
            next_token = events_data.get('continuationToken')
            if next_token:
                continuation_token = next_token
                continue
            else:
                # No more pages
                break

        # Save state only if there are more pages to continue in next run
        # If we hit MAX_PAGES and there's still a continuation token, save it
        # Otherwise, clear the state (set to None)
        save_state(bucket, state_key, {
            'continuationToken': continuation_token if pages >= max_pages and continuation_token else None
        })

        print(f'Successfully processed {total_events} events across {pages} pages')
        print(f'Files written: {len(written_files)}')

    except Exception as e:
        print(f'Error processing events: {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),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

def get_oauth_token(identity_url, client_id, client_secret):
    """Get OAuth 2.0 access token from Bitwarden."""
    body_data = {
        'grant_type': 'client_credentials',
        'scope': 'api.organization',
        'client_id': client_id,
        'client_secret': client_secret
    }

    encoded_data = '&'.join([f'{k}={v}' for k, v in body_data.items()]).encode('utf-8')

    response = http.request(
        'POST',
        identity_url,
        body=encoded_data,
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )

    if response.status != 200:
        raise Exception(f'Failed to get OAuth token: {response.status} {response.data.decode("utf-8")}')

    token_data = json.loads(response.data.decode('utf-8'))
    return token_data['access_token']

def fetch_events(api_base, access_token, continuation_token=None):
    """Fetch events from Bitwarden API with pagination."""
    url = f'{api_base}/public/events'
    if continuation_token:
        url += f'?continuationToken={continuation_token}'

    response = http.request(
        'GET',
        url,
        headers={
            'Authorization': f'Bearer {access_token}',
            'Accept': 'application/json'
        }
    )

    if response.status == 429:
        retry_after = int(response.headers.get('Retry-After', '60'))
        print(f'Rate limited (429). Retrying after {retry_after}s...')
        import time
        time.sleep(retry_after)
        return fetch_events(api_base, access_token, continuation_token)

    if response.status != 200:
        raise Exception(f'Failed to fetch events: {response.status} {response.data.decode("utf-8")}')

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

def write_events_jsonl(bucket, events, prefix, run_timestamp, page_index):
    """
    Write events in JSONL format (one JSON object per line).
    Only writes if there are events to write.
    Returns the GCS key of the written file.
    """
    if not events:
        return None

    # Build JSONL content: one event per line
    lines = [json.dumps(event, separators=(',', ':')) for event in events]
    jsonl_content = '\n'.join(lines) + '\n'  # JSONL format with trailing newline

    # Generate unique filename with page number to avoid conflicts
    timestamp_str = datetime.fromtimestamp(run_timestamp, tz=timezone.utc).strftime('%Y/%m/%d/%H%M%S')
    key = f'{prefix}/{timestamp_str}-page{page_index:05d}-bitwarden-events.jsonl'

    blob = bucket.blob(key)
    blob.upload_from_string(
        jsonl_content,
        content_type='application/x-ndjson'
    )

    print(f'Wrote {len(events)} events to {key}')
    return key

Segundo arquivo: requirements.txt:

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

Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).

Observação: a configuração do gatilho do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.
Observação: a API pública do Bitwarden impõe limites de taxa e pode retornar respostas HTTP 429 em alto volume de solicitações. Se você encontrar respostas 429, a função vai tentar de novo automaticamente com espera exponencial. Considere ajustar MAX_PAGES ou a frequência de programação se a limitação de taxa ocorrer com frequência.

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.

No Console do GCP, acesse o Cloud Scheduler.
Clique em Criar job.

Informe os seguintes detalhes de configuração:

Configuração	Valor
Nome	`bitwarden-events-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	Selecione o tópico do Pub/Sub (`bitwarden-events-trigger`).
Corpo da mensagem	`{}` (objeto JSON vazio)

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

No console do Cloud Scheduler, encontre seu job.
Clique em Executar à força para acionar o job manualmente.
Aguarde alguns segundos.
Acesse Cloud Run > Serviços.
Clique no nome da função (bitwarden-events-collector).
Clique na guia Registros.

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

Processing events with continuation token: None
Page 1: Retrieved X events
Wrote X events to bitwarden/events/YYYY/MM/DD/HHMMSS-page00000-bitwarden-events.jsonl
Successfully processed X events across 1 pages

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Navegue até a pasta de prefixo (bitwarden/events/).
Verifique se um novo arquivo .jsonl foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

HTTP 401: verifique as credenciais da API nas variáveis de ambiente
HTTP 403: verifique se a conta tem as permissões necessárias e se o recurso "Eventos" está ativado nas configurações da organização.
HTTP 429: limitação de taxa. A função vai tentar novamente automaticamente com espera.
Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.

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

Acesse Configurações do SIEM > Feeds.
Clique em Adicionar novo feed.
Clique em Configurar um único feed.
No campo Nome do feed, insira um nome para o feed (por exemplo, Bitwarden Events).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Eventos do Bitwarden como o Tipo de registro.
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
```
Copie esse endereço de e-mail para usar na próxima etapa.

Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.

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.

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Acesse a guia Permissões.
Clique em Conceder acesso.
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.
Clique em Salvar.

Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar um feed no Google SecOps para ingerir registros de eventos do Bitwarden Enterprise

Acesse Configurações do SIEM > Feeds.
Clique em Adicionar novo feed.
Clique em Configurar um único feed.
No campo Nome do feed, insira um nome para o feed (por exemplo, Bitwarden Events).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Eventos do Bitwarden como o Tipo de registro.
Clique em Próxima.
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://bitwarden-events/bitwarden/events/
```
  - Substitua:
    - bitwarden-events: o nome do bucket do GCS.
    - bitwarden/events/: prefixo/caminho da pasta onde os registros são armazenados.
  Observação: sempre inclua a barra (/) no final do URI.
- 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.
    
    Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.
- Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
- Namespace do recurso: o namespace do recurso.
- Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.
Clique em Próxima.
Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Tabela de mapeamento do UDM

Campo de registro	Mapeamento do UDM	Lógica
actingUserId	target.user.userid	Se "enriched.actingUser.userId" estiver vazio ou nulo, esse campo será usado para preencher o campo "target.user.userid".
collectionID	security_result.detection_fields.key	Preenche o campo "key" em "detection_fields" em "security_result".
collectionID	security_result.detection_fields.value	Preenche o campo "value" em "detection_fields" em "security_result".
data	metadata.event_timestamp	Analisado e convertido para um formato de carimbo de data/hora e mapeado para "event_timestamp".
enriched.actingUser.accessAll	security_result.rule_labels.key	Define o valor como "Access_All" em rule_labels em security_result.
enriched.actingUser.accessAll	security_result.rule_labels.value	Preenche o campo "value" em "rule_labels" em "security_result" com o valor de "enriched.actingUser.accessAll" convertido em string.
enriched.actingUser.email	target.user.email_addresses	Preenche o campo "email_addresses" em "target.user".
enriched.actingUser.id	metadata.product_log_id	Preenche o campo "product_log_id" nos metadados.
enriched.actingUser.id	target.labels.key	Define o valor como "ID" em target.labels.
enriched.actingUser.id	target.labels.value	Preenche o campo "value" em "target.labels" com o valor de "enriched.actingUser.id".
enriched.actingUser.name	target.user.user_display_name	Preenche o campo user_display_name em target.user.
enriched.actingUser.object	target.labels.key	Define o valor como "Object" em target.labels.
enriched.actingUser.object	target.labels.value	Preenche o campo "value" em "target.labels" com o valor de "enriched.actingUser.object".
enriched.actingUser.resetPasswordEnrolled	target.labels.key	Define o valor como "ResetPasswordEnrolled" em target.labels.
enriched.actingUser.resetPasswordEnrolled	target.labels.value	Preenche o campo "value" em target.labels com o valor de enriched.actingUser.resetPasswordEnrolled convertido em string.
enriched.actingUser.twoFactorEnabled	security_result.rule_labels.key	Define o valor como "Two Factor Enabled" em rule_labels em security_result.
enriched.actingUser.twoFactorEnabled	security_result.rule_labels.value	Preenche o campo "value" em "rule_labels" em "security_result" com o valor de "enriched.actingUser.twoFactorEnabled" convertido em string.
enriched.actingUser.userId	target.user.userid	Preenche o campo "userid" em "target.user".
enriched.collection.id	additional.fields.key	Define o valor como "ID da coleção" em additional.fields.
enriched.collection.id	additional.fields.value.string_value	Preenche o campo "string_value" em "additional.fields" com o valor de "enriched.collection.id".
enriched.collection.object	additional.fields.key	Define o valor como "Objeto de coleta" em additional.fields.
enriched.collection.object	additional.fields.value.string_value	Preenche o campo "string_value" em "additional.fields" com o valor de "enriched.collection.object".
enriched.type	metadata.product_event_type	Preenche o campo "product_event_type" nos metadados.
groupId	target.user.group_identifiers	Adiciona o valor à matriz group_identifiers em target.user.
ipAddress	principal.ip	Extraiu o endereço IP do campo e mapeou para principal.ip.
N/A	extensions.auth	Um objeto vazio é criado pelo analisador.
N/A	metadata.event_type	Determinado com base em enriched.type e na presença de informações principais e de destino. Valores possíveis: USER_LOGIN, STATUS_UPDATE, GENERIC_EVENT.
N/A	security_result.action	Determinado com base em "enriched.type". Valores possíveis: ALLOW, BLOCK.
objeto	additional.fields.key	Define o valor como "Object" em additional.fields.
objeto	additional.fields.value	Preenche o campo "value" em "additional.fields" com o valor do objeto.

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