企业官网建设流程全解析

• 人工智能
• AI Agent
• 网络安全
• 后端
• 前端

本文以 AiSOC 的 Prisma Cloud 连接器文档为骨架，讲清一件事：如何用一个连接器实例，把 Palo Alto Networks Prisma Cloud 上 CSPM、CWPP、CIEM、DSPM 四个面的开放告警拉进 AiSOC，并完成鉴权、时间窗轮询、字段归一化、严重度映射、OCSF 归类与排障。读完后你能独立完成控制台侧的配置、理解每一次 poll 到底发了什么请求、看懂落到 AiSOC 里的每条告警字段从哪来，并在could not exchange access keys for prisma cloud JWT、HTTP 403这类报错前快速定位原因。

为什么一个连接器就能覆盖四个产品面

Prisma Cloud 的/alert/v1/alertREST 端点把所有模块的发现聚合在同一个返回体里。因此 AiSOC 不需要为 CSPM、CWPP、CIEM、DSPM 各写一套配置，也不存在"某个子产品忘了开"的问题——租户上没授权的模块，自然就不会有对应类型的告警返回，AiSOC 侧没有任何开关需要关掉。

这条设计在源码里是一句注释就说清的：prisma_cloud.py 的文件头明确写着 "A single connector instance covers every Prisma Cloud surface — CSPM, CWPP, CIEM, DSPM — because the/alert/v1/alertendpoint aggregates findings across all of them."

事件归一化后统一打上source: prisma_cloud，同时把 Prisma Cloud 的原始报文完整保留在raw_event里。这一点很关键：playbook 想按 policy 类型或 finding 子类型做精细化处置时，不需要 AiSOC 为每个字段都建索引，直接读raw_event即可。

连接器在 AiSOC 里的身份：schema、能力与调度

Prisma Cloud 连接器是一个标准的BaseConnector子类，在 prisma_cloud.py 中声明了三个身份属性：

connector_id = "prisma_cloud" connector_name = "Prisma Cloud" connector_category = "cloud"

并在 connectors/init.py 中被显式导入注册。AiSOC 的连接器注册表刻意采用饿汉式导入而不是pkgutil动态发现，理由写在init.py：导入关系在 code review 里可审计，导入错误在服务启动时暴露而不是在第一次请求时暴露，同时对 mypy 和 IDE 跳转友好。

能力声明只有一项：

@classmethod def capabilities(cls) -> tuple[Capability, ...]: return (Capability.PULL_ALERTS,)

PULL_ALERTS属于Capability枚举中的 READ 类被动拉取动词。base.py 里解释了为什么把PULL_*与QUERY_*、ENRICH_*分开命名：PULL_ALERTS表示"数据湖里已经落地的"，QUERY_LOGS表示"现在就去问源端"，ENRICH_USER表示"单实体一次往返"。这个区分让 Agent 在规划动作时能估算成本与延迟。

没有ISOLATE_HOST、BLOCK_IP这类动词，等于在类型层面声明了"只读"——这也是文档中 "Live actions" 一节的实现依据。若某实例被运营者用allowed_capabilities白名单收窄，effective_capabilities()会与类声明取交集，交集为空表示该实例仅供轮询、不供 Agent 使用。

schema 由类方法自描述，直接喂给前端"Add connector"向导，因此向导表单不会和后端漂移：

ConnectorSchema( connector_id="prisma_cloud", connector_name="Prisma Cloud", category="cloud", description="... Pulls open alerts across CSPM, CWPP, CIEM, and DSPM surfaces via the v1 REST API.", docs_url="/docs/connectors/prisma-cloud", fields=[...], )

字段的Field定义在 base.py，to_dict()会剥掉值为None的键，所以线上 JSON 很紧凑，前端不必特判"缺失"与"null"。

前置条件：访问密钥与区域 API 端点

开工前需要三样东西：

• 一个有管理员或 System Admin 权限的 Prisma Cloud 租户。
• 一对 Prisma Cloud访问密钥：
  • accessKeyId—— 访问密钥 ID（UUID 格式）。
  • secretKey—— 与之匹配的密钥。
