Recopila registros de auditoría de Harness IO

Compatible con:

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

Antes de comenzar

  • Instancia de Google SecOps
  • Acceso privilegiado a Harness (clave de API y ID de cuenta)
  • Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)

Obtén la clave de API y el ID de cuenta de Harness para una cuenta personal

  1. Accede a la IU web de Harness.
  2. Ve a tu Perfil de usuario > Mis claves de API.
  3. Selecciona Clave de API.
  4. Ingresa un Nombre para la clave de API.
  5. Haz clic en Guardar.
  6. Selecciona Token en tu nueva clave de API.
  7. Ingresa un nombre para el token.
  8. Haz clic en Generate Token.
  9. Copia y guarda el token en una ubicación segura.
  10. Copia y guarda tu ID de cuenta (aparece en la URL de Harness y en Configuración de la cuenta).

Opcional: Obtén la clave de API y el ID de la cuenta de Harness para una cuenta de servicio

  1. Accede a la IU web de Harness.
  2. Crea una cuenta de servicio.
  3. Ve a Configuración de la cuenta > Control de acceso.
  4. Selecciona Cuentas de servicio > selecciona la cuenta de servicio para la que deseas crear una clave de API.
  5. En API Keys, selecciona API Key.
  6. Ingresa un Nombre para la clave de API.
  7. Haz clic en Guardar.
  8. Selecciona Token en la nueva clave de API.
  9. Ingresa un nombre para el token.
  10. Haz clic en Generate Token.
  11. Copia y guarda el token en una ubicación segura.
  12. Copia y guarda tu ID de cuenta (aparece en la URL de Harness y en Configuración de la cuenta).

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

  1. Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
  2. Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, harness-io).
  3. Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
  4. Selecciona el usuario creado.
  5. Selecciona la pestaña Credenciales de seguridad.
  6. Haz clic en Crear clave de acceso en la sección Claves de acceso.
  7. Selecciona Servicio de terceros como el Caso de uso.
  8. Haz clic en Siguiente.
  9. Opcional: Agrega una etiqueta de descripción.
  10. Haz clic en Crear clave de acceso.
  11. Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
  12. Haz clic en Listo.
  13. Selecciona la pestaña Permisos.
  14. Haz clic en Agregar permisos en la sección Políticas de permisos.
  15. Selecciona Agregar permisos.
  16. Selecciona Adjuntar políticas directamente.
  17. Busca y selecciona la política AmazonS3FullAccess.
  18. Haz clic en Siguiente.
  19. Haz clic en Agregar permisos.

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

  1. En la consola de AWS, ve a IAM > Políticas > Crear política > pestaña JSON.

  2. Ingresa la siguiente política:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutHarnessObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::harness-io/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::harness-io/harness/audit/state.json"
    }
  ]
}
    • Reemplaza harness-io si ingresaste un nombre de bucket diferente):

  3. Haz clic en Siguiente > Crear política.

  4. Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.

  5. Adjunta la política recién creada.

  6. Asigna el nombre WriteHarnessToS3Role al rol y haz clic en Crear rol.

Crea la función Lambda

  1. En la consola de AWS, ve a Lambda > Functions > Create function.
  2. Haz clic en Crear desde cero.

  3. Proporciona los siguientes detalles de configuración:

    Configuración Valor
    Nombre harness_io_to_s3
    Tiempo de ejecución Python 3.13
    Arquitectura x86_64
    Rol de ejecución WriteHarnessToS3Role

  4. Después de crear la función, abre la pestaña Código, borra el código auxiliar y, luego, ingresa el siguiente código (harness_io_to_s3.py).

    #!/usr/bin/env python3

import os, json, time, urllib.parse
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

API_BASE = os.environ.get("HARNESS_API_BASE", "https://app.harness.io").rstrip("/")
ACCOUNT_ID = os.environ["HARNESS_ACCOUNT_ID"]
API_KEY = os.environ["HARNESS_API_KEY"]  # x-api-key token
BUCKET = os.environ["S3_BUCKET"]
PREFIX = os.environ.get("S3_PREFIX", "harness/audit/").strip("/")
STATE_KEY = os.environ.get("STATE_KEY", "harness/audit/state.json")
PAGE_SIZE = min(int(os.environ.get("PAGE_SIZE", "100")), 100)  # <=100
START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60"))

s3 = boto3.client("s3")
HDRS = {"x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json"}

def _read_state():
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        j = json.loads(obj["Body"].read())
        return j.get("since"), j.get("pageToken")
    except Exception:
        return None, None

