Collecter les journaux Atlassian Confluence

Compatible avec :

Google SecOps SIEM

Ce document explique comment ingérer des journaux Atlassian Confluence dans Google Security Operations. L'analyseur tente d'abord d'extraire les champs du message de journal brut à l'aide d'expressions régulières (modèles Grok) conçues pour les journaux Atlassian Confluence. Si l'analyse grok échoue ou si le journal est au format JSON, le code tente ensuite d'analyser le message au format JSON. Enfin, les champs extraits sont mappés au schéma UDM Google SecOps et enrichis avec un contexte supplémentaire.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

Une instance Google SecOps
Un compte Atlassian Confluence Cloud avec accès aux journaux d'audit OU un compte Confluence Data Center/Server avec accès administrateur
Pour la méthode basée sur GCP : accès privilégié à GCP (GCS, IAM, Cloud Run, Pub/Sub, Cloud Scheduler)
Pour la méthode Bindplane : Windows Server 2016 ou version ultérieure, ou hôte Linux avec systemd

Présentation des options d'intégration

Ce guide propose deux méthodes d'intégration :

Option 1 : Confluence Data Center/Server via Bindplane + Syslog
Option 2 : Journaux d'audit Confluence Cloud via la fonction GCP Cloud Run + GCS (format JSON)

Choisissez l'option qui correspond le mieux à votre type de déploiement et à votre infrastructure Confluence.

Option 1 : Confluence Data Center/Server via Bindplane + Syslog

Cette option configure Confluence Data Center ou Server pour envoyer des journaux via syslog à un agent Bindplane, qui les transmet ensuite à Google SecOps.

Obtenir le fichier d'authentification d'ingestion Google SecOps

Connectez-vous à la console Google SecOps.
Accédez à Paramètres du SIEM > Agents de collecte.
Cliquez sur Télécharger pour télécharger le fichier d'authentification pour l'ingestion.
Enregistrez le fichier de manière sécurisée sur le système sur lequel l'agent Bindplane sera installé.

Remarque : Ce fichier JSON contient les identifiants permettant d'authentifier l'agent Bindplane auprès de Google SecOps.

Obtenir l'ID client Google SecOps

Connectez-vous à la console Google SecOps.
Accédez à Paramètres SIEM> Profil.
Copiez et enregistrez le numéro client de la section Informations sur l'organisation.

Remarque : L'ID client est requis pour configurer l'exportateur de l'agent Bindplane.

Installer l'agent BindPlane

Installez l'agent Bindplane sur votre système d'exploitation Windows ou Linux en suivant les instructions ci-dessous.

Installation de fenêtres

Ouvrez l'invite de commandes ou PowerShell en tant qu'administrateur.

Exécutez la commande suivante :

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

Attendez la fin de l'installation.
Vérifiez l'installation en exécutant la commande suivante :
```
sc query observiq-otel-collector
```

