Coletar registros de segurança do Keeper Enterprise
Este documento explica como ingerir registros de segurança do Keeper Enterprise no Google Security Operations usando o Google Cloud Storage V2.
O Keeper Enterprise Security é um gerenciador de senhas corporativo e uma plataforma de gerenciamento de acesso privilegiado que protege credenciais, segredos e dados sensíveis. Ela gera registros de auditoria para acesso ao cofre, mudanças de senha, eventos de compartilhamento e ações administrativas usando a API Reporting and Alerts.
Antes de começar
Verifique se você tem os pré-requisitos a seguir:
- Uma instância do Google SecOps
- Um projeto do GCP com a API Storage do Cloud ativada
- Permissões para criar e gerenciar intervalos 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
- Acesso privilegiado ao Keeper Admin Console com permissões de administrador
- Um plano Keeper Enterprise ou Business com o módulo de relatórios e alertas avançados (ARAM) ativado
Criar um bucket do Google Cloud Storage
- Acesse o Console do Google Cloud.
- Selecione ou crie um projeto.
- 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 globalmente exclusivo, por exemplo, keeper-audit-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 Clique em Criar.
Coletar credenciais da API Keeper Enterprise Security
Criar chave de API
- Faça login no Keeper Admin Console.
- Acesse Administrador > Relatórios e alertas.
- Clique em Chaves de API ou navegue até a seção de gerenciamento de chaves de API.
- Clique em Criar chave de API.
- Insira um nome para a chave de API (por exemplo,
Google Security Operations Integration). - Copie e salve os seguintes detalhes em um local seguro:
- Chave de API: o valor da chave de API gerada
- Chave privada: faça o download do arquivo de chave privada (usado para geração de token).
Anote seu ID da empresa do Keeper Enterprise no Admin Console.
Determinar o URL de base da API
O URL base da API Keeper depende da região do seu data center:
Região URL base da API EUA https://keepersecurity.comUE https://keepersecurity.euAustrália https://keepersecurity.com.auCA https://keepersecurity.caJapão https://keepersecurity.jpGovCloud dos EUA https://govcloud.keepersecurity.us
Verifique as permissões
Para verificar se a conta tem as permissões necessárias:
- Faça login no Keeper Admin Console.
- Acesse Administrador > Relatórios e alertas.
- Se você conseguir ver o painel Relatórios e alertas e a seção Chaves de API, significa que você tem as permissões necessárias.
- Se essa opção não estiver disponível, entre em contato com o administrador do Keeper para ativar o módulo ARAM e conceder acesso de administrador.
Testar o acesso à API
Teste suas credenciais antes de prosseguir com a integração:
# Replace with your actual credentials KEEPER_API_KEY="your-api-key" KEEPER_BASE="https://keepersecurity.com" # Test API access - get audit events (requires signed token) # Note: Keeper uses a signed JWT for authentication. # Refer to Keeper Commander CLI for testing: keeper audit-report --format json --limit 1
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
keeper-logs-collector-sa - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect Keeper Enterprise Security logs
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
- Clique em Selecionar papel.
- Pesquise e selecione Administrador de objetos do Storage.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Run.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador de funções do Cloud.
- Clique em Continuar.
- Clique em Concluído.
Essas funções são necessárias 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:
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket (por exemplo,
keeper-audit-logs). - 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,
keeper-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Atribuir papéis: selecione Administrador de objetos do Storage.
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
- 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
keeper-logs-trigger - Não mude as outras configurações.
- ID do tópico: insira
- 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 Keeper Reporting and Alerts 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 keeper-logs-collectorRegião Selecione a região correspondente 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 Gatilho (opcional):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico
keeper-logs-trigger. - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Confira o Identity and Access Management (IAM).
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
keeper-logs-collector-sa.
- Conta de serviço: selecione a conta de serviço
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 Descrição GCS_BUCKETkeeper-audit-logsNome do bucket do GCS GCS_PREFIXkeeperPrefixo para arquivos de registro STATE_KEYkeeper/state.jsonCaminho do arquivo de estado KEEPER_API_BASEhttps://keepersecurity.comURL base da API Keeper. KEEPER_API_KEYyour-api-keyChave de API do Keeper KEEPER_PRIVATE_KEYbase64-encoded-private-keyChave privada codificada em base64 KEEPER_ENTERPRISE_ID12345ID da empresa MAX_RECORDS5000Máximo de registros por execução PAGE_SIZE1000Registros por página LOOKBACK_HOURS24Período de lookback inicial Na seção Variáveis e secrets, role a tela para baixo até Solicitações:
- Tempo limite da solicitação: insira
600segundos (10 minutos)
- Tempo limite da solicitação: insira
Acesse a guia Configurações:
- Na seção Recursos:
- Memória: selecione 512 MiB ou mais.
- CPU: selecione 1
- Na seção Recursos:
Na seção Escalonamento de revisão:
- Número mínimo de instâncias: insira
0 - Número máximo de instâncias: insira
100ou ajuste com base na carga esperada.
- Número mínimo de instâncias: insira
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço for criado, o editor de código inline será aberto automaticamente.
Adicionar código da função
- Insira main no campo Ponto de entrada.
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 base64 import hashlib import hmac # 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', 'keeper') STATE_KEY = os.environ.get('STATE_KEY', 'keeper/state.json') KEEPER_API_BASE = os.environ.get('KEEPER_API_BASE', 'https://keepersecurity.com') KEEPER_API_KEY = os.environ.get('KEEPER_API_KEY') KEEPER_PRIVATE_KEY = os.environ.get('KEEPER_PRIVATE_KEY') KEEPER_ENTERPRISE_ID = os.environ.get('KEEPER_ENTERPRISE_ID') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) def to_unix_millis(dt: datetime) -> int: """Convert datetime to Unix epoch milliseconds.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return int(dt.timestamp() * 1000) 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) def get_auth_token(): """ Generate authentication token for Keeper Reporting API. Uses HMAC-SHA512 signing with the private key. """ api_base = KEEPER_API_BASE.rstrip('/') token_url = f"{api_base}/api/rest/enterprise/auth/token" # Build token request timestamp = str(int(time.time())) message = f"{KEEPER_API_KEY}:{timestamp}" # Decode private key from base64 private_key_bytes = base64.b64decode(KEEPER_PRIVATE_KEY) # Sign with HMAC-SHA512 signature = hmac.new( private_key_bytes, message.encode('utf-8'), hashlib.sha512 ).digest() signature_b64 = base64.b64encode(signature).decode('utf-8') body = json.dumps({ 'api_key': KEEPER_API_KEY, 'timestamp': timestamp, 'signature': signature_b64, 'enterprise_id': KEEPER_ENTERPRISE_ID }) headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } backoff = 1.0 for attempt in range(3): response = http.request('POST', token_url, body=body, 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 auth token: {response.status} - {response.data.decode('utf-8')}") data = json.loads(response.data.decode('utf-8')) return data.get('token', data.get('access_token')) raise RuntimeError("Failed to get auth token after 3 retries") @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Keeper Enterprise Security audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, KEEPER_API_KEY, KEEPER_PRIVATE_KEY, KEEPER_ENTERPRISE_ID]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(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"]) # Overlap by 2 minutes to catch any delayed events 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 auth token token = get_auth_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, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, now.isoformat()) 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 records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest event time 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(records)} 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, 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}") def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int): """ Fetch audit event logs from Keeper Reporting and Alerts API with pagination and rate limiting. Args: token: Authentication 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) """ api_base = KEEPER_API_BASE.rstrip('/') endpoint = f"{api_base}/api/rest/enterprise/audit-events" headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-KeeperCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 # Convert to Unix epoch seconds for Keeper API start_epoch = int(start_time.timestamp()) end_epoch = int(end_time.timestamp()) cursor = None while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break # Build request body body = { 'start_time': start_epoch, 'end_time': end_epoch, 'limit': min(page_size, max_records - len(records)), 'enterprise_id': KEEPER_ENTERPRISE_ID } if cursor: body['cursor'] = cursor try: response = http.request( 'POST', endpoint, body=json.dumps(body), headers=headers ) # Handle rate limiting with exponential backoff 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('audit_events', data.get('events', [])) 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_ts = event.get('timestamp') or event.get('created') if event_ts: if isinstance(event_ts, (int, float)): event_dt = datetime.fromtimestamp(event_ts, tz=timezone.utc) event_time = event_dt.isoformat() else: event_time = str(event_ts) 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 for more results cursor = data.get('cursor') or data.get('next_cursor') if not cursor: print("No more pages (no next cursor)") break 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_timeSegundo 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).
Criar o job do Cloud Scheduler
O Cloud Scheduler vai publicar 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 keeper-logs-collector-hourlyRegiã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 keeper-logs-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 Forçar execução para acionar o job manualmente.
- Aguarde alguns segundos.
- Acesse Cloud Run > Serviços.
- Clique em
keeper-logs-collector. - Clique na guia Registros.
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://keeper-audit-logs/keeper/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsAcesse Cloud Storage > Buckets.
Clique no nome do bucket (
keeper-audit-logs).Navegue até a pasta
keeper/.Verifique se um novo arquivo
.ndjsonfoi 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 do ARAM no Keeper Admin Console.
- HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera.
- Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.
Configurar um feed no Google SecOps para ingerir registros de segurança do Keeper 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,
Keeper Enterprise Security Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Keeper Enterprise Security 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.comCopie o endereço de e-mail.
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://keeper-audit-logs/keeper/- Substitua:
keeper-audit-logs: o nome do bucket do GCS.keeper: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
- Substitua:
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.
Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência.
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
Clique em Próxima.
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 Leitor de objetos do Storage no seu bucket do GCS.
- Acesse Cloud Storage > Buckets.
- Clique no nome do seu bucket.
- Acesse a guia Permissões.
- Clique em Conceder acesso.
- Informe os seguintes detalhes de configuração:
- Adicionar principais: cole o e-mail da conta de serviço do Google SecOps.
- Atribuir papéis: selecione Leitor de objetos do Storage.
- Clique em Salvar.
Tabela de mapeamento de UDM
| Campo de registro | Mapeamento do UDM | Lógica |
|---|---|---|
| logData.channel | extensions.auth.type | Definido como valor em maiúsculas se estiver em ["SSO", "MACHINE", "VPN", "PHYSICAL", "TACACS"], caso contrário, "AUTHTYPE_UNSPECIFIED" |
| logData.timestamp | metadata.event_timestamp | Convertido usando o formato ISO8601 |
| logData.audit_event, logData.remote_address | metadata.event_type | Definido como "USER_LOGIN" se audit_event corresponder a "login" e remote_address não estiver vazio. Caso contrário, "STATUS_UPDATE" se remote_address não estiver vazio. Caso contrário, "GENERIC_EVENT". |
| logData.audit_event | metadata.product_event_type | Valor copiado diretamente |
| logData.enterprise_id | metadata.product_log_id | Convertido em string |
| metadata.product_name | Definido como "KEEPER" | |
| metadata.vendor_name | Definido como "KEEPER" | |
| logData.client_version | network.http.parsed_user_agent | Valor copiado diretamente e convertido em parseduseragent. |
| logData.client_version | network.http.user_agent | Valor copiado diretamente |
| logData.remote_address | principal.asset.ip | Valor copiado diretamente |
| logData.remote_address | principal.ip | Valor copiado diretamente |
| logData.category | security_result.category_details | Unido da origem |
| logData.shared_folder_uid, logData.folder_uid | security_result.detection_fields | Unidos como rótulos com as chaves "shared_folder_uid" e "folder_uid" |
| logData.email | target.user.email_addresses | Unido da origem |
| logData.username | target.user.userid | Valor copiado diretamente |
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.