企业官网建设流程全解析

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

本文以 AiSOC 的 Microsoft 365 Audit 连接器为主线，讲清楚它如何通过 Office 365 Management Activity API 订阅并拉取 Exchange、SharePoint/OneDrive、Teams、Azure AD 等统一审计事件，如何配置应用权限与租户审计开关，以及事件在normalize()与检查点机制下如何变成可检测、可重放的 AiSOC 事件。读完后你能独立完成该连接器从授权、接入、验证到排障的全流程，并理解它与源码实现之间的细节差异。

一、这个连接器到底拉取什么

M365 连接器对接的是Office 365 Management Activity API，而不是 Microsoft Graph。这两者的差别是理解整个接入流程的关键：Graph 提供的是"成品告警"（如 Defender 的 alert 对象），而 Management Activity API 提供的是统一审计日志（Unified Audit Log）的原始事件流，需要按"发布/轮询"模式自己去取。

文档列出的内容类型如下：

事件入库后会被归一化为 AiSOC 事件模型，source固定为m365_audit，并带上由工作负载派生出的workload与event_type字段（见 m365_audit.py）。

关于"订阅五种内容类型"的一处重要事实修正

文档原文写道"The connector subscribes to all five content types"，但在当前源码中，实际默认订阅的是四种：

# services/connectors/app/connectors/m365_audit.py _DEFAULT_CONTENT_TYPES = ( "Audit.AzureActiveDirectory", "Audit.Exchange", "Audit.SharePoint", "Audit.General", )

模块 docstring 明确解释了原因：DLP.All是"噪音较大"的流，默认关闭以控制数据量，未来会通过connector_config让分析员按实例显式开启（见 m365_audit.py）。因此在排障时不要误判"漏订阅DLP.All"为故障——这是设计选择，不是缺陷。如果你确实需要 DLP 事件，需要在代码层面扩展默认集合，而不能仅靠控制台开关。

二、接入前的准备条件

你可以复用已有的 Entra 连接器的应用注册——两者共享同一套客户端凭据（client credentials）授权模型，只是权限资源不同。若复用，只需补加下面的应用权限。

从零创建时需要：

• 一个Azure AD 应用注册（Entra ID 中的 App registration）；
• 一个客户端密钥（client secret）；
• 在Office 365 Management APIs资源（不是Microsoft Graph）上授予应用权限ActivityFeed.Read，如需 DLP 事件再加ActivityFeed.ReadDlp；
• 对这些权限完成管理员同意（Admin consent）；
• 租户内的审计日志必须已启用。

这一步最容易被做错：在 Graph 上加权限不会生效，因为 Management Activity API 走的是另一套资源标识。源码里的令牌请求也直接印证了这一点：scope 是https://manage.office.com/.default，而非 Graph 的 scope：

resp = await client.post( _TOKEN_URL.format(tenant_id=self._tenant_id), data={ "client_id": self._client_id, "client_secret": self._client_secret, "scope": "https://manage.office.com/.default", "grant_type": "client_credentials", }, )

令牌端点同样来自https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token，与 Entra 连接器一致（见 m365_audit.py、m365_audit.py）。

三、配置步骤

1. 添加 Management API 权限

1. Entra 管理中心 → 应用注册 → 你的应用 → API 权限；
2. 添加权限 → Office 365 Management APIs → 应用程序权限；
3. 勾选ActivityFeed.Read（需要时再加ActivityFeed.ReadDlp）；
4. 点击为 <租户> 授予管理员同意。

2. 确认统一审计日志已开启

在已连接 Exchange Online 的 PowerShell 会话中执行：

Get-AdminAuditLogConfig | Select-Object UnifiedAuditLogIngestionEnabled

若返回False，执行：

Set-AdminAuditLogConfig -UnifiedAuditLogIngestionEnabled $true

注意：只是"有权限"并不代表"有数据"。租户审计开关关闭时，API 调用会成功但事件为空，这是最常见的"配置看起来都对却没数据"的原因。

