Recopila registros de Atlassian Confluence

Se admite en los siguientes sistemas operativos:

Google SecOps SIEM

En este documento, se explica cómo transferir registros de Atlassian Confluence a Google Security Operations. Primero, el analizador intenta extraer campos del mensaje de registro sin procesar con expresiones regulares (patrones de Grok) diseñadas para los registros de Atlassian Confluence. Si falla el análisis de grok o el registro está en formato JSON, el código intenta analizar el mensaje como JSON. Por último, los campos extraídos se asignan al esquema de UDM de Google SecOps y se enriquecen con contexto adicional.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

Una instancia de Google SecOps
Una cuenta de Atlassian Confluence Cloud con acceso a los registros de auditoría O Confluence Data Center/Server con acceso de administrador
Para el método basado en GCP: Acceso privilegiado a GCP (GCS, IAM, Cloud Run, Pub/Sub, Cloud Scheduler)
Para el método de Bindplane: Windows Server 2016 o versiones posteriores, o host de Linux con systemd

Descripción general de las opciones de integración

En esta guía, se proporcionan dos rutas de integración:

Opción 1: Confluence Data Center o Server a través de Bindplane y Syslog
Opción 2: Registros de auditoría de Confluence Cloud a través de la función de Cloud Run de GCP y GCS (formato JSON)

Elige la opción que mejor se adapte a tu tipo de implementación e infraestructura de Confluence.

Opción 1: Confluence Data Center o Server a través de Bindplane y Syslog

Esta opción configura Confluence Data Center o Server para que envíe registros a través de syslog a un agente de Bindplane, que luego los reenvía a Google SecOps.

Obtén el archivo de autenticación de transferencia de Google SecOps

Accede a la consola de Google SecOps.
Ve a Configuración de SIEM > Agentes de recopilación.
Haz clic en Descargar para descargar el archivo de autenticación de la transferencia.
Guarda el archivo de forma segura en el sistema en el que se instalará el agente de Bindplane.

Nota: Este archivo JSON contiene credenciales para autenticar el agente de Bindplane en Google SecOps.

Obtén el ID de cliente de Google SecOps

Accede a la consola de Google SecOps.
Ve a Configuración de SIEM > Perfil.
Copia y guarda el ID de cliente de la sección Detalles de la organización.

Nota: Se requiere el ID de cliente para configurar el exportador del agente de Bindplane.

Instala el agente de BindPlane

Instala el agente de Bindplane en tu sistema operativo Windows o Linux según las siguientes instrucciones.

Instalación en Windows

Abre el símbolo del sistema o PowerShell como administrador.

Ejecuta el siguiente comando:

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

Espera a que se complete la instalación.
Ejecute el siguiente comando para verificar la instalación:
```
sc query observiq-otel-collector
```

El servicio debería mostrarse como RUNNING.

Instalación en Linux

Abre una terminal con privilegios de administrador o sudo.

Ejecuta el siguiente comando:

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

Espera a que se complete la instalación.
Ejecute el siguiente comando para verificar la instalación:
```
sudo systemctl status observiq-otel-collector
```

El servicio debería mostrarse como activo (en ejecución).

Recursos de instalación adicionales

Para obtener más opciones de instalación y solucionar problemas, consulta la guía de instalación del agente de Bindplane.

Configura el agente de BindPlane para transferir el syslog y enviarlo a Google SecOps

Ubica el archivo de configuración

Linux:

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

Windows:

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

Edita el archivo de configuración

Reemplaza todo el contenido de config.yaml con la siguiente configuración:

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 configuración

Reemplaza los marcadores de posición que se indican más abajo:
- listen_address: Reemplaza el puerto y la dirección IP según sea necesario en tu infraestructura. Usa 0.0.0.0:514 para escuchar en todas las interfaces del puerto 514.
- creds_file_path: Actualiza la ruta de acceso en la que se guardó el archivo de autenticación:
  - Linux: /etc/bindplane-agent/ingestion-auth.json
  - Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