• 与你租户所在区域对应的 API 端点，例如https://api.prismacloud.io、https://api2.eu.prismacloud.io、https://api.gov.prismacloud.io。这个值在 Prisma Cloud 控制台的System → API Endpoints里查。

访问密钥在控制台的Settings → Access Control → Access Keys → Add New创建。密钥只在创建时展示一次，务必当场复制到密码管理器。

一个常见误解值得单独点名：控制台 URL 不是 API URL。把浏览器地址栏里的域名填进api_url是最典型的一类配置错误。

配置步骤

第一步：创建访问密钥

1. 在 Prisma Cloud 控制台进入Settings → Access Control → Access Keys → Add New。
2. 给密钥起个可追溯的名字，比如aisoc-prod，并选择有效期——与你们的密钥轮换策略对齐。
3. 把Access Key ID与Secret Key存入密码管理器。
4. 确认密钥绑定的角色至少对Alerts资源有读权限。内置的System Admin和Account Group Admin都能用；若要最小权限，自建一个仅含Alerts: View的自定义角色。

第二步：在 AiSOC 中添加连接器

1. Connectors → Add connector → Prisma Cloud。
2. API URL—— 填System → API Endpoints里给出、与你区域匹配的地址（如https://api.prismacloud.io）。
3. Access Key ID与Secret Key—— 粘贴第一步拿到的值。
4. Compute API URL—— 留空。AiSOC 只消费统一的/alert端点，该端点已经把 Compute（Twistlock）的运行时发现收敛进来。
5. Test connection—— AiSOC 会用访问密钥通过POST /login换取一个短期 JWT，并确认鉴权成功。
6. Save。

配置字段参考

下表由连接器的schema()生成，是"Add connector"向导渲染的同一份来源：

compute_url的占位符在源码里给出了格式提示：https://us-east1.cloud.twistlock.com/<tenant-id>；它是一个可选字段，required=False，单元测试 test_prisma_cloud.py 明确断言了这一点，注释写着 "compute_url must be optional; v7.1.0 only consumes the unified /alert endpoint"。

:::note 凭据处理secret_key在写入数据库行之前就被凭据保险库静态加密，只在轮询时解密；该值不会被日志记录，也不会被 API 返回。详见 凭据保险库。 :::

凭据加密与"测试连接"的真实链路

secret_key之所以标为secret类型，不只是给前端渲染一个掩码输入框。AiSOC 在应用层用 Fernet（AES-128-CBC + HMAC-SHA256）加密连接器凭据，密文存放在connector_instances.auth_config里，主密钥来自AISOC_CREDENTIAL_KEY环境变量。设计意图写在 credentials.md：数据库自身加密只防磁盘失窃，不防只读角色泄露、备份泄露或副本配置错误；应用层加密则让一份pg_dump在没有主密钥时毫无价值。

这意味着两个服务必须共享同一个AISOC_CREDENTIAL_KEY：API 服务负责写密文，连接器服务在取数时解密。密钥不一致时轮询会以InvalidToken失败。

向导里的"Test connection"按钮并不在连接器服务内部完成全部工作，它走的是连接器服务的一个无状态路由 router.py：

1. API 服务查出实例行（或直接使用向导里尚未落库的新凭据），解密auth_config。
2. 把明文凭据与connector_config一起 POST 到连接器服务。
3. 连接器服务把两个字典合并成构造参数（{**payload.auth_config, **payload.connector_config}，后者不能覆盖前者，属于纵深防御），在构造对象之前、在任何 socket 打开之前调用refuse_internal_destinations(kwargs)。
4. 构造PrismaCloudConnector(**kwargs)并await connector.test_connection()。

第 3 步值得展开：它会遍历所有 URL 形态的参数，交给 ssrf_guard 校验。任何指向云元数据、回环地址或部署内网服务的api_url都会被直接拒绝并返回 422，而不是先请求再把响应反射回来。注意其细节——DNS 解析失败（socket.gaierror）被视为"解析不了"而非"策略拒绝"，因此不会误判。

