Esta página foi traduzida pela API Cloud Translation.

Coletar registros do Digital Shadows SearchLight

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros do Digital Shadows SearchLight no Google Security Operations usando o Amazon S3. O analisador extrai dados de ocorrência de segurança dos registros JSON. Ele inicializa campos do modelo de dados unificado (UDM), analisa o payload JSON, mapeia campos relevantes para o esquema do UDM, extrai entidades como e-mail e nome do host usando padrões grok e cria os objetos security_result e metadata no evento do UDM.

Antes de começar

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

Uma instância do Google SecOps.
Acesso privilegiado ao locatário do Digital Shadows SearchLight.
Acesso privilegiado à AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

Coletar os pré-requisitos do Digital Shadows SearchLight (IDs, chaves de API, IDs da organização, tokens)

Faça login no portal do Digital Shadows SearchLight.
Acesse Configurações > Credenciais da API.
Crie um novo cliente de API ou par de chaves.
Copie e salve em um local seguro os seguintes detalhes:
- Chave de API
- API Secret
- ID da conta
- URL base da API: https://api.searchlight.app/v1 ou https://portal-digitalshadows.com/api/v1

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

Crie um bucket do Amazon S3 seguindo este guia do usuário: Como criar um bucket
Salve o Nome e a Região do bucket para referência futura (por exemplo, digital-shadows-logs).
Crie um usuário seguindo este guia: Como criar um usuário do IAM.
Selecione o usuário criado.
Selecione a guia Credenciais de segurança.
Clique em Criar chave de acesso na seção Chaves de acesso.
Selecione Serviço de terceiros como Caso de uso.
Clique em Próxima.
Opcional: adicione uma tag de descrição.
Clique em Criar chave de acesso.
Clique em Fazer o download do arquivo .CSV para salvar a chave de acesso e a chave de acesso secreta para referência futura.
Clique em Concluído.
Selecione a guia Permissões.
Clique em Adicionar permissões na seção Políticas de permissões.
Selecione Adicionar permissões.
Selecione Anexar políticas diretamente.
Pesquise a política AmazonS3FullAccess.
Selecione a política.
Clique em Próxima.
Clique em Adicionar permissões

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

No console da AWS, acesse IAM > Políticas.
Clique em Criar política > guia JSON.
Copie e cole a política a seguir.

JSON da política (substitua digital-shadows-logs se você tiver inserido um nome de bucket diferente):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::digital-shadows-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::digital-shadows-logs/digital-shadows-searchlight/state.json"
    }
  ]
}

Clique em Próxima > Criar política.
Acesse IAM > Funções > Criar função > Serviço da AWS > Lambda.
Anexe a política recém-criada.
Nomeie a função como digital-shadows-lambda-role e clique em Criar função.

Criar a função Lambda

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

Configuração Valor

Nome digital-shadows-collector

Ambiente de execução Python 3.13

Arquitetura x86_64

Função de execução digital-shadows-lambda-role

Configuração	Valor
Nome	`digital-shadows-collector`
Ambiente de execução	Python 3.13
Arquitetura	x86_64
Função de execução	`digital-shadows-lambda-role`

Depois que a função for criada, abra a guia Código, exclua o stub e cole o código a seguir (digital-shadows-collector.py).

import json
import os
import base64
import logging
import time
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode

import boto3
import urllib3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

HTTP = urllib3.PoolManager(retries=False)

def _basic_auth_header(key: str, secret: str) -> str:
    token = base64.b64encode(f"{key}:{secret}".encode("utf-8")).decode("utf-8")
    return f"Basic {token}"

def _load_state(s3, bucket, key, default_days=30) -> str:
    """Return ISO8601 checkpoint (UTC)."""
    try:
        obj = s3.get_object(Bucket=bucket, Key=key)
        state = json.loads(obj["Body"].read().decode("utf-8"))
        ts = state.get("last_timestamp")
        if ts:
            return ts
    except s3.exceptions.NoSuchKey:
        pass
    except Exception as e:
        logger.warning(f"State read error: {e}")
    return (datetime.now(timezone.utc) - timedelta(days=default_days)).isoformat()

def _save_state(s3, bucket, key, ts: str) -> None:
    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=json.dumps({"last_timestamp": ts}).encode("utf-8"),
        ContentType="application/json",
    )

def _get_json(url: str, headers: dict, params: dict, backoff_s=2, max_retries=3) -> dict:
    qs = f"?{urlencode(params)}" if params else ""
    for attempt in range(max_retries):
        r = HTTP.request("GET", f"{url}{qs}", headers=headers)
        if r.status == 200:
            return json.loads(r.data.decode("utf-8"))
        if r.status in (429, 500, 502, 503, 504):
            wait = backoff_s * (2 ** attempt)
            logger.warning(f"HTTP {r.status} from DS API, retrying in {wait}s")
            time.sleep(wait)
            continue
        raise RuntimeError(f"DS API error {r.status}: {r.data[:200]}")
    raise RuntimeError("Exceeded retry budget for DS API")

def _collect(api_base, headers, path, since_ts, account_id, page_size, max_pages, time_param):
    items = []
    for page in range(max_pages):
        params = {
            "limit": page_size,
            "offset": page * page_size,
            time_param: since_ts,
        }
        if account_id:
            params["account-id"] = account_id

        data = _get_json(f"{api_base}/{path}", headers, params)
        batch = data.get("items") or data.get("data") or []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < page_size:
            break
    return items

