Questa pagina è stata tradotta dall'API Cloud Translation.

Raccogliere i log di PingOne Advanced Identity Cloud

Supportato in:

Google SecOps SIEM

Questo documento spiega come importare i log di PingOne Advanced Identity Cloud in Google Security Operations utilizzando Amazon S3.

Prima di iniziare

Istanza Google SecOps
Accesso con privilegi al tenant PingOne Advanced Identity Cloud
Accesso privilegiato ad AWS (S3, IAM, Lambda, EventBridge)

Recuperare la chiave API PingOne e l'FQDN del tenant

Accedi alla console di amministrazione di Advanced Identity Cloud.
Fai clic sull'icona dell'utente e poi su Impostazioni tenant.
Nella scheda Impostazioni globali, fai clic su Registra chiavi API.
Fai clic su Nuova chiave API log e fornisci un nome per la chiave.
Fai clic su Crea chiave.
Copia e salva i valori di api_key_id e api_key_secret in un luogo sicuro. Il valore di api_key_secret non viene più visualizzato.
Fai clic su Fine.
Vai a Impostazioni tenant > Dettagli e trova il FQDN del tenant (ad esempio, example.tomcat.pingone.com).

Configura il bucket AWS S3 e IAM per Google SecOps

Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket
Salva il nome e la regione del bucket per riferimento futuro (ad esempio, pingone-aic-logs).
Crea un utente seguendo questa guida: Creazione di un utente IAM.
Seleziona l'utente creato.
Seleziona la scheda Credenziali di sicurezza.
Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
Seleziona Servizio di terze parti come Caso d'uso.
Fai clic su Avanti.
(Facoltativo) Aggiungi un tag di descrizione.
Fai clic su Crea chiave di accesso.
Fai clic su Scarica file CSV per salvare la chiave di accesso e la chiave di accesso segreta per un utilizzo successivo.
Fai clic su Fine.
Seleziona la scheda Autorizzazioni.
Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
Seleziona Aggiungi autorizzazioni.
Seleziona Allega direttamente i criteri.
Cerca e seleziona il criterio AmazonS3FullAccess.
Fai clic su Avanti.
Fai clic su Aggiungi autorizzazioni.

Configura il ruolo e il criterio IAM per i caricamenti S3

Nella console AWS, vai a IAM > Policy > Crea policy > Scheda JSON.

Inserisci la seguente policy.

{
  "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"
    }
  ]
}

Sostituisci pingone-aic-logs se hai inserito un nome bucket diverso.

Fai clic su Avanti > Crea policy.
Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.
Allega il criterio appena creato.
Assegna al ruolo il nome WritePingOneAICToS3Role e fai clic su Crea ruolo.

Crea la funzione Lambda

Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
Fai clic su Crea autore da zero.
Fornisci i seguenti dettagli di configurazione:

Impostazione Valore

Nome pingone_aic_to_s3

Tempo di esecuzione Python 3.13

Architettura x86_64

Ruolo di esecuzione WritePingOneAICToS3Role

Impostazione	Valore
Nome	`pingone_aic_to_s3`
Tempo di esecuzione	Python 3.13
Architettura	x86_64
Ruolo di esecuzione	`WritePingOneAICToS3Role`

Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e inserisci il seguente codice (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())

Vai a Configurazione > Variabili di ambiente > Modifica > Aggiungi nuova variabile di ambiente.

Inserisci le seguenti variabili di ambiente, sostituendole con i tuoi valori:

Chiave	Esempio
`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`

Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).
Seleziona la scheda Configurazione.
Nel riquadro Configurazione generale, fai clic su Modifica.
Modifica Timeout impostandolo su 5 minuti (300 secondi) e fai clic su Salva.

Nota: il timeout è un limite superiore rigido, il che significa che se la funzione è ancora in esecuzione quando il timer raggiunge i 5 minuti, AWS Lambda la terminerà. Qualsiasi esecuzione più breve termina normalmente e ti viene addebitato solo il tempo di esecuzione effettivo.

Creare una pianificazione EventBridge

Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
Fornisci i seguenti dettagli di configurazione:
- Programma ricorrente: Tariffa (1 hour).
- Destinazione: la tua funzione Lambda.
- Nome: pingone-aic-1h
Fai clic su Crea pianificazione.

(Facoltativo) Crea chiavi e utente IAM di sola lettura per Google SecOps

Nella console AWS, vai a IAM > Utenti, poi fai clic su Aggiungi utenti.
Fornisci i seguenti dettagli di configurazione:
- Utente: inserisci un nome univoco (ad esempio secops-reader)
- Tipo di accesso: seleziona Chiave di accesso - Accesso programmatico
- Fai clic su Crea utente.
Allega criterio per la lettura minimi (personalizzati): Utenti > seleziona secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Allega criteri direttamente > Crea criteri

Nell'editor JSON, inserisci la seguente policy:

{
  "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>"
    }
  ]
}

Imposta il nome su secops-reader-policy.
Vai a Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.
Vai a Credenziali di sicurezza > Chiavi di accesso > Crea chiave di accesso.
Scarica il file CSV (questi valori vengono inseriti nel feed).

Configura un feed in Google SecOps per importare i log di PingOne Advanced Identity Cloud

Vai a Impostazioni SIEM > Feed.
Fai clic su Aggiungi nuovo feed.
Nel campo Nome feed, inserisci un nome per il feed (ad esempio, PingOne Advanced Identity Cloud).
Seleziona Amazon S3 V2 come Tipo di origine.
Seleziona PingOne Advanced Identity Cloud come Tipo di log.
Fai clic su Avanti.
Specifica i valori per i seguenti parametri di input:
- URI S3: s3://pingone-aic-logs/pingone-aic/logs/
- Opzioni di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.
- Durata massima del file: 180 giorni per impostazione predefinita.
- ID chiave di accesso: chiave di accesso utente con accesso al bucket S3.
- Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3.
- Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.
- Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.
Fai clic su Avanti.
Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.