Coletar registros de contexto da entidade do Duo

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir dados de contexto de entidade do Duo no Google Security Operations usando o Google Cloud Storage. O analisador transforma os registros JSON em um modelo de dados unificado (UDM) extraindo primeiro os campos do JSON bruto e mapeando esses campos para atributos do UDM. Ele processa vários cenários de dados, incluindo informações de usuários e recursos, detalhes de software e rótulos de segurança, garantindo uma representação abrangente no esquema do UDM.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

• Uma instância do Google SecOps
• Acesso privilegiado ao locatário do Duo (aplicativo da API Admin com privilégios administrativos suficientes para gerenciar aplicativos)
• Um projeto do GCP com a API Cloud Storage ativada
• Permissões para criar e gerenciar buckets do GCS
• Permissões para gerenciar políticas do IAM em buckets do GCS
• Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler

Configurar o aplicativo da API Admin do Duo

1. Faça login no painel de administração do Duo.
2. Acesse Aplicativos > Proteger um aplicativo.
3. Pesquise API Admin e clique em Proteger.
4. Anote os seguintes valores:
   • Chave de integração (ikey)
   • Chave secreta (skey)
   • Nome do host da API (por exemplo, api-XXXXXXXX.duosecurity.com)
5. Em Permissões, ative Conceder recurso - leitura (para ler usuários, grupos, smartphones, endpoints, tokens e credenciais WebAuthn).

6. Clique em Salvar.

   Observação: você precisa ter privilégios administrativos suficientes (por exemplo, Proprietário ou um papel equivalente com permissão para gerenciar aplicativos) no Duo para criar ou modificar aplicativos da API Admin.

Criar um bucket do Google Cloud Storage

1. Acesse o Console do Google Cloud.
2. Selecione seu projeto ou crie um novo.
3. No menu de navegação, acesse Cloud Storage > Buckets.
4. Clique em Criar bucket.

5. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nomeie seu bucket	Insira um nome exclusivo globalmente, por exemplo, duo-context.
   Tipo de local	Escolha com base nas suas necessidades (região, birregional, multirregional)
   Local	Selecione o local (por exemplo, us-central1).
   Classe de armazenamento	Padrão (recomendado para registros acessados com frequência)
   Controle de acesso	Uniforme (recomendado)
   Ferramentas de proteção	Opcional: ativar o controle de versões de objetos ou a política de retenção

6. Clique em Criar.

7. Salve o nome e a região do bucket para referência futura.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS e ser invocada pelo Pub/Sub.

Criar conta de serviço

1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
2. Clique em Criar conta de serviço.
3. Informe os seguintes detalhes de configuração:
   • Nome da conta de serviço: insira duo-entity-context-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Duo entity context data.
4. Clique em Criar e continuar.
5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
   1. Clique em Selecionar papel.
   2. Pesquise e selecione Administrador de objetos do Storage.
   3. Clique em + Adicionar outro papel.
   4. Pesquise e selecione Invocador do Cloud Run.
   5. Clique em + Adicionar outro papel.
   6. Pesquise e selecione Invocador do Cloud Functions.
6. Clique em Continuar.
7. Clique em Concluído.

Esses papéis são necessários para:

• Administrador de objetos do Storage: gravar registros em um bucket do GCS
• Invocador do Cloud Run: permite que o Pub/Sub invoque a função
• Invocador do Cloud Functions: permite a invocação de funções

Conceder permissões do IAM no bucket do GCS

Conceda permissões de gravação à conta de serviço no bucket do GCS:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, duo-entity-context-sa@PROJECT_ID.iam.gserviceaccount.com).
   • Atribuir papéis: selecione Administrador de objetos do Storage.
6. Clique em Salvar.

Criar tópico Pub/Sub

Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

1. No Console do GCP, acesse Pub/Sub > Tópicos.
2. Selecione Criar tópico.
3. Informe os seguintes detalhes de configuração:
   • ID do tópico: insira duo-entity-context-trigger.
   • Não altere as outras configurações.
4. Clique em Criar.

Criar uma função do Cloud Run para coletar dados de contexto de entidade

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar dados de contexto de entidade da API Duo Admin e gravá-los no GCS.

1. No console do GCP, acesse o Cloud Run.
2. Clique em Criar serviço.
3. Selecione Função (use um editor in-line para criar uma função).