3. 在 AiSOC 中添加连接器

1. Connectors → Add connector → Microsoft 365 Audit；
2. 填入tenant_id、client_id、client_secret（与 Entra 连接器相同的值，或新签发的值）；
3. 点击Test connection验证；
4. 点击Save。

关于 Test connection 的实际行为

文档描述为"在Audit.General上启动订阅后立即停止"，但当前实现并不是这样。源码中的探测方式是列出当前订阅：

resp = await client.get( f"{self._base()}/subscriptions/list", headers=self._headers(), ) resp.raise_for_status() return { "success": True, "connector": self.connector_id, "tenant_id": self._tenant_id, "active_subscriptions": [s.get("contentType") for s in resp.json() if s.get("status") == "enabled"], }

也就是说，test_connection()只做两件事：先用客户端凭据换 token，再GET /subscriptions/list。它在零订阅的情况下也会返回 200，而在缺少ActivityFeed.Read管理员同意时返回 403——这正是它作为"最廉价鉴权探针"的原因（见 m365_audit.py，以及测试用例 test_saas_connectors.py）。

另外，该接口不会创建或停止任何订阅，测试过程中不会对租户产生副作用。返回值中active_subscriptions只包含status == "enabled"的条目，禁用状态会被过滤掉。

四、配置字段参考

字段定义来自连接器的schema()，其中tenant_id还带有占位符00000000-0000-0000-0000-000000000000便于表单提示（见 m365_audit.py、plugin.yaml）。

schema()还声明了 OAuth 提示：supported_in_hosted=True，scope 为https://manage.office.com/ActivityFeed.Read。按基类注释，这只是给前端预留"托管 OAuth 即将支持"的渲染信息，前端据此展示徽标，当前实际的认证路径仍然是客户端凭据（见 base.py）。

凭据处理

client_secret在写入数据库行之前就由凭据保险库加密，只在轮询时解密；值不会被写日志，也不会通过 API 回读返回。具体机制、主密钥生成与轮换方式见 凭据保险库文档。连接器服务侧的解密发生在调度器为每个实例构造连接器之前（见 scheduler.py）。

五、轮询机制与检查点

轮询周期

默认间隔300 秒，由调度器统一管理。可覆盖的配置键是connector_config.poll_interval_seconds，取值范围被钳制在30 秒到 86400 秒（24 小时）之间，非法值回落为默认值（见 scheduler.py）。

poll_interval_seconds、checkpoint、filter_rules属于调度器专用配置键，由连接器服务从运行期 kwargs 中剔除，不会传入连接器构造函数（见 router.py）。

拉取流程

一次fetch_alerts()的步骤：

1. 若无 token，先认证；
2. 计算时间窗：end = now(UTC)，start = end - min(since_seconds, 24h)，格式化为%Y-%m-%dT%H:%M:%S；
3. 对四种内容类型幂等地调用POST /subscriptions/start；
4. 对每种类型GET /subscriptions/content?contentType=...&startTime=...&endTime=...列出可用的内容块（blob）；
5. 逐个GET contentUri拉取 blob，把内部事件数组拼接起来；
6. 先过apply_checkpoint()去重，再逐条normalize()。

代码里两处细节值得生产环境注意：

• 24 小时上限：Management API 规范禁止时间窗超过 24 小时，源码用min(since_seconds, 24 * 3600)做了保护，防止运维误配超长窗口。
• 时间格式不带时区后缀：使用%Y-%m-%dT%H:%M:%S输出朴素 UTC 字符串，这是该 API 的既有约定，不要自行追加Z。

订阅逻辑把 400 也视为成功，因为"已启用"的订阅再次 start 会返回 400；网络异常只记录 warning，不阻断本次拉取，下个周期会重试：

if resp.status_code not in (200, 400): logger.warning("m365_audit.subscribe_unexpected_status", ...)

（见 m365_audit.py）

容错与重试

