Recopilar registros de la plataforma Swimlane

Disponible en:

SecOps de Google SIEM

Nota: Esta función está cubierta por los Términos de las Ofertas Previas a la Disponibilidad General de los Términos Específicos del Servicio de Google Security Operations. Las funciones previas a la disponibilidad general pueden tener una asistencia limitada y los cambios que se produzcan en ellas podrían no ser compatibles con otras versiones previas a la disponibilidad general. Para obtener más información, consulta las directrices del servicio de asistencia técnica de Google SecOps y los Términos Específicos de los Servicios de Google SecOps.

En este documento se explica cómo ingerir registros de la plataforma Swimlane en Google Security Operations mediante Amazon S3.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

• Una instancia de Google SecOps
• Acceso privilegiado a Swimlane (administrador de la cuenta que puede generar un token de acceso personal)
• Acceso privilegiado a AWS (S3, IAM, Lambda y EventBridge)

Recopilar los requisitos previos de la plataforma Swimlane (IDs, claves de API, IDs de organización y tokens)

1. Inicia sesión en la plataforma Swimlane como administrador de cuenta.
2. Ve a Opciones de perfil.
3. Haz clic en Perfil para abrir el editor de perfiles.
4. Ve a la sección Token de acceso personal.
5. Haz clic en Generar token para crear un token de acceso personal.
6. Copia el token inmediatamente y guárdalo de forma segura (no se volverá a mostrar).
7. Anota los siguientes detalles de la integración:
   • Token de acceso personal (PAT): se usa en el encabezado Private-Token para las llamadas a la API.
   • ID de cuenta: es obligatorio para la ruta de la API Audit Log /api/public/audit/account/{ACCOUNT_ID}/auditlogs. Si no sabes cuál es tu ID de cuenta, ponte en contacto con tu administrador de Swimlane.
   • URL base: tu dominio de Swimlane (por ejemplo, https://eu.swimlane.app o https://us.swimlane.app).

Configurar un segmento de AWS S3 y IAM para Google SecOps

1. Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
2. Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, swimlane-audit).
3. Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
4. Selecciona el Usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. En la sección Claves de acceso, haz clic en Crear clave de acceso.
7. Selecciona Servicio de terceros como Caso práctico.
8. Haz clic en Siguiente.
9. Opcional: añade 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. En la sección Políticas de permisos, haz clic en Añadir permisos.
15. Selecciona Añadir permisos.
16. Seleccione Adjuntar políticas directamente.
17. Busca y selecciona la política AmazonS3FullAccess.
18. Haz clic en Siguiente.
19. Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas de S3

1. En la consola de AWS, ve a IAM > Policies > Create policy > pestaña JSON.

2. Introduce la siguiente política:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutSwimlaneAuditObjects",
         "Effect": "Allow",
         "Action": ["s3:PutObject"],
         "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/*"
       },
       {
         "Sid": "AllowStateReadWrite",
         "Effect": "Allow",
         "Action": ["s3:GetObject", "s3:PutObject"],
         "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/state.json"
       }
     ]
   }
   

   • Sustituye swimlane-audit si has introducido otro nombre de segmento.

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

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

5. Adjunta la política que acabas de crear y la política gestionada por AWS:

   • La política personalizada creada anteriormente
   • service-role/AWSLambdaBasicExecutionRole (CloudWatch Logs)

6. Dale el nombre WriteSwimlaneAuditToS3Role al rol y haz clic en Crear rol.

Crear la función Lambda

1. En la consola de AWS, ve a Lambda > Funciones > Crear función.
2. Haz clic en Crear desde cero.

3. Proporciona los siguientes detalles de configuración:

   Ajuste	Valor
   Nombre	swimlane_audit_to_s3
   Tiempo de ejecución	Python 3.13
   Arquitectura	x86_64
   Rol de ejecución	WriteSwimlaneAuditToS3Role

4. Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (swimlane_audit_to_s3.py):

   #!/usr/bin/env python3
   import os, json, gzip, io, uuid, datetime as dt, urllib.parse, urllib.request
   import boto3
   
   # ---- Environment ----
   S3_BUCKET = os.environ["S3_BUCKET"]
   S3_PREFIX = os.environ.get("S3_PREFIX", "swimlane/audit/")
   STATE_KEY = os.environ.get("STATE_KEY", S3_PREFIX + "state.json")
   BASE_URL = os.environ["SWIMLANE_BASE_URL"].rstrip("/")  # e.g., https://eu.swimlane.app
   ACCOUNT_ID = os.environ["SWIMLANE_ACCOUNT_ID"]
   TENANT_LIST = os.environ.get("SWIMLANE_TENANT_LIST", "")  # comma-separated; optional
   INCLUDE_ACCOUNT = os.environ.get("INCLUDE_ACCOUNT", "true").lower() == "true"
   PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "100"))  # max 100
   WINDOW_MINUTES = int(os.environ.get("WINDOW_MINUTES", "15"))  # time range per run
   PAT_TOKEN = os.environ["SWIMLANE_PAT_TOKEN"]  # Personal Access Token
   TIMEOUT = int(os.environ.get("TIMEOUT", "30"))
   
   AUDIT_URL = f"{BASE_URL}/api/public/audit/account/{ACCOUNT_ID}/auditlogs"
   
   s3 = boto3.client("s3")
   
   # ---- Helpers ----
   
   def _http(req: urllib.request.Request):
       return urllib.request.urlopen(req, timeout=TIMEOUT)
   
   def _now():
       return dt.datetime.utcnow()
   
   def get_state() -> dict:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           return json.loads(obj["Body"].read())
       except Exception:
           return {}
   
   def put_state(state: dict) -> None:
       state["updated_at"] = _now().isoformat() + "Z"
       s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state).encode())
   
   def build_url(from_dt: dt.datetime, to_dt: dt.datetime, page: int) -> str:
       params = {
           "pageNumber": str(page),
           "pageSize": str(PAGE_SIZE),
           "includeAccount": str(INCLUDE_ACCOUNT).lower(),
           "fromdate": from_dt.replace(microsecond=0).isoformat() + "Z",
           "todate": to_dt.replace(microsecond=0).isoformat() + "Z",
       }
       if TENANT_LIST:
           params["tenantList"] = TENANT_LIST
       return AUDIT_URL + "?" + urllib.parse.urlencode(params)
   
   def fetch_page(url: str) -> dict:
       headers = {
           "Accept": "application/json",
           "Private-Token": PAT_TOKEN,
       }
       req = urllib.request.Request(url, headers=headers)
       with _http(req) as r:
           return json.loads(r.read())
   
   def write_chunk(items: list[dict], ts: dt.datetime) -> str:
       key = f"{S3_PREFIX}{ts:%Y/%m/%d}/swimlane-audit-{uuid.uuid4()}.json.gz"
       buf = io.BytesIO()
       with gzip.GzipFile(fileobj=buf, mode="w") as gz:
           for rec in items:
               gz.write((json.dumps(rec) + "n").encode())
       buf.seek(0)
       s3.upload_fileobj(buf, S3_BUCKET, key)
       return key
   
   def lambda_handler(event=None, context=None):
       state = get_state()
   
       # determine window
       to_dt = _now()
       from_dt = to_dt - dt.timedelta(minutes=WINDOW_MINUTES)
       if (prev := state.get("last_to_dt")):
           try:
               from_dt = dt.datetime.fromisoformat(prev.replace("Z", "+00:00"))
           except Exception:
               pass
   
       page = int(state.get("page", 1))
       total_written = 0
   
       while True:
           url = build_url(from_dt, to_dt, page)
           resp = fetch_page(url)
           items = resp.get("auditlogs", []) or []
           if items:
               write_chunk(items, _now())
               total_written += len(items)
           next_path = resp.get("next")
           if not next_path:
               break
           page += 1
           state["page"] = page
   
       # advance state window
       state["last_to_dt"] = to_dt.replace(microsecond=0).isoformat() + "Z"
       state["page"] = 1
       put_state(state)
   
       return {"ok": True, "written": total_written, "from": from_dt.isoformat() + "Z", "to": to_dt.isoformat() + "Z"}
   
   if __name__ == "__main__":
       print(lambda_handler())
   

5. Vaya a Configuración > Variables de entorno.

6. Haz clic en Editar > Añadir nueva variable de entorno.

7. Introduce las siguientes variables de entorno y sustituye los valores por los tuyos.

   Clave	Valor de ejemplo
   S3_BUCKET	swimlane-audit
   S3_PREFIX	swimlane/audit/
   STATE_KEY	swimlane/audit/state.json
   SWIMLANE_BASE_URL	https://eu.swimlane.app
   SWIMLANE_ACCOUNT_ID	xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
   SWIMLANE_TENANT_LIST	tenantA,tenantB (opcional)
   INCLUDE_ACCOUNT	true
   PAGE_SIZE	100
   WINDOW_MINUTES	15
   SWIMLANE_PAT_TOKEN	<your-personal-access-token>
   TIMEOUT	30

8. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).

9. Seleccione la pestaña Configuración.

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

11. Cambia Tiempo de espera a 5 minutos (300 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 ejecutándose cuando el temporizador llegue a los 5 minutos, AWS Lambda la finalizará. Si la ejecución es más corta, se completará con normalidad y solo se te cobrará por el tiempo de ejecución real.

Crear una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
2. Proporcione los siguientes detalles de configuración:
   • Programación periódica: tarifa (15 min)
   • Destino: tu función Lambda swimlane_audit_to_s3
   • Nombre: swimlane-audit-schedule-15min
3. Haz clic en Crear programación.

Opcional: Crear un usuario y claves de gestión de identidades y accesos de solo lectura para Google SecOps

1. En la consola de AWS, ve a IAM > Usuarios > Añadir usuarios.
2. Haz clic en Add users (Añadir usuarios).
3. Proporcione los siguientes detalles de configuración:
   • Usuario: secops-reader
   • Tipo de acceso: clave de acceso (acceso programático)
4. Haz clic en Crear usuario.
5. Asigna una política de lectura mínima (personalizada): Usuarios > secops-reader > Permisos > Añadir permisos > Asignar políticas directamente > Crear política.

6. En el editor de JSON, introduce la siguiente política:

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

7. Asigna el nombre secops-reader-policy.

8. Ve a Crear política > busca o selecciona > Siguiente > Añadir permisos.

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

10. Descarga el archivo CSV (estos valores se introducen en el feed).

Configurar un feed en Google SecOps para ingerir registros de la plataforma Swimlane

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en + Añadir nuevo feed.
3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, Swimlane Platform logs).
4. Selecciona Amazon S3 V2 como Tipo de fuente.
5. Seleccione Plataforma de carril como Tipo de registro.
6. Haz clic en Siguiente.
7. Especifique los valores de los siguientes parámetros de entrada:
   • URI de S3: s3://swimlane-audit/swimlane/audit/
   • Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
   • Antigüedad máxima del archivo: incluye los archivos modificados en los últimos días. Valor predeterminado: 180 días.
   • ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
   • Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
   • Espacio de nombres de recursos: el espacio de nombres de recursos.
   • Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.