4. Na seção Configurar, forneça os seguintes detalhes de configuração:

   Configuração	Valor
   Nome do serviço	duo-entity-context-collector
   Região	Selecione a região que corresponde ao seu bucket do GCS (por exemplo, us-central1).
   Ambiente de execução	Selecione Python 3.12 ou uma versão mais recente.

5. Na seção Acionador (opcional):

   1. Clique em + Adicionar gatilho.
   2. Selecione Cloud Pub/Sub.
   3. Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico do Pub/Sub (duo-entity-context-trigger).
   4. Clique em Salvar.

6. Na seção Autenticação:

   1. Selecione Exigir autenticação.
   2. Confira o Identity and Access Management (IAM).
   Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.

7. Role a tela para baixo e abra Contêineres, rede, segurança.

8. Acesse a guia Segurança:

   • Conta de serviço: selecione a conta de serviço (duo-entity-context-sa).

9. Acesse a guia Contêineres:

   1. Clique em Variáveis e secrets.
   2. Clique em + Adicionar variável para cada variável de ambiente:

   Nome da variável	Valor de exemplo
   GCS_BUCKET	duo-context
   GCS_PREFIX	duo/context/
   DUO_IKEY	DIXYZ...
   DUO_SKEY	****************
   DUO_API_HOSTNAME	api-XXXXXXXX.duosecurity.com
   LIMIT	100
   RESOURCES	users,groups,phones,endpoints,tokens,webauthncredentials

10. Na seção Variáveis e secrets, role a tela até Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos).

11. Acesse a guia Configurações em Contêineres:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
    • Clique em Concluído.

12. Role até Ambiente de execução:

    • Selecione Padrão (recomendado).

13. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 ou ajuste com base na carga esperada.

14. Clique em Criar.

15. Aguarde a criação do serviço (1 a 2 minutos).

16. Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

1. Insira main em Ponto de entrada da função.

2. No editor de código em linha, crie dois arquivos:

   • Primeiro arquivo: main.py::

   import functions_framework
   from google.cloud import storage
   import json
   import os
   import time
   import hmac
   import hashlib
   import base64
   import email.utils
   import urllib.parse
   from urllib.request import Request, urlopen
   
   # Environment variables
   DUO_IKEY = os.environ["DUO_IKEY"]
   DUO_SKEY = os.environ["DUO_SKEY"]
   DUO_API_HOSTNAME = os.environ["DUO_API_HOSTNAME"].strip()
   GCS_BUCKET = os.environ["GCS_BUCKET"]
   GCS_PREFIX = os.environ.get("GCS_PREFIX", "duo/context/")
   
   # Default resources can be adjusted via ENV
   RESOURCES = [r.strip() for r in os.environ.get("RESOURCES", "users,groups,phones,endpoints,tokens,webauthncredentials,desktop_authenticators").split(",") if r.strip()]
   
   # Duo paging: default 100; max varies by endpoint
   LIMIT = int(os.environ.get("LIMIT", "100"))
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   def _canon_params(params: dict) -> str:
       """RFC3986 encoding with '~' unescaped, keys sorted lexicographically."""
       if not params:
           return ""
       parts = []
       for k in sorted(params.keys()):
           v = params[k]
           if v is None:
               continue
           ks = urllib.parse.quote(str(k), safe="~")
           vs = urllib.parse.quote(str(v), safe="~")
           parts.append(f"{ks}={vs}")
       return "&".join(parts)
   
   def _sign(method: str, host: str, path: str, params: dict) -> dict:
       """Construct Duo Admin API Authorization + Date headers (HMAC-SHA1)."""
       now = email.utils.formatdate()
       canon = "\n".join([
           now,
           method.upper(),
           host.lower(),
           path,
           _canon_params(params)
       ])
       sig = hmac.new(
           DUO_SKEY.encode("utf-8"),
           canon.encode("utf-8"),
           hashlib.sha1
       ).hexdigest()
       auth = base64.b64encode(f"{DUO_IKEY}:{sig}".encode("utf-8")).decode("utf-8")
       return {
           "Date": now,
           "Authorization": f"Basic {auth}"
       }
   
   def _call(method: str, path: str, params: dict) -> dict:
       host = DUO_API_HOSTNAME
       assert host.startswith("api-") and host.endswith(".duosecurity.com"), \
           "DUO_API_HOSTNAME must be e.g. api-XXXXXXXX.duosecurity.com"
   
       qs = _canon_params(params)
       url = f"https://{host}{path}" + (f"?{qs}" if method.upper() == "GET" and qs else "")
   
       req = Request(url, method=method.upper())
       for k, v in _sign(method, host, path, params).items():
           req.add_header(k, v)
   
       with urlopen(req, timeout=60) as r:
           return json.loads(r.read().decode("utf-8"))
   
   def _write_json(obj: dict, when: float, resource: str, page: int) -> str:
       bucket = storage_client.bucket(GCS_BUCKET)
       prefix = GCS_PREFIX.strip("/") + "/" if GCS_PREFIX else ""
       key = f"{prefix}{time.strftime('%Y/%m/%d', time.gmtime(when))}/duo-{resource}-{page:05d}.json"
   
       blob = bucket.blob(key)
       blob.upload_from_string(
           json.dumps(obj, separators=(",", ":")),
           content_type="application/json"
       )
       return key
   
   def _fetch_resource(resource: str) -> dict:
       """Fetch all pages for a list endpoint using limit/offset + metadata.next_offset."""
       path = f"/admin/v1/{resource}"
       offset = 0
       page = 0
       now = time.time()
       total_items = 0
   
       while True:
           params = {"limit": LIMIT, "offset": offset}
           data = _call("GET", path, params)
           _write_json(data, now, resource, page)
           page += 1
   
           resp = data.get("response")
           # most endpoints return a list; if not a list, count as 1 object page
           if isinstance(resp, list):
               total_items += len(resp)
           elif resp is not None:
               total_items += 1
   
           meta = data.get("metadata") or {}
           next_offset = meta.get("next_offset")
           if next_offset is None:
               break
   
           # Duo returns next_offset as int
           try:
               offset = int(next_offset)
           except Exception:
               break
   
       return {
           "resource": resource,
           "pages": page,
           "objects": total_items
       }
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch Duo entity context data and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
       results = []
       for res in RESOURCES:
           print(f"Fetching resource: {res}")
           result = _fetch_resource(res)
           results.append(result)
           print(f"Completed {res}: {result['pages']} pages, {result['objects']} objects")
   
       print(f"All resources fetched successfully: {results}")
   

   • Segundo arquivo: requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   

