Se usó la API de Cloud Translation para traducir esta página.

Recopila registros de auditoría de DigiCert

Compatible con:

Google SecOps SIEM

En este documento, se explica cómo transferir registros de auditoría de DigiCert a Google Security Operations con Amazon S3.

Antes de comenzar

Instancia de Google SecOps
Acceso privilegiado a DigiCert CertCentral (clave de API con rol de administrador)
Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)

Obtén la clave de API y el ID del informe de DigiCert

En CertCentral, ve a Account > API Keys y crea la clave de API (X-DC-DEVKEY).
En Informes > Biblioteca de informes, crea un informe de Registro de auditoría con formato JSON y anota su ID de informe (UUID).
También puedes encontrar el ID de un informe existente en el historial de informes.

Configura el bucket de AWS S3 y el IAM para Google SecOps

Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, digicert-logs).
Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
Selecciona el usuario creado.
Selecciona la pestaña Credenciales de seguridad.
Haz clic en Crear clave de acceso en la sección Claves de acceso.
Selecciona Servicio de terceros como el Caso de uso.
Haz clic en Siguiente.
Opcional: Agrega una etiqueta de descripción.
Haz clic en Crear clave de acceso.
Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
Haz clic en Listo.
Selecciona la pestaña Permisos.
Haz clic en Agregar permisos en la sección Políticas de permisos.
Selecciona Agregar permisos.
Selecciona Adjuntar políticas directamente.
Busca y selecciona la política AmazonS3FullAccess.
Haz clic en Siguiente.
Haz clic en Agregar permisos.

Configura la política y el rol de IAM para las cargas de S3

Ve a la consola de AWS > IAM > Políticas > Crear política > pestaña JSON.

Ingresa la siguiente política:

{
  "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"
    }
  ]
}

Reemplaza digicert-logs si ingresaste un nombre de bucket diferente.

Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunta la política recién creada.
Asigna el nombre WriteDigicertToS3Role al rol y haz clic en Crear rol.

Crea la función Lambda

En la consola de AWS, ve a Lambda > Functions > Create function.
Haz clic en Crear desde cero.
Proporciona los siguientes detalles de configuración:

Configuración Valor

Nombre digicert_audit_logs_to_s3

Tiempo de ejecución Python 3.13

Arquitectura x86_64

Rol de ejecución WriteDigicertToS3Role

Configuración	Valor
Nombre	`digicert_audit_logs_to_s3`
Tiempo de ejecución	Python 3.13
Arquitectura	x86_64
Rol de ejecución	`WriteDigicertToS3Role`

Después de crear la función, abre la pestaña Code, borra el código auxiliar y, luego, ingresa el siguiente código (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}

Ve a Configuración > Variables de entorno > Editar > Agregar nueva variable de entorno.

Ingresa las siguientes variables de entorno y reemplaza los valores por los tuyos:

Clave	Ejemplo
`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`

Después de crear la función, permanece en su página (o abre Lambda > Functions > tu-función).
Selecciona la pestaña Configuración.
En el panel Configuración general, haz clic en Editar.
Cambia Tiempo de espera a 15 minutos (900 segundos) y haz clic en Guardar.

Nota: El tiempo de espera es un límite superior estricto, lo que significa que, si la función sigue en ejecución cuando el temporizador llega a los 15 minutos, AWS Lambda la finalizará. Cualquier ejecución más corta simplemente finaliza de forma normal, y solo se te factura el tiempo de ejecución real.

Crea una programación de EventBridge

Ve a Amazon EventBridge > Scheduler > Create schedule.
Proporciona los siguientes detalles de configuración:
- Programación recurrente: Frecuencia (1 hour).
- Destino: Tu función Lambda.
- Nombre: digicert-audit-1h.
Haz clic en Crear programación.

Opcional: Crea un usuario y claves de IAM de solo lectura para Google SecOps

En la consola de AWS, ve a IAM > Usuarios y, luego, haz clic en Agregar usuarios.
Proporciona los siguientes detalles de configuración:
- Usuario: Ingresa un nombre único (por ejemplo, secops-reader).
- Tipo de acceso: Selecciona Clave de acceso: Acceso programático.
- Haz clic en Crear usuario.
Adjunta una política de lectura mínima (personalizada): Usuarios > selecciona secops-reader > Permisos > Agregar permisos > Adjuntar políticas directamente > Crear política

En el editor de JSON, ingresa la siguiente política:

{
  "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>"
    }
  ]
}

Nombre = secops-reader-policy.
Haz clic en Crear política > busca o selecciona > Siguiente > Agregar permisos.
Crea una clave de acceso para secops-reader: Credenciales de seguridad > Claves de acceso > Crear clave de acceso Crear clave de acceso**.
Descarga el archivo CSV (estos valores se ingresan en el feed).

Configura un feed en Google SecOps para transferir registros de DigiCert

Ve a SIEM Settings > Feeds.
Haz clic en Agregar feed nuevo.
En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, DigiCert Audit Logs).
Selecciona Amazon S3 V2 como el Tipo de fuente.
Selecciona Digicert como el Tipo de registro.
Haz clic en Siguiente.
Especifica valores para los siguientes parámetros de entrada:
- URI de S3: s3://digicert-logs/digicert/logs/
- Opciones de borrado de la fuente: Selecciona la opción de borrado según tu preferencia.
- Antigüedad máxima del archivo: 180 días de forma predeterminada
- ID de clave de acceso: Clave de acceso del usuario con acceso al bucket de S3.
- Clave de acceso secreta: Clave secreta del usuario con acceso al bucket de S3.
- 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.

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