Collecter les journaux d'audit DigiCert

Compatible avec :

Ce document explique comment ingérer des journaux d'audit DigiCert dans Google Security Operations à l'aide d'Amazon S3.

Avant de commencer

  • Instance Google SecOps
  • Accès privilégié à DigiCert CertCentral (clé API avec rôle d'administrateur)
  • Accès privilégié à AWS (S3, IAM, Lambda, EventBridge)

Obtenir la clé API et l'ID de rapport DigiCert

  1. Dans CertCentral, accédez à Compte > Clés API et créez la clé API (X-DC-DEVKEY).
  2. Dans Rapports > Bibliothèque de rapports, créez un rapport Journal d'audit au format JSON et notez son ID de rapport (UUID).
  3. Vous pouvez également trouver l'ID d'un rapport existant à l'aide de l'historique des rapports.

Configurer un bucket AWS S3 et IAM pour Google SecOps

  1. Créez un bucket Amazon S3 en suivant ce guide de l'utilisateur : Créer un bucket.
  2. Enregistrez le nom et la région du bucket pour référence ultérieure (par exemple, digicert-logs).
  3. Créez un utilisateur en suivant ce guide : Créer un utilisateur IAM.
  4. Sélectionnez l'utilisateur créé.
  5. Sélectionnez l'onglet Informations d'identification de sécurité.
  6. Cliquez sur Créer une clé d'accès dans la section Clés d'accès.
  7. Sélectionnez Service tiers comme Cas d'utilisation.
  8. Cliquez sur Suivant.
  9. Facultatif : ajoutez un tag de description.
  10. Cliquez sur Créer une clé d'accès.
  11. Cliquez sur Télécharger le fichier CSV pour enregistrer la clé d'accès et la clé d'accès secrète pour une utilisation ultérieure.
  12. Cliquez sur OK.
  13. Sélectionnez l'onglet Autorisations.
  14. Cliquez sur Ajouter des autorisations dans la section Règles d'autorisation.
  15. Sélectionnez Ajouter des autorisations.
  16. Sélectionnez Joindre directement des règles.
  17. Recherchez et sélectionnez la règle AmazonS3FullAccess.
  18. Cliquez sur Suivant.
  19. Cliquez sur Ajouter des autorisations.

Configurer la stratégie et le rôle IAM pour les importations S3

  1. Accédez à la console AWS> IAM> Policies> Create policy> onglet JSON.

  2. Saisissez la règle suivante :

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutDigiCertObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::digicert-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::digicert-logs/digicert/logs/state.json"
    }
  ]
}
    • Remplacez digicert-logs si vous avez saisi un autre nom de bucket.

  3. Cliquez sur Suivant > Créer une règle.

  4. Accédez à IAM > Rôles > Créer un rôle > Service AWS > Lambda.

  5. Associez la règle que vous venez de créer.

  6. Nommez le rôle WriteDigicertToS3Role, puis cliquez sur Créer un rôle.

Créer la fonction Lambda

  1. Dans la console AWS, accédez à Lambda > Fonctions > Créer une fonction.
  2. Cliquez sur Créer à partir de zéro.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom digicert_audit_logs_to_s3
    Durée d'exécution Python 3.13
    Architecture x86_64
    Rôle d'exécution WriteDigicertToS3Role

  4. Une fois la fonction créée, ouvrez l'onglet Code, supprimez le stub et saisissez le code suivant (digicert_audit_logs_to_s3.py) :

    #!/usr/bin/env python3

import datetime as dt, gzip, io, json, os, time, uuid, zipfile
from typing import Any, Dict, Iterable, List, Tuple
from urllib import request, parse, error
import boto3
from botocore.exceptions import ClientError

API_BASE = "https://api.digicert.com/reports/v1"
USER_AGENT = "secops-digicert-reports/1.0"
s3 = boto3.client("s3")

def _now() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)

def _http(method: str, url: str, api_key: str, body: bytes | None = None,
          timeout: int = 30, max_retries: int = 5) -> Tuple[int, Dict[str,str], bytes]:
    headers = {"X-DC-DEVKEY": api_key, "Content-Type": "application/json", "User-Agent": USER_AGENT}
    attempt, backoff = 0, 1.0
    while True:
        req = request.Request(url=url, method=method, headers=headers, data=body)
        try:
            with request.urlopen(req, timeout=timeout) as resp:
                status, h = resp.status, {k.lower(): v for k, v in resp.headers.items()}
                data = resp.read()
                if 500 <= status <= 599 and attempt < max_retries:
                    attempt += 1; time.sleep(backoff); backoff *= 2; continue
                return status, h, data
        except error.HTTPError as e:
            status, h = e.code, {k.lower(): v for k, v in (e.headers or {}).items()}
            if status == 429 and attempt < max_retries:
                ra = h.get("retry-after"); delay = float(ra) if ra and ra.isdigit() else backoff
                attempt += 1; time.sleep(delay); backoff *= 2; continue
            if 500 <= status <= 599 and attempt < max_retries:
                attempt += 1; time.sleep(backoff); backoff *= 2; continue
            raise
        except error.URLError:
            if attempt < max_retries:
                attempt += 1; time.sleep(backoff); backoff *= 2; continue
            raise

