Esta página foi traduzida pela API Cloud Translation.

Recolha registos do serviço Citrix Monitor

Compatível com:

Google secops SIEM

Este documento explica como carregar registos do serviço Citrix Monitor para o Google Security Operations através do Amazon S3. O analisador transforma os registos formatados JSON não processados num formato estruturado em conformidade com o UDM do Google SecOps. Extrai campos relevantes do registo não processado, mapeia-os para os campos da UDM correspondentes e enriquece os dados com contexto adicional, como informações do utilizador, detalhes da máquina e atividade de rede.

Antes de começar

Certifique-se de que tem os seguintes pré-requisitos:

Instância do Google SecOps
Acesso privilegiado ao inquilino do Citrix Cloud
Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Recolha os pré-requisitos do serviço Citrix Monitor (IDs, chaves da API, IDs da organização e tokens)

Inicie sessão na Citrix Cloud Console.
Aceda a Identity and Access Management > Acesso à API.
Clique em Criar cliente.
Copie e guarde numa localização segura os seguintes detalhes:
- ID de cliente
- Segredo do cliente
- ID do cliente (visível na consola do Citrix Cloud)
- URL base da API:
  - Global: https://api.cloud.com
  - Japão: https://api.citrixcloud.jp

Configure o contentor do AWS S3 e o IAM para o Google SecOps

Crie um contentor do Amazon S3 seguindo este manual do utilizador: Criar um contentor
Guarde o nome e a região do contentor para referência futura (por exemplo, citrix-monitor-logs).
Crie um utilizador seguindo este guia do utilizador: Criar um utilizador do IAM.
Selecione o utilizador criado.
Selecione o separador Credenciais de segurança.
Clique em Criar chave de acesso na secção Chaves de acesso.
Selecione Serviço de terceiros como o Exemplo de utilização.
Clicar em Seguinte.
Opcional: adicione uma etiqueta de descrição.
Clique em Criar chave de acesso.
Clique em Transferir ficheiro CSV para guardar a chave de acesso e a chave de acesso secreta para utilização posterior.
Clique em Concluído.
Selecione o separador Autorizações.
Clique em Adicionar autorizações na secção Políticas de autorizações.
Selecione Adicionar autorizações.
Selecione Anexar políticas diretamente
Pesquise e selecione a política AmazonS3FullAccess.
Clicar em Seguinte.
Clique em Adicionar autorizações.

Configure a política e a função de IAM para carregamentos do S3

Na consola da AWS, aceda a IAM > Políticas > Criar política > separador JSON.

Introduza a seguinte 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"
    }
  ]
}

Substitua citrix-monitor-logs se tiver introduzido um nome de contentor diferente.

Clique em Seguinte > Criar política.
Aceda a IAM > Funções > Criar função > Serviço AWS > Lambda.
Anexe a política recém-criada e a política gerida AWSLambdaBasicExecutionRole.
Dê o nome CitrixMonitorLambdaRole à função e clique em Criar função.

Crie a função Lambda

Na consola da AWS, aceda a Lambda > Functions > Create function.
Clique em Criar do zero.
Faculte os seguintes detalhes de configuração:

Definição Valor

Nome CitrixMonitorCollector

Runtime Python 3.13

Arquitetura x86_64

Função de execução CitrixMonitorLambdaRole

Definição	Valor
Nome	`CitrixMonitorCollector`
Runtime	Python 3.13
Arquitetura	x86_64
Função de execução	`CitrixMonitorLambdaRole`

Depois de criar a função, abra o separador Código, elimine o fragmento e introduza o seguinte 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
        })
    }

Aceda a Configuração > Variáveis de ambiente.
Clique em Editar > Adicionar nova variável de ambiente.

Introduza as seguintes variáveis de ambiente, substituindo-as pelos seus valores:

Chave	Valor de exemplo
`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`

Depois de criar a função, permaneça na respetiva página (ou abra Lambda > Functions > CitrixMonitorCollector).
Selecione o separador Configuração.
No painel Configuração geral, clique em Editar.
Altere Tempo limite para 5 minutos (300 segundos) e clique em Guardar.

Nota: o limite de tempo é um limite superior rígido, o que significa que, se a função ainda estiver em execução quando o temporizador atingir os 5 minutos, o AWS Lambda termina-a. Qualquer execução mais curta termina normalmente. Só lhe é cobrado o tempo de execução real.

Crie um horário do EventBridge

Aceda a Amazon EventBridge > Scheduler > Create schedule.
Indique os seguintes detalhes de configuração:
- Horário recorrente: Taxa (1 hour)
- Alvo: a sua função Lambda CitrixMonitorCollector
- Nome: CitrixMonitorCollector-1h
Clique em Criar programação.

Opcional: crie um utilizador e chaves da IAM só de leitura para o Google SecOps

Na consola da AWS, aceda a IAM > Utilizadores > Adicionar utilizadores.
Clique em Adicionar utilizadores.
Indique os seguintes detalhes de configuração:
- Utilizador: secops-reader
- Tipo de acesso: chave de acesso – acesso programático
Clique em Criar utilizador.
Anexe a política de leitura mínima (personalizada): Users > secops-reader > Permissions > Add permissions > Attach policies directly > Create policy.

No editor JSON, introduza a seguinte 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"
    }
  ]
}

Defina o nome como secops-reader-policy.
Aceda a Criar política > pesquise/selecione > Seguinte > Adicionar autorizações.
Aceda a Credenciais de segurança > Chaves de acesso > Criar chave de acesso.
Transfira o CSV (estes valores são introduzidos no feed).

Configure um feed no Google SecOps para carregar registos do serviço Citrix Monitor

Aceda a Definições do SIEM > Feeds.
Clique em + Adicionar novo feed.
No campo Nome do feed, introduza um nome para o feed (por exemplo, Citrix Monitor Service logs).
Selecione Amazon S3 V2 como o Tipo de origem.
Selecione Citrix Monitor como o Tipo de registo.
Clicar em Seguinte.
Especifique valores para os seguintes parâmetros de entrada:
- URI do S3: s3://citrix-monitor-logs/citrix_monitor/
- Opções de eliminação de origens: selecione a opção de eliminação de acordo com a sua preferência.
- Idade máxima do ficheiro: inclua ficheiros modificados no último número de dias. Predefinição de 180 dias.
- ID da chave de acesso: chave de acesso do utilizador com acesso ao contentor do S3.
- Chave de acesso secreta: chave secreta do utilizador com acesso ao contentor do S3.
- Espaço de nomes do recurso: o espaço de nomes do recurso.
- Etiquetas de carregamento: a etiqueta aplicada aos eventos deste feed.
Clicar em Seguinte.
Reveja a nova configuração do feed no ecrã Finalizar e, de seguida, clique em Enviar.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais da Google SecOps.