Recopila registros del servicio Citrix Monitor
En este documento, se explica cómo transferir registros del servicio Citrix Monitor a Google Security Operations con Amazon S3. El analizador transforma los registros sin procesar con formato JSON en un formato estructurado que cumple con el UDM de Google SecOps. Extrae los campos pertinentes del registro sin procesar, los asigna a los campos correspondientes del UDM y enriquece los datos con contexto adicional, como información del usuario, detalles de la máquina y actividad de la red.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos previos:
- Instancia de Google SecOps
- Acceso privilegiado al arrendatario de Citrix Cloud
- Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)
Recopila los requisitos previos del servicio Citrix Monitor (IDs, claves de API, IDs de organización, tokens)
- Accede a la consola de Citrix Cloud.
- Ve a Identity and Access Management > API Access.
- Haz clic en Crear cliente.
- Copia y guarda en una ubicación segura los siguientes detalles:
- Client ID (ID de cliente)
- Client Secret (Secreto del cliente)
- ID de cliente (visible en la consola de Citrix Cloud)
- URL base de la API:
- Global:
https://api.cloud.com
- Japón:
https://api.citrixcloud.jp
- Global:
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,
citrix-monitor-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
- En la consola de AWS, ve a IAM > Políticas > Crear política > pestaña JSON.
Ingresa la siguiente política:
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::citrix-monitor-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::citrix-monitor-logs/citrix_monitor/state.json" } ] }
- Reemplaza
citrix-monitor-logs
si ingresaste un nombre de bucket diferente.
- Reemplaza
Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunta la política recién creada y la política administrada AWSLambdaBasicExecutionRole.
Asigna el nombre
CitrixMonitorLambdaRole
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 CitrixMonitorCollector
Tiempo de ejecución Python 3.13 Arquitectura x86_64 Rol de ejecución CitrixMonitorLambdaRole
Después de crear la función, abre la pestaña Code, borra el código auxiliar y, luego, ingresa el siguiente código (
CitrixMonitorCollector.py
):import os import json import uuid import datetime import urllib.parse import urllib.request import urllib.error import boto3 import botocore # Citrix Cloud OAuth2 endpoint template TOKEN_URL_TMPL = "{api_base}/cctrustoauth2/{customerid}/tokens/clients" DEFAULT_API_BASE = "https://api.cloud.com" MONITOR_BASE_PATH = "/monitorodata" s3 = boto3.client("s3") def http_post_form(url, data_dict): """POST form data to get authentication token.""" data = urllib.parse.urlencode(data_dict).encode("utf-8") req = urllib.request.Request(url, data=data, headers={ "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded", }) with urllib.request.urlopen(req, timeout=45) as resp: return json.loads(resp.read().decode("utf-8")) def http_get_json(url, headers): """GET JSON data from API endpoint.""" req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=90) as resp: return json.loads(resp.read().decode("utf-8")) def get_citrix_token(api_base, customer_id, client_id, client_secret): """Get Citrix Cloud authentication token.""" url = TOKEN_URL_TMPL.format(api_base=api_base.rstrip("/"), customerid=customer_id) payload = { "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, } response = http_post_form(url, payload) return response["access_token"] def build_entity_url(api_base, entity, filter_query=None, top=None): """Build OData URL with optional filter and pagination.""" base = api_base.rstrip("/") + MONITOR_BASE_PATH + "/" + entity params = [] if filter_query: params.append("$filter=" + urllib.parse.quote(filter_query, safe="()= ':-TZ0123456789")) if top: params.append("$top=" + str(top)) return base + ("?" + "&".join(params) if params else "") def fetch_entity_rows(entity, start_iso=None, end_iso=None, page_size=1000, headers=None, api_base=DEFAULT_API_BASE): """Fetch entity data with optional time filtering and pagination.""" # Try ModifiedDate filter if timestamps are provided first_url = None if start_iso and end_iso: filter_query = f"(ModifiedDate ge {start_iso} and ModifiedDate lt {end_iso})" first_url = build_entity_url(api_base, entity, filter_query, page_size) else: first_url = build_entity_url(api_base, entity, None, page_size) url = first_url while url: try: data = http_get_json(url, headers) items = data.get("value", []) for item in items: yield item url = data.get("@odata.nextLink") except urllib.error.HTTPError as e: # If ModifiedDate filtering fails, fall back to unfiltered query if e.code == 400 and start_iso and end_iso: print(f"ModifiedDate filter not supported for {entity}, falling back to unfiltered query") url = build_entity_url(api_base, entity, None, page_size) continue else: raise def read_state_file(bucket, key): """Read the last processed timestamp from S3 state file.""" try: obj = s3.get_object(Bucket=bucket, Key=key) content = obj["Body"].read().decode("utf-8") state = json.loads(content) timestamp_str = state.get("last_hour_utc") if timestamp_str: return datetime.datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")).replace(tzinfo=None) except botocore.exceptions.ClientError as e: if e.response["Error"]["Code"] == "NoSuchKey": return None raise return None def write_state_file(bucket, key, dt_utc): """Write the current processed timestamp to S3 state file.""" state = {"last_hour_utc": dt_utc.isoformat() + "Z"} s3.put_object( Bucket=bucket, Key=key, Body=json.dumps(state, separators=(",", ":")), ContentType="application/json" ) def write_ndjson_to_s3(bucket, key, rows): """Write rows as NDJSON to S3.""" body_lines = [] for row in rows: json_line = json.dumps(row, separators=(",", ":"), ensure_ascii=False) body_lines.append(json_line) body = ("n".join(body_lines) + "n").encode("utf-8") s3.put_object( Bucket=bucket, Key=key, Body=body, ContentType="application/x-ndjson" ) def lambda_handler(event, context): """Main Lambda handler function.""" # Environment variables bucket = os.environ["S3_BUCKET"] prefix = os.environ.get("S3_PREFIX", "citrix_monitor").strip("/") state_key = os.environ.get("STATE_KEY") or f"{prefix}/state.json" customer_id = os.environ["CITRIX_CUSTOMER_ID"] client_id = os.environ["CITRIX_CLIENT_ID"] client_secret = os.environ["CITRIX_CLIENT_SECRET"] api_base = os.environ.get("API_BASE", DEFAULT_API_BASE) entities = [e.strip() for e in os.environ.get("ENTITIES", "Machines,Sessions,Connections,Applications,Users").split(",") if e.strip()] page_size = int(os.environ.get("PAGE_SIZE", "1000")) lookback_minutes = int(os.environ.get("LOOKBACK_MINUTES", "75")) use_time_filter = os.environ.get("USE_TIME_FILTER", "true").lower() == "true" # Time window calculation now = datetime.datetime.utcnow() fallback_hour = (now - datetime.timedelta(minutes=lookback_minutes)).replace(minute=0, second=0, microsecond=0) last_processed = read_state_file(bucket, state_key) target_hour = (last_processed + datetime.timedelta(hours=1)) if last_processed else fallback_hour start_iso = target_hour.isoformat() + "Z" end_iso = (target_hour + datetime.timedelta(hours=1)).isoformat() + "Z" # Authentication token = get_citrix_token(api_base, customer_id, client_id, client_secret) headers = { "Authorization": f"CWSAuth bearer={token}", "Citrix-CustomerId": customer_id, "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br", "User-Agent": "citrix-monitor-s3-collector/1.0" } total_records = 0 # Process each entity type for entity in entities: rows_batch = [] try: entity_generator = fetch_entity_rows( entity=entity, start_iso=start_iso if use_time_filter else None, end_iso=end_iso if use_time_filter else None, page_size=page_size, headers=headers, api_base=api_base ) for row in entity_generator: # Store raw Citrix data directly for proper parser recognition rows_batch.append(row) # Write in batches to avoid memory issues if len(rows_batch) >= 1000: s3_key = f"{prefix}/{entity}/year={target_hour.year:04d}/month={target_hour.month:02d}/day={target_hour.day:02d}/hour={target_hour.hour:02d}/part-{uuid.uuid4().hex}.ndjson" write_ndjson_to_s3(bucket, s3_key, rows_batch) total_records += len(rows_batch) rows_batch = [] except Exception as ex: print(f"Error processing entity {entity}: {str(ex)}") continue # Write remaining records if rows_batch: s3_key = f"{prefix}/{entity}/year={target_hour.year:04d}/month={target_hour.month:02d}/day={target_hour.day:02d}/hour={target_hour.hour:02d}/part-{uuid.uuid4().hex}.ndjson" write_ndjson_to_s3(bucket, s3_key, rows_batch) total_records += len(rows_batch) # Update state file write_state_file(bucket, state_key, target_hour) return { "statusCode": 200, "body": json.dumps({ "success": True, "hour_collected": start_iso, "records_written": total_records, "entities_processed": entities }) }
Ve a Configuration > Environment variables.
Haz clic en Editar > Agregar nueva variable de entorno.
Ingresa las siguientes variables de entorno y reemplázalas por tus valores:
Clave Valor de ejemplo S3_BUCKET
citrix-monitor-logs
S3_PREFIX
citrix_monitor
STATE_KEY
citrix_monitor/state.json
CITRIX_CLIENT_ID
your-client-id
CITRIX_CLIENT_SECRET
your-client-secret
CITRIX_CUSTOMER_ID
your-customer-id
API_BASE
https://api.cloud.com
ENTITIES
Machines,Sessions,Connections,Applications,Users
PAGE_SIZE
1000
LOOKBACK_MINUTES
75
USE_TIME_FILTER
true
Después de crear la función, permanece en su página (o abre Lambda > Functions > CitrixMonitorCollector).
Selecciona la pestaña Configuración.
En el panel Configuración general, haz clic en Editar.
Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.
Crea una programación de EventBridge
- Ve a Amazon EventBridge > Scheduler > Create schedule.
- Proporciona los siguientes detalles de configuración:
- Programa recurrente: Frecuencia (
1 hour
) - Destino: Tu función Lambda
CitrixMonitorCollector
- Nombre:
CitrixMonitorCollector-1h
- Programa recurrente: Frecuencia (
- 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 > Agregar usuarios.
- Haz clic en Agregar usuarios.
- Proporciona los siguientes detalles de configuración:
- Usuario:
secops-reader
- Tipo de acceso: Clave de acceso: Acceso programático
- Usuario:
- Haz clic en Crear usuario.
- Adjunta la política de lectura mínima (personalizada): Usuarios > 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:::citrix-monitor-logs/*" }, { "Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": "arn:aws:s3:::citrix-monitor-logs" } ] }
Configura el nombre como
secops-reader-policy
.Ve a Crear política > busca o selecciona > Siguiente > Agregar permisos.
Ve a Credenciales de seguridad > Claves 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 del servicio Citrix Monitor
- Ve a Configuración de SIEM > Feeds.
- Haz clic en + Agregar feed nuevo.
- En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo,
Citrix Monitor Service logs
). - Selecciona Amazon S3 V2 como el Tipo de fuente.
- Selecciona Citrix Monitor como el Tipo de registro.
- Haz clic en Siguiente.
- Especifica valores para los siguientes parámetros de entrada:
- URI de S3:
s3://citrix-monitor-logs/citrix_monitor/
- Opciones de borrado de la fuente: Selecciona la opción de borrado según tu preferencia.
- Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es de 180 días.
- 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 aplica a los eventos de este feed.
- URI de S3:
- 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.