def start_report_run(api_key: str, report_id: str, timeout: int) -> None:
    st, _, body = _http("POST", f"{API_BASE}/report/{report_id}/run", api_key, b"{}", timeout)
    if st not in (200, 201): raise RuntimeError(f"Start run failed: {st} {body[:200]!r}")

def list_report_history(api_key: str, *, status_filter: str | None = None, report_type: str | None = None,
                        limit: int = 100, sort_by: str = "report_start_date", sort_direction: str = "DESC",
                        timeout: int = 30, offset: int = 0) -> Dict[str, Any]:
    qs = {"limit": str(limit), "offset": str(offset), "sort_by": sort_by, "sort_direction": sort_direction}
    if status_filter: qs["status"] = status_filter
    if report_type: qs["report_type"] = report_type
    st, _, body = _http("GET", f"{API_BASE}/report/history?{parse.urlencode(qs)}", api_key, timeout=timeout)
    if st != 200: raise RuntimeError(f"History failed: {st} {body[:200]!r}")
    return json.loads(body.decode("utf-8"))

def find_ready_run(api_key: str, report_id: str, started_not_before: dt.datetime,
                  timeout: int, max_wait_seconds: int, poll_interval: int) -> str:
    deadline = time.time() + max_wait_seconds
    while time.time() < deadline:
        hist = list_report_history(api_key, status_filter="READY", report_type="audit-logs",
                                  limit=200, timeout=timeout).get("report_history", [])
        for it in hist:
            if it.get("report_identifier") != report_id or not it.get("report_run_identifier"): continue
            try:
                rsd = dt.datetime.strptime(it.get("report_start_date",""), "%Y-%m-%d %H:%M:%S").replace(tzinfo=dt.timezone.utc)
            except Exception:
                rsd = started_not_before
            if rsd + dt.timedelta(seconds=60) >= started_not_before:
                return it["report_run_identifier"]
        time.sleep(poll_interval)
    raise TimeoutError("READY run not found in time")

def get_json_rows(api_key: str, report_id: str, run_id: str, timeout: int) -> List[Dict[str, Any]]:
    st, h, body = _http("GET", f"{API_BASE}/report/{report_id}/{run_id}/json", api_key, timeout=timeout)
    if st != 200: raise RuntimeError(f"Get JSON failed: {st} {body[:200]!r}")
    if "application/zip" in h.get("content-type","").lower() or body[:2] == b"PK":
        with zipfile.ZipFile(io.BytesIO(body)) as zf:
            name = next((n for n in zf.namelist() if n.lower().endswith(".json")), None)
            if not name: raise RuntimeError("ZIP has no JSON")
            rows = json.loads(zf.read(name).decode("utf-8"))
    else:
        rows = json.loads(body.decode("utf-8"))
    if not isinstance(rows, list): raise RuntimeError("Unexpected JSON format")
    return rows

def load_state(bucket: str, key: str) -> Dict[str, Any]:
    try:
        return json.loads(s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode("utf-8"))
    except ClientError as e:
        if e.response["Error"]["Code"] in ("NoSuchKey","404"): return {}
        raise

def save_state(bucket: str, key: str, state: Dict[str, Any]) -> None:
    s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(state).encode("utf-8"), ContentType="application/json")

