Coletar registros de auditoria da DigiCert

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir registros de auditoria da DigiCert no Google Security Operations usando o Google Cloud Storage. O DigiCert CertCentral é uma plataforma de gerenciamento do ciclo de vida de certificados que fornece registros de auditoria para operações de certificados, atividades de usuários e ações administrativas.

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
• Acesso privilegiado ao DigiCert CertCentral (chave de API com função de administrador)

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, digicert-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 da API DigiCert

Receber chave da API DigiCert

1. Faça login no DigiCert CertCentral.
2. Acesse Conta > Chaves de API.
3. Clique em Criar chave de API.
4. Informe os seguintes detalhes de configuração:
   • Nome: insira um nome descritivo, por exemplo, Chronicle Integration.
   • Função: selecione Administrador.
5. Clique em Criar.
6. Copie e salve a chave de API (X-DC-DEVKEY). Esse valor não será mostrado novamente.

Receber o ID do relatório da DigiCert

1. No DigiCert CertCentral, acesse Relatórios > Biblioteca de relatórios.
2. Clique em Criar relatório.
3. Informe os seguintes detalhes de configuração:
   • Tipo de relatório: selecione Registro de auditoria.
   • Formato: selecione JSON.
   • Nome: insira um nome descritivo, por exemplo, Chronicle Audit Logs.
4. Clique em Criar.

5. Copie e salve o ID da denúncia (formato UUID).

   Observação: o relatório de registros de auditoria precisa ser configurado com JSON como um formato compatível antes de usar o endpoint da API. Também é possível encontrar o ID de um relatório no histórico.

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 digicert-logs-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect DigiCert audit 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 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:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, digicert-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 digicert-audit-trigger.
   • Não altere as outras configurações.
4. 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 DigiCert 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	digicert-audit-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 o tópico do Pub/Sub (digicert-audit-trigger).
   4. Clique em Salvar.

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

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

