Recopila registros de auditoría de Harness IO
En este documento, se explica cómo transferir registros de auditoría de Harness IO a las Operaciones de seguridad de Google con Amazon S3.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos previos:
- Una instancia de Google SecOps
- Acceso privilegiado a Harness con los siguientes permisos:
- Crea claves de API
- Accede a los registros de auditoría
- Ver la configuración de la cuenta
- Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)
Recopila credenciales de la API de Harness
Crea una clave de API en Harness
- Accede a la plataforma de Harness.
- Haz clic en tu perfil de usuario.
- Ve a Mis claves de API.
- Haz clic en + API Key.
- Proporciona los siguientes detalles de configuración:
- Nombre: Ingresa un nombre descriptivo (por ejemplo,
Google SecOps Integration). - Descripción: Descripción opcional.
- Nombre: Ingresa un nombre descriptivo (por ejemplo,
- Haz clic en Guardar.
- Haz clic en + Token para crear un token nuevo.
- Proporciona los siguientes detalles de configuración:
- Nombre: Ingresa
Chronicle Feed Token. - Set Expiration: Selecciona un tiempo de vencimiento adecuado o No Expiration (para uso en producción).
- Nombre: Ingresa
- Haz clic en Generar token.
Copia y guarda el valor del token de forma segura. Este token se usará como el valor del encabezado
x-api-key.
Obtén el ID de la cuenta de Harness
- En la Plataforma de Harness, toma nota del ID de cuenta de la URL.
- URL de ejemplo:
https://app.harness.io/ng/account/YOUR_ACCOUNT_ID/... - La parte
YOUR_ACCOUNT_IDes el identificador de tu cuenta.
- URL de ejemplo:
- También puedes ir a Configuración de la cuenta > Resumen para ver tu identificador de cuenta.
Copia y guarda el ID de la cuenta para usarlo en la función de Lambda.
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,
harness-io-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 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 consultarlas en el futuro.
- 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 la política AmazonS3FullAccess.
- Selecciona la política.
- Haz clic en Siguiente.
- Haz clic en Agregar permisos.
Configura la política y el rol de IAM para las cargas de Lambda en S3
- En la consola de AWS, ve a IAM > Policies > Create policy > pestaña JSON.
Copia y pega la siguiente política:
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutHarnessObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/state.json" } ] }- Reemplaza
harness-io-logssi ingresaste un nombre de bucket diferente.
- Reemplaza
Haz clic en Siguiente.
Asigna el nombre
HarnessToS3Policya la política y haz clic en Crear política.Ve a IAM > Roles > Crear rol.
Selecciona Servicio de AWS como el tipo de entidad de confianza.
Selecciona Lambda como el caso de uso.
Haz clic en Siguiente.
Busca y selecciona las siguientes políticas:
HarnessToS3Policy(la política que acabas de crear)AWSLambdaBasicExecutionRole(para CloudWatch Logs)
Haz clic en Siguiente.
Nombra el rol
HarnessAuditLambdaRoley 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 harness-audit-to-s3Tiempo de ejecución Python 3.13 Arquitectura x86_64 Rol de ejecución HarnessAuditLambdaRoleHaz clic en Crear función.
Después de crear la función, abre la pestaña Código.
Borra el código de stub predeterminado y, luego, ingresa el siguiente código de la función Lambda:
Código de la función Lambda (
harness_audit_to_s3.py)#!/usr/bin/env python3 """ Harness.io Audit Logs to S3 Lambda Fetches audit logs from Harness API and writes to S3 for Chronicle ingestion. """ import os import json import time import uuid import logging import urllib.parse from datetime import datetime, timedelta, timezone from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError import boto3 # Configuration from Environment Variables 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"] 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", "50")), 100) START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60")) # Optional filters (NEW) FILTER_MODULES = os.environ.get("FILTER_MODULES", "").split(",") if os.environ.get("FILTER_MODULES") else None FILTER_ACTIONS = os.environ.get("FILTER_ACTIONS", "").split(",") if os.environ.get("FILTER_ACTIONS") else None STATIC_FILTER = os.environ.get("STATIC_FILTER") # e.g., "EXCLUDE_LOGIN_EVENTS" MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3")) # AWS clients s3 = boto3.client("s3") # HTTP headers for Harness API HDRS = { "x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json", } # Logging configuration logger = logging.getLogger() logger.setLevel(logging.INFO) # ============================================ # State Management Functions # ============================================ def _read_state(): """Read checkpoint state from S3.""" try: obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY) state = json.loads(obj["Body"].read()) since_ms = state.get("since") page_token = state.get("pageToken") logger.info(f"State loaded: since={since_ms}, pageToken={page_token}") return since_ms, page_token except s3.exceptions.NoSuchKey: logger.info("No state file found, starting fresh collection") start_time = datetime.now(timezone.utc) - timedelta(minutes=START_MINUTES_BACK) since_ms = int(start_time.timestamp() * 1000) logger.info(f"Initial since timestamp: {since_ms} ({start_time.isoformat()})") return since_ms, None except Exception as e: logger.error(f"Error reading state: {e}") raise def _write_state(since_ms: int, page_token: str = None): """Write checkpoint state to S3.""" state = { "since": since_ms, "pageToken": page_token, "lastRun": int(time.time() * 1000), "lastRunISO": datetime.now(timezone.utc).isoformat() } try: s3.put_object( Bucket=BUCKET, Key=STATE_KEY, Body=json.dumps(state, indent=2).encode(), ContentType="application/json" ) logger.info(f"State saved: since={since_ms}, pageToken={page_token}") except Exception as e: logger.error(f"Error writing state: {e}") raise # ============================================ # Harness API Functions # ============================================ def _fetch_harness_audits(since_ms: int, page_token: str = None, retry_count: int = 0): """ Fetch audit logs from Harness API with retry logic. API Endpoint: POST /audit/api/audits/listV2 Documentation: https://apidocs.harness.io/audit/getauditeventlistv2 """ try: # Build URL with query parameters url = ( f"{API_BASE}/audit/api/audits/listV2" f"?accountIdentifier={urllib.parse.quote(ACCOUNT_ID)}" f"&pageSize={PAGE_SIZE}" ) if page_token: url += f"&pageToken={urllib.parse.quote(page_token)}" logger.info(f"Fetching from: {url[:100]}...") # Build request body with time filter and optional filters body_data = { "startTime": since_ms, "endTime": int(time.time() * 1000), "filterType": "Audit" } if FILTER_MODULES: body_data["modules"] = [m.strip() for m in FILTER_MODULES if m.strip()] logger.info(f"Applying module filter: {body_data['modules']}") if FILTER_ACTIONS: body_data["actions"] = [a.strip() for a in FILTER_ACTIONS if a.strip()] logger.info(f"Applying action filter: {body_data['actions']}") if STATIC_FILTER: body_data["staticFilter"] = STATIC_FILTER logger.info(f"Applying static filter: {STATIC_FILTER}") logger.debug(f"Request body: {json.dumps(body_data)}") # Make POST request req = Request( url, data=json.dumps(body_data).encode('utf-8'), headers=HDRS, method="POST" ) resp = urlopen(req, timeout=30) resp_text = resp.read().decode('utf-8') resp_data = json.loads(resp_text) if "status" not in resp_data: logger.warning(f"Response missing 'status' field: {resp_text[:200]}") # Check response status if resp_data.get("status") != "SUCCESS": error_msg = resp_data.get("message", "Unknown error") raise Exception(f"API returned status: {resp_data.get('status')} - {error_msg}") # Extract data from response structure data_obj = resp_data.get("data", {}) if not data_obj: logger.warning("Response 'data' object is empty or missing") events = data_obj.get("content", []) has_next = data_obj.get("hasNext", False) next_token = data_obj.get("pageToken") logger.info(f"API response: {len(events)} events, hasNext={has_next}, pageToken={next_token}") if not events and data_obj: logger.info(f"Empty events but data present. Data keys: {list(data_obj.keys())}") return { "events": events, "hasNext": has_next, "pageToken": next_token } except HTTPError as e: error_body = e.read().decode() if hasattr(e, 'read') else '' if e.code == 401: logger.error("Authentication failed: Invalid API key") raise Exception("Invalid Harness API key. Check HARNESS_API_KEY environment variable.") elif e.code == 403: logger.error("Authorization failed: Insufficient permissions") raise Exception("API key lacks required audit:read permissions") elif e.code == 429: retry_after = int(e.headers.get("Retry-After", "60")) logger.warning(f"Rate limit exceeded. Retry after {retry_after} seconds (attempt {retry_count + 1}/{MAX_RETRIES})") if retry_count < MAX_RETRIES: logger.info(f"Waiting {retry_after} seconds before retry...") time.sleep(retry_after) logger.info(f"Retrying request (attempt {retry_count + 2}/{MAX_RETRIES})") return _fetch_harness_audits(since_ms, page_token, retry_count + 1) else: raise Exception(f"Max retries ({MAX_RETRIES}) exceeded for rate limiting") elif e.code == 400: logger.error(f"Bad request: {error_body}") raise Exception(f"Invalid request parameters: {error_body}") else: logger.error(f"HTTP {e.code}: {e.reason} - {error_body}") raise Exception(f"Harness API error {e.code}: {e.reason}") except URLError as e: logger.error(f"Network error: {e.reason}") raise Exception(f"Failed to connect to Harness API: {e.reason}") except json.JSONDecodeError as e: logger.error(f"Invalid JSON response: {e}") logger.error(f"Response text (first 500 chars): {resp_text[:500] if 'resp_text' in locals() else 'N/A'}") raise Exception("Harness API returned invalid JSON") except Exception as e: logger.error(f"Unexpected error in _fetch_harness_audits: {e}", exc_info=True) raise # ============================================ # S3 Upload Functions # ============================================ def _upload_to_s3(events: list) -> str: """ Upload audit events to S3 in JSONL format. Each line is a complete JSON object (one event per line). """ if not events: logger.info("No events to upload") return None try: # Create JSONL content (one JSON object per line) jsonl_lines = [json.dumps(event) for event in events] jsonl_content = "\n".join(jsonl_lines) # Generate S3 key with timestamp and UUID timestamp = datetime.now(timezone.utc) key = ( f"{PREFIX}/" f"{timestamp:%Y/%m/%d}/" f"harness-audit-{timestamp:%Y%m%d-%H%M%S}-{uuid.uuid4()}.jsonl" ) # Upload to S3 s3.put_object( Bucket=BUCKET, Key=key, Body=jsonl_content.encode('utf-8'), ContentType="application/x-ndjson", Metadata={ "event-count": str(len(events)), "source": "harness-audit-lambda", "collection-time": timestamp.isoformat() } ) logger.info(f"Uploaded {len(events)} events to s3://{BUCKET}/{key}") return key except Exception as e: logger.error(f"Error uploading to S3: {e}", exc_info=True) raise # ============================================ # Main Orchestration Function # ============================================ def fetch_and_store(): """ Main function to fetch audit logs from Harness and store in S3. Handles pagination and state management. """ logger.info("=== Harness Audit Collection Started ===") logger.info(f"Configuration: API_BASE={API_BASE}, ACCOUNT_ID={ACCOUNT_ID[:8]}..., PAGE_SIZE={PAGE_SIZE}") if FILTER_MODULES: logger.info(f"Module filter enabled: {FILTER_MODULES}") if FILTER_ACTIONS: logger.info(f"Action filter enabled: {FILTER_ACTIONS}") if STATIC_FILTER: logger.info(f"Static filter enabled: {STATIC_FILTER}") try: # Step 1: Read checkpoint state since_ms, page_token = _read_state() if page_token: logger.info(f"Resuming pagination from saved pageToken") else: since_dt = datetime.fromtimestamp(since_ms / 1000, tz=timezone.utc) logger.info(f"Starting new collection from: {since_dt.isoformat()}") # Step 2: Collect all events with pagination all_events = [] current_page_token = page_token page_count = 0 max_pages = 100 # Safety limit has_next = True while has_next and page_count < max_pages: page_count += 1 logger.info(f"--- Fetching page {page_count} ---") # Fetch one page of results result = _fetch_harness_audits(since_ms, current_page_token) # Extract events events = result.get("events", []) all_events.extend(events) logger.info(f"Page {page_count}: {len(events)} events (total: {len(all_events)})") # Check pagination status has_next = result.get("hasNext", False) current_page_token = result.get("pageToken") if not has_next: logger.info("Pagination complete (hasNext=False)") break if not current_page_token: logger.warning("hasNext=True but no pageToken, stopping pagination") break # Small delay between pages to avoid rate limiting time.sleep(0.5) if page_count >= max_pages: logger.warning(f"Reached max pages limit ({max_pages}), stopping") # Step 3: Upload collected events to S3 if all_events: s3_key = _upload_to_s3(all_events) logger.info(f"Successfully uploaded {len(all_events)} total events") else: logger.info("No new events to upload") s3_key = None # Step 4: Update checkpoint state if not has_next: # Pagination complete - update since to current time for next run new_since = int(time.time() * 1000) _write_state(new_since, None) logger.info(f"Pagination complete, state updated with new since={new_since}") else: # Pagination incomplete - save pageToken for continuation _write_state(since_ms, current_page_token) logger.info(f"Pagination incomplete, saved pageToken for next run") # Step 5: Return result result = { "statusCode": 200, "message": "Success", "eventsCollected": len(all_events), "pagesProcessed": page_count, "paginationComplete": not has_next, "s3Key": s3_key, "filters": { "modules": FILTER_MODULES, "actions": FILTER_ACTIONS, "staticFilter": STATIC_FILTER } } logger.info(f"Collection completed: {json.dumps(result)}") return result except Exception as e: logger.error(f"Collection failed: {e}", exc_info=True) result = { "statusCode": 500, "message": "Error", "error": str(e), "errorType": type(e).__name__ } return result finally: logger.info("=== Harness Audit Collection Finished ===") # ============================================ # Lambda Handler # ============================================ def lambda_handler(event, context): """AWS Lambda handler function.""" return fetch_and_store() # ============================================ # Local Testing # ============================================ if __name__ == "__main__": # For local testing result = lambda_handler(None, None) print(json.dumps(result, indent=2))
Haz clic en Implementar para guardar el código de la función.
Configura las variables de entorno de Lambda
- En la página de la función Lambda, selecciona la pestaña Configuración.
- Haz clic en Variables de entorno en la barra lateral izquierda.
- Haz clic en Editar.
Haz clic en Agregar variable de entorno para cada una de las siguientes opciones:
Variables de entorno obligatorias:
Clave Valor Descripción HARNESS_ACCOUNT_IDTu ID de cuenta de Harness Identificador de la cuenta de Harness HARNESS_API_KEYTu token de clave de API Token con permisos de audit:read S3_BUCKETharness-io-logsNombre del bucket de S3 S3_PREFIXharness/auditPrefijo para objetos de S3 STATE_KEYharness/audit/state.jsonRuta de acceso del archivo de estado en S3 Variables de entorno opcionales:
Clave Valor predeterminado Descripción HARNESS_API_BASEhttps://app.harness.ioURL base de la API de Harness PAGE_SIZE50Eventos por página (máx. 100) START_MINUTES_BACK60Período inicial de visualización en minutos FILTER_MODULESNinguno Módulos separados por comas (p.ej., CD,CI,CEFILTER_ACTIONSNinguno Acciones separadas por comas (p.ej., CREATE,UPDATE,DELETESTATIC_FILTERNinguno Filtro predefinido: EXCLUDE_LOGIN_EVENTSoEXCLUDE_SYSTEM_EVENTSMAX_RETRIES3Cantidad máxima de reintentos para el límite de frecuencia Haz clic en Guardar.
Configura el tiempo de espera y la memoria de Lambda
- En la página de la función Lambda, selecciona la pestaña Configuración.
- Haz clic en Configuración general en la barra lateral izquierda.
- Haz clic en Editar.
- Proporciona los siguientes detalles de configuración:
- Memoria:
256 MB(recomendado) - Tiempo de espera:
5 min 0 sec(300 segundos)
- Memoria:
- Haz clic en Guardar.
Crea una programación de EventBridge
- Ve a Amazon EventBridge > Scheduler > Create schedule.
- Proporciona los siguientes detalles de configuración:
- Nombre de la programación: Ingresa
harness-audit-hourly. - Descripción: Descripción opcional.
- Nombre de la programación: Ingresa
- Haz clic en Siguiente.
- En Patrón de programación, selecciona Programación recurrente.
- Selecciona Programación basada en tarifas.
- Proporciona los siguientes detalles de configuración:
- Expresión de la tarifa: Ingresa
1 hour.
- Expresión de la tarifa: Ingresa
- Haz clic en Siguiente.
- En Destino, proporciona los siguientes detalles de configuración:
- API de destino: Selecciona AWS Lambda Invoke.
- Función Lambda: Selecciona tu función
harness-audit-to-s3.
- Haz clic en Siguiente.
- Revisa la configuración de la programación.
- Haz clic en Crear programación.
Crea un usuario de IAM con acceso de lectura para Google SecOps
Este usuario de IAM permite que el equipo de SecOps de Google lea los registros del bucket de S3.
- Ve a Consola de AWS > IAM > Usuarios > Crear usuario.
- Proporciona los siguientes detalles de configuración:
- Nombre de usuario: Ingresa
chronicle-s3-reader.
- Nombre de usuario: Ingresa
- Haz clic en Siguiente.
- Selecciona Adjuntar políticas directamente.
- Haz clic en Crear política.
- Selecciona la pestaña JSON.
Pega la siguiente política:
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*" }, { "Effect": "Allow", "Action": [ "s3:ListBucket" ], "Resource": "arn:aws:s3:::harness-io-logs", "Condition": { "StringLike": { "s3:prefix": "harness/audit/*" } } } ] }Haz clic en Siguiente.
Asigna a la política el nombre
ChronicleHarnessS3ReadPolicy.Haz clic en Crear política.
Regresa a la pestaña de creación de usuarios y actualiza la lista de políticas.
Busca y selecciona
ChronicleHarnessS3ReadPolicy.Haz clic en Siguiente.
Revisa la información y haz clic en Crear usuario.
Crea claves de acceso para el usuario lector
- En la página Usuarios de IAM, selecciona el usuario
chronicle-s3-reader. - Selecciona la pestaña Credenciales de seguridad.
- Haz clic en Crear clave de acceso.
- Selecciona Servicio de terceros como 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 el ID de clave de acceso y la clave de acceso secreta.
- Haz clic en Listo.
Configura un feed en Google SecOps para transferir registros de Harness IO
- Ve a Configuración de SIEM > Feeds.
- Haz clic en Agregar nueva.
- En la siguiente página, haz clic en Configurar un solo feed.
- En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo,
Harness Audit Logs). - Selecciona Amazon S3 V2 como el Tipo de fuente.
- Selecciona Harness IO como el Tipo de registro.
- Haz clic en Siguiente.
Especifica valores para los siguientes parámetros de entrada:
- URI de S3: Ingresa el URI del bucket de S3 con la ruta de acceso del prefijo:
s3://harness-io-logs/harness/audit/ 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 inicialmente).
- On success: Borra todos los archivos y directorios vacíos después de una transferencia exitosa.
Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es 180 días.
ID de clave de acceso: Ingresa el ID de clave de acceso del usuario
chronicle-s3-reader.Clave de acceso secreta: Ingresa la clave de acceso secreta del usuario
chronicle-s3-reader.Espacio de nombres del recurso: Es el espacio de nombres del recurso. Ingresa
harness.audit.Etiquetas de transferencia: Son etiquetas opcionales que se pueden aplicar a los eventos de este feed.
- URI de S3: Ingresa el URI del bucket de S3 con la ruta de acceso del prefijo:
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.