此页面由 Cloud Translation API 翻译。

收集 PingOne Advanced Identity Cloud 日志

支持的语言：

Google SecOps SIEM

本文档介绍了如何使用 Amazon S3 将 PingOne Advanced Identity Cloud 日志注入到 Google Security Operations。

准备工作

Google SecOps 实例
对 PingOne Advanced Identity Cloud 租户的特权访问权限
对 AWS（S3、IAM、Lambda、EventBridge）的特权访问权限

获取 PingOne API 密钥和租户 FQDN

登录 Advanced Identity Cloud 管理控制台。
点击用户图标，然后点击 Tenant Settings（租户设置）。
在全局设置标签页上，点击 Log API Keys。
点击 New Log API Key，然后为该密钥提供一个名称。
点击创建密钥。
复制 api_key_id 和 api_key_secret 值，并将其保存在安全的位置。系统不会再次显示 api_key_secret 值。
点击完成
前往租户设置 > 详细信息，然后找到您的租户 FQDN（例如 example.tomcat.pingone.com）。

为 Google SecOps 配置 AWS S3 存储桶和 IAM

按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
保存存储桶名称和区域以供日后参考（例如 pingone-aic-logs）。
按照以下用户指南创建用户：创建 IAM 用户。
选择创建的用户。
选择安全凭据标签页。
在访问密钥部分中，点击创建访问密钥。
选择第三方服务作为使用情形。
点击下一步。
可选：添加说明标记。
点击创建访问密钥。
点击 Download CSV file（下载 CSV 文件），保存访问密钥和不公开的访问密钥以供日后使用。
点击完成。
选择权限标签页。
在权限政策部分中，点击添加权限。
选择添加权限。
选择直接附加政策
搜索并选择 AmazonS3FullAccess 政策。
点击下一步。
点击添加权限。

为 S3 上传配置 IAM 政策和角色

在 AWS 控制台中，依次前往 IAM > 政策 > 创建政策 > JSON 标签页。

输入以下政策。

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

如果您输入了其他存储桶名称，请替换 pingone-aic-logs。

依次点击下一步 > 创建政策。
依次前往 IAM > 角色 > 创建角色 > AWS 服务 > Lambda。
附加新创建的政策。
将角色命名为 WritePingOneAICToS3Role，然后点击创建角色。

创建 Lambda 函数

在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
点击从头开始创作。
提供以下配置详细信息：

设置值

名称 pingone_aic_to_s3

运行时 Python 3.13

架构 x86_64

执行角色 WritePingOneAICToS3Role

设置	值
名称	`pingone_aic_to_s3`
运行时	Python 3.13
架构	x86_64
执行角色	`WritePingOneAICToS3Role`

创建函数后，打开 Code 标签页，删除桩代码并输入以下代码 (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())

依次前往配置 > 环境变量 > 修改 > 添加新的环境变量。

输入以下环境变量，并替换为您的值：

键	示例
`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`

创建函数后，请停留在其页面上（或依次打开 Lambda > 函数 > your-function）。
选择配置标签页。
在常规配置面板中，点击修改。
将超时更改为 5 分钟（300 秒），然后点击保存。

注意：超时时间是一个硬性上限，这意味着如果函数在计时器达到 5 分钟时仍在运行，AWS Lambda 将终止该函数。任何较短的执行时间都会正常结束；您只需为实际运行时间付费。

创建 EventBridge 计划

依次前往 Amazon EventBridge > 调度程序 > 创建计划。
提供以下配置详细信息：
- 周期性安排：费率 (1 hour)。
- 目标：您的 Lambda 函数。
- 名称：pingone-aic-1h。
点击创建时间表。

可选：为 Google SecOps 创建只读 IAM 用户和密钥

在 AWS 控制台中，依次前往 IAM > Users，然后点击 Add users。
提供以下配置详细信息：
- 用户：输入唯一名称（例如 secops-reader）
- 访问类型：选择访问密钥 - 以程序化方式访问
- 点击创建用户。
附加最低限度的读取政策（自定义）：用户 > 选择 secops-reader > 权限 > 添加权限 > 直接附加政策 > 创建政策

在 JSON 编辑器中，输入以下政策：

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

将名称设置为 secops-reader-policy。
依次前往创建政策 > 搜索/选择 > 下一步 > 添加权限。
依次前往安全凭据 > 访问密钥 > 创建访问密钥。
下载 CSV（这些值将输入到 Feed 中）。

在 Google SecOps 中配置 Feed 以注入 PingOne Advanced Identity Cloud 日志

依次前往 SIEM 设置> Feed。
点击添加新 Feed。
在Feed 名称字段中，输入 Feed 的名称（例如 PingOne Advanced Identity Cloud）。
选择 Amazon S3 V2 作为来源类型。
选择 PingOne Advanced Identity Cloud 作为日志类型。
点击下一步。
为以下输入参数指定值：
- S3 URI：s3://pingone-aic-logs/pingone-aic/logs/
- 来源删除选项：根据您的偏好设置选择删除选项。
- 文件最长保留时间：默认值为 180 天。
- 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
- 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥。
- 资产命名空间：资产命名空间。
- 注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。