Coletar registros do Atlassian Confluence

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros do Atlassian Confluence no Google Security Operations. Primeiro, o analisador tenta extrair campos da mensagem de registro bruta usando expressões regulares (padrões grok) projetadas para registros do Atlassian Confluence. Se a análise grok falhar ou o registro estiver no formato JSON, o código tentará analisar a mensagem como JSON. Por fim, os campos extraídos são mapeados para o esquema UDM do Google SecOps e enriquecidos com mais contexto.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

Uma instância do Google SecOps
Uma conta do Atlassian Confluence Cloud com acesso ao registro de auditoria OU Confluence Data Center/Server com acesso administrativo
Para o método baseado no GCP: acesso privilegiado ao GCP (GCS, IAM, Cloud Run, Pub/Sub, Cloud Scheduler)
Para o método do Bindplane: Windows Server 2016 ou mais recente ou host Linux com systemd

Visão geral das opções de integração

Este guia oferece dois caminhos de integração:

Opção 1: Confluence Data Center/Server via Bindplane + Syslog
Opção 2: registros de auditoria do Confluence Cloud usando a função do Cloud Run do GCP + GCS (formato JSON)

Escolha a opção que melhor se adapta ao tipo de implantação e à infraestrutura do Confluence.

Opção 1: Confluence Data Center/Server via Bindplane + Syslog

Essa opção configura o Confluence Data Center ou Server para enviar registros via syslog a um agente do Bindplane, que os encaminha para o Google SecOps.

Receber o arquivo de autenticação de ingestão do Google SecOps

Faça login no console do Google SecOps.
Acesse Configurações do SIEM > Agentes de coleta.
Clique em Fazer o download para baixar o arquivo de autenticação de ingestão.
Salve o arquivo com segurança no sistema em que o agente do Bindplane será instalado.

Observação: esse arquivo JSON contém credenciais para autenticar o agente do Bindplane no Google SecOps.

Receber o ID de cliente do Google SecOps

Faça login no console do Google SecOps.
Acesse Configurações do SIEM > Perfil.
Copie e salve o ID do cliente na seção Detalhes da organização.

Observação: o ID do cliente é necessário para configurar o exportador do agente Bindplane.

Instalar o agente do BindPlane

Instale o agente do Bindplane no seu sistema operacional Windows ou Linux de acordo com as instruções a seguir.

Instalação do Windows

Abra o prompt de comando ou o PowerShell como administrador.

Execute este comando:

msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quiet

Aguarde a conclusão da instalação.
Execute o seguinte comando para confirmar a instalação:
```
sc query observiq-otel-collector
```

O serviço vai aparecer como EM EXECUÇÃO.

Instalação do Linux

Abra um terminal com privilégios de root ou sudo.

Execute este comando:

sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.sh

Aguarde a conclusão da instalação.
Execute o seguinte comando para confirmar a instalação:
```
sudo systemctl status observiq-otel-collector
```

O serviço vai aparecer como ativo (em execução).

Outros recursos de instalação

Para mais opções de instalação e solução de problemas, consulte o guia de instalação do agente do Bindplane.

Configurar o agente do Bindplane para ingerir syslog e enviar ao Google SecOps

Localizar o arquivo de configuração

Linux:

sudo nano /etc/bindplane-agent/config.yaml

Windows:

notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"

Editar o arquivo de configuração

Substitua todo o conteúdo de config.yaml pela seguinte configuração:

receivers:
  udplog:
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/confluence_logs:
    compression: gzip
    creds_file_path: '/etc/bindplane-agent/ingestion-auth.json'
    customer_id: 'YOUR_CUSTOMER_ID'
    endpoint: malachiteingestion-pa.googleapis.com
    log_type: ATLASSIAN_CONFLUENCE
    raw_log_field: body
    ingestion_labels:
      service: confluence

service:
  pipelines:
    logs/confluence:
      receivers:
        - udplog
      exporters:
        - chronicle/confluence_logs

Parâmetros de configuração

Substitua os seguintes marcadores de posição:
- listen_address: substitua a porta e o endereço IP conforme necessário na sua infraestrutura. Use 0.0.0.0:514 para detectar todas as interfaces na porta 514.
- creds_file_path: atualize o caminho em que o arquivo de autenticação foi salvo:
  - Linux: /etc/bindplane-agent/ingestion-auth.json
  - Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