def lambda_handler(event, context):
    # Required
    s3_bucket  = os.environ["S3_BUCKET"]
    api_key    = os.environ["DS_API_KEY"]
    api_secret = os.environ["DS_API_SECRET"]

    # Optional / defaults
    s3_prefix  = os.environ.get("S3_PREFIX", "digital-shadows-searchlight/")
    state_key  = os.environ.get("STATE_KEY", "digital-shadows-searchlight/state.json")
    api_base   = os.environ.get("API_BASE", "https://api.searchlight.app/v1")
    account_id = os.environ.get("DS_ACCOUNT_ID", "")
    page_size  = int(os.environ.get("PAGE_SIZE", "100"))
    max_pages  = int(os.environ.get("MAX_PAGES", "10"))

    s3 = boto3.client("s3")
    last_ts = _load_state(s3, s3_bucket, state_key)
    logger.info(f"Checkpoint: {last_ts}")

    headers = {
        "Authorization": _basic_auth_header(api_key, api_secret),
        "Accept": "application/json",
        "User-Agent": "Chronicle-DigitalShadows-S3/1.0",
    }

    records = []

    # Incidents (time filter often 'published-after' or 'updated-since' depending on tenancy)
    incidents = _collect(api_base, headers, "incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
    for incident in incidents:
        incident['_source_type'] = 'incident'
    records.extend(incidents)

    # Intelligence incidents (alerts)
    intel_incidents = _collect(api_base, headers, "intel-incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
    for intel in intel_incidents:
        intel['_source_type'] = 'intelligence_incident'
    records.extend(intel_incidents)

    # Indicators (IOCs)
    indicators = _collect(api_base, headers, "indicators", last_ts, account_id, page_size, max_pages, time_param="lastUpdated-after")
    for indicator in indicators:
        indicator['_source_type'] = 'ioc'
    records.extend(indicators)

    if records:
        # Choose newest timestamp seen in this batch
        newest = max(
            (r.get("updated") or r.get("raised") or r.get("lastUpdated") or last_ts)
            for r in records
        )
        key = f"{s3_prefix}digital_shadows_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
        body = "\n".join(json.dumps(r, separators=(",", ":")) for r in records).encode("utf-8")

        s3.put_object(
            Bucket=s3_bucket,
            Key=key,
            Body=body,
            ContentType="application/x-ndjson",
        )
        _save_state(s3, s3_bucket, state_key, newest)
        msg = f"Wrote {len(records)} records to s3://{s3_bucket}/{key}"
    else:
        msg = "No new records"

    logger.info(msg)
    return {"statusCode": 200, "body": msg}

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

Insira as variáveis de ambiente fornecidas na tabela a seguir, substituindo os valores de exemplo pelos seus.

Variáveis de ambiente

Chave	Valor de exemplo
`S3_BUCKET`	`digital-shadows-logs`
`S3_PREFIX`	`digital-shadows-searchlight/`
`STATE_KEY`	`digital-shadows-searchlight/state.json`
`DS_API_KEY`	`<your-6-character-api-key>`
`DS_API_SECRET`	`<your-32-character-api-secret>`
`API_BASE`	`https://api.searchlight.app/v1` (ou `https://portal-digitalshadows.com/api/v1`)
`DS_ACCOUNT_ID`	`<your-account-id>` (obrigatório para a maioria dos locatários)
`PAGE_SIZE`	`100`
`MAX_PAGES`	`10`

Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > sua-função.
Selecione a guia Configuração.
No painel Configuração geral, clique em Editar.
Mude Tempo limite para 5 minutos (300 segundos) e clique em Salvar.

Observação: o tempo limite é um limite superior fixo. Isso significa que, se a função ainda estiver em execução quando o timer atingir 5 minutos, o AWS Lambda vai encerrá-la. Execuções mais curtas terminam normalmente, e você só recebe cobranças pelo tempo de execução real.

Criar uma programação do EventBridge

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

(Opcional) Criar um usuário e chaves do IAM somente leitura para o Google SecOps

Acesse Console da AWS > IAM > Usuários.
Clique em Add users.
Informe os seguintes detalhes de configuração:
- Usuário: insira secops-reader.
- Tipo de acesso: selecione Chave de acesso – Acesso programático.
Clique em Criar usuário.
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.

JSON:

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

Name = secops-reader-policy.
Clique em Criar política > pesquisar/selecionar > Próxima > Adicionar permissões.
Crie uma chave de acesso para secops-reader: Credenciais de segurança > Chaves de acesso.
Clique em Criar chave de acesso.
Faça o download do .CSV. Cole esses valores no feed.

Configurar um feed no Google SecOps para ingerir registros do Digital Shadows SearchLight

Acesse Configurações do SIEM > Feeds.
Clique em + Adicionar novo feed.
No campo Nome do feed, insira um nome para o feed (por exemplo, Digital Shadows SearchLight logs).
Selecione Amazon S3 V2 como o Tipo de origem.
Selecione Digital Shadows SearchLight como o Tipo de registro.
Clique em Próxima.
Especifique valores para os seguintes parâmetros de entrada:
- URI do S3: s3://digital-shadows-logs/digital-shadows-searchlight/
- 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. O 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.
Clique em Próxima.
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.