Coletar registros do Citrix Analytics

Compatível com:

Este documento explica como ingerir registros do Citrix Analytics no Google Security Operations usando o Amazon S3.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Instância do Google SecOps
  • Acesso privilegiado ao locatário do Citrix Analytics for Performance
  • Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Coletar pré-requisitos do Citrix Analytics

  1. Faça login no Citrix Cloud Console (em inglês).
  2. Acesse Identity and Access Management > Acesso à API.
  3. Clique em Criar cliente.
  4. Copie e salve em um local seguro os seguintes detalhes:
    • ID do cliente
    • Client Secret
    • ID do cliente (localizado no URL do Citrix Cloud ou na página do IAM)
    • URL base da API: https://api.cloud.com/casodata

Configurar o bucket do AWS S3 e o IAM para o Google SecOps

  1. Crie um bucket do Amazon S3 seguindo este guia do usuário: Como criar um bucket
  2. Salve o Nome e a Região do bucket para referência futura (por exemplo, citrix-analytics-logs).
  3. Crie um usuário seguindo este guia: Como criar um usuário do IAM.
  4. Selecione o usuário criado.
  5. Selecione a guia Credenciais de segurança.
  6. Clique em Criar chave de acesso na seção Chaves de acesso.
  7. Selecione Serviço de terceiros como o Caso de uso.
  8. Clique em Próxima.
  9. Opcional: adicione uma tag de descrição.
  10. Clique em Criar chave de acesso.
  11. Clique em Fazer o download do arquivo CSV para salvar a chave de acesso e a chave de acesso secreta para uso posterior.
  12. Clique em Concluído.
  13. Selecione a guia Permissões.
  14. Clique em Adicionar permissões na seção Políticas de permissões.
  15. Selecione Adicionar permissões.
  16. Selecione Anexar políticas diretamente.
  17. Pesquise e selecione a política AmazonS3FullAccess.
  18. Clique em Próxima.
  19. Clique em Adicionar permissões

Configurar a política e o papel do IAM para uploads do S3

  1. No console da AWS, acesse IAM > Políticas > Criar política > guia JSON.
  2. Insira a seguinte política:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::citrix-analytics-logs/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::citrix-analytics-logs/citrix_analytics/state.json"
        }
      ]
    }
    
    • Substitua citrix-analytics-logs se você tiver inserido um nome de bucket diferente.
  3. Clique em Próxima > Criar política.

  4. Acesse IAM > Funções > Criar função > Serviço da AWS > Lambda.

  5. Anexe a política recém-criada.

  6. Nomeie a função como CitrixAnalyticsLambdaRole e clique em Criar função.