- customer_id: Reemplaza YOUR_CUSTOMER_ID por el ID de cliente real del paso anterior.
- endpoint: URL del extremo regional:
  - EE.UU.: malachiteingestion-pa.googleapis.com
  - Europa: europe-malachiteingestion-pa.googleapis.com
  - Asia: asia-southeast1-malachiteingestion-pa.googleapis.com

Guarda el archivo de configuración

Después de editarlo, guarda el archivo:

Linux: Presiona Ctrl+O, luego Enter y, después, Ctrl+X.
Windows: Haz clic en Archivo > Guardar

Reinicia el agente de Bindplane para aplicar los cambios

Cómo reiniciar el agente de Bindplane en Linux

Para reiniciar el agente de Bindplane en Linux, ejecuta el siguiente comando:
```
sudo systemctl restart observiq-otel-collector
```

Verifica que el servicio esté en ejecución:

sudo systemctl status observiq-otel-collector

Revisa los registros en busca de errores:

sudo journalctl -u observiq-otel-collector -f

Cómo reiniciar el agente de Bindplane en Windows

Para reiniciar el agente de Bindplane en Windows, elige una de las siguientes opciones:
- Con el símbolo del sistema o PowerShell como administrador, haz lo siguiente:
```
net stop observiq-otel-collector && net start observiq-otel-collector
```
- Usa la consola de Services:
  1. Presiona Win+R, escribe services.msc y presiona Intro.
  2. Busca observIQ OpenTelemetry Collector.
  3. Haz clic con el botón derecho y selecciona Reiniciar.
  4. Verifica que el servicio esté en ejecución:
```
sc query observiq-otel-collector
```
  5. Revisa los registros en busca de errores:
```
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
```

Configura el reenvío de Syslog en Confluence Data Center o Server

Opción A: Configura rsyslog para que reenvíe archivos de registro locales (recomendado)

Configura Confluence para que escriba registros en archivos (comportamiento predeterminado).

Instala rsyslog si no está presente:

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

Crea el archivo de configuración de 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

Reemplaza BINDPLANE_AGENT_IP por la dirección IP del agente de Bindplane (por ejemplo, 192.168.1.100).
Ajusta las rutas de acceso de los archivos de registro según tu instalación de Confluence:
- Por lo general, los registros de la aplicación son <confluence-install>/logs/ o <local-home>/logs/.
- Registros de auditoría: <confluence-home-directory>/log/audit/ (formato JSON)
- Para encontrar el directorio principal de Confluence, ve a Configuración > Configuración general > Información del sistema y busca Confluence Home o Local Home.

Reinicia rsyslog:
```
sudo systemctl restart rsyslog
```

Opción B: Configura el reenvío de Syslog de Log4j2

Esta opción requiere que se modifique la configuración de Log4j2. Se recomienda la opción A (rsyslog) por su simplicidad.

Accede a tu servidor de Confluence a través de SSH o RDP.
Ubica el archivo de configuración de Log4j2 en la siguiente ubicación:
```
<confluence-install>/confluence/WEB-INF/classes/log4j2.xml
```
Nota: Ajusta la ruta de acceso según tu instalación de Confluence.

Edita el archivo de configuración para agregar un appender de 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>

Reemplaza BINDPLANE_AGENT_IP por la dirección IP del agente de Bindplane (por ejemplo, 192.168.1.100).

Reinicia Confluence para aplicar los cambios:
```
sudo systemctl restart confluence
```

Opción 2: Registros de auditoría de Confluence Cloud a través de la función de Cloud Run de GCP y GCS

Este método usa la función de Cloud Run de GCP para recuperar periódicamente los registros de auditoría a través de la API de REST de Confluence Audit y almacenarlos en GCS para la transferencia de datos de SecOps de Google.

Recopila credenciales de la API de Confluence Cloud