3. Clique em Implantar para salvar e implantar a função.

4. Aguarde a conclusão da implantação (2 a 3 minutos).

   Observação: a configuração do gatilho do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.
   Observação: diferentes endpoints da API Admin têm valores máximos diferentes. Usar um valor maior do que o aceito para um determinado endpoint vai resultar em um erro 400. O valor padrão de 100 é seguro para todos os endpoints.

Criar o job do Cloud Scheduler

O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

1. No Console do GCP, acesse o Cloud Scheduler.
2. Clique em Criar job.

3. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nome	duo-entity-context-hourly
   Região	Selecione a mesma região da função do Cloud Run
   Frequência	0 * * * * (a cada hora, na hora)
   Fuso horário	Selecione o fuso horário (UTC recomendado)
   Tipo de destino	Pub/Sub
   Tópico	Selecione o tópico do Pub/Sub (duo-entity-context-trigger).
   Corpo da mensagem	{} (objeto JSON vazio)

4. Clique em Criar.

Opções de frequência de programação

• Escolha a frequência com base nos requisitos de atualização de dados:

  Frequência	Expressão Cron	Caso de uso
  A cada hora	0 * * * *	Padrão (recomendado)
  A cada 2 horas	0 */2 * * *	Atualização moderada
  A cada 6 horas	0 */6 * * *	Atualizações de baixa frequência
  Diário	0 0 * * *	Atualizações mínimas

  Observação: considere os limites de taxa da API Admin do Duo ao escolher a frequência da programação. Uma coleta mais frequente pode exigir o ajuste do parâmetro LIMIT ou a implementação de outra lógica de limitação de taxa.

Testar o job do programador

1. No console do Cloud Scheduler, encontre seu job (duo-entity-context-hourly).
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > duo-entity-context-collector > Registros.
4. Verifique se a função foi executada com sucesso.
5. Verifique o bucket do GCS para confirmar se os dados de contexto da entidade foram gravados.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, Duo Entity Context).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Dados de contexto da entidade do Duo como o Tipo de registro.

7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:

   [e-mail]
   

8. Copie esse endereço de e-mail para usar na próxima etapa.

   Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
   • Atribuir papéis: selecione Leitor de objetos do Storage.