轮询机制：每次 poll 重新换 JWT

时间与频率的约定

• 默认间隔：300 秒。
• 每次 poll 都会调用POST /login换取一个新的 JWT。Prisma Cloud 的令牌是滑动 TTL，重新鉴权成本很低；每次 poll 都换，可以彻底避开过期窗口附近的缓存失效问题。
• 取数窗口是"最近interval秒"，使用timeRange.type = "absolute"配合 epoch 毫秒的startTime/endTime。
• 只拉status = "open"的告警；resolved / snoozed / dismissed 在源端就被过滤掉。
• 结果集上限200 条/次。如果租户在 5 分钟内稳定产出超过这个量，应调小轮询间隔而不是调大上限——这样每次请求都稳稳落在 Prisma Cloud 的 API 超时之内。

这三条在源码里都能对上。默认值与上下界定义在 scheduler.py：

_DEFAULT_POLL_INTERVAL_S = 300 _MIN_POLL_INTERVAL_S = 30 _MAX_POLL_INTERVAL_S = 86400 # 24 hours def _coerce_poll_interval(connector_config: dict[str, Any]) -> int: raw = connector_config.get("poll_interval_seconds", _DEFAULT_POLL_INTERVAL_S) try: seconds = int(raw) except (TypeError, ValueError): return _DEFAULT_POLL_INTERVAL_S if seconds < _MIN_POLL_INTERVAL_S: return _MIN_POLL_INTERVAL_S if seconds > _MAX_POLL_INTERVAL_S: return _MAX_POLL_INTERVAL_S return seconds

于是poll_interval_seconds被钳制在30 秒到 86400 秒之间，非法值（比如"not-a-number"）回落到 300。这些边界在 test_scheduler.py 里有对应测试。

关键的一点：调度器把轮询间隔同时当作取数窗口。它在 scheduler.py 里写下：

since_seconds = _coerce_poll_interval(target.connector_config) raw_events = await connector.fetch_alerts(since_seconds=since_seconds)

所以 300 秒一次 poll，取的就是最近 300 秒；把间隔改成 60 秒，窗口也同步变成 60 秒。此外还有一条"故障恢复回填"路径：当连接器中断至少 30 分钟后恢复，会触发一次带扩展窗口的一次性 poll，窗口上限 24 小时（_MAX_BACKFILL_LOOKBACK_S），并叠加 5 分钟缓冲（_BACKFILL_LOOKBACK_BUFFER_S）覆盖"最后一次成功 poll"与last_outage_at之间的空隙。上限的存在是为了避免一次三天中断变成一次三天跨度的查询、被源端限流或直接拒绝。

一次 poll 的完整调用链

fetch_alerts()在 prisma_cloud.py 中的行为可以逐步还原：

1. 调_login()换 JWT；失败则返回空列表而不是抛异常，让轮询循环优雅降级，由调度器在下一个周期重试。
2. 计算end = now(UTC)、start = end - timedelta(seconds=since_seconds)。
3. 组装如下请求体（结构与源码逐字段一致）：

{ "timeRange": { "type": "absolute", "value": { "startTime": 1735689300000, "endTime": 1735689600000 } }, "filters": [ { "name": "alert.status", "operator": "=", "value": "open" } ], "sortBy": ["alertTime:desc"], "limit": 200 }

4. POST {api_url}/alert/v1/alert，请求头带x-redlock-auth: <token>（这就是 Prisma Cloud 的鉴权头名，见_headers()）。
5. 超时时间为30 秒（登录请求为15 秒）。非 200 时记录prisma_cloud.fetch_failed并返回空列表。
6. 解析返回体后逐条normalize()。

第 6 步有一个容易被忽略的兼容处理：/alert/v1/alert在不同区域的租户上，返回体可能是裸数组，也可能是{"items": [...]}信封，源码还额外接受alerts键：

if isinstance(payload, list): items = payload elif isinstance(payload, dict): items = payload.get("items") or payload.get("alerts") or [] else: items = []