Accede a tu cuenta de Atlassian.
Ve a https://id.atlassian.com/manage-profile/security/api-tokens.
Haz clic en Crear token de API.
Ingresa una etiqueta para el token (por ejemplo, Google Security Operations Integration).
Haz clic en Crear.
Copia y guarda el token de API de forma segura.
Anota la URL de tu sitio de Confluence Cloud (por ejemplo, https://yoursite.atlassian.net).
Anota la dirección de correo electrónico de tu cuenta de Atlassian (se usa para la autenticación).

Nota: El token de API requiere el permiso global de "Administrador de Confluence" para acceder a los registros de auditoría.

Verifica los permisos

Para verificar que la cuenta tenga los permisos necesarios, sigue estos pasos:

Accede a Confluence Cloud.
Haz clic en el ícono de Configuración (⚙️) en la esquina superior derecha.
Si ves Monitoring > Registro de auditoría en el panel de navegación de la izquierda, tienes los permisos necesarios.
Si no ves esta opción, comunícate con tu administrador para que te otorgue permiso de administrador de Confluence.

Prueba el acceso a la API

Prueba tus credenciales antes de continuar con la integración:

# 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"

Crea un bucket de Google Cloud Storage

Ve a Google Cloud Console.
Selecciona tu proyecto o crea uno nuevo.
En el menú de navegación, ve a Cloud Storage > Buckets.
Haz clic en Crear bucket.

Proporciona los siguientes detalles de configuración:

Configuración	Valor
Asigna un nombre a tu bucket	Ingresa un nombre global único (por ejemplo, `confluence-audit-logs`).
Tipo de ubicación	Elige según tus necesidades (región, birregional, multirregional)
Ubicación	Selecciona la ubicación (por ejemplo, `us-central1`).
Clase de almacenamiento	Estándar (recomendado para los registros a los que se accede con frecuencia)
Control de acceso	Uniforme (recomendado)
Herramientas de protección	Opcional: Habilita el control de versiones de objetos o la política de retención

Haz clic en Crear.

Crea una cuenta de servicio para la Cloud Run Function

La Cloud Run Function necesita una cuenta de servicio con permisos para escribir en el bucket de GCS y ser invocada por Pub/Sub.

Crear cuenta de servicio

En GCP Console, ve a IAM y administración > Cuentas de servicio.
Haz clic en Crear cuenta de servicio.
Proporciona los siguientes detalles de configuración:
- Nombre de la cuenta de servicio: Ingresa confluence-audit-collector-sa.
- Descripción de la cuenta de servicio: Ingresa Service account for Cloud Run function to collect Confluence Cloud audit logs.
Haz clic en Crear y continuar.
En la sección Otorga a esta cuenta de servicio acceso al proyecto, agrega los siguientes roles:
1. Haz clic en Selecciona un rol.
2. Busca y selecciona Administrador de objetos de almacenamiento.
3. Haz clic en + Agregar otra función.
4. Busca y selecciona Invocador de Cloud Run.
5. Haz clic en + Agregar otra función.
6. Busca y selecciona Cloud Functions Invoker.
Haz clic en Continuar.
Haz clic en Listo.

Estos roles son necesarios para las siguientes acciones:

Administrador de objetos de almacenamiento: Escribe registros en el bucket de GCS y administra archivos de estado
Invocador de Cloud Run: Permite que Pub/Sub invoque la función
Cloud Functions Invoker: Permite la invocación de funciones

Otorga permisos de IAM en el bucket de GCS

Otorga permisos de escritura a la cuenta de servicio en el bucket de GCS:

Ve a Cloud Storage > Buckets.
Haz clic en el nombre de tu bucket.
Ve a la pestaña Permisos.
Haz clic en Otorgar acceso.
Proporciona los siguientes detalles de configuración:
- Agregar principales: Ingresa el correo electrónico de la cuenta de servicio (por ejemplo, confluence-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
- Asignar roles: Selecciona Administrador de objetos de Storage.
Haz clic en Guardar.

Crear tema de Pub/Sub

Crea un tema de Pub/Sub en el que Cloud Scheduler publicará y al que se suscribirá la función de Cloud Run.

En GCP Console, ve a Pub/Sub > Temas.
Haz clic en Crear un tema.
Proporciona los siguientes detalles de configuración:
- ID del tema: Ingresa confluence-audit-trigger.
- Deja el resto de la configuración con sus valores predeterminados.
Haz clic en Crear.

Crea una función de Cloud Run para recopilar registros

La función de Cloud Run se activa con los mensajes de Pub/Sub de Cloud Scheduler para recuperar registros de la API de Confluence Cloud Audit y escribirlos en GCS.

En GCP Console, ve a Cloud Run.
Haz clic en Crear servicio.
Selecciona Función (usa un editor intercalado para crear una función).

En la sección Configurar, proporciona los siguientes detalles de configuración:

Configuración	Valor
Nombre del servicio	`confluence-audit-collector`
Región	Selecciona la región que coincida con tu bucket de GCS (por ejemplo, `us-central1`).
Tiempo de ejecución	Selecciona Python 3.12 o una versión posterior.

En la sección Activador (opcional), haz lo siguiente:
1. Haz clic en + Agregar activador.
2. Selecciona Cloud Pub/Sub.
3. En Selecciona un tema de Cloud Pub/Sub, elige confluence-audit-trigger.
4. Haz clic en Guardar.
En la sección Autenticación, haz lo siguiente:
1. Selecciona Solicitar autenticación.
2. Verifica Identity and Access Management (IAM).
Nota: Pub/Sub controlará automáticamente la autenticación cuando invoque la función.
Desplázate hacia abajo y expande Contenedores, redes y seguridad.
Ve a la pestaña Seguridad:
- Cuenta de servicio: Selecciona confluence-audit-collector-sa.

Ve a la pestaña Contenedores:

Haz clic en Variables y secretos.
Haz clic en + Agregar variable para cada variable de entorno:

Nombre de la variable	Valor de ejemplo	Descripción
`GCS_BUCKET`	`confluence-audit-logs`	Nombre del bucket de GCS
`GCS_PREFIX`	`confluence-audit`	Prefijo para los archivos de registro
`STATE_KEY`	`confluence-audit/state.json`	Ruta de acceso al archivo de estado
`CONFLUENCE_URL`	`https://yoursite.atlassian.net`	URL del sitio de Confluence
`CONFLUENCE_EMAIL`	`your-email@example.com`	Correo electrónico de la cuenta de Atlassian
`CONFLUENCE_API_TOKEN`	`your-api-token-here`	Token de API
`MAX_RECORDS`	`1000`	Cantidad máxima de registros por ejecución

En la sección Variables y Secrets, desplázate hacia abajo hasta Solicitudes:
- Tiempo de espera de la solicitud: Ingresa 600 segundos (10 minutos).
Ve a la pestaña Configuración:
- En la sección Recursos, haz lo siguiente:
  - Memoria: Selecciona 512 MiB o más.
  - CPU: Selecciona 1.
En la sección Ajuste de escala de revisión, haz lo siguiente:
- Cantidad mínima de instancias: Ingresa 0.
- Cantidad máxima de instancias: Ingresa 100 (o ajusta según la carga esperada).
Haz clic en Crear.
Espera a que se cree el servicio (de 1 a 2 minutos).
Después de crear el servicio, se abrirá automáticamente el editor de código intercalado.

Agregar el código de función

Ingresa main en Punto de entrada de la función.

En el editor de código intercalado, crea dos archivos:

Primer archivo: 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 archivo: requirements.txt:

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

Haz clic en Implementar para guardar y, luego, implementar la función.
Espera a que se complete la implementación (de 2 a 3 minutos).

Nota: La configuración del activador de Pub/Sub crea automáticamente las suscripciones y los permisos necesarios.

Crea un trabajo de Cloud Scheduler

Cloud Scheduler publica mensajes en el tema de Pub/Sub a intervalos regulares, lo que activa la función de Cloud Run.

En GCP Console, ve a Cloud Scheduler.
Haz clic en Crear trabajo.

Proporciona los siguientes detalles de configuración:

Configuración	Valor
Nombre	`confluence-audit-collector-hourly`
Región	Selecciona la misma región que la función de Cloud Run
Frecuencia	`0 * * * *` (cada hora, en punto)
Zona horaria	Selecciona la zona horaria (se recomienda UTC)
Tipo de orientación	Pub/Sub
Tema	Seleccionar `confluence-audit-trigger`
Cuerpo del mensaje	`{}` (objeto JSON vacío)

Haz clic en Crear.

Opciones de frecuencia de programación

Elige la frecuencia según los requisitos de latencia y volumen de registros:

Frecuencia	Expresión cron	Caso de uso
Cada 5 minutos	`/5 * * *`	Alto volumen y baja latencia
Cada 15 minutos	`/15 * * *`	Volumen medio
Cada 1 hora	`0 * * * *`	Estándar (opción recomendada)
Cada 6 horas	`0 /6 * *`	Procesamiento por lotes y volumen bajo
Diario	`0 0 * * *`	Recopilación de datos históricos

Prueba la integración

En la consola de Cloud Scheduler, busca tu trabajo.
Haz clic en Forzar ejecución para activar el trabajo de forma manual.
Espera unos segundos.
Ve a Cloud Run > Servicios.
Haz clic en confluence-audit-collector:
Haz clic en la pestaña Registros.

Verifica que la función se haya ejecutado correctamente. Busca lo siguiente:

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

Ve a Cloud Storage > Buckets.
Haz clic en el nombre de tu bucket.
Navega a la carpeta confluence-audit/.
Verifica que se haya creado un archivo .ndjson nuevo con la marca de tiempo actual.

Si ves errores en los registros, haz lo siguiente:

HTTP 401: Verifica las credenciales de la API en las variables de entorno
HTTP 403: Verifica que la cuenta tenga permisos de administrador de Confluence
HTTP 429: Limitación de frecuencia. La función volverá a intentarlo automáticamente con una espera exponencial.
Faltan variables de entorno: Verifica que estén configuradas todas las variables requeridas.

Recupera la cuenta de servicio de Google SecOps

Las Operaciones de seguridad de Google usan una cuenta de servicio única para leer datos de tu bucket de GCS. Debes otorgar acceso a tu bucket a esta cuenta de servicio.

Obtén el correo electrónico de la cuenta de servicio

Ve a Configuración de SIEM > Feeds.
Haz clic en Agregar feed nuevo.
Haz clic en Configura un feed único.
En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Confluence Cloud Audit Logs).
Selecciona Google Cloud Storage V2 como el Tipo de fuente.
Selecciona Atlassian Confluence como el Tipo de registro.
Haz clic en Obtener cuenta de servicio. Se mostrará un correo electrónico único de la cuenta de servicio, por ejemplo:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
Copia esta dirección de correo electrónico para usarla en el siguiente paso.

Nota: Cada instancia de Google SecOps tiene una cuenta de servicio única. No uses cuentas de servicio de otros ejemplos o documentación.

Otorga permisos de IAM a la cuenta de servicio de Google SecOps

La cuenta de servicio de Google SecOps necesita el rol de visualizador de objetos de almacenamiento en tu bucket de GCS.

Ve a Cloud Storage > Buckets.
Haz clic en el nombre de tu bucket.
Ve a la pestaña Permisos.
Haz clic en Otorgar acceso.
Proporciona los siguientes detalles de configuración:
- Agregar principales: Pega el correo electrónico de la cuenta de servicio de Google SecOps.
- Asignar roles: Selecciona Visualizador de objetos de Storage.
Haz clic en Guardar.

Nota: Si planeas usar la opción de eliminación "Borrar archivos transferidos" o "Borrar archivos transferidos y directorios vacíos", otorga el rol de Administrador de objetos de Storage en lugar del rol de Visualizador de objetos de Storage.

Configura un feed en Google SecOps para transferir registros de Confluence

Ve a Configuración de SIEM > Feeds.
Haz clic en Agregar feed nuevo.
Haz clic en Configura un feed único.
En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Confluence Cloud Audit Logs).
Selecciona Google Cloud Storage V2 como el Tipo de fuente.
Selecciona Atlassian Confluence como el Tipo de registro.
Haz clic en Siguiente.
Especifica valores para los siguientes parámetros de entrada:
- URL del bucket de almacenamiento: Ingresa el URI del bucket de GCS con la ruta de acceso del prefijo:
```
gs://confluence-audit-logs/confluence-audit/
```
  - Reemplaza lo siguiente:
    - confluence-audit-logs: Es el nombre de tu bucket de GCS.
    - confluence-audit: Es el prefijo o la ruta de carpeta opcionales en los que se almacenan los registros (déjalo vacío para la raíz).
  - Ejemplos:
    - Bucket raíz: gs://company-logs/
    - Con prefijo: gs://company-logs/confluence-audit/
    - Con subcarpeta: gs://company-logs/confluence/audit/
  Nota: Siempre incluye la barra diagonal final (/) al final del URI.