- customer_id: substitua YOUR_CUSTOMER_ID pelo ID do cliente real da etapa anterior.
- endpoint: URL do endpoint regional:
  - EUA: malachiteingestion-pa.googleapis.com
  - Europa: europe-malachiteingestion-pa.googleapis.com
  - Ásia: asia-southeast1-malachiteingestion-pa.googleapis.com

Salvar o arquivo de configuração

Depois de editar, salve o arquivo:

Linux: pressione Ctrl+O, Enter e Ctrl+X.
Windows: clique em Arquivo > Salvar.

Reinicie o agente do Bindplane para aplicar as mudanças

Reiniciar o agente do Bindplane no Linux

Para reiniciar o agente do Bindplane no Linux, execute o seguinte comando:
```
sudo systemctl restart observiq-otel-collector
```

Verifique se o serviço está em execução:

sudo systemctl status observiq-otel-collector

Verifique se há erros nos registros:

sudo journalctl -u observiq-otel-collector -f

Reiniciar o agente do Bindplane no Windows

Para reiniciar o agente do Bindplane em Windows, escolha uma das seguintes opções:
- Usando o prompt de comando ou o PowerShell como administrador:
```
net stop observiq-otel-collector && net start observiq-otel-collector
```
- Usando o console do Services:
  1. Pressione Win+R, digite services.msc e pressione Enter.
  2. Localize o Coletor do OpenTelemetry da observIQ.
  3. Clique com o botão direito do mouse e selecione Reiniciar.
  4. Verifique se o serviço está em execução:
```
sc query observiq-otel-collector
```
  5. Verifique se há erros nos registros:
```
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
```

Configurar o encaminhamento de syslog no Confluence Data Center/Server

Opção A: configurar o rsyslog para encaminhar arquivos de registros locais (recomendado)

Configure o Confluence para gravar registros em arquivos (comportamento padrão).

Instale o rsyslog se ele não estiver presente:

sudo apt-get install rsyslog  # Debian/Ubuntu
sudo yum install rsyslog      # RHEL/CentOS

Crie o arquivo de configuração do rsyslog /etc/rsyslog.d/confluence.conf:

# Forward Confluence logs to Bindplane
$ModLoad imfile

# Application logs
$InputFileName /opt/atlassian/confluence/logs/atlassian-confluence.log
$InputFileTag confluence-app:
$InputFileStateFile stat-confluence-app
$InputFileSeverity info
$InputFileFacility local0
$InputRunFileMonitor

# Audit logs (JSON format in DC/Server)
$InputFileName <confluence-home-directory>/log/audit/audit.log
$InputFileTag confluence-audit:
$InputFileStateFile stat-confluence-audit
$InputFileSeverity info
$InputFileFacility local1
$InputRunFileMonitor

# Forward to Bindplane agent
*.* @@BINDPLANE_AGENT_IP:514

Substitua BINDPLANE_AGENT_IP pelo endereço IP do agente do Bindplane (por exemplo, 192.168.1.100).
Ajuste os caminhos dos arquivos de registro com base na sua instalação do Confluence:
- Os registros de aplicativos normalmente: <confluence-install>/logs/ ou <local-home>/logs/
- Registros de auditoria: <confluence-home-directory>/log/audit/ (formato JSON)
- Para encontrar o diretório inicial do Confluence, acesse Configurações > Configuração geral > Informações do sistema e procure por Confluence Home ou Local Home.

Reinicie o rsyslog:
```
sudo systemctl restart rsyslog
```

Opção B: configurar o encaminhamento do Syslog do Log4j2

Essa opção exige a modificação da configuração do Log4j2. A opção A (rsyslog) é recomendada para simplificação.

Faça login no servidor do Confluence via SSH ou RDP.
Localize o arquivo de configuração do Log4j2 em:
```
<confluence-install>/confluence/WEB-INF/classes/log4j2.xml
```
Observação: ajuste o caminho com base na sua instalação do Confluence.

Edite o arquivo de configuração para adicionar um appender Syslog:

<Configuration>
  <Appenders>
    <!-- Existing appenders -->
    <Syslog name="SyslogAppender" 
            host="BINDPLANE_AGENT_IP" 
            port="514" 
            protocol="UDP"
            format="RFC5424"
            facility="LOCAL0">
      <PatternLayout pattern="%d{ISO8601} %p [%t] [%c{1}] %m%n"/>
    </Syslog>
  </Appenders>

  <Loggers>
    <Root level="info">
      <AppenderRef ref="SyslogAppender"/>
      <!-- Other appender refs -->
    </Root>

    <!-- Audit logger -->
    <Logger name="com.atlassian.confluence.event.events.security.AuditEvent" 
            level="info" 
            additivity="false">
      <AppenderRef ref="SyslogAppender"/>
    </Logger>
  </Loggers>
</Configuration>

Substitua BINDPLANE_AGENT_IP pelo endereço IP do agente do Bindplane (por exemplo, 192.168.1.100).

Reinicie o Confluence para aplicar as mudanças:
```
sudo systemctl restart confluence
```

Opção 2: registros de auditoria do Confluence Cloud usando uma função do Cloud Run do GCP e o GCS

Esse método usa a função do Cloud Run do GCP para buscar periodicamente registros de auditoria pela API REST de auditoria do Confluence e armazená-los no GCS para ingestão do Google SecOps.

Coletar credenciais da API do Confluence Cloud

