Coletar registros do PingOne Advanced Identity Cloud

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de ofertas pré-GA dos Termos específicos do 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 do serviço do Google SecOps.

Este documento explica como ingerir registros do PingOne Advanced Identity Cloud no Google Security Operations usando o Amazon S3.

Antes de começar

• Instância do Google SecOps
• Acesso privilegiado ao locatário do PingOne Advanced Identity Cloud
• Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Receber a chave de API do PingOne e o FQDN do locatário

1. Faça login no Admin Console do Advanced Identity Cloud.
2. Clique no ícone do usuário e em Configurações do locatário.
3. Na guia Configurações globais, clique em Registrar chaves de API.
4. Clique em Nova chave de API de registro e dê um nome para a chave.
5. Clique em Criar chave.
6. Copie e salve os valores api_key_id e api_key_secret em um local seguro. O valor de api_key_secret não será mostrado novamente.
7. Clique em Concluir.
8. Acesse Configurações do locatário > Detalhes e encontre o FQDN do locatário (por exemplo, example.tomcat.pingone.com).

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, pingone-aic-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": "AllowPutPingOneAICObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::pingone-aic-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::pingone-aic-logs/pingone-aic/logs/state.json"
       }
     ]
   }
   

   • Substitua pingone-aic-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 WritePingOneAICToS3Role 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	pingone_aic_to_s3
   Ambiente de execução	Python 3.13
   Arquitetura	x86_64
   Função de execução	WritePingOneAICToS3Role