- Opción de borrado de la fuente: Selecciona la opción de borrado según tu preferencia:
  - Nunca: Nunca borra ningún archivo después de las transferencias (se recomienda para las pruebas).
  - Borrar archivos transferidos: Borra los archivos después de la transferencia exitosa.
  - Borrar los archivos transferidos y los directorios vacíos: Borra los archivos y los directorios vacíos después de la transferencia exitosa.
    
    Nota: Si seleccionas una opción de eliminación, la cuenta de servicio debe tener el rol de administrador de objetos de Storage en lugar del rol de visualizador de objetos de Storage. Actualiza los permisos de IAM según corresponda.
- Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es de 180 días.
- Espacio de nombres del recurso: Es el espacio de nombres del recurso.
- Etiquetas de transmisión: Es la etiqueta que se aplicará a los eventos de este feed.
Haz clic en Siguiente.
Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.

Tabla de asignación de UDM

Campo de registro	Asignación de UDM	Lógica
agente	read_only_udm.network.http.user_agent	Valor tomado del campo "agent".
app_protocol	read_only_udm.network.application_protocol	Se deriva del campo "app_protocol". Si "app_protocol" contiene "HTTPS", "HTTP", "SSH" o "RDP", se usa el protocolo correspondiente. De lo contrario, el valor predeterminado es "UNKNOWN_APPLICATION_PROTOCOL".
app_protocol	read_only_udm.network.application_protocol_version	Es el valor que se toma del campo "app_protocol".
auditType.action	read_only_udm.security_result.action	Se deriva del campo "auditType.action". Si "auditType.action" contiene "successful", el valor se establece en "ALLOW". Si contiene "restricted", el valor se establece en "BLOCK".
auditType.action	read_only_udm.security_result.summary	Valor que se toma del campo "auditType.action" cuando "auditType" no está vacío y "auditType_area" es "SECURITY".
auditType.actionI18nKey	read_only_udm.metadata.product_event_type	Valor que se toma del campo "auditType.actionI18nKey" cuando "auditType" no está vacío.
auditType.area	read_only_udm.security_result.detection_fields.value	El valor se toma del campo "auditType.area" y se asigna al campo "value" de un campo de detección con el campo "key" establecido en "auditType area". Esta asignación se realiza cuando "auditType" no está vacío.
auditType.category	read_only_udm.security_result.category_details	Valor que se toma del campo "auditType.category" cuando "auditType" no está vacío.
auditType.categoryI18nKey	read_only_udm.security_result.detection_fields.value	Valor tomado del campo "auditType.categoryI18nKey" y asignado al campo "value" de un campo de detección con el campo "key" establecido en "auditType categoryI18nKey". Esta asignación se realiza cuando "auditType" no está vacío.
auditType.level	read_only_udm.security_result.detection_fields.value	Valor tomado del campo "auditType.level" y asignado al campo "value" de un campo de detección con el campo "key" establecido en "auditType level". Esta asignación se realiza cuando "auditType" no está vacío.
author.displayName	read_only_udm.principal.user.user_display_name	Valor tomado del campo "author.displayName".
author.externalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	El valor se toma del campo "author.externalCollaborator" y se asigna al campo "value" de una etiqueta con el campo "key" establecido en "externalCollaborator".
author.id	read_only_udm.principal.user.userid	Valor que se toma del campo "author.id" cuando "author.type" es "user" y "principal_user_present" es "false".
author.isExternalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	El valor se toma del campo "author.isExternalCollaborator" y se asigna al campo "value" de una etiqueta con el campo "key" establecido en "isExternalCollaborator".
author.name	read_only_udm.principal.user.user_display_name	El valor se toma del campo "author.name" cuando "author.type" es "user" y "principal_user_present" es "false".
bytes_in	read_only_udm.network.received_bytes	Valor tomado del campo "bytes_in" si contiene dígitos. De lo contrario, el valor predeterminado es 0.
categoría	read_only_udm.security_result.category_details	Valor tomado del campo "categoría".
changedValues	read_only_udm.principal.resource.attribute.labels	Itera por cada elemento de "changedValues" y crea etiquetas con claves como "changedValue [índice] [clave]" y valores de los valores correspondientes en el array "changedValues".
creationDate	read_only_udm.metadata.event_timestamp	Valor tomado del campo "creationDate", analizado como marca de tiempo UNIX o UNIX_MS.
extraAttributes	read_only_udm.principal.resource.attribute.labels	Itera por cada elemento de "extraAttributes" y crea etiquetas con claves basadas en los campos "name" y "nameI18nKey", y los valores del campo "value" correspondiente.
http_verb	read_only_udm.network.http.method	Es el valor que se toma del campo "http_verb".
ip	read_only_udm.target.ip	Valor tomado del campo "ip".
principal_host	read_only_udm.principal.hostname	Valor tomado del campo "principal_host".
referral_url	read_only_udm.network.http.referral_url	Valor tomado del campo "referral_url".
remoteAddress	read_only_udm.principal.ip	Valor tomado del campo "remoteAddress", analizado como una dirección IP.
response_code	read_only_udm.network.http.response_code	Es el valor que se toma del campo "response_code".
session_duration	read_only_udm.additional.fields.value.string_value	Valor tomado del campo "session_duration" y asignado al campo "string_value" de una etiqueta con el campo "key" establecido en "Duración de la sesión".
source	read_only_udm.principal.ip	Valor tomado del campo "source" y analizado como una dirección IP.
src_ip	read_only_udm.principal.ip	Valor que se toma del campo "src_ip" si "remoteAddress" está vacío.
resumen	read_only_udm.security_result.summary	Valor tomado del campo "summary".
sysAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	El valor se toma del campo "sysAdmin" y se asigna al campo "value" de una etiqueta con el campo "key" establecido en "sysAdmin".
superAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	Valor tomado del campo "superAdmin" y asignado al campo "value" de una etiqueta con el campo "key" establecido en "superAdmin".
target_url	read_only_udm.target.url	Valor tomado del campo "target_url".
timestamp	read_only_udm.metadata.event_timestamp	Valor tomado del campo "timestamp", analizado como una cadena de fecha y hora.
user_id	read_only_udm.principal.user.userid	Valor tomado del campo "user_id".
	read_only_udm.metadata.event_type	El valor de este campo se determina mediante una serie de verificaciones y, de forma predeterminada, se establece en "GENERIC_EVENT". Se establece en valores específicos, como "NETWORK_HTTP", "USER_UNCATEGORIZED" o "STATUS_UPDATE", según la presencia y el contenido de otros campos, como "principal_host", "user_id", "has_principal" y "author.type".
	read_only_udm.metadata.vendor_name	Se debe establecer en "ATLASSIAN".
	read_only_udm.metadata.product_name	Se establece en "CONFLUENCE".
	read_only_udm.metadata.log_type	Se debe establecer en "ATLASSIAN_CONFLUENCE".
	read_only_udm.principal.user.user_display_name	El valor de este campo puede provenir de "author.displayName" o "affectedObject.name", según el contexto.
	read_only_udm.target.process.pid	El valor de este campo puede provenir de "principal_host" o "pid", según el contexto.
	read_only_udm.principal.resource.attribute.labels	Este campo se propaga con varias etiquetas derivadas de campos como "affectedObjects", "changedValues" y "extraAttributes". Las claves y los valores de estas etiquetas se generan de forma dinámica según el contenido específico de estos campos.

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.