人工智能AI Agent网络安全后端前端【免费下载链接】AiSOCOpen-source AI Security Operations Center: alert fusion, LLM-agent triage, MITRE ATTCK investigation, and a replayable decision ledger for every agent step. Self-hostable, runs with no API keys, MIT licensed. Ships an MCP server for Claude, Cursor and Continue.项目地址https://gitcode.com/GitHub_Trending/ai/AiSOC点击查看免费下载GCP Security Command CenterSCC是 Google Cloud 上唯一的组织级安全信号聚合面把漏洞、错误配置、运行时威胁与合规姿态统一成 findings。本文以 AiSOC 官方连接器文档为骨架结合services/connectors/app/connectors/gcp_scc.py的真实实现逐层拆解这个连接器的权限模型、配置字段、轮询窗口、JWT 认证链路、严重度与分类归一化以及排障时应该看哪个日志字段。读完你可以独立完成组织级 SCC 接入并能解释“为什么我拿到了 0 条 finding”“为什么 critical 没有降级成 high”这类一线问题。连接器的定位把「已经判过的结论」搬进 AiSOCSCC connector 拉取的并不是原始遥测而是findings——Google 从 Cloud Audit Logs、VPC Flow Logs、Container Threat Detection、Web Security Scanner 以及合作方集成中综合出的、已经带有结论的安全信号。连接器源码模块注释把这一点讲得很直白SCC 相当于 GCP 侧的 AWS Security Hub 或 Microsoft Defender for Cloud是 GCP 资产中信号密度最高的一路数据源。在 AiSOC 的能力模型里这个连接器只声明了一个能力能力声明位置含义PULL_ALERTScapabilities()gcp_scc.py被动轮询已有告警流不包含QUERY_LOGS、ENRICH_*或任何阻断类动作这意味着 SCC 连接器是只读的。它不会替你关掉一个公开的 GCS bucket也不会隔离实例——这类动作需要由 playbook 走 Actions 服务且需要目标连接器显式声明对应能力。能力的粒度设计在 base.py 中有说明PULL_ALERTS表示“已经落到数据湖里的东西”与QUERY_LOGS现在就去源端查严格区分便于 agent 在规划时估算成本与延迟。能拿到什么原文档列出的四类 finding 与 SCC 的发现来源一一对应VulnerabilityCVE、泄露的凭据、公开可读的 bucketMisconfiguration来自 Security Health Analytics 的错误配置Threat来自 Event Threat Detection 的异常 IAM 行为、暴力破解、挖矿Posture来自 Posture Management 的合规姿态违规进入 AiSOC 之后被归到哪一类这一点原文档只给了一句category: cloud_posture/category: threat但落地细节在 ingest 服务里值得展开。在 OCSF 归类表 ocsf_classes.go 中gcp_scc被显式映射为2001 Security Finding并附了一条理由说明gcp_scc: {genericReason: Security Command Center findings: already judged by the vendor},2001 属于 OCSFcategory 2Findings类目推导见 ocsf_classes.gocategory_uid class_uid / 1000。这条归属决定的实战含义是SCC 事件会无条件进入 Findings 类目并参与提升promotion而不是像原始日志那样先被降级过滤再等待匹配。原因写在注释里——vendor 已经判过了AiSOC 不该再判一次“值不值得看”。因此一次典型的处理链路是SCC findings.list (org 级) │ fetch_alerts(since_seconds) ▼ GCPSCCConnector.normalize() # severity / event_type / resource_* 落字段 │ IngestClient.post(batch, tenant header) ▼ services/ingest OCSF 归一化 # gcp_scc → class_uid 2001, category_uid 2 ▼ fusion / detection / agent triage前置条件SCC 连接器对权限的要求比 Cloud Audit Logs 严格得多因为它读的是组织级资源。下列四项缺一不可Security Command Center 必须在组织级启用。项目级 SCC 不会暴露这里用到的 API。一个组织 ID必须是数字形式用gcloud organizations list查。一个服务账号在组织作用域上被授予Security Center Findings Viewerroles/securitycenter.findingsViewer。该服务账号的JSON key。源码 schema 的字段帮助文本也重复了这条约束gcp_scc.pyNumeric GCP organization ID. SCC is org-scoped; the service account must have findingsViewer at this level (or on a parent folder/source).注意括号里的补充除了组织级父 folder 或单个 source上的授权也能生效。这在你需要把接入范围收窄到某个 folder 时很有用但代价是组织内其他 folder 的 finding 将不可见。配置实战从零到「测试连接通过」1. 确认 SCC 已启用gcloud scc settings services list --organizationORG_ID输出中应能看到SECURITY_HEALTH_ANALYTICS、EVENT_THREAT_DETECTION等服务处于启用状态。如果为空或未启用需要在组织级开通 SCC Standard 或 Premium。2. 创建服务账号GCP Console → IAM Admin → Service Accounts选择任意项目建议单独建一个security-tools项目专门存放只读安全凭据。Create service account命名aisoc-scc-reader。3. 在组织级授予角色Console → IAM Admin → IAM切换到组织视图使用组织切换器。Grant access → principal 填新建服务账号的邮箱 → 角色选Security Center Findings Viewer。这一步必须在组织级完成不能在项目级。SCC findings 是组织作用域资源项目级授权只会得到一个看似配置正确、实际永远 403 的连接器。4. 生成 JSON key 并填入 AiSOC服务账号 → Keys → Add key → Create new key →JSON浏览器会下载密钥文件。AiSOC → Connectors → Add connector →GCP Security Command Center。organization_id填数字组织 ID例如123456789012而不是example.com。service_account_json粘贴整份JSON 文件内容。Test connection。Save。关于第 5 步这里有一个值得说明的实现细节原文档描述为“调用findings.listpageSize1”而当前源码实际调用的是sources.listresp await client.get( f{_SCC_BASE}/organizations/{self._organization_id}/sources, headersself._headers(), params{pageSize: 1}, )见 gcp_scc.py紧邻的注释解释了选择理由列 source 是最廉价的校验方式一次调用同时验证了三件事——token 有效、组织 ID 正确、SCC 已在组织上启用。如果你在按文档与代码比对行为以sources端点为准两者在“单次最小开销探活”这个意图上是一致的。测试成功时返回结构包含success、connector、organization_id以及服务账号邮箱{ success: true, connector: gcp_scc, organization_id: 123456789012, service_account: aisoc-scc-readersecurity-tools.iam.gserviceaccount.com }失败则返回{success: false, connector: gcp_scc, error: HTTP 403: ...}error会截断到 300 字符以内避免把整段 HTML 错误页塞进连接器健康状态字段。配置字段参考下表由scripts/generate_connector_docs.py从连接器schema()生成并补入了源码中的类型与提示信息字段标签类型必填默认值说明organization_idOrganization IDstring是—数字形式的 GCP 组织 ID。SCC 是组织级资源服务账号必须在该层级或父 folder / source 上拥有 findingsViewer。schema 中的 placeholder 为1234567890。service_account_jsonService account JSON keysecret是—静态加密存储· 粘贴完整 JSON 密钥文件写入前由凭据保险库加密。两个字段都声明为必填required: [organization_id, service_account_json]见 plugin.yaml因此 API 侧校验与前端表单一一致不会出现“只在 UI 上必填”的空档。:::note 凭据处理service_account_json在写入数据库行之前就由凭据保险库加密仅在轮询时解密。该值不会被记录到日志也不会由 API 返回。详见 凭据保险库与密钥管理。 :::加密在链路上是怎么发生的保险库的密钥来自部署级环境变量AISOC_CREDENTIAL_KEY实现位于 credential_vault.py。调度器在每次 tick 时对它做解密把明文只留在内存中一次fetch_alerts()的生命周期里。这意味着两件运维上很重要的事一份pg_dump泄露而不泄露AISOC_CREDENTIAL_KEY等于拿不到服务账号私钥API 与 connectors 两个服务必须共享同一个AISOC_CREDENTIAL_KEY否则解密会因完整性校验失败而报错源码中的错误提示明确指向“key mismatch between API and connectors services”。还有一个容易被忽略的 schema 细节OAuthHints在这个连接器上被显式声明为supported_in_hostedFalse只填了token_url与scopesoauthOAuthHints( supported_in_hostedFalse, authorize_urlNone, token_url_TOKEN_URL, scopes[_SCOPE], )也就是说即使在托管部署里你也必须自带服务账号 JSON不存在“点一下 Connect 就授权”的路径。这是设计使然SCC 需要组织级 IAM 授权SaaS 厂商代为申请的 OAuth 应用无法覆盖客户的 org IAM 边界。轮询与增量窗口文档的 600 秒与代码的 300 秒原文档写默认间隔600 秒10 分钟理由是 SCC findings 的更新频率低于审计日志且 API 配额较紧。这个 600 秒在实现里不是硬编码——它通过连接器配置里的poll_interval_seconds表达interval _coerce_poll_interval(target.connector_config)而_coerce_poll_interval()的行为在 scheduler.py 中定义得非常明确参数值说明_DEFAULT_POLL_INTERVAL_S300未配置poll_interval_seconds时的默认值_MIN_POLL_INTERVAL_S30低于此值被抬到 30 秒_MAX_POLL_INTERVAL_S86400高于此值被压到 24 小时所以要把 SCC 调到文档推荐的 10 分钟需要在连接器配置中显式写入poll_interval_seconds: 600。若不写实际生效的是调度器的 300 秒默认值——这是一个容易被误判为“文档写错了”的点实际是“文档给的是推荐值默认值是另一层”。顺便说把间隔设得很小对 SCC 基本没有收益findings 的eventTime更新粒度远大于分钟级30 秒轮询只会消耗配额并把大量重复窗口重新拉一遍。since窗口如何与轮询间隔对齐调度器把轮询间隔直接当作回看窗口传给连接器# We deliberately default to the same poll interval, so a 5-min # poll fetches the last 5 min of events. since_seconds _coerce_poll_interval(target.connector_config) raw_events await connector.fetch_alerts(since_secondssince_seconds)见 scheduler.py。所以一个 600 秒的连接器每次拉取的是“最近 600 秒内eventTime落在窗口里的 ACTIVE findings”。发生故障恢复时行为不同worker 会传入显式的backfill_seconds覆盖整个中断时长并被_MAX_BACKFILL_LOOKBACK_S截断。这个上限的存在有现实原因——SCC 的查询窗口过大时可能被限流甚至拒绝。过滤表达式连接器构造的 filter 是scc_filter fstateACTIVE AND eventTime {since}两点值得注意stateACTIVE表示只取当前仍活跃的 finding已静音muted和已解决resolved的会被刻意排除。这不是遗漏而是设计——排障章节里专门解释了为什么“已知有漏洞的组织却拿到空结果”往往就是这个条件造成的。比较符是而原文档写的是。边界上会多包含一条恰好等于窗口起点的 finding实践中不会造成重复风暴下一轮窗口已前移但你在做精确去重分析时应知道这一点。分页大小_PAGE_SIZE 200。当前fetch_alerts()只发一次请求没有实现nextPageToken的循环翻页——这一点从函数体可以直接读出来请求一次、取listFindingsResults、归一化返回。对大多数组织的增量窗口来说 200 条足够但如果你在首次接入一个历史积累了大量 finding 的组织或者把窗口设得很大就应该预期“单轮只拿到前 200 条”而后续轮次靠eventTime窗口自然推进。需要一次性全量回填时用gcloud scc findings list或调整窗口分段更可靠。认证链路手写 JWT 而不是依赖 SDK这个连接器没有引入google-auth而是自己用cryptography签了一个 RS256 的 JWT再去https://oauth2.googleapis.com/token换 access token。相关的常量集中在文件头部_TOKEN_URL https://oauth2.googleapis.com/token _SCC_BASE https://securitycenter.googleapis.com/v1 _SCOPE https://www.googleapis.com/auth/cloud-platform三个关键环节1. 密钥解析要早失败_parse_sa()对 JSON 解析失败和缺字段给出的是可直接指引用户修正的错误gcp_scc.pyraise ValueError(service_account_json is not valid JSON. Paste the entire key file contents.) raise ValueError(fservice_account_json missing required field: {required})必检字段是client_email、private_key、token_uri。这三者缺任意一个都不可能换到 token所以在构造器阶段就拦掉而不是等到第一次轮询才 500。2. JWT claims 的最小集合claims { iss: self._sa_info[client_email], scope: _SCOPE, aud: self._sa_info.get(token_uri, _TOKEN_URL), iat: now, exp: now 3600, }aud优先取密钥文件里自带的token_uri取不到才回落到默认端点——这一点对私有云或代理场景更友好。签名用的是 PKCS#1 v1.5 SHA-256即 RSA 服务账号密钥的标准形态。3. token 缓存与 401 重试if self._access_token and time.time() self._token_expiry - 60: return self._access_token用提前 60 秒过期的方式避免边界竞态token 在还剩 60 秒时就被视为不可用宁可多换一次也不要在请求途中失效。换取的 grant_type 是urn:ietf:params:oauth:grant-type:jwt-bearer。在fetch_alerts()与get_resource_config()里还有一个额外的 401 分支清空缓存 token、重新认证、重放一次请求。只重放一次是有意的——如果第二次还是 401那问题在权限或密钥本身重试再多次也不会变。顺带说明一个设计取舍SCC 与gcp_cloud_audit的 JWT 逻辑是复制而非共享模块的。原因写在模块头部注释里——连接器要保持可独立提取到 marketplace宁可接受短期重复也不要为了复用而引入跨连接器依赖。如果将来抽出gcp_auth公共 helper两边需要同步切换。归一化severity、event_type 与资源上下文normalize()是整个连接器的语义核心负责把 SCC 的 finding 形状压成 AiSOC 的事件形状。它先做了一个形状兼容finding raw.get(finding, {}) or raw # tolerate either shape resource raw.get(resource, {}) or {}findings.list返回的是包了一层的{finding: {...}, resource: {...}}但get_resource_config()拿到的是裸 finding 对象两种形态都要能吃下。severity 映射SCC severityAiSOC severityCRITICALcriticalHIGHhighMEDIUMmediumLOWlowunset /SEVERITY_UNSPECIFIED/ 其他info这里有一处必须点明的文档与实现差异。原文档表格写的是“unset →low”而当前代码的回落值是infoseverity { CRITICAL: critical, HIGH: high, MEDIUM: medium, LOW: low, }.get(scc_sev, info)对应的测试 test_gcp_connector.py 明确断言了这一行为def test_scc_normalize_unknown_severity_falls_back_to_info(fake_sa_json): raw {finding: {name: x, category: UNKNOWN_THING}} out connector.normalize(raw) assert out[severity] info从源码注释看这个选择的意图是保留 SCC 的原始优先级不要把一个 P1 finding 压成highMap directly to AiSOCs 5-tier ladder (info | low | medium | high | critical) so P1 SCC findings keep their original priority rather than getting collapsed intohigh.未知严重度回落到info与这条原则是自洽的编译器在不知道严重度时既不该猜测成high制造噪声也不该断言成low暗示风险可忽略。运维上你需要知道的实际后果是——如果某个 SCC 来源不填 severity这些 finding 在 AiSOC 里是info级不会自动触发 P1 流程。category / findingClass 映射原文档给出的findingClass映射表如下SCCfindingClassAiSOC categoryTHREATthreatVULNERABILITYcloud_postureMISCONFIGURATIONcloud_postureOBSERVATION、SCC_ERRORcloud_posture这条映射表达的是语义意图威胁类 finding 进入threat通道漏洞与配置偏差进入cloud_posture通道。但在当前normalize()的返回字典中并没有写出category字段也没有读取findingClass返回值见 gcp_scc.py字段为source、external_id、title、description、severity、actor、actor_email、event_type、raw_event、created_at、resource_name、resource_type。从源码结构看分类信号实际由两处承载event_type形如gcp.scc.category 小写例如MALWARE_BAD_DOMAIN会变成gcp.scc.malware_bad_domainingest 侧的 OCSF 归类整路gcp_scc统一落在 2001 / category 2理由为“已由 vendor 判定”。所以如果你要按“威胁 vs 姿态”分流到不同的 detection 或 playbook当前最稳的做法是基于event_type前缀加上 finding 的category值原始值保留在raw_event里而不是依赖事件上存在一个category: cloud_posture字段。这是文档语义与代码现状之间需要读者自己对齐的一处细节也是版本升级时值得复查的位置。完整字段清单AiSOC 字段来源说明source常量固定gcp_scc即connector_idexternal_idfinding.name完整资源名可直接回灌get_resource_config()titlefinding.category例如PUBLIC_BUCKET_ACL、IAM_ANOMALY缺失时回落GCP SCC Findingdescriptionfinding.description→finding.category→ 固定串三级回落保证永远非空severityfinding.severity见上一节映射actor/actor_emailfinding.access.principalEmail威胁类 finding 里最有价值的实体用于实体关联与 UEBAevent_typegcp.scc.categorycategory缺失时用finding兜底created_atfinding.eventTime→finding.createTimeeventTime更贴近“事情发生”的时间resource_nameresource.name→finding.resourceName完整资源路径resource_typeresource.type例如google.cloud.storage.Bucketraw_event原始对象完整保留供 playbook 与取证使用raw_event的完整保留是关键设计SCC finding 里有大量 AiSOC 顶层 schema 不建模的字段sourceProperties、indicator、vulnerability、securityMarks等保留原始对象意味着后续不必改归一化层就能在 playbook 里取用。溯源能力get_resource_config()与姿态快照这是文档没有展开、但对“可复盘”很重要的一个方法。它接受external_id即 finding 的完整资源名拼成https://securitycenter.googleapis.com/v1/{full_resource_name}去取详情返回{ raw: { ...: 完整 finding 详情 }, snapshot_freshness: live, snapshot_requested_at: 2026-01-10T00:00:00Z }行为上有几条硬约束都能在 gcp_scc.py 与测试中确认空 ID 或纯空白直接返回{}不发请求非 200 响应返回{}而不是抛异常并记录一条gcp_scc.get_resource_config_failed日志finding字段做过去除换行的截断防止日志注入401 时重认证并重放一次测试test_scc_get_resource_config_401_then_retry覆盖了这条路径传入at_ts时只把它作为snapshot_requested_at回填因为 SCC 提供的是当前状态不存在时间点回溯——所以snapshot_freshness恒为live。这是诚实的表达不要指望 SCC 给你历史快照。此外还有一个哨兵 ID当resource_id POSTURE_SNAPSHOT_ID值为__posture_snapshot__时方法会转而调用 GCP 身份姿态采集器 posture/gcp.py返回一份 IAM 姿态快照而不是单条 finding。该采集器会沿projects.getAncestry读取组织与 folder 层继承的绑定并在无权读取时写下 note 而不是把“读不到”渲染成“没有绑定”——因为后者会让一个实际拥有roles/owner的主体看起来毫无权限。故障排查PERMISSION_DENIED: ... does not have securitycenter.findings.list ... on organization ...服务账号没有组织级权限。回到组织 IAM 页面重新授予roles/securitycenter.findingsViewer不要在任何项目级页面授权。注意 SCC 的权限检查发生在 org 资源上项目级即使显示“已有该角色”也不生效。NOT_FOUND: Organization organizations/123456789012 was not found组织 ID 写错或该组织从未启用过 SCC。用下面的命令确认组织侧的 SCC 配置是否存在gcloud scc settings describe --organizationORG_ID同时确认你填的是数字 ID而不是域名、项目 ID 或项目编号。已知存在漏洞的组织却拿到空结果先用 CLI 验证 findings 确实存在gcloud scc findings list organizations/ORG_ID如果 CLI 有结果但连接器返回 0第一嫌疑是stateACTIVE这个条件——连接器刻意过滤掉了 muted 的 finding。请在 Console 里确认这些 finding 没有被静音。第二嫌疑是时间窗口eventTime必须落在since之内一个几周前产生、之后没有更新eventTime的 finding 不会出现在增量窗口里。源码层面还能补充的两个坑service_account_json不是合法 JSON报错文本会明确要求粘贴整份文件内容。常见成因是粘贴时漏掉结尾的}或把文件路径当成了内容粘贴进去。轮询失败时返回空数组而非抛错fetch_alerts()在非 200 时打一条gcp_scc.fetch_failed警告日志带status与截断到 300 字符的body然后return []。这意味着调度器层面这一轮会被视作成功、只是拿到了 0 条事件。所以当你在 UI 上看到“连接器健康、最近同步正常、事件数 0”时必须去 connectors 服务日志里搜gcp_scc.fetch_failed才能看到真正的 HTTP 状态码。这是排障时最容易漏看的一层。与 GCP Cloud Audit Logs 的配合两个连接器的分工可以一句话概括SCC 告诉你哪里不对Cloud Audit Logs 告诉你谁在什么时候改的。维度GCP SCCGCP Cloud Audit Logs作用域组织级单个项目数据性质vendor 已判定的 finding原始 API 调用审计事件归一化落点OCSF 2001category 2OCSF 6003 API Activity典型用途姿态与威胁的入口、优先级排序变更溯源、责任主体定位权限roles/securitycenter.findingsViewerorgroles/logging.viewer/privateLogViewer配套使用的方式很自然SCC 报出“某个 bucket 变成 public”PUBLIC_BUCKET_ACL其中resource_name指向具体 bucket接着用 Cloud Audit Logs 查该 bucket 在对应时间窗内的storage.setIamPermissions调用定位操作者 IP 与 principal。SCC 事件里的actor_email只在威胁类 finding 上有值来自finding.access.principalEmail配置类 finding 往往需要靠审计日志补齐主体信息。配置页见 GCP Cloud Audit Logs凭据模型见 凭据保险库与密钥管理。延伸阅读与源码索引连接器实现services/connectors/app/connectors/gcp_scc.py连接器基类与能力枚举services/connectors/app/connectors/base.py轮询调度与间隔边界services/connectors/app/scheduler.pyOCSF 归类表gcp_scc→ 2001 Security Findingservices/ingest/internal/normalizer/ocsf_classes.go单元测试schema、severity、认证、401 重试、详情拉取services/connectors/tests/test_gcp_connectors.py插件清单plugins/gcp-scc/plugin.yamlGCP 身份姿态采集器services/connectors/app/connectors/posture/gcp.py连接器出口校验SSRF 防护services/connectors/app/security/egress.py连接器总览与统一轮询模型apps/docs/docs/connectors/index.md需要特别提醒的一点本文中的行为描述以当前仓库快照为准。normalize()未输出category、未知 severity 回落info、test_connection()使用sources.list、单次请求不翻页、默认轮询 300 秒——这几处都可能随版本演进调整升级时建议对照上述源码位置复查而不是只看文档表格。赞分享人工智能AI Agent网络安全后端前端【免费下载链接】AiSOCOpen-source AI Security Operations Center: alert fusion, LLM-agent triage, MITRE ATTCK investigation, and a replayable decision ledger for every agent step. Self-hostable, runs with no API keys, MIT licensed. Ships an MCP server for Claude, Cursor and Continue.项目地址https://gitcode.com/GitHub_Trending/ai/AiSOC点击查看免费下载相关推荐AiSOC AWS GuardDuty 连接器实战直连原生 GuardDuty API 的威胁发现拉取、归一化与轮询机制详解AiSOC AWS GuardDuty 连接器实战直连原生 GuardDuty API 的威胁发现拉取、归一化与轮询机制详解 本文基于 AiSOC 官方文档人工智能AI Agent网络安全后端前端AiSOC Cisco Umbrella 连接器接入 DNS 与 Secure Web Gateway 事件、凭据管理与轮询机制详解AiSOC Cisco Umbrella 连接器接入 DNS 与 Secure Web Gateway 事件、凭据管理与轮询机制详解 本篇基于 AiSOC 仓人工智能AI Agent网络安全后端前端AiSOC 接入 Palo Alto Cortex XSIAM连接器配置、Fernet 凭证加密与轮询拉取机制详解AiSOC 接入 Palo Alto Cortex XSIAM连接器配置、Fernet 凭证加密与轮询拉取机制详解 本文以 AiSOC 官方连接器文档 cor人工智能AI Agent网络安全后端前端上一篇gogcli 幻灯片表格行操作指南用 gog slides table row 插入、删除与调整行高下一篇Agent OS Grid Balancing Swarm 实战100 个 DER 自治代理的电网平衡与治理演示创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
人工智能AI Agent网络安全后端前端【免费下载链接】AiSOCOpen-source AI Security Operations Center: alert fusion, LLM-agent triage, MITRE ATTCK investigation, and a replayable decision ledger for every agent step. Self-hostable, runs with no API keys, MIT licensed. Ships an MCP server for Claude, Cursor and Continue.项目地址https://gitcode.com/GitHub_Trending/ai/AiSOC点击查看免费下载GCP Security Command CenterSCC是 Google Cloud 上唯一的组织级安全信号聚合面把漏洞、错误配置、运行时威胁与合规姿态统一成 findings。本文以 AiSOC 官方连接器文档为骨架结合services/connectors/app/connectors/gcp_scc.py的真实实现逐层拆解这个连接器的权限模型、配置字段、轮询窗口、JWT 认证链路、严重度与分类归一化以及排障时应该看哪个日志字段。读完你可以独立完成组织级 SCC 接入并能解释“为什么我拿到了 0 条 finding”“为什么 critical 没有降级成 high”这类一线问题。连接器的定位把「已经判过的结论」搬进 AiSOCSCC connector 拉取的并不是原始遥测而是findings——Google 从 Cloud Audit Logs、VPC Flow Logs、Container Threat Detection、Web Security Scanner 以及合作方集成中综合出的、已经带有结论的安全信号。连接器源码模块注释把这一点讲得很直白SCC 相当于 GCP 侧的 AWS Security Hub 或 Microsoft Defender for Cloud是 GCP 资产中信号密度最高的一路数据源。在 AiSOC 的能力模型里这个连接器只声明了一个能力能力声明位置含义PULL_ALERTScapabilities()gcp_scc.py被动轮询已有告警流不包含QUERY_LOGS、ENRICH_*或任何阻断类动作这意味着 SCC 连接器是只读的。它不会替你关掉一个公开的 GCS bucket也不会隔离实例——这类动作需要由 playbook 走 Actions 服务且需要目标连接器显式声明对应能力。能力的粒度设计在 base.py 中有说明PULL_ALERTS表示“已经落到数据湖里的东西”与QUERY_LOGS现在就去源端查严格区分便于 agent 在规划时估算成本与延迟。能拿到什么原文档列出的四类 finding 与 SCC 的发现来源一一对应VulnerabilityCVE、泄露的凭据、公开可读的 bucketMisconfiguration来自 Security Health Analytics 的错误配置Threat来自 Event Threat Detection 的异常 IAM 行为、暴力破解、挖矿Posture来自 Posture Management 的合规姿态违规进入 AiSOC 之后被归到哪一类这一点原文档只给了一句category: cloud_posture/category: threat但落地细节在 ingest 服务里值得展开。在 OCSF 归类表 ocsf_classes.go 中gcp_scc被显式映射为2001 Security Finding并附了一条理由说明gcp_scc: {genericReason: Security Command Center findings: already judged by the vendor},2001 属于 OCSFcategory 2Findings类目推导见 ocsf_classes.gocategory_uid class_uid / 1000。这条归属决定的实战含义是SCC 事件会无条件进入 Findings 类目并参与提升promotion而不是像原始日志那样先被降级过滤再等待匹配。原因写在注释里——vendor 已经判过了AiSOC 不该再判一次“值不值得看”。因此一次典型的处理链路是SCC findings.list (org 级) │ fetch_alerts(since_seconds) ▼ GCPSCCConnector.normalize() # severity / event_type / resource_* 落字段 │ IngestClient.post(batch, tenant header) ▼ services/ingest OCSF 归一化 # gcp_scc → class_uid 2001, category_uid 2 ▼ fusion / detection / agent triage前置条件SCC 连接器对权限的要求比 Cloud Audit Logs 严格得多因为它读的是组织级资源。下列四项缺一不可Security Command Center 必须在组织级启用。项目级 SCC 不会暴露这里用到的 API。一个组织 ID必须是数字形式用gcloud organizations list查。一个服务账号在组织作用域上被授予Security Center Findings Viewerroles/securitycenter.findingsViewer。该服务账号的JSON key。源码 schema 的字段帮助文本也重复了这条约束gcp_scc.pyNumeric GCP organization ID. SCC is org-scoped; the service account must have findingsViewer at this level (or on a parent folder/source).注意括号里的补充除了组织级父 folder 或单个 source上的授权也能生效。这在你需要把接入范围收窄到某个 folder 时很有用但代价是组织内其他 folder 的 finding 将不可见。配置实战从零到「测试连接通过」1. 确认 SCC 已启用gcloud scc settings services list --organizationORG_ID输出中应能看到SECURITY_HEALTH_ANALYTICS、EVENT_THREAT_DETECTION等服务处于启用状态。如果为空或未启用需要在组织级开通 SCC Standard 或 Premium。2. 创建服务账号GCP Console → IAM Admin → Service Accounts选择任意项目建议单独建一个security-tools项目专门存放只读安全凭据。Create service account命名aisoc-scc-reader。3. 在组织级授予角色Console → IAM Admin → IAM切换到组织视图使用组织切换器。Grant access → principal 填新建服务账号的邮箱 → 角色选Security Center Findings Viewer。这一步必须在组织级完成不能在项目级。SCC findings 是组织作用域资源项目级授权只会得到一个看似配置正确、实际永远 403 的连接器。4. 生成 JSON key 并填入 AiSOC服务账号 → Keys → Add key → Create new key →JSON浏览器会下载密钥文件。AiSOC → Connectors → Add connector →GCP Security Command Center。organization_id填数字组织 ID例如123456789012而不是example.com。service_account_json粘贴整份JSON 文件内容。Test connection。Save。关于第 5 步这里有一个值得说明的实现细节原文档描述为“调用findings.listpageSize1”而当前源码实际调用的是sources.listresp await client.get( f{_SCC_BASE}/organizations/{self._organization_id}/sources, headersself._headers(), params{pageSize: 1}, )见 gcp_scc.py紧邻的注释解释了选择理由列 source 是最廉价的校验方式一次调用同时验证了三件事——token 有效、组织 ID 正确、SCC 已在组织上启用。如果你在按文档与代码比对行为以sources端点为准两者在“单次最小开销探活”这个意图上是一致的。测试成功时返回结构包含success、connector、organization_id以及服务账号邮箱{ success: true, connector: gcp_scc, organization_id: 123456789012, service_account: aisoc-scc-readersecurity-tools.iam.gserviceaccount.com }失败则返回{success: false, connector: gcp_scc, error: HTTP 403: ...}error会截断到 300 字符以内避免把整段 HTML 错误页塞进连接器健康状态字段。配置字段参考下表由scripts/generate_connector_docs.py从连接器schema()生成并补入了源码中的类型与提示信息字段标签类型必填默认值说明organization_idOrganization IDstring是—数字形式的 GCP 组织 ID。SCC 是组织级资源服务账号必须在该层级或父 folder / source 上拥有 findingsViewer。schema 中的 placeholder 为1234567890。service_account_jsonService account JSON keysecret是—静态加密存储· 粘贴完整 JSON 密钥文件写入前由凭据保险库加密。两个字段都声明为必填required: [organization_id, service_account_json]见 plugin.yaml因此 API 侧校验与前端表单一一致不会出现“只在 UI 上必填”的空档。:::note 凭据处理service_account_json在写入数据库行之前就由凭据保险库加密仅在轮询时解密。该值不会被记录到日志也不会由 API 返回。详见 凭据保险库与密钥管理。 :::加密在链路上是怎么发生的保险库的密钥来自部署级环境变量AISOC_CREDENTIAL_KEY实现位于 credential_vault.py。调度器在每次 tick 时对它做解密把明文只留在内存中一次fetch_alerts()的生命周期里。这意味着两件运维上很重要的事一份pg_dump泄露而不泄露AISOC_CREDENTIAL_KEY等于拿不到服务账号私钥API 与 connectors 两个服务必须共享同一个AISOC_CREDENTIAL_KEY否则解密会因完整性校验失败而报错源码中的错误提示明确指向“key mismatch between API and connectors services”。还有一个容易被忽略的 schema 细节OAuthHints在这个连接器上被显式声明为supported_in_hostedFalse只填了token_url与scopesoauthOAuthHints( supported_in_hostedFalse, authorize_urlNone, token_url_TOKEN_URL, scopes[_SCOPE], )也就是说即使在托管部署里你也必须自带服务账号 JSON不存在“点一下 Connect 就授权”的路径。这是设计使然SCC 需要组织级 IAM 授权SaaS 厂商代为申请的 OAuth 应用无法覆盖客户的 org IAM 边界。轮询与增量窗口文档的 600 秒与代码的 300 秒原文档写默认间隔600 秒10 分钟理由是 SCC findings 的更新频率低于审计日志且 API 配额较紧。这个 600 秒在实现里不是硬编码——它通过连接器配置里的poll_interval_seconds表达interval _coerce_poll_interval(target.connector_config)而_coerce_poll_interval()的行为在 scheduler.py 中定义得非常明确参数值说明_DEFAULT_POLL_INTERVAL_S300未配置poll_interval_seconds时的默认值_MIN_POLL_INTERVAL_S30低于此值被抬到 30 秒_MAX_POLL_INTERVAL_S86400高于此值被压到 24 小时所以要把 SCC 调到文档推荐的 10 分钟需要在连接器配置中显式写入poll_interval_seconds: 600。若不写实际生效的是调度器的 300 秒默认值——这是一个容易被误判为“文档写错了”的点实际是“文档给的是推荐值默认值是另一层”。顺便说把间隔设得很小对 SCC 基本没有收益findings 的eventTime更新粒度远大于分钟级30 秒轮询只会消耗配额并把大量重复窗口重新拉一遍。since窗口如何与轮询间隔对齐调度器把轮询间隔直接当作回看窗口传给连接器# We deliberately default to the same poll interval, so a 5-min # poll fetches the last 5 min of events. since_seconds _coerce_poll_interval(target.connector_config) raw_events await connector.fetch_alerts(since_secondssince_seconds)见 scheduler.py。所以一个 600 秒的连接器每次拉取的是“最近 600 秒内eventTime落在窗口里的 ACTIVE findings”。发生故障恢复时行为不同worker 会传入显式的backfill_seconds覆盖整个中断时长并被_MAX_BACKFILL_LOOKBACK_S截断。这个上限的存在有现实原因——SCC 的查询窗口过大时可能被限流甚至拒绝。过滤表达式连接器构造的 filter 是scc_filter fstateACTIVE AND eventTime {since}两点值得注意stateACTIVE表示只取当前仍活跃的 finding已静音muted和已解决resolved的会被刻意排除。这不是遗漏而是设计——排障章节里专门解释了为什么“已知有漏洞的组织却拿到空结果”往往就是这个条件造成的。比较符是而原文档写的是。边界上会多包含一条恰好等于窗口起点的 finding实践中不会造成重复风暴下一轮窗口已前移但你在做精确去重分析时应知道这一点。分页大小_PAGE_SIZE 200。当前fetch_alerts()只发一次请求没有实现nextPageToken的循环翻页——这一点从函数体可以直接读出来请求一次、取listFindingsResults、归一化返回。对大多数组织的增量窗口来说 200 条足够但如果你在首次接入一个历史积累了大量 finding 的组织或者把窗口设得很大就应该预期“单轮只拿到前 200 条”而后续轮次靠eventTime窗口自然推进。需要一次性全量回填时用gcloud scc findings list或调整窗口分段更可靠。认证链路手写 JWT 而不是依赖 SDK这个连接器没有引入google-auth而是自己用cryptography签了一个 RS256 的 JWT再去https://oauth2.googleapis.com/token换 access token。相关的常量集中在文件头部_TOKEN_URL https://oauth2.googleapis.com/token _SCC_BASE https://securitycenter.googleapis.com/v1 _SCOPE https://www.googleapis.com/auth/cloud-platform三个关键环节1. 密钥解析要早失败_parse_sa()对 JSON 解析失败和缺字段给出的是可直接指引用户修正的错误gcp_scc.pyraise ValueError(service_account_json is not valid JSON. Paste the entire key file contents.) raise ValueError(fservice_account_json missing required field: {required})必检字段是client_email、private_key、token_uri。这三者缺任意一个都不可能换到 token所以在构造器阶段就拦掉而不是等到第一次轮询才 500。2. JWT claims 的最小集合claims { iss: self._sa_info[client_email], scope: _SCOPE, aud: self._sa_info.get(token_uri, _TOKEN_URL), iat: now, exp: now 3600, }aud优先取密钥文件里自带的token_uri取不到才回落到默认端点——这一点对私有云或代理场景更友好。签名用的是 PKCS#1 v1.5 SHA-256即 RSA 服务账号密钥的标准形态。3. token 缓存与 401 重试if self._access_token and time.time() self._token_expiry - 60: return self._access_token用提前 60 秒过期的方式避免边界竞态token 在还剩 60 秒时就被视为不可用宁可多换一次也不要在请求途中失效。换取的 grant_type 是urn:ietf:params:oauth:grant-type:jwt-bearer。在fetch_alerts()与get_resource_config()里还有一个额外的 401 分支清空缓存 token、重新认证、重放一次请求。只重放一次是有意的——如果第二次还是 401那问题在权限或密钥本身重试再多次也不会变。顺带说明一个设计取舍SCC 与gcp_cloud_audit的 JWT 逻辑是复制而非共享模块的。原因写在模块头部注释里——连接器要保持可独立提取到 marketplace宁可接受短期重复也不要为了复用而引入跨连接器依赖。如果将来抽出gcp_auth公共 helper两边需要同步切换。归一化severity、event_type 与资源上下文normalize()是整个连接器的语义核心负责把 SCC 的 finding 形状压成 AiSOC 的事件形状。它先做了一个形状兼容finding raw.get(finding, {}) or raw # tolerate either shape resource raw.get(resource, {}) or {}findings.list返回的是包了一层的{finding: {...}, resource: {...}}但get_resource_config()拿到的是裸 finding 对象两种形态都要能吃下。severity 映射SCC severityAiSOC severityCRITICALcriticalHIGHhighMEDIUMmediumLOWlowunset /SEVERITY_UNSPECIFIED/ 其他info这里有一处必须点明的文档与实现差异。原文档表格写的是“unset →low”而当前代码的回落值是infoseverity { CRITICAL: critical, HIGH: high, MEDIUM: medium, LOW: low, }.get(scc_sev, info)对应的测试 test_gcp_connector.py 明确断言了这一行为def test_scc_normalize_unknown_severity_falls_back_to_info(fake_sa_json): raw {finding: {name: x, category: UNKNOWN_THING}} out connector.normalize(raw) assert out[severity] info从源码注释看这个选择的意图是保留 SCC 的原始优先级不要把一个 P1 finding 压成highMap directly to AiSOCs 5-tier ladder (info | low | medium | high | critical) so P1 SCC findings keep their original priority rather than getting collapsed intohigh.未知严重度回落到info与这条原则是自洽的编译器在不知道严重度时既不该猜测成high制造噪声也不该断言成low暗示风险可忽略。运维上你需要知道的实际后果是——如果某个 SCC 来源不填 severity这些 finding 在 AiSOC 里是info级不会自动触发 P1 流程。category / findingClass 映射原文档给出的findingClass映射表如下SCCfindingClassAiSOC categoryTHREATthreatVULNERABILITYcloud_postureMISCONFIGURATIONcloud_postureOBSERVATION、SCC_ERRORcloud_posture这条映射表达的是语义意图威胁类 finding 进入threat通道漏洞与配置偏差进入cloud_posture通道。但在当前normalize()的返回字典中并没有写出category字段也没有读取findingClass返回值见 gcp_scc.py字段为source、external_id、title、description、severity、actor、actor_email、event_type、raw_event、created_at、resource_name、resource_type。从源码结构看分类信号实际由两处承载event_type形如gcp.scc.category 小写例如MALWARE_BAD_DOMAIN会变成gcp.scc.malware_bad_domainingest 侧的 OCSF 归类整路gcp_scc统一落在 2001 / category 2理由为“已由 vendor 判定”。所以如果你要按“威胁 vs 姿态”分流到不同的 detection 或 playbook当前最稳的做法是基于event_type前缀加上 finding 的category值原始值保留在raw_event里而不是依赖事件上存在一个category: cloud_posture字段。这是文档语义与代码现状之间需要读者自己对齐的一处细节也是版本升级时值得复查的位置。完整字段清单AiSOC 字段来源说明source常量固定gcp_scc即connector_idexternal_idfinding.name完整资源名可直接回灌get_resource_config()titlefinding.category例如PUBLIC_BUCKET_ACL、IAM_ANOMALY缺失时回落GCP SCC Findingdescriptionfinding.description→finding.category→ 固定串三级回落保证永远非空severityfinding.severity见上一节映射actor/actor_emailfinding.access.principalEmail威胁类 finding 里最有价值的实体用于实体关联与 UEBAevent_typegcp.scc.categorycategory缺失时用finding兜底created_atfinding.eventTime→finding.createTimeeventTime更贴近“事情发生”的时间resource_nameresource.name→finding.resourceName完整资源路径resource_typeresource.type例如google.cloud.storage.Bucketraw_event原始对象完整保留供 playbook 与取证使用raw_event的完整保留是关键设计SCC finding 里有大量 AiSOC 顶层 schema 不建模的字段sourceProperties、indicator、vulnerability、securityMarks等保留原始对象意味着后续不必改归一化层就能在 playbook 里取用。溯源能力get_resource_config()与姿态快照这是文档没有展开、但对“可复盘”很重要的一个方法。它接受external_id即 finding 的完整资源名拼成https://securitycenter.googleapis.com/v1/{full_resource_name}去取详情返回{ raw: { ...: 完整 finding 详情 }, snapshot_freshness: live, snapshot_requested_at: 2026-01-10T00:00:00Z }行为上有几条硬约束都能在 gcp_scc.py 与测试中确认空 ID 或纯空白直接返回{}不发请求非 200 响应返回{}而不是抛异常并记录一条gcp_scc.get_resource_config_failed日志finding字段做过去除换行的截断防止日志注入401 时重认证并重放一次测试test_scc_get_resource_config_401_then_retry覆盖了这条路径传入at_ts时只把它作为snapshot_requested_at回填因为 SCC 提供的是当前状态不存在时间点回溯——所以snapshot_freshness恒为live。这是诚实的表达不要指望 SCC 给你历史快照。此外还有一个哨兵 ID当resource_id POSTURE_SNAPSHOT_ID值为__posture_snapshot__时方法会转而调用 GCP 身份姿态采集器 posture/gcp.py返回一份 IAM 姿态快照而不是单条 finding。该采集器会沿projects.getAncestry读取组织与 folder 层继承的绑定并在无权读取时写下 note 而不是把“读不到”渲染成“没有绑定”——因为后者会让一个实际拥有roles/owner的主体看起来毫无权限。故障排查PERMISSION_DENIED: ... does not have securitycenter.findings.list ... on organization ...服务账号没有组织级权限。回到组织 IAM 页面重新授予roles/securitycenter.findingsViewer不要在任何项目级页面授权。注意 SCC 的权限检查发生在 org 资源上项目级即使显示“已有该角色”也不生效。NOT_FOUND: Organization organizations/123456789012 was not found组织 ID 写错或该组织从未启用过 SCC。用下面的命令确认组织侧的 SCC 配置是否存在gcloud scc settings describe --organizationORG_ID同时确认你填的是数字 ID而不是域名、项目 ID 或项目编号。已知存在漏洞的组织却拿到空结果先用 CLI 验证 findings 确实存在gcloud scc findings list organizations/ORG_ID如果 CLI 有结果但连接器返回 0第一嫌疑是stateACTIVE这个条件——连接器刻意过滤掉了 muted 的 finding。请在 Console 里确认这些 finding 没有被静音。第二嫌疑是时间窗口eventTime必须落在since之内一个几周前产生、之后没有更新eventTime的 finding 不会出现在增量窗口里。源码层面还能补充的两个坑service_account_json不是合法 JSON报错文本会明确要求粘贴整份文件内容。常见成因是粘贴时漏掉结尾的}或把文件路径当成了内容粘贴进去。轮询失败时返回空数组而非抛错fetch_alerts()在非 200 时打一条gcp_scc.fetch_failed警告日志带status与截断到 300 字符的body然后return []。这意味着调度器层面这一轮会被视作成功、只是拿到了 0 条事件。所以当你在 UI 上看到“连接器健康、最近同步正常、事件数 0”时必须去 connectors 服务日志里搜gcp_scc.fetch_failed才能看到真正的 HTTP 状态码。这是排障时最容易漏看的一层。与 GCP Cloud Audit Logs 的配合两个连接器的分工可以一句话概括SCC 告诉你哪里不对Cloud Audit Logs 告诉你谁在什么时候改的。维度GCP SCCGCP Cloud Audit Logs作用域组织级单个项目数据性质vendor 已判定的 finding原始 API 调用审计事件归一化落点OCSF 2001category 2OCSF 6003 API Activity典型用途姿态与威胁的入口、优先级排序变更溯源、责任主体定位权限roles/securitycenter.findingsViewerorgroles/logging.viewer/privateLogViewer配套使用的方式很自然SCC 报出“某个 bucket 变成 public”PUBLIC_BUCKET_ACL其中resource_name指向具体 bucket接着用 Cloud Audit Logs 查该 bucket 在对应时间窗内的storage.setIamPermissions调用定位操作者 IP 与 principal。SCC 事件里的actor_email只在威胁类 finding 上有值来自finding.access.principalEmail配置类 finding 往往需要靠审计日志补齐主体信息。配置页见 GCP Cloud Audit Logs凭据模型见 凭据保险库与密钥管理。延伸阅读与源码索引连接器实现services/connectors/app/connectors/gcp_scc.py连接器基类与能力枚举services/connectors/app/connectors/base.py轮询调度与间隔边界services/connectors/app/scheduler.pyOCSF 归类表gcp_scc→ 2001 Security Findingservices/ingest/internal/normalizer/ocsf_classes.go单元测试schema、severity、认证、401 重试、详情拉取services/connectors/tests/test_gcp_connectors.py插件清单plugins/gcp-scc/plugin.yamlGCP 身份姿态采集器services/connectors/app/connectors/posture/gcp.py连接器出口校验SSRF 防护services/connectors/app/security/egress.py连接器总览与统一轮询模型apps/docs/docs/connectors/index.md需要特别提醒的一点本文中的行为描述以当前仓库快照为准。normalize()未输出category、未知 severity 回落info、test_connection()使用sources.list、单次请求不翻页、默认轮询 300 秒——这几处都可能随版本演进调整升级时建议对照上述源码位置复查而不是只看文档表格。赞分享人工智能AI Agent网络安全后端前端【免费下载链接】AiSOCOpen-source AI Security Operations Center: alert fusion, LLM-agent triage, MITRE ATTCK investigation, and a replayable decision ledger for every agent step. Self-hostable, runs with no API keys, MIT licensed. Ships an MCP server for Claude, Cursor and Continue.项目地址https://gitcode.com/GitHub_Trending/ai/AiSOC点击查看免费下载相关推荐AiSOC AWS GuardDuty 连接器实战直连原生 GuardDuty API 的威胁发现拉取、归一化与轮询机制详解AiSOC AWS GuardDuty 连接器实战直连原生 GuardDuty API 的威胁发现拉取、归一化与轮询机制详解 本文基于 AiSOC 官方文档人工智能AI Agent网络安全后端前端AiSOC Cisco Umbrella 连接器接入 DNS 与 Secure Web Gateway 事件、凭据管理与轮询机制详解AiSOC Cisco Umbrella 连接器接入 DNS 与 Secure Web Gateway 事件、凭据管理与轮询机制详解 本篇基于 AiSOC 仓人工智能AI Agent网络安全后端前端AiSOC 接入 Palo Alto Cortex XSIAM连接器配置、Fernet 凭证加密与轮询拉取机制详解AiSOC 接入 Palo Alto Cortex XSIAM连接器配置、Fernet 凭证加密与轮询拉取机制详解 本文以 AiSOC 官方连接器文档 cor人工智能AI Agent网络安全后端前端上一篇gogcli 幻灯片表格行操作指南用 gog slides table row 插入、删除与调整行高下一篇Agent OS Grid Balancing Swarm 实战100 个 DER 自治代理的电网平衡与治理演示创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
)
)
)