6. Clique em Salvar.

   Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar um feed no Google SecOps para ingerir dados de contexto de entidade do Duo

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, Duo Entity Context).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Dados de contexto da entidade do Duo como o Tipo de registro.
7. Clique em Próxima.

8. Especifique valores para os seguintes parâmetros de entrada:

   • URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:

     gs://duo-context/duo/context/
     

     • Substitua:

       • duo-context: o nome do bucket do GCS.
       • duo/context/: prefixo/caminho da pasta em que os registros são armazenados (precisa corresponder à variável de ambiente GCS_PREFIX).
     Observação: sempre inclua a barra (/) no final do URI.

   • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

     • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
     • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.

     • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.

       Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.

   • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.

   • Namespace do recurso: o namespace do recurso.

   • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

9. Clique em Próxima.

10. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Tabela de mapeamento do UDM

Campo de registro	Mapeamento do UDM	Lógica
Ativado	entity.asset.deployment_status	Se "activated" for falso, defina como "DECOMISSIONED". Caso contrário, defina como "ACTIVE".
browsers.browser_family	entity.asset.software.name	Extraído da matriz "browsers" no registro bruto.
browsers.browser_version	entity.asset.software.version	Extraído da matriz "browsers" no registro bruto.
device_name	entity.asset.hostname	Mapeado diretamente do registro bruto.
disk_encryption_status	entity.asset.attribute.labels.key: "disk_encryption_status", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
e-mail	entity.user.email_addresses	Mapeado diretamente do registro bruto se ele contiver "@". Caso contrário, usa "username" ou "username1" se eles contiverem "@".
criptografado	entity.asset.attribute.labels.key: "Encrypted", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
epkey	entity.asset.product_object_id	Usado como "product_object_id" se estiver presente. Caso contrário, usa "phone_id" ou "token_id".
impressão digital	entity.asset.attribute.labels.key: "Finger Print", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
firewall_status	entity.asset.attribute.labels.key: "firewall_status", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
hardware_uuid	entity.asset.asset_id	Usado como "asset_id" se estiver presente. Caso contrário, usa "user_id".
last_seen	entity.asset.last_discover_time	Analisado como um carimbo de data/hora ISO8601 e mapeado.
modelo	entity.asset.hardware.model	Mapeado diretamente do registro bruto.
número	entity.user.phone_numbers	Mapeado diretamente do registro bruto.
os_family	entity.asset.platform_software.platform	Mapeado para "WINDOWS", "LINUX" ou "MAC" com base no valor, sem diferenciar maiúsculas e minúsculas.
versão_so	entity.asset.platform_software.platform_version	Mapeado diretamente do registro bruto.
password_status	entity.asset.attribute.labels.key: "password_status", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
phone_id	entity.asset.product_object_id	Usado como "product_object_id" se "epkey" não estiver presente. Caso contrário, usa "token_id".
security_agents.security_agent	entity.asset.software.name	Extraído da matriz "security_agents" no registro bruto.
security_agents.version	entity.asset.software.version	Extraído da matriz "security_agents" no registro bruto.
timestamp	entity.metadata.collected_timestamp	Preenche o campo "collected_timestamp" no objeto "metadata".
token_id	entity.asset.product_object_id	Usado como "product_object_id" se "epkey" e "phone_id" não estiverem presentes.
trusted_endpoint	entity.asset.attribute.labels.key: "trusted_endpoint", entity.asset.attribute.labels.value	Mapeado diretamente do registro bruto e convertido em letras minúsculas.
tipo	entity.asset.type	Se o "type" do registro bruto contiver "mobile" (sem diferenciação de maiúsculas e minúsculas), defina como "MOBILE". Caso contrário, defina como "LAPTOP".
user_id	entity.asset.asset_id	Usado como "asset_id" se "hardware_uuid" não estiver presente.
users.email	entity.user.email_addresses	Usado como "email_addresses" se for o primeiro usuário na matriz "users" e contiver "@".
users.username	entity.user.userid	Nome de usuário extraído antes de "@" e usado como "userid" se for o primeiro usuário na matriz "users".
	entity.metadata.vendor_name	"Duo"
	entity.metadata.product_name	"Dados de contexto da entidade do Duo"
	entity.metadata.entity_type	RECURSO
	entity.relations.entity_type	USUÁRIO
	entity.relations.relationship	OWNS

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