• 单个内容类型列表请求返回 401 时，会重新认证并重试一次；其他非 200 只记 warning 并跳过该类型，不影响其他类型。
• blob 拉取非 200 时记录blob_id与状态码后跳过。
• blob 响应既可能是数组，也可能是{"value": [...]}包装，源码两种都兼容。

检查点：为什么重启不会重复灌数据

连接器声明了两个游标字段：

checkpoint_time_field = ("CreationTime",) checkpoint_id_field = ("Id",)

它们被基类的通用检查点机制消费：按(CreationTime, Id)排序，丢弃游标及之前的所有行，并把新的游标暂存在_next_checkpoint上。关键设计是"读时暂存、入库后才持久化"——只有当 ingest 确认整批事件落盘，调度器才会推进游标，因此被 ingest 拒绝的批次会被重读，而不是被跳过（见 base.py）。

Id作为同秒并列的决胜键不是可选项：同一秒内的两条事件很常见，只用时间做游标要么丢第二条，要么永远重放第一条（基类注释对此有专门说明）。检查点契约测试把m365_audit的("CreationTime",), ("Id",)组合固定在断言里（见 test_checkpoint_contract.py）。

数据延迟

Management API 本身有最长约60 分钟的投递延迟。订阅已启动但events_added: 0并不必然代表故障，应至少等待一个完整的轮询周期后再观察。对有延迟容忍的事件源，把poll_interval_seconds设得更短并不能降低延迟，只会增加空轮询。

六、归一化与严重度推断

Management API 不提供严重度字段，AiSOC 依据Operation与ResultStatus推断。

高风险操作匹配表

源码中_HIGH_RISK_OPERATIONS采用不区分大小写的子串匹配，命中即high：

（见 m365_audit.py）

严重度判定顺序

if is_high_risk: severity = "high" elif result_status in ("failed", "failure") and "login" in op_lower: severity = "low" elif result_status in ("failed", "failure"): severity = "low" else: severity = "info"

随后还有一次"管理员加成"：当UserType为2（Admin）或3（DcAdmin）且当前严重度为info时，提升为low。原因是特权操作者的日常动作风险基线高于普通用户噪音（见 m365_audit.py）。

UserType取值含义在源码注释中给出：0 普通用户、1 保留、2 管理员、3 全局/域控管理员、4 系统、5 应用、6 服务主体。

与文档严重度表的差异（务必注意）

文档中的表格与当前实现存在若干不一致，落地调优时以源码为准：

这意味着：如果你依赖"连接器会自动丢弃常规文件访问"，事件量会显著高于预期。正确的降噪位置是检测规则与融合层，而不是期待连接器过滤。仓库中针对此类信号的检测规则确实是按product: m365与具体操作名来写的，例如基于New-InboxRule/Set-InboxRule的罕见客户端检测（见 ta-m365-mailbox-rule-from-rare-client.yaml）和针对AnonymousLinkCreated的敏感文件检测（见 app-onedrive-anonymous-link-sensitive.yaml）。

输出事件结构

归一化后每条事件包含：

{ "source": "m365_audit", "external_id": "<Id>", "title": "<Operation> 或 'M365 <Workload> event'", "description": "workload=<Workload>; operation=<Operation>; result=<success|failed>", "severity": "high|low|info", "actor": "<UserId|UserKey|unknown>", "actor_email": "含 @ 时填充，否则 null", "src_ip": "<ClientIP>", "workload": "<Workload>", "event_type": "m365.<workload 小写>.<operation 去空格小写>", "raw_event": { "...": "原始审计记录" }, "created_at": "<CreationTime>" }

event_type的构造规则值得记住：Operation中的空格会被直接删除再转小写，所以FileAccessed在 SharePoint 工作负载下产出m365.sharepoint.fileaccessed。这决定了你写检测条件时的字符串形态（见 m365_audit.py，测试断言见 test_saas_connectors.py）。

此外raw_event保留了完整原始记录，因此即使某个字段没有进入标准模型（例如 SharePoint 的敏感度标签），下游规则仍可从原始事件中取值——这也解释了上面那条检测规则为何能引用sensitivity_label。