def write_ndjson_gz(bucket: str, prefix: str, rows: Iterable[Dict[str, Any]], run_id: str) -> str:
    ts = _now().strftime("%Y/%m/%d/%H%M%S")
    key = f"{prefix}/{ts}-digicert-audit-{run_id[:8]}-{uuid.uuid4().hex}.json.gz"
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
        for r in rows:
            gz.write((json.dumps(r, separators=(',',':')) + "\n").encode("utf-8"))
    s3.put_object(Bucket=bucket, Key=key, Body=buf.getvalue(),
                  ContentType="application/x-ndjson", ContentEncoding="gzip")
    return key

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    api_key  = os.environ["DIGICERT_API_KEY"]
    report_id = os.environ["DIGICERT_REPORT_ID"]
    bucket   = os.environ["S3_BUCKET"]
    prefix   = os.environ.get("S3_PREFIX", "digicert/logs").rstrip("/")
    state_key = os.environ.get("STATE_KEY", f"{prefix}/state.json")
    max_wait = int(os.environ.get("MAX_WAIT_SECONDS", "300"))
    poll_int = int(os.environ.get("POLL_INTERVAL", "10"))
    timeout  = int(os.environ.get("REQUEST_TIMEOUT", "30"))

    state = load_state(bucket, state_key) if state_key else {}
    last_run = state.get("last_run_id")

    started = _now()
    start_report_run(api_key, report_id, timeout)
    run_id = find_ready_run(api_key, report_id, started, timeout, max_wait, poll_int)
    if last_run and last_run == run_id:
        return {"status":"skip", "report_run_identifier": run_id}

    rows = get_json_rows(api_key, report_id, run_id, timeout)
    key = write_ndjson_gz(bucket, prefix, rows, run_id)
    if state_key:
        save_state(bucket, state_key, {"last_run_id": run_id, "last_success_at": _now().isoformat(),
                                      "last_s3_key": key, "rows_count": len(rows)})
    return {"status":"ok", "report_identifier": report_id, "report_run_identifier": run_id,
            "rows": len(rows), "s3_key": key}

  5. Accédez à Configuration> Variables d'environnement> Modifier> Ajouter une variable d'environnement.

  6. Saisissez les variables d'environnement suivantes en remplaçant les valeurs par les vôtres :

    Clé Exemple
    S3_BUCKET digicert-logs
    S3_PREFIX digicert/logs/
    STATE_KEY digicert/logs/state.json
    DIGICERT_API_KEY xxxxxxxxxxxxxxxxxxxxxxxx
    DIGICERT_REPORT_ID 88de5e19-ec57-4d70-865d-df953b062574
    REQUEST_TIMEOUT 30
    POLL_INTERVAL 10
    MAX_WAIT_SECONDS 300

  7. Une fois la fonction créée, restez sur sa page (ou ouvrez Lambda > Fonctions > votre-fonction).

  8. Accédez à l'onglet Configuration.

  9. Dans le panneau Configuration générale, cliquez sur Modifier.

  10. Définissez Délai avant expiration sur 15 minutes (900 secondes), puis cliquez sur Enregistrer.

Créer une programmation EventBridge

  1. Accédez à Amazon EventBridge> Scheduler> Create schedule (Créer une programmation).
  2. Fournissez les informations de configuration suivantes :
    • Planning récurrent : Tarif (1 hour).
    • Cible : votre fonction Lambda.
    • Nom : digicert-audit-1h.
  3. Cliquez sur Créer la programmation.

Facultatif : Créez un utilisateur et des clés IAM en lecture seule pour Google SecOps

  1. Dans la console AWS, accédez à IAM> Utilisateurs, puis cliquez sur Ajouter des utilisateurs.
  2. Fournissez les informations de configuration suivantes :
    • Utilisateur : saisissez un nom unique (par exemple, secops-reader).
    • Type d'accès : sélectionnez Clé d'accès – Accès programmatique.
    • Cliquez sur Créer un utilisateur.
  3. Associez une règle de lecture minimale (personnalisée) : Utilisateurs> sélectionnez secops-reader > Autorisations> Ajouter des autorisations> Associer des règles directement> Créer une règle

  4. Dans l'éditeur JSON, saisissez la stratégie suivante :

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::<your-bucket>/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::<your-bucket>"
    }
  ]
}

  5. Nom = secops-reader-policy.

  6. Cliquez sur Créer une règle> recherchez/sélectionnez > Suivant> Ajouter des autorisations.

  7. Créez une clé d'accès pour secops-reader : Identifiants de sécurité> Clés d'accès> Créer une clé d'accès.

  8. Téléchargez le CSV (ces valeurs sont saisies dans le flux).

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

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Add New Feed (Ajouter un flux).
  3. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, DigiCert Audit Logs).
  4. Sélectionnez Amazon S3 V2 comme type de source.
  5. Sélectionnez Digicert comme Type de journal.
  6. Cliquez sur Suivant.
  7. Spécifiez les valeurs des paramètres d'entrée suivants :
    • URI S3 : s3://digicert-logs/digicert/logs/
    • Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
    • Âge maximal des fichiers : 180 jours par défaut.
    • ID de clé d'accès : clé d'accès utilisateur ayant accès au bucket S3.
    • Clé d'accès secrète : clé secrète de l'utilisateur ayant accès au bucket S3.
    • 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.
  8. Cliquez sur Suivant.
  9. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

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