Le service doit être indiqué comme RUNNING (EN COURS D'EXÉCUTION).

Installation de Linux

Ouvrez un terminal avec les droits root ou sudo.

Exécutez la commande suivante :

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

Attendez la fin de l'installation.
Vérifiez l'installation en exécutant la commande suivante :
```
sudo systemctl status observiq-otel-collector
```

Le service doit être indiqué comme actif (en cours d'exécution).

Autres ressources d'installation

Pour obtenir d'autres options d'installation et de dépannage, consultez le guide d'installation de l'agent Bindplane.

Configurer l'agent Bindplane pour ingérer les journaux syslog et les envoyer à Google SecOps

Localiser le fichier de configuration

Linux :

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

Windows :

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

Modifiez le fichier de configuration

Remplacez l'intégralité du contenu de config.yaml par la configuration suivante :

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

Paramètres de configuration

Remplacez les espaces réservés suivants :
- listen_address : remplacez le port et l'adresse IP selon les besoins de votre infrastructure. Utilisez 0.0.0.0:514 pour écouter sur toutes les interfaces sur le port 514.
- creds_file_path : mettez à jour le chemin d'accès où le fichier d'authentification a été enregistré :
  - Linux : /etc/bindplane-agent/ingestion-auth.json
  - Windows : C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
- customer_id : remplacez YOUR_CUSTOMER_ID par le numéro client réel de l'étape précédente.
- endpoint : URL du point de terminaison régional :
  - États-Unis : malachiteingestion-pa.googleapis.com
  - Europe : europe-malachiteingestion-pa.googleapis.com
  - Asie : asia-southeast1-malachiteingestion-pa.googleapis.com

Enregistrez le fichier de configuration.

Après avoir modifié le fichier, enregistrez-le :

Linux : appuyez sur Ctrl+O, puis sur Enter, puis sur Ctrl+X.
Windows : cliquez sur Fichier > Enregistrer.

Redémarrez l'agent Bindplane pour appliquer les modifications.

Redémarrer l'agent Bindplane sous Linux

Pour redémarrer l'agent Bindplane sous Linux, exécutez la commande suivante :
```
sudo systemctl restart observiq-otel-collector
```
Vérifiez que le service est en cours d'exécution :
```
sudo systemctl status observiq-otel-collector
```

Recherchez les erreurs dans les journaux :

sudo journalctl -u observiq-otel-collector -f

Redémarrer l'agent Bindplane sous Windows

Pour redémarrer l'agent Bindplane dans Windows, choisissez l'une des options suivantes :
- À l'aide de l'invite de commandes ou de PowerShell en tant qu'administrateur :
```
net stop observiq-otel-collector && net start observiq-otel-collector
```
- Utiliser la console Services :
  1. Appuyez sur Win+R, saisissez services.msc, puis appuyez sur Entrée.
  2. Localisez observIQ OpenTelemetry Collector.
  3. Effectuez un clic droit, puis sélectionnez Redémarrer.
  4. Vérifiez que le service est en cours d'exécution :
```
sc query observiq-otel-collector
```
  5. Recherchez les erreurs dans les journaux :
```
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
```

Configurer le transfert Syslog sur Confluence Data Center/Server

Option A : Configurer rsyslog pour transférer les fichiers journaux locaux (recommandé)

Configurez Confluence pour qu'il écrive les journaux dans des fichiers (comportement par défaut).

Installez rsyslog si ce n'est pas déjà fait :

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

Créez le fichier de configuration 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

Remplacez BINDPLANE_AGENT_IP par l'adresse IP de l'agent Bindplane (par exemple, 192.168.1.100).
Ajustez les chemins d'accès aux fichiers journaux en fonction de votre installation Confluence :
- Journaux d'application : <confluence-install>/logs/ ou <local-home>/logs/
- Journaux d'audit : <confluence-home-directory>/log/audit/ (format JSON)
- Pour trouver votre répertoire personnel Confluence, accédez à Paramètres > Configuration générale > Informations système, puis recherchez Répertoire personnel Confluence ou Répertoire personnel local.

Redémarrez rsyslog :
```
sudo systemctl restart rsyslog
```

Option B : Configurer le transfert Syslog Log4j2

Cette option nécessite de modifier la configuration Log4j2. L'option A (rsyslog) est recommandée pour sa simplicité.

Connectez-vous à votre serveur Confluence via SSH ou RDP.
Localisez le fichier de configuration Log4j2 à l'adresse suivante :
```
<confluence-install>/confluence/WEB-INF/classes/log4j2.xml
```
Remarque : Ajustez le chemin d'accès en fonction de votre installation Confluence.

Modifiez le fichier de configuration pour ajouter un 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>

Remplacez BINDPLANE_AGENT_IP par l'adresse IP de l'agent Bindplane (par exemple, 192.168.1.100).

Redémarrez Confluence pour appliquer les modifications :
```
sudo systemctl restart confluence
```

Option 2 : Journaux d'audit Confluence Cloud via une fonction GCP Cloud Run et GCS

Cette méthode utilise la fonction GCP Cloud Run pour extraire régulièrement les journaux d'audit via l'API REST Confluence Audit et les stocker dans GCS pour l'ingestion Google SecOps.

Collecter les identifiants de l'API Confluence Cloud

Connectez-vous à votre compte Atlassian.
Accédez à https://id.atlassian.com/manage-profile/security/api-tokens.
Cliquez sur Create API token (Créer un jeton d'API).
Saisissez un libellé pour le jeton (par exemple, Google Security Operations Integration).
Cliquez sur Créer.
Copiez et enregistrez le jeton d'API de manière sécurisée.
Notez l'URL de votre site Confluence Cloud (par exemple, https://yoursite.atlassian.net).
Notez l'adresse e-mail de votre compte Atlassian (utilisée pour l'authentification).

Remarque : Le jeton d'API nécessite l'autorisation globale "Administrateur Confluence" pour accéder aux journaux d'audit.

Vérifier les autorisations

Pour vérifier que le compte dispose des autorisations requises :

Connectez-vous à Confluence Cloud.
Cliquez sur l'icône Paramètres (⚙️) en haut à droite.
Si vous voyez Monitoring > Journal d'audit dans le panneau de navigation de gauche, cela signifie que vous disposez des autorisations requises.
Si vous ne voyez pas cette option, contactez votre administrateur pour qu'il vous accorde l'autorisation Administrateur Confluence.

Tester l'accès à l'API

Testez vos identifiants avant de procéder à l'intégration :

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

Créer un bucket Google Cloud Storage

Accédez à la console Google Cloud.
Sélectionnez votre projet ou créez-en un.
Dans le menu de navigation, accédez à Cloud Storage> Buckets.
Cliquez sur Créer un bucket.

Fournissez les informations de configuration suivantes :

Paramètre	Valeur
Nommer votre bucket	Saisissez un nom unique (par exemple, `confluence-audit-logs`).
Type d'emplacement	Choisissez en fonction de vos besoins (région, birégion ou multirégion).
Emplacement	Sélectionnez l'emplacement (par exemple, `us-central1`).
Classe de stockage	Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
Access control (Contrôle des accès)	Uniforme (recommandé)
Outils de protection	Facultatif : Activer la gestion des versions des objets ou la règle de conservation

Cliquez sur Créer.

Créer un compte de service pour la fonction Cloud Run

La fonction Cloud Run a besoin d'un compte de service disposant des autorisations nécessaires pour écrire dans le bucket GCS et être appelée par Pub/Sub.

Créer un compte de service

Dans la console GCP, accédez à IAM et administration > Comptes de service.
Cliquez sur Créer un compte de service.
Fournissez les informations de configuration suivantes :
- Nom du compte de service : saisissez confluence-audit-collector-sa.
- Description du compte de service : saisissez Service account for Cloud Run function to collect Confluence Cloud audit logs.
Cliquez sur Créer et continuer.
Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
1. Cliquez sur Sélectionner un rôle.
2. Recherchez et sélectionnez Administrateur des objets de l'espace de stockage.
3. Cliquez sur + Ajouter un autre rôle.
4. Recherchez et sélectionnez Demandeur Cloud Run.
5. Cliquez sur + Ajouter un autre rôle.
6. Recherchez et sélectionnez Demandeur Cloud Functions.
Cliquez sur Continuer.
Cliquez sur OK.

Ces rôles sont requis pour :

Administrateur des objets Storage : écrire des journaux dans le bucket GCS et gérer les fichiers d'état
Demandeur Cloud Run : autorise Pub/Sub à appeler la fonction
Demandeur Cloud Functions : autorise l'appel de fonctions

Accorder des autorisations IAM sur un bucket GCS

Accordez au compte de service des autorisations d'écriture sur le bucket GCS :

Accédez à Cloud Storage > Buckets.
Cliquez sur le nom de votre bucket.
Accédez à l'onglet Autorisations.
Cliquez sur Accorder l'accès.
Fournissez les informations de configuration suivantes :
- Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (par exemple, confluence-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
- Attribuer des rôles : sélectionnez Administrateur des objets Storage.
Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

Créez un sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.

Dans la console GCP, accédez à Pub/Sub > Sujets.
Cliquez sur Create topic (Créer un sujet).
Fournissez les informations de configuration suivantes :
- ID du sujet : saisissez confluence-audit-trigger.
- Conservez les valeurs par défaut des autres paramètres.
Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

La fonction Cloud Run est déclenchée par les messages Pub/Sub de Cloud Scheduler pour extraire les journaux de l'API Confluence Cloud Audit et les écrire dans GCS.

Dans la console GCP, accédez à Cloud Run.
Cliquez sur Créer un service.
Sélectionnez Fonction (utilisez un éditeur intégré pour créer une fonction).

Dans la section Configurer, fournissez les informations de configuration suivantes :

Paramètre	Valeur
Nom du service	`confluence-audit-collector`
Région	Sélectionnez la région correspondant à votre bucket GCS (par exemple, `us-central1`).
Runtime (durée d'exécution)	Sélectionnez Python 3.12 ou version ultérieure.

Dans la section Déclencheur (facultatif) :
1. Cliquez sur + Ajouter un déclencheur.
2. Sélectionnez Cloud Pub/Sub.
3. Dans Sélectionner un sujet Cloud Pub/Sub, choisissez confluence-audit-trigger.
4. Cliquez sur Enregistrer.
Dans la section Authentification :
1. Sélectionnez Exiger l'authentification.
2. Consultez Identity and Access Management (IAM).
Remarque : Pub/Sub gère automatiquement l'authentification lors de l'appel de la fonction.
Faites défiler la page vers le bas, puis développez Conteneurs, mise en réseau, sécurité.
Accédez à l'onglet Sécurité :
- Compte de service : sélectionnez confluence-audit-collector-sa.

Accédez à l'onglet Conteneurs :

Cliquez sur Variables et secrets.
Cliquez sur + Ajouter une variable pour chaque variable d'environnement :

Nom de la variable	Exemple de valeur	Description
`GCS_BUCKET`	`confluence-audit-logs`	Nom du bucket GCS
`GCS_PREFIX`	`confluence-audit`	Préfixe des fichiers journaux
`STATE_KEY`	`confluence-audit/state.json`	Chemin d'accès au fichier d'état
`CONFLUENCE_URL`	`https://yoursite.atlassian.net`	URL du site Confluence
`CONFLUENCE_EMAIL`	`your-email@example.com`	Adresse e-mail du compte Atlassian
`CONFLUENCE_API_TOKEN`	`your-api-token-here`	Jeton d'API
`MAX_RECORDS`	`1000`	Nombre maximal d'enregistrements par exécution

Dans la section Variables et secrets, faites défiler la page jusqu'à Requêtes :
- Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).
Accédez à l'onglet Paramètres :
- Dans la section Ressources :
  - Mémoire : sélectionnez 512 Mio ou plus.
  - CPU : sélectionnez 1.
Dans la section Scaling de révision :
- Nombre minimal d'instances : saisissez 0.
- Nombre maximal d'instances : saisissez 100 (ou ajustez en fonction de la charge attendue).
Cliquez sur Créer.
Attendez que le service soit créé (1 à 2 minutes).
Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter un code de fonction

Saisissez main dans Point d'entrée de la fonction.

Dans l'éditeur de code intégré, créez deux fichiers :

Premier fichier : 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

Deuxième fichier : requirements.txt:

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

Cliquez sur Déployer pour enregistrer et déployer la fonction.
Attendez la fin du déploiement (deux à trois minutes).

Remarque : La configuration du déclencheur Pub/Sub crée automatiquement les abonnements et les autorisations nécessaires.

Créer une tâche Cloud Scheduler

Cloud Scheduler publie des messages sur le sujet Pub/Sub à intervalles réguliers, ce qui déclenche la fonction Cloud Run.

Dans la console GCP, accédez à Cloud Scheduler.
Cliquez sur Créer une tâche.

Fournissez les informations de configuration suivantes :

Paramètre	Valeur
Nom	`confluence-audit-collector-hourly`
Région	Sélectionnez la même région que la fonction Cloud Run.
Fréquence	`0 * * * *` (toutes les heures)
Fuseau horaire	Sélectionnez un fuseau horaire (UTC recommandé).
Type de cible	Pub/Sub
Topic	Sélectionner `confluence-audit-trigger`
Corps du message	`{}` (objet JSON vide)

Cliquez sur Créer.

Options de fréquence de planification

Choisissez la fréquence en fonction du volume de journaux et des exigences de latence :

Fréquence	Expression Cron	Cas d'utilisation
Toutes les 5 minutes	`/5 * * *`	Volume élevé, faible latence
Toutes les 15 minutes	`/15 * * *`	Volume moyen
Toutes les heures	`0 * * * *`	Standard (recommandé)
Toutes les 6 heures	`0 /6 * *`	Traitement par lot à faible volume
Tous les jours	`0 0 * * *`	Collecte de données historiques

Tester l'intégration

Dans la console Cloud Scheduler, recherchez votre job.
Cliquez sur Exécuter de force pour déclencher le job manuellement.
Patientez pendant quelques secondes.
Accédez à Cloud Run > Services.
Cliquez sur confluence-audit-collector.
Cliquez sur l'onglet Journaux.

Vérifiez que la fonction s'est exécutée correctement. Recherchez les éléments suivants :

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

Accédez à Cloud Storage > Buckets.
Cliquez sur le nom de votre bucket.
Accédez au dossier confluence-audit/ :
Vérifiez qu'un fichier .ndjson a été créé avec le code temporel actuel.

Si vous constatez des erreurs dans les journaux :

HTTP 401 : vérifiez les identifiants de l'API dans les variables d'environnement
HTTP 403 : vérifiez que le compte dispose des autorisations d'administrateur Confluence.
HTTP 429 : limitation du débit. La fonction effectuera automatiquement une nouvelle tentative avec un intervalle de temps.
Variables d'environnement manquantes : vérifiez que toutes les variables requises sont définies.

Récupérer le compte de service Google SecOps

Google SecOps utilise un compte de service unique pour lire les données de votre bucket GCS. Vous devez accorder à ce compte de service l'accès à votre bucket.

Obtenir l'adresse e-mail du compte de service

Accédez à Paramètres SIEM> Flux.
Cliquez sur Add New Feed (Ajouter un flux).
Cliquez sur Configurer un flux unique.
Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Confluence Cloud Audit Logs).
Sélectionnez Google Cloud Storage V2 comme Type de source.
Sélectionnez Atlassian Confluence comme Type de journal.
Cliquez sur Obtenir un compte de service. Une adresse e-mail unique pour le compte de service s'affiche, par exemple :
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
Copiez cette adresse e-mail pour l'utiliser à l'étape suivante.

Remarque : Chaque instance Google SecOps possède un compte de service unique. N'utilisez pas de comptes de service provenant d'autres documents ou exemples.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps a besoin du rôle Lecteur des objets Storage sur votre bucket GCS.

Accédez à Cloud Storage > Buckets.
Cliquez sur le nom de votre bucket.
Accédez à l'onglet Autorisations.
Cliquez sur Accorder l'accès.
Fournissez les informations de configuration suivantes :
- Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
- Attribuez des rôles : sélectionnez Lecteur des objets de l'espace de stockage.
Cliquez sur Enregistrer.

Remarque : Si vous prévoyez d'utiliser l'option de suppression "Supprimer les fichiers transférés" ou "Supprimer les fichiers transférés et les répertoires vides", accordez le rôle Administrateur des objets Storage au lieu de Lecteur des objets Storage.

Configurer un flux dans Google SecOps pour ingérer les journaux Confluence

Accédez à Paramètres SIEM> Flux.
Cliquez sur Add New Feed (Ajouter un flux).
Cliquez sur Configurer un flux unique.
Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Confluence Cloud Audit Logs).
Sélectionnez Google Cloud Storage V2 comme Type de source.
Sélectionnez Atlassian Confluence comme Type de journal.
Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
- URL du bucket Storage : saisissez l'URI du bucket GCS avec le préfixe du chemin d'accès :
```
gs://confluence-audit-logs/confluence-audit/
```
  - Remplacez :
    - confluence-audit-logs : nom de votre bucket GCS.
    - confluence-audit : préfixe/chemin d'accès au dossier facultatif où les journaux sont stockés (laisser vide pour la racine).
  - Exemples :
    - Bucket racine : gs://company-logs/
    - Avec préfixe : gs://company-logs/confluence-audit/
    - Avec un sous-dossier : gs://company-logs/confluence/audit/
  Remarque : Incluez toujours la barre oblique (/) à la fin de l'URI.
- Option de suppression de la source : sélectionnez l'option de suppression de votre choix :
  - Jamais : ne supprime jamais aucun fichier après les transferts (recommandé pour les tests).
  - Supprimer les fichiers transférés : supprime les fichiers après un transfert réussi.
  - Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides après un transfert réussi.
    
    Remarque : Si vous sélectionnez une option de suppression, le compte de service doit disposer du rôle Administrateur des objets Storage au lieu de celui de lecteur. Mettez à jour les autorisations IAM en conséquence.
- Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours. La valeur par défaut est de 180 jours.
- Espace de noms de l'élément : espace de noms de l'élément.
- Libellés d'ingestion : libellé à appliquer aux événements de ce flux.
Cliquez sur Suivant.
Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Table de mappage UDM

Champ de journal	Mappage UDM	Logique
agent	read_only_udm.network.http.user_agent	Valeur extraite du champ "agent".
app_protocol	read_only_udm.network.application_protocol	Dérivé du champ "app_protocol". Si "app_protocol" contient "HTTPS", "HTTP", "SSH" ou "RDP", le protocole correspondant est utilisé. Sinon, la valeur par défaut est "UNKNOWN_APPLICATION_PROTOCOL".
app_protocol	read_only_udm.network.application_protocol_version	Valeur extraite du champ "app_protocol".
auditType.action	read_only_udm.security_result.action	Dérivé du champ "auditType.action". Si "auditType.action" contient "successful", la valeur est définie sur "ALLOW". Si elle contient "restricted" (restreint), la valeur est définie sur "BLOCK" (BLOQUER).
auditType.action	read_only_udm.security_result.summary	Valeur extraite du champ "auditType.action" lorsque "auditType" n'est pas vide et que "auditType_area" est défini sur "SECURITY".
auditType.actionI18nKey	read_only_udm.metadata.product_event_type	Valeur extraite du champ "auditType.actionI18nKey" lorsque "auditType" n'est pas vide.
auditType.area	read_only_udm.security_result.detection_fields.value	Valeur extraite du champ "auditType.area" et attribuée au champ "value" d'un champ de détection dont le champ "key" est défini sur "auditType area". Ce mappage est effectué lorsque "auditType" n'est pas vide.
auditType.category	read_only_udm.security_result.category_details	Valeur extraite du champ "auditType.category" lorsque "auditType" n'est pas vide.
auditType.categoryI18nKey	read_only_udm.security_result.detection_fields.value	Valeur extraite du champ "auditType.categoryI18nKey" et attribuée au champ "value" d'un champ de détection dont le champ "key" est défini sur "auditType categoryI18nKey". Ce mappage est effectué lorsque "auditType" n'est pas vide.
auditType.level	read_only_udm.security_result.detection_fields.value	Valeur extraite du champ "auditType.level" et attribuée au champ "value" d'un champ de détection dont le champ "key" est défini sur "auditType level". Ce mappage est effectué lorsque "auditType" n'est pas vide.
author.displayName	read_only_udm.principal.user.user_display_name	Valeur extraite du champ "author.displayName".
author.externalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	Valeur extraite du champ "author.externalCollaborator" et attribuée au champ "value" d'un libellé dont le champ "key" est défini sur "externalCollaborator".
author.id	read_only_udm.principal.user.userid	Valeur extraite du champ "author.id" lorsque "author.type" est défini sur "user" et "principal_user_present" sur "false".
author.isExternalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	Valeur extraite du champ "author.isExternalCollaborator" et attribuée au champ "value" d'un libellé dont le champ "key" est défini sur "isExternalCollaborator".
author.name	read_only_udm.principal.user.user_display_name	Valeur extraite du champ "author.name" lorsque "author.type" est défini sur "user" et "principal_user_present" sur "false".
bytes_in	read_only_udm.network.received_bytes	Valeur extraite du champ "bytes_in" s'il contient des chiffres. Sinon, la valeur par défaut est 0.
catégorie	read_only_udm.security_result.category_details	Valeur extraite du champ "category" (catégorie).
changedValues	read_only_udm.principal.resource.attribute.labels	Parcourt chaque élément de "changedValues" et crée des libellés avec des clés telles que "changedValue [index] [key]" et des valeurs issues des valeurs correspondantes du tableau "changedValues".
date de création	read_only_udm.metadata.event_timestamp	Valeur extraite du champ "creationDate", analysée en tant que code temporel UNIX ou UNIX_MS.
extraAttributes	read_only_udm.principal.resource.attribute.labels	Parcourt chaque élément de "extraAttributes" et crée des libellés avec des clés basées sur les champs "name" et "nameI18nKey", et des valeurs issues du champ "value" correspondant.
http_verb	read_only_udm.network.http.method	Valeur extraite du champ "http_verb".
ip	read_only_udm.target.ip	Valeur extraite du champ "ip".
principal_host	read_only_udm.principal.hostname	Valeur extraite du champ "principal_host".
referral_url	read_only_udm.network.http.referral_url	Valeur extraite du champ "referral_url".
remoteAddress	read_only_udm.principal.ip	Valeur extraite du champ "remoteAddress", analysée en tant qu'adresse IP.
response_code	read_only_udm.network.http.response_code	Valeur extraite du champ "response_code".
session_duration	read_only_udm.additional.fields.value.string_value	Valeur extraite du champ "session_duration" et attribuée au champ "string_value" d'un libellé dont le champ "key" est défini sur "Session Duration" (Durée de la session).
source	read_only_udm.principal.ip	Valeur extraite du champ "source" et analysée en tant qu'adresse IP.
src_ip	read_only_udm.principal.ip	Valeur extraite du champ "src_ip" si "remoteAddress" est vide.
résumé	read_only_udm.security_result.summary	Valeur extraite du champ "summary" (résumé).
sysAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	Valeur extraite du champ "sysAdmin" et attribuée au champ "value" d'un libellé dont le champ "key" est défini sur "sysAdmin".
superAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	Valeur extraite du champ "superAdmin" et attribuée au champ "value" d'un libellé dont le champ "key" est défini sur "superAdmin".
target_url	read_only_udm.target.url	Valeur extraite du champ "target_url".
timestamp	read_only_udm.metadata.event_timestamp	Valeur extraite du champ "timestamp", analysée sous forme de chaîne de date et d'heure.
user_id	read_only_udm.principal.user.userid	Valeur extraite du champ "user_id".
	read_only_udm.metadata.event_type	La valeur de ce champ est déterminée par une série de vérifications et est définie par défaut sur "GENERIC_EVENT". Elle est définie sur des valeurs spécifiques telles que "NETWORK_HTTP", "USER_UNCATEGORIZED" ou "STATUS_UPDATE" en fonction de la présence et du contenu d'autres champs tels que "principal_host", "user_id", "has_principal" et "author.type".
	read_only_udm.metadata.vendor_name	Défini sur "ATLASSIAN".
	read_only_udm.metadata.product_name	Définissez-le sur "CONFLUENCE".
	read_only_udm.metadata.log_type	Définissez-le sur "ATLASSIAN_CONFLUENCE".
	read_only_udm.principal.user.user_display_name	La valeur de ce champ peut provenir de "author.displayName" ou de "affectedObject.name", selon le contexte.
	read_only_udm.target.process.pid	La valeur de ce champ peut provenir de "principal_host" ou de "pid" selon le contexte.
	read_only_udm.principal.resource.attribute.labels	Ce champ est renseigné avec différents libellés dérivés de champs tels que "affectedObjects", "changedValues" et "extraAttributes". Les clés et les valeurs de ces libellés sont générées de manière dynamique en fonction du contenu spécifique de ces champs.

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.