4. Depois que a função for criada, abra a guia Código, exclua o stub e insira o seguinte código (pingone_aic_to_s3.py):

   #!/usr/bin/env python3
   
   import os, json, time, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError, URLError
   import boto3
   
   FQDN = os.environ["AIC_TENANT_FQDN"].strip("/")
   API_KEY_ID = os.environ["AIC_API_KEY_ID"]
   API_KEY_SECRET = os.environ["AIC_API_SECRET"]
   S3_BUCKET = os.environ["S3_BUCKET"]
   S3_PREFIX = os.environ.get("S3_PREFIX", "pingone-aic/logs/").strip("/")
   SOURCES = [s.strip() for s in os.environ.get("SOURCES", "am-everything,idm-everything").split(",") if s.strip()]
   PAGE_SIZE = min(int(os.environ.get("PAGE_SIZE", "500")), 1000)  # hard cap per docs
   MAX_PAGES = int(os.environ.get("MAX_PAGES", "20"))
   STATE_KEY = os.environ.get("STATE_KEY", "pingone-aic/logs/state.json")
   LOOKBACK_SECONDS = int(os.environ.get("LOOKBACK_SECONDS", "3600"))
   
   s3 = boto3.client("s3")
   
   def _headers():
       return {"x-api-key": API_KEY_ID, "x-api-secret": API_KEY_SECRET}
   
   def _iso(ts: float) -> str:
       return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))
   
   def _http_get(url: str, timeout: int = 60, max_retries: int = 5) -> dict:
       attempt, backoff = 0, 1.0
       while True:
           req = Request(url, method="GET", headers=_headers())
           try:
               with urlopen(req, timeout=timeout) as r:
                   data = r.read()
                   return json.loads(data.decode("utf-8"))
           except HTTPError as e:
               # 429: respect X-RateLimit-Reset (epoch seconds) if present
               if e.code == 429 and attempt < max_retries:
                   reset = e.headers.get("X-RateLimit-Reset")
                   now = int(time.time())
                   delay = max(1, int(reset) - now) if (reset and reset.isdigit()) else int(backoff)
                   time.sleep(delay); attempt += 1; backoff *= 2; continue
               if 500 <= e.code <= 599 and attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
           except URLError:
               if attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
   
   def _load_state() -> dict:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           return json.loads(obj["Body"].read())
       except Exception:
           return {"sources": {}}
   
   def _save_state(state: dict):
       s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY,
                     Body=json.dumps(state, separators=(",", ":")).encode("utf-8"),
                     ContentType="application/json")
   
   def _write_page(payload: dict, source: str) -> str:
       ts = time.gmtime()
       key = f"{S3_PREFIX}/{time.strftime('%Y/%m/%d/%H%M%S', ts)}-pingone-aic-{source}.json"
       s3.put_object(Bucket=S3_BUCKET, Key=key,
                     Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
                     ContentType="application/json")
       return key
   
   def _bounded_begin_time(last_ts: str | None, now: float) -> str:
       # beginTime must be <= 24h before endTime (now if endTime omitted)
       # if last_ts older than 24h → cap to now-24h; else use last_ts; else lookback
       twenty_four_h_ago = now - 24*3600
       if last_ts:
           try:
               t_struct = time.strptime(last_ts[:19] + "Z", "%Y-%m-%dT%H:%M:%SZ")
               t_epoch = int(time.mktime(t_struct))
           except Exception:
               t_epoch = int(now - LOOKBACK_SECONDS)
           begin_epoch = max(t_epoch, int(twenty_four_h_ago))
       else:
           begin_epoch = max(int(now - LOOKBACK_SECONDS), int(twenty_four_h_ago))
       return _iso(begin_epoch)
   
   def fetch_source(source: str, last_ts: str | None):
       base = f"https://{FQDN}/monitoring/logs"
       now = time.time()
       params = {
           "source": source,
           "_pageSize": str(PAGE_SIZE),
           "_sortKeys": "timestamp",
           "beginTime": _bounded_begin_time(last_ts, now)
       }
   
       pages = 0
       written = 0
       newest_ts = last_ts
       cookie = None
   
       while pages < MAX_PAGES:
           if cookie:
               params["_pagedResultsCookie"] = cookie
           qs = urllib.parse.urlencode(params, quote_via=urllib.parse.quote)
           data = _http_get(f"{base}?{qs}")
           _write_page(data, source)
   
           results = data.get("result") or data.get("results") or []
           for item in results:
               t = item.get("timestamp") or item.get("payload", {}).get("timestamp")
               if t and (newest_ts is None or t > newest_ts):
                   newest_ts = t
   
           written += len(results)
           cookie = data.get("pagedResultsCookie")
           pages += 1
           if not cookie:
               break
   
       return {"source": source, "pages": pages, "written": written, "newest_ts": newest_ts}
   
   def lambda_handler(event=None, context=None):
       state = _load_state()
       state.setdefault("sources", {})
       summary = []
       for source in SOURCES:
           last_ts = state["sources"].get(source, {}).get("last_ts")
           res = fetch_source(source, last_ts)
           if res.get("newest_ts"):
               state["sources"][source] = {"last_ts": res["newest_ts"]}
           summary.append(res)
       _save_state(state)
       return {"ok": True, "summary": summary}
   
   if __name__ == "__main__":
       print(lambda_handler())
   
   

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	Exemplo
   S3_BUCKET	pingone-aic-logs
   S3_PREFIX	pingone-aic/logs/
   STATE_KEY	pingone-aic/logs/state.json
   AIC_TENANT_FQDN	example.tomcat.pingone.com
   AIC_API_KEY_ID	<api_key_id>
   AIC_API_SECRET	<api_key_secret>
   SOURCES	am-everything,idm-everything
   PAGE_SIZE	500
   MAX_PAGES	20
   LOOKBACK_SECONDS	3600

7. Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > sua-função.

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.

    Observação: o tempo limite é um limite máximo 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

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.
   • Nome: pingone-aic-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 e clique em Adicionar usuários.
2. Informe os seguintes detalhes de configuração:
   • Usuário: insira um nome exclusivo (por exemplo, secops-reader)
   • Tipo de acesso: selecione Chave de acesso - Acesso programático.
   • Clique em Criar usuário.
3. Anexe a política de leitura mínima (personalizada): Usuários > selecione secops-reader > Permissões > Adicionar permissões > Anexar políticas diretamente > Criar política

4. No editor JSON, insira a seguinte política:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::<your-bucket>/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::<your-bucket>"
       }
     ]
   }
   

5. Defina o nome como secops-reader-policy.

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

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

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

Configurar um feed no Google SecOps para ingerir registros do Advanced Identity Cloud do PingOne

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, PingOne Advanced Identity Cloud).
4. Selecione Amazon S3 V2 como o Tipo de origem.
5. Selecione Advanced Identity Cloud do PingOne como o Tipo de registro.
6. Clique em Próxima.
7. Especifique valores para os seguintes parâmetros de entrada:
   • URI do S3: s3://pingone-aic-logs/pingone-aic/logs/
   • 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: 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 a ser 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.