Coletar registros do NetDocuments
Este documento explica como ingerir registros do NetDocuments no Google Security Operations usando o Google Cloud Storage V2.
A NetDocuments é uma plataforma de gerenciamento de documentos baseada na nuvem projetada para organizações de serviços jurídicos e profissionais. Ele gera registros de auditoria que rastreiam o acesso, as modificações, o compartilhamento e a atividade administrativa de documentos pela API REST do NetDocuments.
Antes de começar
Verifique se você atende aos seguintes pré-requisitos:
- 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 NetDocuments (papel de administrador do repositório ou da organização)
- Um registro de aplicativo do NetDocuments com credenciais OAuth2 (ID do cliente, chave secreta do cliente e token de atualização)
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, netdocuments-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 NetDocuments
Registrar um aplicativo
- Faça login no Portal do desenvolvedor do NetDocuments com uma conta de administrador.
- Acesse Aplicativos e clique em Registrar aplicativo.
- Insira um nome para o aplicativo (por exemplo,
Google Security Operations Integration). - Defina o Tipo de aplicativo como Servidor (confidencial).
- Observe as seguintes credenciais:
- ID do cliente: o identificador do cliente OAuth2.
- Chave secreta do cliente: a chave secreta do cliente OAuth2.
Gerar token de atualização
- Use o fluxo de autorização OAuth2 do NetDocuments para receber um token de atualização.
- Autorize o aplicativo com os escopos necessários:
read(acesso de leitura a documentos e registros de auditoria)full(acesso total, se exigido pela sua organização)
Conclua o fluxo do código de autorização do OAuth2 e salve o token de atualização com segurança.
Determinar o URL de base da API
O URL de base da API NetDocuments depende da região do seu data center:
| Região | URL base da API |
|---|---|
| EUA | https://api.netdocuments.com/v2 |
| UE | https://api.eu.netdocuments.com/v2 |
| Austrália | https://api.au.netdocuments.com/v2 |
Verifique as permissões
Para verificar se a conta tem as permissões necessárias:
- Faça login no portal de administrador do NetDocuments.
- Acesse Administrador do repositório > Registros de atividade.
- Se a seção "Registro de atividades" aparecer, você tem as permissões necessárias.
- Se essa opção não estiver disponível, entre em contato com o administrador do NetDocuments para receber acesso de administrador do repositório ou da organização.
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
netdocuments-logs-collector-sa - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect NetDocuments 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 do Cloud Functions.
- 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,
netdocuments-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,
netdocuments-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
netdocuments-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 REST do NetDocuments 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 netdocuments-logs-collectorRegião Selecione a região correspondente ao 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):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico
netdocuments-logs-trigger. - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Verifique 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
netdocuments-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_BUCKETnetdocuments-audit-logsNome do bucket do GCS GCS_PREFIXnetdocumentsPrefixo para arquivos de registro STATE_KEYnetdocuments/state.jsonCaminho do arquivo de estado ND_API_BASEhttps://api.netdocuments.com/v2URL de base da API NetDocuments ND_CLIENT_IDyour-client-idID do cliente OAuth 2.0 ND_CLIENT_SECRETyour-client-secretChave secreta do cliente OAuth2 ND_REFRESH_TOKENyour-refresh-tokenToken de atualização do OAuth2 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 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:
- 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 # 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', 'netdocuments') STATE_KEY = os.environ.get('STATE_KEY', 'netdocuments/state.json') ND_API_BASE = os.environ.get('ND_API_BASE', 'https://api.netdocuments.com/v2') ND_CLIENT_ID = os.environ.get('ND_CLIENT_ID') ND_CLIENT_SECRET = os.environ.get('ND_CLIENT_SECRET') ND_REFRESH_TOKEN = os.environ.get('ND_REFRESH_TOKEN') 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_access_token(): """ Obtain an OAuth2 access token using the refresh token. """ token_url = "https://api.netdocuments.com/v1/OAuth" # Build Basic auth header credentials = base64.b64encode( f"{ND_CLIENT_ID}:{ND_CLIENT_SECRET}".encode('utf-8') ).decode('utf-8') headers = { 'Authorization': f'Basic {credentials}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } body = f"grant_type=refresh_token&refresh_token={ND_REFRESH_TOKEN}" 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 access token: {response.status} - {response.data.decode('utf-8')}") data = json.loads(response.data.decode('utf-8')) return data['access_token'] raise RuntimeError("Failed to get access token after 3 retries") @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch NetDocuments audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ND_CLIENT_ID, ND_CLIENT_SECRET, ND_REFRESH_TOKEN]): 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 access token token = get_access_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 logs from NetDocuments REST API with pagination and rate limiting. Args: token: OAuth2 access 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 = ND_API_BASE.rstrip('/') endpoint = f"{api_base}/AuditLog/search" headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-NetDocumentsCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 offset = 0 while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break # Build request body body = { 'startDate': start_time.strftime('%Y-%m-%dT%H:%M:%SZ'), 'endDate': end_time.strftime('%Y-%m-%dT%H:%M:%SZ'), '$top': min(page_size, max_records - len(records)), '$skip': offset } 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('results', data.get('value', [])) 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('date') or event.get('timestamp') or event.get('created') if event_ts: 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 offset += len(page_results) if len(page_results) < page_size: print("No more pages (partial page received)") 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_time- requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Clique 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 netdocuments-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 Selecione o fuso horário (UTC recomendado) Tipo de destino Pub/Sub Tópico Selecione o tópico netdocuments-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 * * * |
Processamento em lote de baixo volume |
| 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
netdocuments-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://netdocuments-audit-logs/netdocuments/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsAcesse Cloud Storage > Buckets.
Clique no nome do bucket (
netdocuments-audit-logs).Navegue até a pasta
netdocuments/.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 do OAuth2 nas variáveis de ambiente
- HTTP 403: verifique se a conta tem as permissões necessárias de administrador do repositório ou da organização.
- 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 do NetDocuments
- 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,
NetDocuments Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione NetDocuments 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://netdocuments-audit-logs/netdocuments/- Substitua:
netdocuments-audit-logs: o nome do bucket do GCS.netdocuments: prefixo/caminho de pasta opcional em que 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 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
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 |
|---|---|---|
| description, name | metadata.description | Uma descrição legível por humanos do evento. |
| data | metadata.event_timestamp | O horário em que o evento ocorreu. |
| metadata.event_type | O tipo de evento. | |
| source.id | principal.asset.hostname | O nome do host do recurso associado ao principal. |
| host | principal.asset.ip | O endereço IP do recurso associado ao principal. |
| source.id | principal.hostname | O nome do host associado ao principal. |
| host | principal.ip | O endereço IP associado ao principal. |
| source.name | principal.resource.attribute.labels | Um mapa de rótulos para o recurso do principal. |
| user.email | principal.user.email_addresses | Os endereços de e-mail associados ao usuário. |
| user.memberType | principal.user.role_name | O nome da função do usuário. |
| user.name | principal.user.user_display_name | O nome de exibição do usuário. |
| user.id | principal.user.userid | O ID do usuário do principal. |
| storageObject.fileExtension | target.file.mime_type | O tipo MIME do arquivo de destino. |
| storageObject.name | target.file.names | Os nomes dos arquivos de destino. |
| storageObject.size | target.file.size | O tamanho do arquivo de destino em bytes. |
| storageObject.version, storageObject.collabSpace, storageObject.NetBinder, storageObject.cabinet.name, storageObject.cabinet.id | target.resource.attribute.labels | Um mapa de rótulos para o recurso de destino. |
| storageObject.docId | target.resource.product_object_id | O ID do objeto específico do produto do recurso de destino. |
Registro de alterações
Confira o registro de alterações deste analisador
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.