七、验证接入是否正确

用测试用例反推预期行为

该连接器的行为被一组单元测试完整覆盖，测试同时是"预期语义"的文档（respx模拟了令牌、订阅、列表与 blob 四层 HTTP）：

cd services/connectors && python -m pytest tests/test_saas_connectors.py -k m365 -v

覆盖点包括：

• schema 必需字段与client_secret的secret类型；
• schema 声明托管 OAuth 且 scope 含ActivityFeed.Read；
• 高危操作 →high；
• New-InboxRule→high（BEC 指标）；
• 失败登录 →low（保留给下游"喷洒"检测做乘法放大）；
• 管理员例行操作info → low；
• 普通用户FileAccessed→info且event_type以m365.sharepoint.开头；
• test_connection成功路径只回传 enabled 订阅；
• 403 时返回success: False且错误信息含403；
• fetch_alerts能穿透 blob 层拉出事件并归一化。

（见 test_saas_connectors.py）

手动验证顺序

1. 先测鉴权：点 Test connection，返回success: true说明令牌与subscriptions/list已通；
2. 再看订阅：若active_subscriptions为空，等一个轮询周期后再看，首轮拉取会幂等地启动四个订阅；
3. 最后看事件量：确认UnifiedAuditLogIngestionEnabled为True，并容忍最长 60 分钟延迟。

八、故障排查

AAD: Tenant ... does not have a service principal for resource manage.office.com应用只有 Microsoft Graph 权限，缺少 Office 365 Management API 权限。必须在 Office 365 Management APIs 资源上添加ActivityFeed.Read并授予管理员同意。在 Graph 上加同名权限不会解决。

订阅已启动但events_added: 0Management API 首次事件最长有 60 分钟延迟。等一个完整轮询周期再看；同时确认UnifiedAuditLogIngestionEnabled为True。若两者都正常，检查fetch_alerts日志中是否出现m365_audit.list_failed（某内容类型列表失败被跳过）或m365_audit.blob_fetch_failed（blob 拉取失败）。

InvalidPropertyValueException: tenant ID has not been onboarded需要先通过POST /subscriptions/start?contentType=...启动订阅。连接器会在首次轮询时自动完成，因此等一个周期即可；若持续报错，通常意味着订阅请求本身被拒（如权限或租户状态问题），可查看m365_audit.subscribe_unexpected_status日志中的状态码与响应体。

Test connection 返回 401 或超时401 通常指向client_secret过期或值填错（Entra 页面上的"Secret ID"不是可用值，要用"Value"）；超时则多与出网策略有关。连接器侧refuse_internal_destinations会在建立连接前拒绝指向云元数据、回环或内网服务的地址，避免 SSRF（见 router.py）。

重复事件先确认检查点是否在推进。若 ingest 拒收批次，游标不会前进，下一轮会重读同一窗口——这是刻意的"安全方向"设计，结果为重复而非丢数。可用checkpoint配置键（调度器专用）观察持久化值。

九、与其它 Microsoft 连接器的分工

• Microsoft Entra ID：目录级审计与风险登录。M365 审计提供工作负载级的深度。
• Microsoft Defender (XDR)：成品告警。M365 审计提供这些告警背后的原始事件。

三者可以共享同一个 Azure AD 应用注册，但权限授予的资源对象各不相同：Entra 走 Microsoft Graph 的AuditLog.Read.All等权限，M365 走manage.office.com的ActivityFeed.Read。把这两组权限同时授予同一个应用，是微软环境中最省事的部署形态。

十、实现要点速查

整体来看，这个连接器的复杂度几乎全部来自 Management Activity API 的"订阅—列块—取块"三段式模型，而不是认证；理解了这一点，再对照源码中的_ensure_subscriptions、fetch_alerts与apply_checkpoint三个方法，就能准确判断任何"没数据""有重复""严重度不对"的现象究竟发生在链路的哪一环。

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