8. Acesse a guia Segurança:

   • Conta de serviço: selecione a conta de serviço (digicert-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
   GCS_BUCKET	digicert-logs
   GCS_PREFIX	digicert/logs
   STATE_KEY	digicert/logs/state.json
   DIGICERT_API_KEY	xxxxxxxxxxxxxxxxxxxxxxxx
   DIGICERT_REPORT_ID	88de5e19-ec57-4d70-865d-df953b062574
   REQUEST_TIMEOUT	30
   POLL_INTERVAL	10
   MAX_WAIT_SECONDS	300

10. Role a tela para baixo na guia Variáveis e secrets até Solicitações:

    • Tempo limite da solicitação: insira 900 segundos (15 minutos).

11. Acesse a guia Configurações em Contêineres:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
    • Clique em Concluído.

12. Role até Ambiente de execução:

    • Selecione Padrão (recomendado).

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

14. Clique em Criar.

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

16. Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

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

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
   import io
   import gzip
   import zipfile
   import uuid
   
   # Initialize HTTP client
   http = urllib3.PoolManager()
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   API_BASE = "https://api.digicert.com/reports/v1"
   USER_AGENT = "secops-digicert-reports/1.0"
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch DigiCert audit logs 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', 'digicert/logs').rstrip('/')
       state_key = os.environ.get('STATE_KEY', f'{prefix}/state.json')
       api_key = os.environ.get('DIGICERT_API_KEY')
       report_id = os.environ.get('DIGICERT_REPORT_ID')
       max_wait = int(os.environ.get('MAX_WAIT_SECONDS', '300'))
       poll_int = int(os.environ.get('POLL_INTERVAL', '10'))
       timeout = int(os.environ.get('REQUEST_TIMEOUT', '30'))
   
       if not all([bucket_name, api_key, report_id]):
           print('Error: Missing required environment variables')
           return
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(bucket_name)
   
           # Load state
           state = load_state(bucket, state_key)
           last_run = state.get('last_run_id')
   
           # Start report run
           started = datetime.now(timezone.utc)
           start_report_run(api_key, report_id, timeout)
   
           # Wait for report to be ready
           run_id = find_ready_run(api_key, report_id, started, timeout, max_wait, poll_int)
   
           # Skip if same run as last time
           if last_run and last_run == run_id:
               print(f'Skipping duplicate run: {run_id}')
               return
   
           # Get report data
           rows = get_json_rows(api_key, report_id, run_id, timeout)
   
           # Write to GCS
           key = write_ndjson_gz(bucket, prefix, rows, run_id)
   
           # Update state
           save_state(bucket, state_key, {
               'last_run_id': run_id,
               'last_success_at': datetime.now(timezone.utc).isoformat(),
               'last_s3_key': key,
               'rows_count': len(rows)
           })
   
           print(f'Successfully processed {len(rows)} logs to {key}')
   
       except Exception as e:
           print(f'Error processing logs: {str(e)}')
           raise
   
   def http_request(method, url, api_key, body=None, timeout=30, max_retries=5):
       """Make HTTP request with retry logic."""
       headers = {
           'X-DC-DEVKEY': api_key,
           'Content-Type': 'application/json',
           'User-Agent': USER_AGENT
       }
   
       attempt, backoff = 0, 1.0
   
       while True:
           try:
               response = http.request(
                   method,
                   url,
                   headers=headers,
                   body=body,
                   timeout=timeout
               )
   
               status = response.status
   
               # Retry on server errors
               if 500 <= status <= 599 and attempt < max_retries:
                   attempt += 1
                   time.sleep(backoff)
                   backoff *= 2
                   continue
   
               # Retry on rate limit
               if status == 429 and attempt < max_retries:
                   retry_after = response.headers.get('Retry-After')
                   delay = float(retry_after) if retry_after and retry_after.isdigit() else backoff
                   attempt += 1
                   time.sleep(delay)
                   backoff *= 2
                   continue
   
               if status not in (200, 201):
                   raise RuntimeError(f'HTTP {status}: {response.data[:200]}')
   
               return status, response.headers, response.data
   
           except urllib3.exceptions.HTTPError as e:
               if attempt < max_retries:
                   attempt += 1
                   time.sleep(backoff)
                   backoff *= 2
                   continue
               raise
   
   def start_report_run(api_key, report_id, timeout):
       """Start a new report run."""
       status, _, body = http_request(
           'POST',
           f'{API_BASE}/report/{report_id}/run',
           api_key,
           b'{}',
           timeout
       )
       if status not in (200, 201):
           raise RuntimeError(f'Start run failed: {status} {body[:200]}')
   
   def list_report_history(api_key, status_filter=None, report_type=None, limit=100, timeout=30):
       """List report history."""
       params = {
           'limit': str(limit),
           'offset': '0',
           'sort_by': 'report_start_date',
           'sort_direction': 'DESC'
       }
       if status_filter:
           params['status'] = status_filter
       if report_type:
           params['report_type'] = report_type
   
       query_string = '&'.join([f'{k}={v}' for k, v in params.items()])
       url = f'{API_BASE}/report/history?{query_string}'
   
       status, _, body = http_request('GET', url, api_key, timeout=timeout)
       if status != 200:
           raise RuntimeError(f'History failed: {status} {body[:200]}')
   
       return json.loads(body.decode('utf-8'))
   
   def find_ready_run(api_key, report_id, started_not_before, timeout, max_wait_seconds, poll_interval):
       """Find a ready report run."""
       deadline = time.time() + max_wait_seconds
   
       while time.time() < deadline:
           hist = list_report_history(
               api_key,
               status_filter='READY',
               limit=200,
               timeout=timeout
           ).get('report_history', [])
   
           for item in hist:
               if item.get('report_identifier') != report_id:
                   continue
               if not item.get('report_run_identifier'):
                   continue
   
               try:
                   rsd = datetime.strptime(
                       item.get('report_start_date', ''),
                       '%Y-%m-%d %H:%M:%S'
                   ).replace(tzinfo=timezone.utc)
               except Exception:
                   rsd = started_not_before
   
               if rsd + timedelta(seconds=60) >= started_not_before:
                   return item['report_run_identifier']
   
           time.sleep(poll_interval)
   
       raise TimeoutError('READY run not found in time')
   
   def get_json_rows(api_key, report_id, run_id, timeout):
       """Get JSON rows from report."""
       status, headers, body = http_request(
           'GET',
           f'{API_BASE}/report/{report_id}/{run_id}/json',
           api_key,
           timeout=timeout
       )
   
       if status != 200:
           raise RuntimeError(f'Get JSON failed: {status} {body[:200]}')
   
       # Check if response is ZIP
       content_type = headers.get('content-type', '').lower()
       if 'application/zip' in content_type or body[:2] == b'PK':
           with zipfile.ZipFile(io.BytesIO(body)) as zf:
               json_files = [n for n in zf.namelist() if n.lower().endswith('.json')]
               if not json_files:
                   raise RuntimeError('ZIP has no JSON')
               rows = json.loads(zf.read(json_files[0]).decode('utf-8'))
       else:
           rows = json.loads(body.decode('utf-8'))
   
       if not isinstance(rows, list):
           raise RuntimeError('Unexpected JSON format')
   
       return rows
   
   def write_ndjson_gz(bucket, prefix, rows, run_id):
       """Write NDJSON gzipped file to GCS."""
       ts = datetime.now(timezone.utc).strftime('%Y/%m/%d/%H%M%S')
       key = f'{prefix}/{ts}-digicert-audit-{run_id[:8]}-{uuid.uuid4().hex}.json.gz'
   
       buf = io.BytesIO()
       with gzip.GzipFile(fileobj=buf, mode='wb') as gz:
           for r in rows:
               gz.write((json.dumps(r, separators=(',', ':')) + '\n').encode('utf-8'))
   
       blob = bucket.blob(key)
       blob.upload_from_string(
           buf.getvalue(),
           content_type='application/x-ndjson',
           content_encoding='gzip'
       )
   
       return key
   
   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)}')
   

   • 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	digicert-audit-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 (digicert-audit-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
  Diário	0 0 * * *	Coleta de dados históricos

Testar o job do programador

1. No console do Cloud Scheduler, encontre seu job.
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > digicert-audit-logs-collector > Registros.
4. Verifique se a função foi executada com sucesso.
5. Verifique o bucket do GCS para confirmar se os registros foram gravados.

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, DigiCert Audit Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Digicert 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:

   [e-mail]
   

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

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
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.

   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 do DigiCert

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, DigiCert Audit Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Digicert como o Tipo de registro.
7. Clique em Próxima.

8. 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://digicert-logs/digicert/logs/
     

     • Substitua:

       • digicert-logs: o nome do bucket do GCS.
       • digicert/logs: 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.

9. Clique em Próxima.

10. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

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