Faça login na sua conta da Atlassian.
Acesse https://id.atlassian.com/manage-profile/security/api-tokens.
Clique em Criar um token de API.
Insira um marcador para o token (por exemplo, Google Security Operations Integration).
Clique em Criar.
Copie e salve o token da API com segurança.
Anote o URL do site do Confluence Cloud (por exemplo, https://yoursite.atlassian.net).
Anote o endereço de e-mail da sua conta da Atlassian (usado para autenticação).

Observação: o token da API exige a permissão global "Administrador do Confluence" para acessar os registros de auditoria.

Verifique as permissões

Para verificar se a conta tem as permissões necessárias:

Faça login no Confluence Cloud.
Clique no ícone Configurações (⚙️) no canto superior direito.
Se você encontrar Monitoring > Registro de auditoria na navegação à esquerda, isso significa que você tem as permissões necessárias.
Se essa opção não aparecer, entre em contato com seu administrador para conceder a permissão de Administrador do Confluence.

Testar o acesso à API

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

# Replace with your actual credentials
CONFLUENCE_EMAIL="your-email@example.com"
CONFLUENCE_API_TOKEN="your-api-token"
CONFLUENCE_URL="https://yoursite.atlassian.net"

# Test API access
curl -u "${CONFLUENCE_EMAIL}:${CONFLUENCE_API_TOKEN}" \
  -H "Accept: application/json" \
  "${CONFLUENCE_URL}/wiki/rest/api/audit"

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, `confluence-audit-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

Clique em Criar.

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 confluence-audit-collector-sa.
- Descrição da conta de serviço: insira Service account for Cloud Run function to collect Confluence Cloud audit 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, confluence-audit-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 confluence-audit-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 Confluence Cloud Audit 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	`confluence-audit-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 confluence-audit-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 confluence-audit-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	Descrição
`GCS_BUCKET`	`confluence-audit-logs`	Nome do bucket do GCS
`GCS_PREFIX`	`confluence-audit`	Prefixo para arquivos de registro
`STATE_KEY`	`confluence-audit/state.json`	Caminho do arquivo de estado
`CONFLUENCE_URL`	`https://yoursite.atlassian.net`	URL do site do Confluence
`CONFLUENCE_EMAIL`	`your-email@example.com`	E-mail da conta da Atlassian
`CONFLUENCE_API_TOKEN`	`your-api-token-here`	Token da API
`MAX_RECORDS`	`1000`	Máximo de registros por execução

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, 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', 'confluence-audit/')
STATE_KEY = os.environ.get('STATE_KEY', 'confluence-audit/state.json')
CONFLUENCE_URL = os.environ.get('CONFLUENCE_URL')
CONFLUENCE_EMAIL = os.environ.get('CONFLUENCE_EMAIL')
CONFLUENCE_API_TOKEN = os.environ.get('CONFLUENCE_API_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))

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)

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Confluence Cloud audit logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([GCS_BUCKET, CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        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=24)

        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

        # Convert to Unix milliseconds
        start_millis = to_unix_millis(last_time)
        end_millis = to_unix_millis(now)

        # Fetch logs
        records, newest_event_time = fetch_logs(
            api_base=CONFLUENCE_URL,
            email=CONFLUENCE_EMAIL,
            api_token=CONFLUENCE_API_TOKEN,
            start_time_ms=start_millis,
            end_time_ms=end_millis,
            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(api_base: str, email: str, api_token: str, start_time_ms: int, end_time_ms: int, max_records: int):
    """
    Fetch logs from Confluence Cloud Audit API with pagination and rate limiting.

    Args:
        api_base: Confluence site URL
        email: Atlassian account email
        api_token: API token
        start_time_ms: Start time in Unix milliseconds
        end_time_ms: End time in Unix milliseconds
        max_records: Maximum total records to fetch

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """
    # Clean up URL
    base_url = api_base.rstrip('/')

    # Build authentication header
    auth_string = f"{email}:{api_token}"
    auth_bytes = auth_string.encode('utf-8')
    auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
    headers = {
        'Authorization': f'Basic {auth_b64}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-ConfluenceCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    start_index = 0

    while True:
        page_num += 1

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

        # Build request URL
        url = f"{base_url}/wiki/rest/api/audit?startDate={start_time_ms}&endDate={end_time_ms}&start={start_index}&limit=100"

        try:
            response = http.request('GET', url, 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', [])

            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:
                    # creationDate is in Unix milliseconds
                    event_time_ms = event.get('creationDate')
                    if event_time_ms:
                        event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc)
                        event_time = event_dt.isoformat()
                        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
            current_size = data.get('size', 0)
            if current_size < 100:
                print(f"Reached last page (size={current_size} < limit=100)")
                break

            start_index += current_size

        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[:max_records], newest_time

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 acionador do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.

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	`confluence-audit-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 `confluence-audit-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 em confluence-audit-collector.
Clique na guia Registros.

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

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://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

Acesse Cloud Storage > Buckets.
Clique no nome do bucket.
Navegue até a pasta confluence-audit/.
Verifique se um novo arquivo .ndjson 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 permissões de administrador do Confluence
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, Confluence Cloud Audit Logs).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Atlassian Confluence 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 do Confluence

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, Confluence Cloud Audit Logs).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Atlassian Confluence 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://confluence-audit-logs/confluence-audit/
```
  - Substitua:
    - confluence-audit-logs: o nome do bucket do GCS.
    - confluence-audit: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
  - Exemplos:
    - Bucket raiz: gs://company-logs/
    - Com prefixo: gs://company-logs/confluence-audit/
    - Com subpasta: gs://company-logs/confluence/audit/
  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
agente	read_only_udm.network.http.user_agent	Valor extraído do campo "agent".
app_protocol	read_only_udm.network.application_protocol	Derivado do campo "app_protocol". Se "app_protocol" contiver "HTTPS", "HTTP", "SSH" ou "RDP", o protocolo correspondente será usado. Caso contrário, o padrão será "UNKNOWN_APPLICATION_PROTOCOL".
app_protocol	read_only_udm.network.application_protocol_version	Valor extraído do campo "app_protocol".
auditType.action	read_only_udm.security_result.action	Derivado do campo "auditType.action". Se "auditType.action" contiver "successful", o valor será definido como "ALLOW". Se ele contiver "restricted", o valor será definido como "BLOCK".
auditType.action	read_only_udm.security_result.summary	Valor extraído do campo "auditType.action" quando "auditType" não está vazio e "auditType_area" é "SECURITY".
auditType.actionI18nKey	read_only_udm.metadata.product_event_type	Valor extraído do campo "auditType.actionI18nKey" quando "auditType" não está vazio.
auditType.area	read_only_udm.security_result.detection_fields.value	Valor extraído do campo "auditType.area" e atribuído ao campo "value" de um campo de detecção com o campo "key" definido como "auditType area". Esse mapeamento é feito quando "auditType" não está vazio.
auditType.category	read_only_udm.security_result.category_details	Valor extraído do campo "auditType.category" quando "auditType" não está vazio.
auditType.categoryI18nKey	read_only_udm.security_result.detection_fields.value	Valor extraído do campo "auditType.categoryI18nKey" e atribuído ao campo "value" de um campo de detecção com o campo "key" definido como "auditType categoryI18nKey". Esse mapeamento é feito quando "auditType" não está vazio.
auditType.level	read_only_udm.security_result.detection_fields.value	Valor extraído do campo "auditType.level" e atribuído ao campo "value" de um campo de detecção com o campo "key" definido como "auditType level". Esse mapeamento é feito quando "auditType" não está vazio.
author.displayName	read_only_udm.principal.user.user_display_name	Valor extraído do campo "author.displayName".
author.externalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	Valor extraído do campo "author.externalCollaborator" e atribuído ao campo "value" de um rótulo com o campo "key" definido como "externalCollaborator".
author.id	read_only_udm.principal.user.userid	Valor extraído do campo "author.id" quando "author.type" é "user" e "principal_user_present" é "false".
author.isExternalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	Valor extraído do campo "author.isExternalCollaborator" e atribuído ao campo "value" de um rótulo com o campo "key" definido como "isExternalCollaborator".
author.name	read_only_udm.principal.user.user_display_name	Valor extraído do campo "author.name" quando "author.type" é "user" e "principal_user_present" é "false".
bytes_in	read_only_udm.network.received_bytes	Valor extraído do campo "bytes_in" se ele contiver dígitos. Caso contrário, o padrão é 0.
categoria	read_only_udm.security_result.category_details	Valor extraído do campo "category".
changedValues	read_only_udm.principal.resource.attribute.labels	Itera por cada elemento em "changedValues" e cria rótulos com chaves como "changedValue [index] [key]" e valores dos valores correspondentes na matriz "changedValues".
creationDate	read_only_udm.metadata.event_timestamp	Valor extraído do campo "creationDate", analisado como carimbo de data/hora UNIX ou UNIX_MS.
extraAttributes	read_only_udm.principal.resource.attribute.labels	Itera por cada elemento em "extraAttributes" e cria rótulos com chaves com base nos campos "name" e "nameI18nKey" e nos valores do campo "value" correspondente.
http_verb	read_only_udm.network.http.method	Valor extraído do campo "http_verb".
ip	read_only_udm.target.ip	Valor extraído do campo "ip".
principal_host	read_only_udm.principal.hostname	Valor extraído do campo "principal_host".
referral_url	read_only_udm.network.http.referral_url	Valor extraído do campo "referral_url".
remoteAddress	read_only_udm.principal.ip	Valor extraído do campo "remoteAddress", analisado como um endereço IP.
response_code	read_only_udm.network.http.response_code	Valor extraído do campo "response_code".
session_duration	read_only_udm.additional.fields.value.string_value	Valor extraído do campo "session_duration" e atribuído ao campo "string_value" de um rótulo com o campo "key" definido como "Session Duration".
source	read_only_udm.principal.ip	Valor extraído do campo "source", analisado como um endereço IP.
src_ip	read_only_udm.principal.ip	Valor extraído do campo "src_ip" se "remoteAddress" estiver vazio.
resumo	read_only_udm.security_result.summary	Valor extraído do campo "summary".
sysAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	Valor extraído do campo "sysAdmin" e atribuído ao campo "value" de um rótulo com o campo "key" definido como "sysAdmin".
superAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	Valor extraído do campo "superAdmin" e atribuído ao campo "value" de um rótulo com o campo "key" definido como "superAdmin".
target_url	read_only_udm.target.url	Valor extraído do campo "target_url".
timestamp	read_only_udm.metadata.event_timestamp	Valor extraído do campo "timestamp", analisado como uma string de data e hora.
user_id	read_only_udm.principal.user.userid	Valor extraído do campo "user_id".
	read_only_udm.metadata.event_type	O valor desse campo é determinado por uma série de verificações e o padrão é "GENERIC_EVENT". Ele é definido como valores específicos, como "NETWORK_HTTP", "USER_UNCATEGORIZED" ou "STATUS_UPDATE", com base na presença e no conteúdo de outros campos, como "principal_host", "user_id", "has_principal" e "author.type".
	read_only_udm.metadata.vendor_name	Defina como "ATLASSIAN".
	read_only_udm.metadata.product_name	Defina como "CONFLUENCE".
	read_only_udm.metadata.log_type	Defina como "ATLASSIAN_CONFLUENCE".
	read_only_udm.principal.user.user_display_name	O valor desse campo pode vir de "author.displayName" ou "affectedObject.name", dependendo do contexto.
	read_only_udm.target.process.pid	O valor desse campo pode vir de "principal_host" ou "pid", dependendo do contexto.
	read_only_udm.principal.resource.attribute.labels	Esse campo é preenchido com vários rótulos derivados de campos como "affectedObjects", "changedValues" e "extraAttributes". As chaves e os valores desses rótulos são gerados dinamicamente com base no conteúdo específico desses campos.

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