Criar a função Lambda

  1. No console da AWS, acesse Lambda > Functions > Create function.
  2. Clique em Criar do zero.
  3. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nome CitrixAnalyticsCollector
    Ambiente de execução Python 3.13
    Arquitetura x86_64
    Função de execução CitrixAnalyticsLambdaRole
  4. Depois que a função for criada, abra a guia Código, exclua o stub e insira o seguinte código (CitrixAnalyticsCollector.py):

    import os
    import json
    import uuid
    import datetime
    import urllib.parse
    import urllib.request
    import boto3
    import botocore
    
    CITRIX_TOKEN_URL_TMPL = "https://api.cloud.com/cctrustoauth2/{customerid}/tokens/clients"
    DEFAULT_API_BASE = "https://api.cloud.com/casodata"
    
    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=30) as response:
            return json.loads(response.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=60) as response:
            return json.loads(response.read().decode("utf-8"))
    
    def get_citrix_token(customer_id, client_id, client_secret):
        """Get Citrix Cloud authentication token."""
        url = CITRIX_TOKEN_URL_TMPL.format(customerid=customer_id)
        payload = {
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        }
        token_response = _http_post_form(url, payload)
        return token_response["access_token"]
    
    def fetch_odata_entity(entity, when_utc, top, headers, api_base):
        """Fetch data from Citrix Analytics OData API with pagination."""
        year = when_utc.year
        month = when_utc.month
        day = when_utc.day
        hour = when_utc.hour
    
        base_url = f"{api_base.rstrip('/')}/{entity}?year={year:04d}&month={month:02d}&day={day:02d}&hour={hour:02d}"
        skip = 0
    
        while True:
            url = f"{base_url}&$top={top}&$skip={skip}"
            data = _http_get_json(url, headers)
            items = data.get("value", [])
    
            if not items:
                break
    
            for item in items:
                yield item
    
            if len(items) < top:
                break
    
            skip += top
    
    def read_state_file(bucket, state_key):
        """Read the last processed timestamp from S3 state file."""
        try:
            obj = s3.get_object(Bucket=bucket, Key=state_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, state_key, dt_utc):
        """Write the current processed timestamp to S3 state file."""
        state_data = {"last_hour_utc": dt_utc.isoformat() + "Z"}
        s3.put_object(
            Bucket=bucket, 
            Key=state_key,
            Body=json.dumps(state_data, separators=(",", ":")),
            ContentType="application/json"
        )
    
    def write_ndjson_to_s3(bucket, key, records):
        """Write records as NDJSON to S3."""
        body_lines = []
        for record in records:
            json_line = json.dumps(record, 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", "").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", "sessions,machines,users").split(",") if e.strip()]
        top_n = int(os.environ.get("TOP_N", "1000"))
        lookback_minutes = int(os.environ.get("LOOKBACK_MINUTES", "75"))
    
        # Determine target hour to collect
        now = datetime.datetime.utcnow()
        fallback_target = (now - datetime.timedelta(minutes=lookback_minutes)).replace(minute=0, second=0, microsecond=0)
    
        last_processed = read_state_file(bucket, state_key)
        if last_processed:
            target_hour = last_processed + datetime.timedelta(hours=1)
        else:
            target_hour = fallback_target
    
        # Get authentication token
        token = get_citrix_token(customer_id, client_id, client_secret)
        headers = {
            "Authorization": f"CwsAuth bearer={token}",
            "Citrix-CustomerId": customer_id,
            "Accept": "application/json",
            "Content-Type": "application/json",
        }
    
        total_records = 0
    
        # Process each entity type
        for entity in entities:
            records = []
    
            for row in fetch_odata_entity(entity, target_hour, top_n, headers, api_base):
                enriched_record = {
                    "citrix_entity": entity,
                    "citrix_hour_utc": target_hour.isoformat() + "Z",
                    "collection_timestamp": datetime.datetime.utcnow().isoformat() + "Z",
                    "raw": row
                }
                records.append(enriched_record)
    
                # Write in batches to avoid memory issues
                if len(records) >= 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, records)
                    total_records += len(records)
                    records = []
    
            # Write remaining records
            if records:
                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, records)
                total_records += len(records)
    
        # Update state file
        write_state_file(bucket, state_key, target_hour)
    
        return {
            "statusCode": 200,
            "body": json.dumps({
                "success": True,
                "hour_collected": target_hour.isoformat() + "Z",
                "records_written": total_records,
                "entities_processed": entities
            })
        }
    
  5. Acesse Configuração > Variáveis de ambiente > Editar > Adicionar nova variável de ambiente.

  6. Insira as seguintes variáveis de ambiente, substituindo pelos seus valores:

    Chave Valor de exemplo
    S3_BUCKET citrix-analytics-logs
    S3_PREFIX citrix_analytics
    STATE_KEY citrix_analytics/state.json
    CITRIX_CLIENT_ID your-client-id
    CITRIX_CLIENT_SECRET your-client-secret
    API_BASE https://api.cloud.com/casodata
    CITRIX_CUSTOMER_ID your-customer-id
    ENTITIES sessions,machines,users
    TOP_N 1000
    LOOKBACK_MINUTES 75
  7. Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > CitrixAnalyticsCollector.

  8. Selecione a guia Configuração.

  9. No painel Configuração geral, clique em Editar.

  10. Mude Tempo limite para 5 minutos (300 segundos) e clique em Salvar.

Criar uma programação do EventBridge

  1. Acesse Amazon EventBridge > Scheduler > Criar programação.
  2. Informe os seguintes detalhes de configuração:
    • Programação recorrente: Taxa (1 hour)
    • Destino: sua função Lambda CitrixAnalyticsCollector
    • Nome: CitrixAnalyticsCollector-1h
  3. Clique em Criar programação.

Opcional: criar um usuário e chaves do IAM somente leitura para o Google SecOps

  1. No console da AWS, acesse IAM > Usuários > Adicionar usuários.
  2. Clique em Add users.
  3. Informe os seguintes detalhes de configuração:
    • Usuário: secops-reader
    • Tipo de acesso: Chave de acesso — Acesso programático
  4. Clique em Criar usuário.
  5. Anexe a política de leitura mínima (personalizada): Usuários > secops-reader > Permissões > Adicionar permissões > Anexar políticas diretamente > Criar política.
  6. No editor JSON, insira a seguinte política:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::citrix-analytics-logs/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::citrix-analytics-logs"
        }
      ]
    }
    
  7. Defina o nome como secops-reader-policy.

  8. Acesse Criar política > pesquise/selecione > Próxima > Adicionar permissões.

  9. Acesse Credenciais de segurança > Chaves de acesso > Criar chave de acesso.

  10. Faça o download do CSV (esses valores são inseridos no feed).

Configurar um feed no Google SecOps para ingerir registros do Citrix Analytics

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em + Adicionar novo feed.
  3. No campo Nome do feed, insira um nome para o feed (por exemplo, Citrix Analytics Performance logs).
  4. Selecione Amazon S3 V2 como o Tipo de origem.
  5. Selecione Citrix Analytics como o Tipo de registro.
  6. Clique em Próxima.
  7. Especifique valores para os seguintes parâmetros de entrada:
    • URI do S3: s3://citrix-analytics-logs/citrix_analytics/
    • Opções de exclusão de fontes: selecione a opção de exclusão de acordo com sua preferência.
    • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. Padrão de 180 dias.
    • ID da chave de acesso: chave de acesso do usuário com acesso ao bucket do S3.
    • Chave de acesso secreta: chave secreta do usuário com acesso ao bucket do S3.
    • Namespace do recurso: o namespace do recurso.
    • Rótulos de ingestão: o rótulo aplicado aos eventos deste feed.
  8. Clique em Próxima.
  9. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

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