对应测试test_fetch_alerts_handles_bare_list_envelope就是在模拟区域差异，注释直言 "Some Prisma Cloud regions return a bare list instead of the{"items": [...]}envelope; the connector must accept both."

字段归一化：落到 AiSOC 里的每条告警长什么样

normalize()把 Prisma Cloud 的告警报文压成 AiSOC 的规范信封。以一条"公开可读的 S3 桶"为例，输入是：

{ "id": "alert-1", "alertTime": 1735689600000, "policy": { "name": "Public S3 bucket", "description": "S3 bucket is publicly readable", "severity": "critical" }, "resource": { "name": "logs-bucket", "rrn": "rrn::s3:us-east-1:123:bucket/logs-bucket", "cloudType": "aws", "regionId": "us-east-1" } }

输出的规范字段与取值来源如下：

时间戳的转换是"防御式"的：alertTime通常是 epoch 毫秒，AiSOC 下游期望 ISO-8601 字符串，因此代码判断类型后再转换；但部分历史报文直接给 ISO 字符串，此时原样透传：

created_at = raw.get("alertTime") or raw.get("firstSeen") if isinstance(created_at, int | float): created_at = datetime.fromtimestamp(created_at / 1000, tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ")

test_normalize_handles_iso_timestamp_passthrough覆盖了后者，test_normalize_critical_preserves_critical覆盖了前者并断言created_at == "2025-01-01T00:00:00Z"。

严重度映射：文档表与当前源码的一处偏差

文档给出的映射是一张 4 档阶梯到 AiSOC 标准档位的对照表：

文档解释的理由是"AiSOC 不暴露独立的 critical 档位"，因此把critical收敛进high，并把原始档位保留在raw_event.policy.severity下。

但当前仓库源码已经不是这个行为。 prisma_cloud.py 里的映射表是：

_SEVERITY_MAP: dict[str, str] = { "critical": "critical", "high": "high", "medium": "medium", "low": "low", "informational": "info", "info": "info", }

2025-01-01 的那条单测直接断言out["severity"] == "critical"，注释写着 "Prisma Cloud 'critical' must map directly to AiSOC 'critical' so P1 SLAs apply"，措辞是 "never silently downgrade tohigh, since AiSOC now exposes a dedicated 5th tier for P1 work"。

交叉验证也支持"5 档"这一现状：告警升级接口 alerts.py 使用的阶梯是["info", "low", "medium", "high", "critical"]；融合服务的推广判定should_promote()把severity_id >= 4视为可达；OCSF 规范侧_canonicalSeverityMap也是critical: 5, high: 4, medium: 3, low: 2, info: 1。

因此实操建议是：以源码为准，critical会保留为critical；文档中那张表应视为旧版（4 档时代）描述。这不影响raw_event.policy.severity的保留承诺——原始档位始终在raw_event里可取。同理，无法识别的严重度一律回落到medium（test_normalize_falls_back_to_medium_for_unknown_severity断言了"never silently drop"），而 "Live actions" 一节标注的 v7.3.1 与compute_url帮助文本里的 v7.1.0，也都是版本演进留下的历史标注；当前仓库 VERSION 为 20.1.0，capabilities()仍只声明PULL_ALERTS，即只读定位没有改变。

覆盖范围：四个模块各自带来什么

统一的/alert端点包含你租户已授权的所有模块的发现：

租户未授权的模块不会产生对应类型的告警，AiSOC 侧无需关闭任何东西。

进入数据湖之后：OCSF 归类与"必然提升"

归一化只是第一站。事件进入 ingest 后要做 OCSF 归类，而归类结果直接决定这条事件是否变成分析师队列里的告警。

ocsf_classes.go 的头部注释讲清了机制：should_promote()在"OCSF category 2（Findings）"或"severity_id >= 4"时提升为告警；category 等于 class uid 除以 1000。也就是说，连接器事件携带的 class 决定了它是进队列还是只留在湖里。

Prisma Cloud 在这张表里被显式归入故意的 2001 Security Finding：

"prisma_cloud": {genericReason: "cloud security alerts: already judged by the vendor"},

同一组里还有 CrowdStrike、SentinelOne、Wiz、Orca、Lacework、AWS GuardDuty、GCP SCC 等。genericReason字段不是"没人看过"的逃生门，而是要求写明理由的字段——这里的理由是：这些连接器返回的是厂商已经判定为发现的结果，所以 category 2 与无条件提升是正确结果，而不是疏漏。

这一点对排障有直接价值：如果你发现 Prisma Cloud 的告警"看起来什么都进得来"，那是设计如此；如果你担心低危噪声淹没队列，正确的手段不是改 class，而是下一节的过滤规则或下游的评分。

去重、重放与游标

PrismaCloudConnector没有声明checkpoint_time_field/checkpoint_id_field，按 base.py 的checkpoints()语义，它的返回值是False——即重启后会重读其重叠窗口，这是安全的方向，代价是长时间中断会丢失窗口之外的事件。

重叠窗口带来的重复投递，由 ingest 侧的确定性事件 ID 收敛。normalizer.go 的generateEventID用(connector_id, tenant_id)加上一个稳定的厂商标识（按external_id→event_id→id→_cd优先级）派生 UUIDv5：

base := fmt.Sprintf("%s:%s", raw.ConnectorID, raw.TenantID) for _, field := range []string{"external_id", "event_id", "id", "_cd"} { if v, ok := raw.Payload[field].(string); ok && v != "" { return uuid.NewSHA1(uuid.NameSpaceOID, []byte(base+":"+field+"="+v)).String() } }

注释特别指出ReceivedAt被刻意排除在这条路径之外——它每次 poll 都会变，以前正是它让 ID 在每次轮询时改变。若确实没有稳定标识，才回退到"事件时间 + payload 内容哈希"，保证字节级重放仍然收敛，而同一时间戳的两条不同内容事件保持可分。

对 Prisma Cloud 而言，normalize()产出的external_id来自raw.id或raw.alertId，因此重叠窗口与故障回填都不会造成重复告警。反过来说，如果某个租户返回的告警缺少id/alertId，去重就会退化为内容哈希，此时需要关注上游字段完整性。

用过滤规则在边缘降噪

调度器在每个 poll 的归一化之后、推送 ingest 之前，会执行连接器配置里的filter_rules。实现见 filter_rules.py，规则形态刻意保持极小：

filter_rules: - field: severity op: eq value: info action: drop - field: rule_name op: contains value: "test alert" action: drop

支持的操作符与federated.query共用一套词汇，避免运营者学两种方言：eq、ne、contains、starts_with、ends_with、in；动作只有drop与keep。语义与 iptables 一致——自上而下、首个匹配生效、无匹配默认保留。缺失字段被视为不匹配而非报错，一个引用了厂商已改名字段的规则不会把轮询循环打挂；未知操作符一律"拒绝丢弃"，宁可多留也不静默丢事件。

对 Prisma Cloud 场景，典型用法是压掉低价值的合规提示（比如某个高频的informational策略），把热门存储留给真正的姿态风险。注意filter_rules属于调度器专用键，router.py 与调度器都会在构造连接器前把它剔除，不会误传给__init__。

只读定位与处置路径

Prisma Cloud 连接器目前是只读的。云侧发现的遏制动作通过其他连接器完成：

• AWS Security Hub —— 仅 AWS 部署下的原生发现聚合。
• AWS GuardDuty —— AWS 威胁发现。
• Cloudflare —— 边缘侧处置。
• GCP Cloud Audit —— GCP 控制面审计。

这个"读一个源、动另一个源"的分工在架构上是刻意的：PrismaCloudConnector的capabilities()只返回PULL_ALERTS，Agent 层拿不到任何动能动词，因此不存在"以为能隔离、实际调不到执行器"的落差。

排障手册

could not exchange access keys for prisma cloud JWT——accessKeyId/secretKey不匹配、访问密钥已过期，或api_url指向的区域与租户不符。这段错误文案就是test_connection()在_login()返回None时给出的固定字符串。直接验证：

curl -X POST "https://api.prismacloud.io/login" \ -H "Content-Type: application/json" \ -d '{"username":"YOUR_ACCESS_KEY_ID","password":"YOUR_SECRET"}'

返回 200 且带{ "token": "..." }，说明凭据可用，问题在 AiSOC 侧字段（最常见的还是区域 URL 填错）。若返回 200 但没有token字段，源码也会判为失败——test_test_connection_handles_login_returning_no_token专门模拟了"密钥被禁用时 Prisma Cloud 返回 200 却无 token"的实测行为。

/alert/v1/alert返回HTTP 403—— 密钥绑定的角色缺少Alerts: View。检查Settings → Access Control → Access Keys → <你的密钥> → Role，确认权限存在。

两次 poll 之间events_added: 0—— 租户在过去 5 分钟内没有产生新的开放告警。resolved / snoozed 已在源端被alert.status = "open"过滤掉，安静租户上这是预期的稳态，不是故障。events_added由调度器在 ingest 接受后写入，见 scheduler.py 与 connector_repo.py：它统计的是被接受的事件数，因此过滤规则丢弃的事件不会出现在这里。

EU / Gov 区域租户取不到数据—— 确认 API URL 与区域匹配：EU 租户用api2.eu.prismacloud.io，Gov 用api.gov.prismacloud.io。再次强调，控制台 URL 不是 API URL。

连接器显示不健康但日志没有明显报错—— 依次怀疑三处：凭据解密失败（connector.scheduler.decrypt_failed，主密钥不一致）、构造参数不匹配（connector.scheduler.bad_config，通常是TypeError）、目标地址被 SSRF 策略拒绝（connector.scheduler.destination_refused）。第三种在企业内网场景需要留意：若api_url落在私有网段，需要显式设置AISOC_SSRF_ALLOW_PRIVATE=1。

保存后轮询没有立即生效—— 调度器以 60 秒为周期重读连接器表（_DEFAULT_RELOAD_INTERVAL_S），并通过配置签名比对决定是否重排任务。新实例最多等一个重载周期；启动时会立即触发一次重载，所以不会等满 60 秒才开始。

本地验证连接器行为

仓库提供了完整单测，用respx拦截 HTTP，不会真的打到 Prisma Cloud：

cd services/connectors python -m pytest tests/test_prisma_cloud.py -v

覆盖的关键断言包括：schema 必含四个字段且secret_key.type == "secret"；compute_url可选；critical保留为critical；informational映射为info；ISO 时间戳透传；未知严重度回落medium；items信封与裸数组两种返回体都能解析；登录失败或告警端点 5xx 时返回空列表而不抛异常。

如果想手工验证一条真实告警的归一化结果，可以直接实例化连接器并调用normalize()：

from app.connectors.prisma_cloud import PrismaCloudConnector c = PrismaCloudConnector("https://api.prismacloud.io", "KEY_ID", "SECRET") out = c.normalize({ "id": "alert-1", "alertTime": 1735689600000, "policy": {"name": "Public S3 bucket", "severity": "critical"}, "resource": {"name": "logs-bucket", "rrn": "rrn::s3:us-east-1:123:bucket/logs-bucket", "cloudType": "aws", "regionId": "us-east-1"}, }) print(out["severity"], out["created_at"], out["cloud_resource"])

这段代码只依赖连接器模块本身，不需要数据库或凭据保险库，适合在接入前确认字段是否符合你的 playbook 预期。

相关连接器

• Wiz —— 另一条 CNAPP 路径，云图上下文更丰富。
• Lacework —— 侧重工作负载与合规的 CNAPP 方案。
• AWS Security Hub —— 纯 AWS 部署下的原生发现聚合。

想了解连接器整体模型（自描述 schema、凭据加密、轮询可观测性），可回到 连接器总览；想了解凭据如何生成、轮换与在 K8s / Fly.io / ECS 上放置主密钥，见 凭据保险库与密钥管理。

• 人工智能
• AI Agent
• 网络安全
• 后端
• 前端