def _write_state(since_ms: int, page_token: str | None):
    body = json.dumps({"since": since_ms, "pageToken": page_token}).encode("utf-8")
    s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")

def _http_post(path: str, body: dict, query: dict, timeout: int = 60, max_retries: int = 5) -> dict:
    qs = urllib.parse.urlencode(query)
    url = f"{API_BASE}{path}?{qs}"
    data = json.dumps(body).encode("utf-8")

    attempt, backoff = 0, 1.0
    while True:
        req = Request(url, data=data, method="POST")
        for k, v in HDRS.items():
            req.add_header(k, v)
        try:
            with urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode("utf-8"))
        except HTTPError as e:
            if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise
        except URLError:
            if attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise

def _write_page(obj: dict, now: float, page_index: int) -> str:
    ts = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime(now))
    key = f"{PREFIX}/{ts}-page{page_index:05d}.json"
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )
    return key

def fetch_and_store():
    now_s = time.time()
    since_ms, page_token = _read_state()
    if since_ms is None:
        since_ms = int((now_s - START_MINUTES_BACK * 60) * 1000)
    until_ms = int(now_s * 1000)

    page_index = 0
    total = 0

    while True:
        body = {"startTime": since_ms, "endTime": until_ms}
        query = {"accountIdentifier": ACCOUNT_ID, "pageSize": PAGE_SIZE}
        if page_token:
            query["pageToken"] = page_token
        else:
            query["pageIndex"] = page_index

        data = _http_post("/audit/api/audits/listV2", body, query)
        _write_page(data, now_s, page_index)

        entries = []
        for key in ("data", "content", "response", "resource", "resources", "items"):
            if isinstance(data.get(key), list):
                entries = data[key]
                break
        total += len(entries) if isinstance(entries, list) else 0

        next_token = (
            data.get("pageToken")
            or (isinstance(data.get("meta"), dict) and data["meta"].get("pageToken"))
            or (isinstance(data.get("metadata"), dict) and data["metadata"].get("pageToken"))
        )

        if next_token:
            page_token = next_token
            page_index += 1
            continue

        if len(entries) < PAGE_SIZE:
            break
        page_index += 1

    _write_state(until_ms, None)
    return {"pages": page_index + 1, "objects_estimate": total}

def lambda_handler(event=None, context=None):
    return fetch_and_store()

if __name__ == "__main__":
    print(lambda_handler())

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

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

    Clave Ejemplo
    S3_BUCKET harness-io
    S3_PREFIX harness/audit/
    STATE_KEY harness/audit/state.json
    HARNESS_ACCOUNT_ID 123456789
    HARNESS_API_KEY harness_xxx_token
    HARNESS_API_BASE https://app.harness.io
    PAGE_SIZE 100
    START_MINUTES_BACK 60

  7. Después de crear la función, permanece en su página (o abre Lambda > Functions > tu‑función).

  8. Selecciona la pestaña Configuración.

  9. En el panel Configuración general, haz clic en Editar.

  10. Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Crea una programación de EventBridge

  1. Ve a Amazon EventBridge > Scheduler > Create schedule.

  2. Proporciona los siguientes detalles de configuración:

    • Programación recurrente: Frecuencia (1 hour).
    • Destino: Tu función Lambda.
    • Nombre: harness-io-1h.

  3. Haz clic en Crear programación.

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

  1. En la consola de AWS, ve a IAM > Usuarios y, luego, haz clic en Agregar usuarios.
  2. 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.
  3. Adjunta una política de lectura mínima (personalizada): Usuarios > selecciona secops-reader > Permisos > Agregar permisos > Adjuntar políticas directamente > Crear política

  4. 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>"
    }
  ]
}

  5. Configura el nombre como secops-reader-policy.

  6. Ve a Crear política > busca o selecciona > Siguiente > Agregar permisos.

  7. Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.

  8. Descarga el archivo CSV (estos valores se ingresan en el feed).

Configura un feed en Google SecOps para transferir registros de Harness IO

  1. Ve a SIEM Settings > Feeds.
  2. Haz clic en Agregar feed nuevo.
  3. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Harness IO).
  4. Selecciona Amazon S3 V2 como el Tipo de fuente.
  5. Selecciona Harness IO como el Tipo de registro.
  6. Haz clic en Siguiente.
  7. Especifica valores para los siguientes parámetros de entrada:
    • URI de S3: s3://harness-io/harness/audit/
    • 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.
  8. Haz clic en Siguiente.
  9. 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.