GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-30 12:36:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(architecture): govern direct egress

Browse Source
This commit is contained in:
jxxghp
2026-08-27 07:28:42 +08:00
parent 3b7f8c357d
commit 47f0de745c
19 changed files with 3598 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test.yml
+1
View File
@@ -46,6 +46,7 @@ jobs:
uv run --locked --no-sync pytest \
tests/test_architecture_dependencies.py \
tests/test_architecture_adapter_imports.py \
tests/test_architecture_egress.py \
tests/test_architecture_contract_baseline.py \
tests/test_architecture_baseline_cli.py -q
uv run --locked --no-sync python \
AGENTS.md
+1 -1
View File
@@ -30,7 +30,7 @@ For work that changes or reviews repository behavior, identify the domains actua
### External Communication and Interfaces
* **Primary Reference:** `docs/rules/09-external-response.md`
* **Required Constraints:** Transport implementations for third-party HTTP must go through `RequestUtils`; this rule does not authorize Application/Chain to import the concrete Adapter. Response formats must use the project's standard schemas. Error handling must follow the per-layer conventions.
* **Required Constraints:** Host-authored ordinary HTTP must go through `RequestUtils`; this rule does not authorize Application/Chain to import the concrete Adapter. Canonical transport, SDK, streaming protocol, contained vendor, diagnostic and control-plane exceptions must match the exact direct-egress policy. Response formats must use the project's standard schemas. Error handling must follow the per-layer conventions.
### Data and Persistence
* **Primary Reference:** `docs/rules/10-data-and-persistent.md`
docs/architecture-optimization-checklist.md
+7 -3
View File
@@ -415,15 +415,19 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
- `app/chain` 有 8 个文件、13 条直接 Adapter 导入，使用 `RequestUtils`、Browser、Cloudflare、
CookieCloud、ServerHelper 等具体能力。
- Passkey Application 服务直接判断 Redis 后端并调用 `RedisHelper.pop()`，安全策略识别了具体实现。
- “所有 HTTP 必须经 RequestUtils”与 LLM streaming、第三方 SDK、移植库中的直接 HTTP 客户端现状
没有一份精确例外表，也没有 zero-growth gate。
- 审计时 LLM streaming、第三方 SDK、移植库和本地控制面没有精确例外表；S0-L2.4b 已建立
66 条完整 egress identity 与 zero-growth policy，其中 11 条普通 HTTP/Session bridge 和 1 条
Application DNS I/O 是清零债务；每条初始边另有独立指纹上界，不能靠同时刷新 baseline/policy
掩盖同一边的调用面增长，债务删除后也不得恢复。
**目标与步骤**
- [x] 建立 Application/Chain 原始 Adapter 直连事实与精确临时 policy，冻结新增、替换和陈旧条目。
- [x] 建立全宿主 direct egress 事实；SDK/stream/vendor/local-control 例外精确到 bindings/uses 指纹。
- [ ] 将 Passkey 原子领取提升为 runtime cache contract，由 Memory/Redis backend 分别实现。
- [ ] 为 Backup 定义 Application-owned artifact store Port，由 startup 注入文件系统实现。
- [ ] 普通外部请求迁移到统一网络能力；SDK transport、streaming 和 vendor code 例外必须精确到路径与原因。
- [ ] 将 policy 中 11 条普通 HTTP/Session bridge 债务迁移到统一网络能力并把目标收缩为空。
- [ ] 为 Application SSRF 校验注入 DNS 解析 Port，清除 `socket.getaddrinfo` 直接 I/O。
- [ ] 命名外部产品、安全敏感能力及通用技术 Adapter 均改为注入 Port；不在 Application/Chain 保留直连例外。
- [ ] 最终把基线收缩到零或少量书面化例外，而不是一次性禁止后再大量豁免。
docs/architecture-overview.md
+6
View File
@@ -673,6 +673,11 @@ flowchart LR
`--write-host` 不会替代人工决策。
- 同一 baseline 的 `direct_adapter_imports` 记录现存原始直连；policy 将其全部标为有 owner 的
`temporary_debt`，并以初始 28 条冻结上界、目标为空集合。新增、替换、删除后未清理 policy 都会失败。
- `direct_egress` 记录全宿主 66 条 raw transport、network SDK 和协议操作 identity；11 条普通 HTTP/
Session bridge 与 1 条 Application DNS I/O 是清零债务，54 条 canonical transport、SDK、
stream/vendor/diagnostic/control-plane 事实是精确 containment。每条初始边的指纹由测试独立冻结，
bindings/uses 变化、分类互换、通配导入和初始边增长都会失败；债务删除时同步删除冻结项以禁止恢复，
`--write-host` 不会改写人工 policy 或冻结上界。
- 任何所有权迁移必须同步更新：canonical 导入、`app/runtime/compat/manifest.py`、
SDK 导出（若公开）、`docs/rules/05-architecture.md` 与上述架构测试。
- 延迟导入不被接受为隐藏循环依赖的手段。
@@ -687,6 +692,7 @@ flowchart LR
| Python 模块 | 835 |
| 内部导入边 | 6,810 |
| 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） |
| Direct egress | 66（12 条待迁移债务，54 条精确 containment） |
| Module Contract V2 spec | 215（其中 214 个进入 `run_module` 观察面） |
| Event Contract | 53 |
| Model/Oper 自动事务与自建 Session | 0 |
docs/architecture-refactor-roadmap.md
+26 -23
View File
@@ -76,8 +76,8 @@ G-ARCH 只有在以下条件全部满足后才可完成：
| S0-L2.1 Host Oper/UoW 规范 | `DELIVERED` | S0-L1 | `3bf94ffed`：宿主无 Session Oper 规范债务归零，远端 `0/0` |
| S0-L2.2 完整宿主 SCC policy | `DELIVERED` | S0-L2.1 | `a884ab5c2`：完整宿主 SCC 精确 policy 生效，远端 `0/0` |
| S0-L2.3 Adapter 直连事实 | `DELIVERED` | S0-L2.1 | `e1483e85d`：锁定 28 条原始 Adapter import 事实，远端 `0/0` |
| S0-L2.4 Adapter zero-growth | `VERIFIED` | S0-L2.3 | 当前直连均登记迁移 owner，新增/替换失败，删除后要求清理陈旧 policy |
| S0-L2.4b HTTP/Egress 事实与政策 | `PLANNED` | S0-L2.4 | direct HTTP/SDK/stream/vendor egress 分类完整，债务与精确例外分别受不可增长政策约束 |
| S0-L2.4 Adapter zero-growth | `DELIVERED` | S0-L2.3 | `2553226f3`：冻结 28 条直连及 owner，收缩/新增/stale policy 门禁生效，远端 `0/0` |
| S0-L2.4b HTTP/Egress 事实与政策 | `VERIFIED` | S0-L2.4 | 66 条事实、12 条债务和 54 条精确例外已冻结；collector、policy、全量测试与本地质量门禁通过，待推送后确认远端 CI |
| S0-L2.5 Event consumer 识别 | `PLANNED` | S0-L2.1 | consumer 只识别可静态证明的 EventManager 注册，动态误报归零 |
| S0-L2.6 事实源与 CI 投影 | `PLANNED` | S0-L2.2,S0-L2.4b,S0-L2.5 | fixture/policy/overview 职责固定，CI 分开报告语义 policy 与快照一致性 |
@@ -107,8 +107,8 @@ G-ARCH 只有在以下条件全部满足后才可完成：
| S2-L3 GlobalVar/provider 注册收口 | `PLANNED` | S2-L1 | `global_vars` canonical 消费清零，provider 注册进入显式装配阶段并可 reset；Legacy 入口精确保留 |
| S2-L4 Passkey 缓存边界 | `PLANNED` | S0-L4 | Application 不识别 Redis；原子 consume 由 runtime cache contract + backend 实现 |
| S2-L5 Backup artifact Port | `PLANNED` | S0-L4 | Application 不构造 `BackupFiles`，文件 I/O 由注入 Adapter 拥有 |
| S2-L6 Application Adapter 债务清零 | `PLANNED` | S2-L4,S2-L5 | Application 到具体 Adapter 的未批准边归零，批准通用机制有精确规则和门禁 |
| S2-L7 Chain Adapter/HTTP 债务清零 | `PLANNED` | S2-L6 | Chain 具体 Adapter 与普通 direct HTTP 依赖归零；SDK/stream/vendor 例外精确 containment |
| S2-L6 Application Adapter/DNS 债务清零 | `PLANNED` | S2-L4,S2-L5 | Application 到具体 Adapter 的未批准边归零，SSRF DNS I/O 进入注入 Port，批准通用机制有精确规则和门禁 |
| S2-L7 Chain Adapter/宿主 HTTP 债务清零 | `PLANNED` | S2-L6 | Chain 具体 Adapter 与 11 条普通 direct HTTP/Session bridge 归零；SDK/stream/vendor 例外保持精确 containment |
### S3：大型编排器职责清零
@@ -154,50 +154,53 @@ G-ARCH 只有在以下条件全部满足后才可完成：
## 4. 当前活动叶子
### S0-L2.4 Adapter zero-growth
### S0-L2.4b HTTP/Egress 事实与政策
**Status:** `VERIFIED`（本地验收完成，等待提交、推送和远端一致性确认）
**Status:** `VERIFIED`（本地验收完成，等待提交、推送和远端 CI 确认）
**Outcome**
把 S0-L2.3 收集的 Application/Chain 到具体 Adapter 直连全部登记为有迁移 owner 的临时债务。
初始 28 条形成只减不增的冻结上界，政策目标固定为空集合；新增、替换和删除后未同步清理的
陈旧登记都必须失败。
扫描宿主 raw transport、network SDK 和库名扫描会漏掉的协议操作，事实保留 import provenance、
稳定 callable/operation，不保存行号。普通 HTTP/Session bridge 是待迁移债务；canonical transport、
SDK、streaming、contained vendor 和 local control-plane 只允许精确 containment。
**Ownership**
- `tests/fixtures/architecture/dependency-policy.json` 的人工 Adapter policy。
- `tests/test_architecture_adapter_imports.py` 的精确集合、owner 和 stale-policy 门禁。
- `AGENTS.md`、`docs/rules/05-architecture.md` 与架构总览中的规范边界。
- `.github/workflows/test.yml` 的快速架构 CI 投影。
- `scripts/architecture/` 的 direct egress AST collector 与 registry。
- `tests/fixtures/architecture/dependency-baseline.json` 的生成事实和
`dependency-policy.json` 的人工分类。
- `tests/test_architecture_egress.py` 的 collector、当前事实、policy 和 zero-growth 门禁。
- HTTP 规范、架构总览、优化清单与快速架构 CI 投影。
- 本路线图的叶子状态和交付记录。
**Excluded**
- 不在本叶迁移 28 条运行时依赖；迁移由 S2-L4 至 S2-L7 按 owner 债务清零。
- 不修改生成事实、完整依赖图、SCC、digest、`app/plugins/**` 或运行时代码。
- 不把 direct HTTP/SDK/stream/vendor egress 混入 Adapter policy；这些由 S0-L2.4b 独立治理。
- 不在本叶迁移生产 HTTP 调用；本叶完成完整事实、分类和不可增长门禁。
- 不修改 Adapter 直连 policy、完整依赖图/SCC/digest、运行时代码或 `app/plugins/**`。
- 未登记 registry 的任意第三方包不能被猜测为网络 SDK；新增 SDK 必须显式扩展 registry 与 policy。
**Acceptance**
```bash
.venv/bin/python -m pytest \
tests/test_architecture_adapter_imports.py \
tests/test_architecture_egress.py \
tests/test_architecture_contract_baseline.py \
tests/test_architecture_dependencies.py \
tests/test_architecture_baseline_cli.py \
tests/test_architecture_ci.py \
tests/test_architecture_documentation.py -q
tests/test_plugin_identity_transitions.py -q
.venv/bin/python scripts/architecture/baseline.py --check-host
.venv/bin/python scripts/architecture/ruff_ratchet.py
.venv/bin/python scripts/architecture/mypy_ratchet.py
.venv/bin/pylint tests/test_architecture_adapter_imports.py \
.venv/bin/pylint scripts/architecture/baseline.py \
scripts/architecture/egress.py \
tests/test_architecture_egress.py \
tests/test_architecture_contract_baseline.py \
tests/test_architecture_dependencies.py \
tests/test_architecture_ci.py
tests/test_architecture_ci.py \
tests/test_plugin_identity_transitions.py
git diff --check
```
**Delivery**
- 单一提交主题：冻结现有 Adapter 直连债务并阻止增长。
- 单一提交主题：建立并冻结 direct egress 事实与精确政策。
- 推送 `origin/v3` 后确认提交祖先关系、远端 SHA 和 ahead/behind `0/0`。
docs/rules/05-architecture.md
+19
View File
@@ -528,6 +528,25 @@ the Port required by its use case, startup injects the concrete Adapter, and Cha
consumes the Application capability or an injected Port. A `canonical capability`
never means permission to import a concrete `app.adapters.*` implementation.
Direct egress is a separate boundary from Adapter imports. The generated
`direct_egress` facts scan the complete host `app` tree except `app.plugins` and
record raw transports, registered network SDKs and exact protocol operations.
Each identity contains import provenance plus stable callable/operation uses and
has no line number. The manual policy classifies every full fingerprint as either
`temporary_debt` with a removal leaf and empty target state, or an
`approved_exception` with an exact owner and reason. Canonical transports, SDKs,
streaming protocols, contained vendor code, diagnostics and control planes are
contained exceptions, not category-wide permissions. In policy, `owner: "$source"`
means the fact's exact `source` module is the owner; it does not authorize sibling
or child modules. Runtime wildcard imports from a registered egress root are
forbidden. Updating the generated baseline never updates this policy; additions,
fact changes, classification swaps and stale entries fail independently. Current
debt may shrink without changing a fixed count, but no initial edge may grow or be
reclassified. Tests independently freeze every initial edge fingerprint, so
refreshing both generated facts and manual policy cannot hide growth on the same
`source/target`; when debt is removed, its frozen edge and fingerprint must be
removed in the same reviewed change so that it cannot return.
## Permitted Call Directions
| Direction | Status |
docs/rules/09-external-response.md
+11 -3
View File
@@ -2,11 +2,19 @@
## HTTP Client Conventions
**Rule:** Host outbound HTTP transport implementations must go through `RequestUtils` from
**Rule:** Host-authored ordinary outbound HTTP must go through `RequestUtils` from
`app/adapters/network/http.py`. This transport rule does not authorize Application or Chain to
import the concrete Adapter; they own/use a Port and receive its implementation from startup.
Plugins import the curated facade from `app.sdk.network`. Do not use `requests`, `httpx`, or
`aiohttp` directly outside a separately reviewed SDK, streaming, or vendored transport boundary.
Plugins import the curated facade from `app.sdk.network`.
Direct `requests`/`httpx` clients, SDK transports, product streaming protocols, contained vendor
code, diagnostics and control-plane access are governed by the exact `direct_egress` facts and
manual policy. An approved fingerprint is containment for that exact source/binding/use identity
only; `owner: "$source"` names that source module and does not authorize another caller or
operation. New SDK roots must be added to the registry and receive explicit policy review.
Ordinary direct HTTP, RequestUtils Session bridges and Application-owned DNS I/O remain
`temporary_debt` with an empty target state. Runtime wildcard imports from registered egress roots
are forbidden.
`RequestUtils` handles:
- Proxy configuration (from `settings.PROXY_*`)
scripts/architecture/baseline.py
+8 -1
View File
@@ -13,10 +13,16 @@ from collections import defaultdict
from pathlib import Path
from typing import Any, Optional
try:
from scripts.architecture.egress import collect_direct_egress
except ModuleNotFoundError:
from egress import collect_direct_egress
PROJECT_ROOT = Path(__file__).resolve().parents[2]
APP_ROOT = PROJECT_ROOT / "app"
BASELINE_ROOT = PROJECT_ROOT / "tests" / "fixtures" / "architecture"
DEPENDENCY_BASELINE_PATH = BASELINE_ROOT / "dependency-baseline.json"
DEPENDENCY_POLICY_PATH = BASELINE_ROOT / "dependency-policy.json"
RUNTIME_BASELINE_PATH = BASELINE_ROOT / "runtime-contract-baseline.json"
TRANSACTION_BASELINE_PATH = BASELINE_ROOT / "transaction-debt-baseline.json"
CONFIGURATION_BASELINE_PATH = BASELINE_ROOT / "configuration-debt-baseline.json"
@@ -440,7 +446,7 @@ def collect_dependency_baseline() -> dict[str, Any]:
{edge["target"] for edge in direct_adapter_imports}
)
return {
"schema_version": 2,
"schema_version": 3,
"scope": "MoviePilot host app excluding app/plugins",
"module_count": len(modules),
"edge_count": len(edges),
@@ -473,6 +479,7 @@ def collect_dependency_baseline() -> dict[str, Any]:
"targets": direct_adapter_targets,
"edges": direct_adapter_imports,
},
"direct_egress": collect_direct_egress(modules),
"boundary_edges": collect_boundary_edges(graph, modules),
}
scripts/architecture/egress.py
+1053
View File
File diff suppressed because it is too large Load Diff
tests/fixtures/architecture/dependency-baseline.json
Vendored
+1263 -1
View File
File diff suppressed because it is too large Load Diff
tests/fixtures/architecture/dependency-policy.json
Vendored
+208 -1
View File
@@ -1,5 +1,5 @@
{
"schema_version": 2,
"schema_version": 3,
"scope": {
"dependency_kind": "static_runtime_imports",
"excluded_roots": [
@@ -213,5 +213,212 @@
}
],
"target_state": "empty"
},
"direct_egress": {
"scope": {
"root": "app",
"excluded_roots": ["app.plugins"],
"runtime_only": true,
"line_numbers": false
},
"registry": {
"raw_transports": [
"requests", "httpx", "httpx2", "aiohttp", "urllib.request", "urllib3",
"http.client", "websocket", "websockets", "curl_cffi", "cloudscraper",
"httplib2", "pycurl"
],
"network_sdks": [
"boto3", "botocore", "cloakbrowser", "ddgs", "discord", "docker",
"google.genai", "langchain_anthropic", "langchain_aws", "langchain_deepseek",
"langchain_google_genai", "langchain_openai", "lark_oapi", "openai", "oss2",
"plexapi", "pywebpush", "qbittorrentapi", "redis", "slack_bolt", "slack_sdk",
"smbclient", "smbprotocol", "telebot", "transmission_rpc"
],
"protocol_operations": [
"asyncio.open_connection", "socket.create_connection", "socket.getaddrinfo",
"socket.gethostbyname", "xmlrpc.client.ServerProxy"
],
"reexports": [
{
"source": "app.adapters.network.http.requests",
"target": "requests"
}
]
},
"groups": [
{
"classification": "temporary_debt",
"reason_code": "direct_http",
"tracking": "S2-L7",
"target_state": "empty",
"reason": "普通宿主 HTTP 与 SDK client factory 尚未收口；provider 的直接请求和注入 client 必须在同一迁移中处理。",
"facts": [
{"source": "app.agent.llm.provider", "target": "httpx", "kind": "raw_transport", "fingerprint": "a68ad216d86b4b33aa8c35572c26f7993245c0c12dd512e962bcacb46d7f15fc"},
{"source": "app.modules.filemanager.storages.alipan", "target": "requests", "kind": "raw_transport", "fingerprint": "82d84425f58ed9664a6e2a4ece53240198c95a0f9fee5d0afd0a11e3c459ecf6"},
{"source": "app.modules.filemanager.storages.u115", "target": "httpx", "kind": "raw_transport", "fingerprint": "f6678109963f9973d5d15e38e5f5e8dca01ec5e0117598ad1ac877b841ff7961"},
{"source": "app.modules.ugreen.api", "target": "requests", "kind": "raw_transport", "fingerprint": "a266dd2b74f216b289682e5d83c0b5a7d0d08b5447d79c165659b6d7302656d6"}
]
},
{
"classification": "temporary_debt",
"reason_code": "requestutils_session_bridge",
"tracking": "S2-L7",
"target_state": "empty",
"reason": "模块构造或接收 requests Session，并管理生命周期或桥接给 RequestUtils；迁移后不再直接拥有 transport。",
"facts": [
{"source": "app.modules.bangumi.bangumi", "target": "requests", "kind": "raw_transport", "fingerprint": "e5baf8a89b8661025169e9eec8d3c13d30660f63b35d94ec4896cd97e7fde3bf"},
{"source": "app.modules.douban.apiv2", "target": "requests", "kind": "raw_transport", "fingerprint": "6e871a27b88263676797edd65093d0b38db48f77fb5db2bb581b7b82e9b2977d"},
{"source": "app.modules.imdb.api", "target": "requests", "kind": "raw_transport", "fingerprint": "3d05532e62d2cddf3bb05495196176b76aa02216c2b0d868cf9281d2257d9627"},
{"source": "app.modules.indexer.parser", "target": "requests", "kind": "raw_transport", "fingerprint": "facdb091442bacba0fc0f631637643e95e18545bd7ab63f1d49d67a9a928d016"},
{"source": "app.modules.musicbrainz", "target": "requests", "kind": "raw_transport", "fingerprint": "c54a10dd672fb0ebb32a536ae42c221ad434313466cc108d43e433dd84cac9a5"},
{"source": "app.modules.plex.plex", "target": "requests", "kind": "raw_transport", "fingerprint": "93f20b9c01571553e9601f4b9f4ee9471073e65f938cb635320f72fa41ec3843"},
{"source": "app.modules.trimemedia.api", "target": "requests", "kind": "raw_transport", "fingerprint": "aecbe00404b2866c918c939bee5f1002045fc72aed4440c85e0811a4379913e4"}
]
},
{
"classification": "temporary_debt",
"reason_code": "direct_dns",
"tracking": "S2-L6",
"target_state": "empty",
"reason": "SSRF 校验仍在 Application 直接执行 DNS I/O；迁移到注入的解析 Port 后清零。",
"facts": [
{"source": "app.application.security.url", "target": "socket.getaddrinfo", "kind": "protocol_operation", "fingerprint": "89c563fdb640259584563fc718f7104a7b4dc6717e97b9c9e449f70f74a9532b"}
]
},
{
"classification": "approved_exception",
"reason_code": "canonical_transport",
"owner": "$source",
"reason": "精确 Adapter source 持有其 Redis、浏览器、HTTP 或 DNS transport；上层只能消费 Port 或统一 Facade。",
"facts": [
{"source": "app.adapters.cache.redis", "target": "redis", "kind": "network_sdk", "fingerprint": "9d455a5298d4373ff18d74a9d498a3dd797bcb5f5543af9776c0c741c30362c7"},
{"source": "app.adapters.network.browser", "target": "cloakbrowser", "kind": "network_sdk", "fingerprint": "15c1777b14eb9147d6cab9783f67577011714f9220600dda5db5269b59726173"},
{"source": "app.adapters.network.doh", "target": "socket.getaddrinfo", "kind": "protocol_operation", "fingerprint": "4ff03419dfacc6bf582b7d4421dd5a0666a63f8ca79be2b1e625f4c8f4c96b71"},
{"source": "app.adapters.network.doh", "target": "urllib.request", "kind": "raw_transport", "fingerprint": "6f5f5fd3da02a9e780ea5e7cc1e47bd962314a1a358f14b4ee698485f96ab52b"},
{"source": "app.adapters.network.http", "target": "httpx2", "kind": "raw_transport", "fingerprint": "a70799e9930ff79cd28ebed92836c8108cd2d18bc668e0081f68533d84f69d0d"},
{"source": "app.adapters.network.http", "target": "requests", "kind": "raw_transport", "fingerprint": "95fb4576cec363355f40aa1758d90fbb09242a2b1d029b7ab5ab65e2aaca12f8"},
{"source": "app.adapters.network.ip", "target": "socket.gethostbyname", "kind": "protocol_operation", "fingerprint": "a43e8969e2f26546fcf925b258728b309f1b9b966ff3c7b48a64273b8c82d048"}
]
},
{
"classification": "approved_exception",
"reason_code": "transport_configuration",
"owner": "$source",
"reason": "统一 HTTP Adapter 仅配置 urllib3 警告策略，不创建 urllib3 网络连接。",
"facts": [
{"source": "app.adapters.network.http", "target": "urllib3", "kind": "raw_transport", "fingerprint": "66cbd8ec4e7552bd458db0baada30f1953e6d0493793822d23a2199567bca98a"}
]
},
{
"classification": "approved_exception",
"reason_code": "sdk_transport",
"owner": "$source",
"reason": "命名 SDK 或显式注入的 SDK transport 由该精确 source 持有，不授权新增调用方。",
"facts": [
{"source": "app.agent.llm.capability", "target": "openai", "kind": "network_sdk", "fingerprint": "60160ab01b60be2c9794b9724a50b90f8875c6a10748ff9392a1e4bc7d48c334"},
{"source": "app.agent.llm.helper", "target": "google.genai", "kind": "network_sdk", "fingerprint": "d41688ecda4c5de296fabb0d0d25c79a7b0f29f67ccdd44569dc1110e1db9b86"},
{"source": "app.agent.llm.helper", "target": "httpx", "kind": "raw_transport", "fingerprint": "caeccd432f48d9a23948dd5b9220ddbebc57d2b2936bfa499fc7d4876e935ea8"},
{"source": "app.agent.llm.helper", "target": "langchain_anthropic", "kind": "network_sdk", "fingerprint": "c8d2c3166c4d42d61e7ef38a25080cac0f93422f47393df6dc19b329a7f72265"},
{"source": "app.agent.llm.helper", "target": "langchain_aws", "kind": "network_sdk", "fingerprint": "51dd02f5e2305cce4e1f6f29b4ed47edf54dde9f6557781178a6edf96bfeeb6b"},
{"source": "app.agent.llm.helper", "target": "langchain_deepseek", "kind": "network_sdk", "fingerprint": "bcf8b1ffae0995cfc33cfe26609cef8014cd2ce036b845c2a88c0cd878b1ebc4"},
{"source": "app.agent.llm.helper", "target": "langchain_google_genai", "kind": "network_sdk", "fingerprint": "79ae88c63908da9ea36dcfc23c4ad5345bc5d1977b80cc2e0b89292d063ed8ee"},
{"source": "app.agent.llm.helper", "target": "langchain_openai", "kind": "network_sdk", "fingerprint": "15775bf1694780d5cf516e6cc3d3b5174ebb25e3038433132d3071e6abceba52"},
{"source": "app.agent.llm.helper", "target": "openai", "kind": "network_sdk", "fingerprint": "dbebd5e9fa38b2a525a5a2e5a92d855ba9d05762410590bfb544e665fe38f620"},
{"source": "app.agent.llm.provider", "target": "boto3", "kind": "network_sdk", "fingerprint": "04ca4b5ecbda1531266d5f53817dde061dcfdd3293cabf9301fc8d8e4c281a88"},
{"source": "app.agent.llm.provider", "target": "botocore", "kind": "network_sdk", "fingerprint": "fbeaa702703665ca60233a153aade0da7e49dfb6fe31f72e5593c883fb7d3f11"},
{"source": "app.agent.llm.provider", "target": "google.genai", "kind": "network_sdk", "fingerprint": "b545cffb169b87f87606c17762b5d9da3f35372bd411c1f4ebba010b3d7c3ebb"},
{"source": "app.agent.llm.provider", "target": "openai", "kind": "network_sdk", "fingerprint": "ffac3b355eb640f6421299cf838cb4e24866a39afabd9dcfd8d0f340925dc1a8"},
{"source": "app.agent.tools.impl.search_web", "target": "ddgs", "kind": "network_sdk", "fingerprint": "377e73bb3be804b825d60ad6e792344ca633de107f8c4dbb45d50a34a4dfa04b"},
{"source": "app.api.endpoints.message", "target": "pywebpush", "kind": "network_sdk", "fingerprint": "7d83ca0dc89dfb6af1cdd7bc90d922105de94222e4e3a44bba4f53bbd97bc31b"},
{"source": "app.modules.discord.discord", "target": "discord", "kind": "network_sdk", "fingerprint": "e1486525fedbfe574aa47d60b483426c8c48e4e27fa35dbe34f6cd2f47d37c86"},
{"source": "app.modules.feishu.feishu", "target": "lark_oapi", "kind": "network_sdk", "fingerprint": "99d16968a59932b5980f6d42ea2bbc8fa35ccdb1a27933beec29b4bdc093d78b"},
{"source": "app.modules.filemanager.storages.smb", "target": "smbclient", "kind": "network_sdk", "fingerprint": "62e0585282ef206ac81b2bf9e93a58423fb058685edc0ef4ee592453bbf30376"},
{"source": "app.modules.filemanager.storages.u115", "target": "oss2", "kind": "network_sdk", "fingerprint": "f7b89c8ae6dad2603f0a9e0caaa159769aef5b3581d7e728f62445b979366eae"},
{"source": "app.modules.plex.plex", "target": "plexapi", "kind": "network_sdk", "fingerprint": "76c1334863dc6c6623ce6ad3415bec5ff92656f51c31e4d982285898f193e57e"},
{"source": "app.modules.qbittorrent.qbittorrent", "target": "qbittorrentapi", "kind": "network_sdk", "fingerprint": "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d"},
{"source": "app.modules.slack.slack", "target": "slack_bolt", "kind": "network_sdk", "fingerprint": "be5dcb032ece8d8627abeb243f98143aaf60f26751ab6e5e098813d4048419e6"},
{"source": "app.modules.slack.slack", "target": "slack_sdk", "kind": "network_sdk", "fingerprint": "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14"},
{"source": "app.modules.telegram.telegram", "target": "telebot", "kind": "network_sdk", "fingerprint": "78f5ab18bfd67ba4fa0f3c0fc4a1a561a7b1d9e81e335fb80de86edb480b84b8"},
{"source": "app.modules.transmission.transmission", "target": "transmission_rpc", "kind": "network_sdk", "fingerprint": "1652e661cb17dbadb039fdc4ab73d6d06e47eb118292ba3693313e45834959cf"},
{"source": "app.modules.webpush", "target": "pywebpush", "kind": "network_sdk", "fingerprint": "389c73b06150e3d5bcaf31f35a25178873d2ed38ef9a28354cb9bab691eeab76"}
]
},
{
"classification": "approved_exception",
"reason_code": "streaming_protocol",
"owner": "$source",
"reason": "产品长连接或 SCGI/XML-RPC transport 由该协议模块精确持有。",
"facts": [
{"source": "app.modules.qqbot.gateway", "target": "websocket", "kind": "raw_transport", "fingerprint": "6f6b7d61f3a95e620e67c450d544f0aa077f087a188d98f32e4501d94c6b37ae"},
{"source": "app.modules.rtorrent.rtorrent", "target": "socket.create_connection", "kind": "protocol_operation", "fingerprint": "878114d6b1bda091bd3b3d8aa819831d84381682feba78267592d2593295af90"},
{"source": "app.modules.rtorrent.rtorrent", "target": "xmlrpc.client.ServerProxy", "kind": "protocol_operation", "fingerprint": "2ac64b5670c930bc28bb2135cc4b73891c1cae01a74a9586f74e302878c45c85"},
{"source": "app.modules.wechat.wechatbot", "target": "websocket", "kind": "raw_transport", "fingerprint": "1bae78270eadce0571e2caaa111a5c0a9065ba2da97ebb26b8a5b76d3ed5eef6"}
]
},
{
"classification": "approved_exception",
"reason_code": "contained_vendor",
"owner": "$source",
"reason": "TMDB 移植客户端在 containment 内创建 Session，并把实际请求交给 RequestUtils。",
"facts": [
{"source": "app.modules.themoviedb.tmdbv3api.tmdb", "target": "requests", "kind": "raw_transport", "fingerprint": "d605eb176a203b3f4d205c5d183469b3002426682eb9cb4e7408bb6013652484"}
]
},
{
"classification": "approved_exception",
"reason_code": "local_control_plane",
"owner": "$source",
"reason": "CLI 只调用本机 MoviePilot 控制面，维持精确运维入口。",
"facts": [
{"source": "app.cli", "target": "urllib.request", "kind": "raw_transport", "fingerprint": "71b3e5be2a7d85fdc7a207d15c0e690d7ab96b335b882a3eaec03a1cb4ec1d1b"}
]
},
{
"classification": "approved_exception",
"reason_code": "diagnostic_probe",
"owner": "$source",
"reason": "doctor 按用户配置探测 PostgreSQL 与后端健康端点，可能指向远端，仅用于显式诊断。",
"facts": [
{"source": "app.doctor.checks", "target": "socket.create_connection", "kind": "protocol_operation", "fingerprint": "dfb34d4710353029dcc06e0e0ce3b299bae1a7e71a5743de1d253ac26380f9b6"},
{"source": "app.doctor.checks", "target": "urllib.request", "kind": "raw_transport", "fingerprint": "d8dd2279263ef58c3b4c59c70451533a4fc27d2ccd76960ca3aff29cd73f3354"}
]
},
{
"classification": "approved_exception",
"reason_code": "type_or_compat_only",
"owner": "$source",
"reason": "运行期 import 仅用于响应类型、协议解析或兼容补丁，uses 为空或不获取网络能力。",
"facts": [
{"source": "app.adapters.external.market", "target": "httpx2", "kind": "raw_transport", "fingerprint": "d4a648e8188818c0465013fc63cc3e49899da4df38541344303c251896564b1f"},
{"source": "app.adapters.external.market", "target": "requests", "kind": "raw_transport", "fingerprint": "ecc5368adfced20741e5ed8696008ea7555a4a81ceda6d7149e09f5f3d0ff7e3"},
{"source": "app.modules.douban.apiv2", "target": "httpx2", "kind": "raw_transport", "fingerprint": "edc337b12976d978cad3f83d0fa6faf71b2af2d814926dff890d5acc94f0a0a6"},
{"source": "app.modules.emby.emby", "target": "requests", "kind": "raw_transport", "fingerprint": "6c4c17fe170226ea1272f5859bc119c2775442cabe8ce2a9edcabaf9c37bec31"},
{"source": "app.modules.filemanager.storages.smb", "target": "smbprotocol", "kind": "network_sdk", "fingerprint": "f02686afd99c59820dffa7b4c2627a0be1ad62980c98a9cd33b7564697842b22"},
{"source": "app.modules.jellyfin.jellyfin", "target": "requests", "kind": "raw_transport", "fingerprint": "5c46d09ca9a4bcc0bae21ca5d554ef09baa3f901c562b27c5f5ee1439c7746b5"},
{"source": "app.modules.rtorrent.rtorrent", "target": "http.client", "kind": "raw_transport", "fingerprint": "6adc93b3bc479dfe81197554c7977930a60abc51714f14ef29298eb40730aaf7"},
{"source": "app.modules.telegram.compat", "target": "urllib3", "kind": "raw_transport", "fingerprint": "18862bbdb252b59573f57ea776b5d64bfb775e7739603fecd23cc8f2c38a7e0d"},
{"source": "app.modules.zspace.zspace", "target": "requests", "kind": "raw_transport", "fingerprint": "9df3fd27b9696d45a72e7c8f67b5a9ad79a7371d1fe690bbaa17485bd1960d51"},
{"source": "app.startup.lifecycle", "target": "urllib3", "kind": "raw_transport", "fingerprint": "cb6f0a314aeb1e2d3e76c240aa20460ac0c36d9f5c18c1a6ea3170f64dd3366b"}
]
},
{
"classification": "approved_exception",
"reason_code": "daemon_control_plane",
"owner": "$source",
"reason": "runtime state 连接配置指定的 Docker daemon 控制面，并由该进程能力 owner 收口。",
"facts": [
{"source": "app.runtime.state", "target": "docker", "kind": "network_sdk", "fingerprint": "20a91ec521f7dfe6a0153dfd8ea49c4bac7f0a16b55f1c0655dfb33f54a01215"}
]
},
{
"classification": "approved_exception",
"reason_code": "test_network_guard",
"owner": "$source",
"reason": "测试守卫包装 getaddrinfo 以拒绝真实外联，不执行新的 DNS 解析策略。",
"facts": [
{"source": "app.testing.network_guard", "target": "socket.getaddrinfo", "kind": "protocol_operation", "fingerprint": "2518de211c9ba32ccbc004cf58fe9c98837fe8ac95fe8ffb6ea82897b28d753f"}
]
}
]
}
}
tests/fixtures/architecture/mypy-baseline.json
Vendored
+1 -2
View File
@@ -1462,8 +1462,7 @@
},
"app/chain/media.py": {
"arg-type": 34,
"assignment": 16,
"call-overload": 1,
"assignment": 15,
"comparison-overlap": 3,
"no-any-return": 3,
"no-untyped-def": 3,
tests/test_architecture_adapter_imports.py
+1 -1
View File
@@ -255,7 +255,7 @@ def test_current_direct_adapter_imports_match_temporary_debt_policy() -> None:
adapter_policy = policy["direct_adapter_imports"]
entries = adapter_policy["entries"]
assert policy["schema_version"] == 2
assert policy["schema_version"] == 3
assert adapter_policy["classification"] == "temporary_debt"
assert adapter_policy["target_state"] == "empty"
assert _adapter_policy_scope_errors(adapter_policy["scope"], contract["scope"]) == []
tests/test_architecture_baseline_cli.py
+24
View File
@@ -451,7 +451,10 @@ def test_architecture_write_host_only_updates_host_files(
dependency_path = tmp_path / "dependency.json"
runtime_path = tmp_path / "runtime.json"
transaction_path = tmp_path / "transaction.json"
configuration_path = tmp_path / "configuration.json"
policy_path = tmp_path / "dependency-policy.json"
plugin_path = tmp_path / "plugin.json"
policy_path.write_text('{"manual": true}\n', encoding="utf-8")
monkeypatch.setattr(
architecture_baseline,
"DEPENDENCY_BASELINE_PATH",
@@ -467,6 +470,16 @@ def test_architecture_write_host_only_updates_host_files(
"TRANSACTION_BASELINE_PATH",
transaction_path,
)
monkeypatch.setattr(
architecture_baseline,
"CONFIGURATION_BASELINE_PATH",
configuration_path,
)
monkeypatch.setattr(
architecture_baseline,
"DEPENDENCY_POLICY_PATH",
policy_path,
)
monkeypatch.setattr(architecture_baseline, "PLUGIN_BASELINE_PATH", plugin_path)
monkeypatch.setattr(
architecture_baseline,
@@ -483,6 +496,11 @@ def test_architecture_write_host_only_updates_host_files(
"collect_transaction_debt_baseline",
lambda: {"scope": "host-transaction"},
)
monkeypatch.setattr(
architecture_baseline,
"collect_configuration_debt_baseline",
lambda: {"scope": "host-configuration"},
)
assert architecture_baseline.main(["--write-host"]) == 0
@@ -491,12 +509,18 @@ def test_architecture_write_host_only_updates_host_files(
assert json.loads(transaction_path.read_text()) == {
"scope": "host-transaction"
}
assert json.loads(configuration_path.read_text()) == {
"scope": "host-configuration"
}
assert json.loads(policy_path.read_text()) == {"manual": True}
assert not plugin_path.exists()
output = capsys.readouterr().out
assert "即将写入" in output
assert "dependency.json" in output
assert "runtime.json" in output
assert "transaction.json" in output
assert "configuration.json" in output
assert "dependency-policy.json" not in output
def test_architecture_write_plugins_only_updates_plugin_file(
tests/test_architecture_ci.py
+1
View File
@@ -32,6 +32,7 @@ def test_unit_test_workflow_has_independent_host_architecture_gate():
assert "push" in workflow["on"]
assert "tests/test_architecture_dependencies.py" in commands
assert "tests/test_architecture_adapter_imports.py" in commands
assert "tests/test_architecture_egress.py" in commands
assert "tests/test_architecture_contract_baseline.py" in commands
assert "scripts/architecture/baseline.py --check-host" in commands
assert "scripts/architecture/ruff_ratchet.py" in commands
tests/test_architecture_contract_baseline.py
+14 -1
View File
@@ -64,7 +64,7 @@ def test_dependency_baseline_records_nonempty_host_graph() -> None:
baseline_path = BASELINE_ROOT / "dependency-baseline.json"
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
assert baseline["schema_version"] == 2
assert baseline["schema_version"] == 3
assert baseline["module_count"] == len(baseline["modules"])
assert baseline["edge_count"] == len(baseline["edges"])
assert baseline["module_count"] > 0
@@ -78,6 +78,19 @@ def test_dependency_baseline_records_nonempty_host_graph() -> None:
}
assert direct_imports["source_count"] == len(direct_imports["sources"])
assert direct_imports["target_count"] == len(direct_imports["targets"])
direct_egress = baseline["direct_egress"]
assert direct_egress["count"] == len(direct_egress["entries"])
assert sum(direct_egress["counts_by_kind"].values()) == direct_egress["count"]
assert set(direct_egress["counts_by_kind"]) == {
"raw_transport",
"network_sdk",
"protocol_operation",
}
assert set(direct_egress["application_chain_counts"]) == {
"app.application",
"app.chain",
}
assert all("line" not in entry for entry in direct_egress["entries"])
def test_official_discovery_plugins_explicitly_keep_host_page_envelope():
tests/test_architecture_dependencies.py
+1 -1
View File
@@ -1087,7 +1087,7 @@ def test_complete_host_sccs_match_reviewed_policy() -> None:
policy = _load_dependency_policy()
entries = policy["allowed_sccs"]
assert policy["schema_version"] == 2
assert policy["schema_version"] == 3
assert policy["scope"] == {
"dependency_kind": "static_runtime_imports",
"excluded_roots": ["app/plugins"],
tests/test_architecture_egress.py
+925
View File
@@ -0,0 +1,925 @@
"""宿主 direct egress 事实与人工政策契约。"""
import json
from pathlib import Path
from scripts.architecture.baseline import collect_dependency_baseline
from scripts.architecture.egress import collect_direct_egress
PROJECT_ROOT = Path(__file__).parents[1]
DEPENDENCY_POLICY_PATH = (
PROJECT_ROOT / "tests" / "fixtures" / "architecture" / "dependency-policy.json"
)
FROZEN_EGRESS_EDGES_BY_REASON = {
"direct_http": {
("app.agent.llm.provider", "httpx"),
("app.modules.filemanager.storages.alipan", "requests"),
("app.modules.filemanager.storages.u115", "httpx"),
("app.modules.ugreen.api", "requests"),
},
"requestutils_session_bridge": {
("app.modules.bangumi.bangumi", "requests"),
("app.modules.douban.apiv2", "requests"),
("app.modules.imdb.api", "requests"),
("app.modules.indexer.parser", "requests"),
("app.modules.musicbrainz", "requests"),
("app.modules.plex.plex", "requests"),
("app.modules.trimemedia.api", "requests"),
},
"direct_dns": {
("app.application.security.url", "socket.getaddrinfo"),
},
"canonical_transport": {
("app.adapters.cache.redis", "redis"),
("app.adapters.network.browser", "cloakbrowser"),
("app.adapters.network.doh", "socket.getaddrinfo"),
("app.adapters.network.doh", "urllib.request"),
("app.adapters.network.http", "httpx2"),
("app.adapters.network.http", "requests"),
("app.adapters.network.ip", "socket.gethostbyname"),
},
"transport_configuration": {
("app.adapters.network.http", "urllib3"),
},
"sdk_transport": {
("app.agent.llm.capability", "openai"),
("app.agent.llm.helper", "google.genai"),
("app.agent.llm.helper", "httpx"),
("app.agent.llm.helper", "langchain_anthropic"),
("app.agent.llm.helper", "langchain_aws"),
("app.agent.llm.helper", "langchain_deepseek"),
("app.agent.llm.helper", "langchain_google_genai"),
("app.agent.llm.helper", "langchain_openai"),
("app.agent.llm.helper", "openai"),
("app.agent.llm.provider", "boto3"),
("app.agent.llm.provider", "botocore"),
("app.agent.llm.provider", "google.genai"),
("app.agent.llm.provider", "openai"),
("app.agent.tools.impl.search_web", "ddgs"),
("app.api.endpoints.message", "pywebpush"),
("app.modules.discord.discord", "discord"),
("app.modules.feishu.feishu", "lark_oapi"),
("app.modules.filemanager.storages.smb", "smbclient"),
("app.modules.filemanager.storages.u115", "oss2"),
("app.modules.plex.plex", "plexapi"),
("app.modules.qbittorrent.qbittorrent", "qbittorrentapi"),
("app.modules.slack.slack", "slack_bolt"),
("app.modules.slack.slack", "slack_sdk"),
("app.modules.telegram.telegram", "telebot"),
("app.modules.transmission.transmission", "transmission_rpc"),
("app.modules.webpush", "pywebpush"),
},
"streaming_protocol": {
("app.modules.qqbot.gateway", "websocket"),
("app.modules.rtorrent.rtorrent", "socket.create_connection"),
("app.modules.rtorrent.rtorrent", "xmlrpc.client.ServerProxy"),
("app.modules.wechat.wechatbot", "websocket"),
},
"contained_vendor": {
("app.modules.themoviedb.tmdbv3api.tmdb", "requests"),
},
"local_control_plane": {
("app.cli", "urllib.request"),
},
"diagnostic_probe": {
("app.doctor.checks", "socket.create_connection"),
("app.doctor.checks", "urllib.request"),
},
"type_or_compat_only": {
("app.adapters.external.market", "httpx2"),
("app.adapters.external.market", "requests"),
("app.modules.douban.apiv2", "httpx2"),
("app.modules.emby.emby", "requests"),
("app.modules.filemanager.storages.smb", "smbprotocol"),
("app.modules.jellyfin.jellyfin", "requests"),
("app.modules.rtorrent.rtorrent", "http.client"),
("app.modules.telegram.compat", "urllib3"),
("app.modules.zspace.zspace", "requests"),
("app.startup.lifecycle", "urllib3"),
},
"daemon_control_plane": {
("app.runtime.state", "docker"),
},
"test_network_guard": {
("app.testing.network_guard", "socket.getaddrinfo"),
},
}
FROZEN_EGRESS_REASON_BY_EDGE = {
edge: reason
for reason, edges in FROZEN_EGRESS_EDGES_BY_REASON.items()
for edge in edges
}
FROZEN_EGRESS_FINGERPRINT_BY_EDGE = {
("app.adapters.cache.redis", "redis"): "9d455a5298d4373ff18d74a9d498a3dd797bcb5f5543af9776c0c741c30362c7",
("app.adapters.external.market", "httpx2"): "d4a648e8188818c0465013fc63cc3e49899da4df38541344303c251896564b1f",
("app.adapters.external.market", "requests"): "ecc5368adfced20741e5ed8696008ea7555a4a81ceda6d7149e09f5f3d0ff7e3",
("app.adapters.network.browser", "cloakbrowser"): "15c1777b14eb9147d6cab9783f67577011714f9220600dda5db5269b59726173",
("app.adapters.network.doh", "socket.getaddrinfo"): "4ff03419dfacc6bf582b7d4421dd5a0666a63f8ca79be2b1e625f4c8f4c96b71",
("app.adapters.network.doh", "urllib.request"): "6f5f5fd3da02a9e780ea5e7cc1e47bd962314a1a358f14b4ee698485f96ab52b",
("app.adapters.network.http", "httpx2"): "a70799e9930ff79cd28ebed92836c8108cd2d18bc668e0081f68533d84f69d0d",
("app.adapters.network.http", "requests"): "95fb4576cec363355f40aa1758d90fbb09242a2b1d029b7ab5ab65e2aaca12f8",
("app.adapters.network.http", "urllib3"): "66cbd8ec4e7552bd458db0baada30f1953e6d0493793822d23a2199567bca98a",
("app.adapters.network.ip", "socket.gethostbyname"): "a43e8969e2f26546fcf925b258728b309f1b9b966ff3c7b48a64273b8c82d048",
("app.agent.llm.capability", "openai"): "60160ab01b60be2c9794b9724a50b90f8875c6a10748ff9392a1e4bc7d48c334",
("app.agent.llm.helper", "google.genai"): "d41688ecda4c5de296fabb0d0d25c79a7b0f29f67ccdd44569dc1110e1db9b86",
("app.agent.llm.helper", "httpx"): "caeccd432f48d9a23948dd5b9220ddbebc57d2b2936bfa499fc7d4876e935ea8",
("app.agent.llm.helper", "langchain_anthropic"): "c8d2c3166c4d42d61e7ef38a25080cac0f93422f47393df6dc19b329a7f72265",
("app.agent.llm.helper", "langchain_aws"): "51dd02f5e2305cce4e1f6f29b4ed47edf54dde9f6557781178a6edf96bfeeb6b",
("app.agent.llm.helper", "langchain_deepseek"): "bcf8b1ffae0995cfc33cfe26609cef8014cd2ce036b845c2a88c0cd878b1ebc4",
("app.agent.llm.helper", "langchain_google_genai"): "79ae88c63908da9ea36dcfc23c4ad5345bc5d1977b80cc2e0b89292d063ed8ee",
("app.agent.llm.helper", "langchain_openai"): "15775bf1694780d5cf516e6cc3d3b5174ebb25e3038433132d3071e6abceba52",
("app.agent.llm.helper", "openai"): "dbebd5e9fa38b2a525a5a2e5a92d855ba9d05762410590bfb544e665fe38f620",
("app.agent.llm.provider", "boto3"): "04ca4b5ecbda1531266d5f53817dde061dcfdd3293cabf9301fc8d8e4c281a88",
("app.agent.llm.provider", "botocore"): "fbeaa702703665ca60233a153aade0da7e49dfb6fe31f72e5593c883fb7d3f11",
("app.agent.llm.provider", "google.genai"): "b545cffb169b87f87606c17762b5d9da3f35372bd411c1f4ebba010b3d7c3ebb",
("app.agent.llm.provider", "httpx"): "a68ad216d86b4b33aa8c35572c26f7993245c0c12dd512e962bcacb46d7f15fc",
("app.agent.llm.provider", "openai"): "ffac3b355eb640f6421299cf838cb4e24866a39afabd9dcfd8d0f340925dc1a8",
("app.agent.tools.impl.search_web", "ddgs"): "377e73bb3be804b825d60ad6e792344ca633de107f8c4dbb45d50a34a4dfa04b",
("app.api.endpoints.message", "pywebpush"): "7d83ca0dc89dfb6af1cdd7bc90d922105de94222e4e3a44bba4f53bbd97bc31b",
("app.application.security.url", "socket.getaddrinfo"): "89c563fdb640259584563fc718f7104a7b4dc6717e97b9c9e449f70f74a9532b",
("app.cli", "urllib.request"): "71b3e5be2a7d85fdc7a207d15c0e690d7ab96b335b882a3eaec03a1cb4ec1d1b",
("app.doctor.checks", "socket.create_connection"): "dfb34d4710353029dcc06e0e0ce3b299bae1a7e71a5743de1d253ac26380f9b6",
("app.doctor.checks", "urllib.request"): "d8dd2279263ef58c3b4c59c70451533a4fc27d2ccd76960ca3aff29cd73f3354",
("app.modules.bangumi.bangumi", "requests"): "e5baf8a89b8661025169e9eec8d3c13d30660f63b35d94ec4896cd97e7fde3bf",
("app.modules.discord.discord", "discord"): "e1486525fedbfe574aa47d60b483426c8c48e4e27fa35dbe34f6cd2f47d37c86",
("app.modules.douban.apiv2", "httpx2"): "edc337b12976d978cad3f83d0fa6faf71b2af2d814926dff890d5acc94f0a0a6",
("app.modules.douban.apiv2", "requests"): "6e871a27b88263676797edd65093d0b38db48f77fb5db2bb581b7b82e9b2977d",
("app.modules.emby.emby", "requests"): "6c4c17fe170226ea1272f5859bc119c2775442cabe8ce2a9edcabaf9c37bec31",
("app.modules.feishu.feishu", "lark_oapi"): "99d16968a59932b5980f6d42ea2bbc8fa35ccdb1a27933beec29b4bdc093d78b",
("app.modules.filemanager.storages.alipan", "requests"): "82d84425f58ed9664a6e2a4ece53240198c95a0f9fee5d0afd0a11e3c459ecf6",
("app.modules.filemanager.storages.smb", "smbclient"): "62e0585282ef206ac81b2bf9e93a58423fb058685edc0ef4ee592453bbf30376",
("app.modules.filemanager.storages.smb", "smbprotocol"): "f02686afd99c59820dffa7b4c2627a0be1ad62980c98a9cd33b7564697842b22",
("app.modules.filemanager.storages.u115", "httpx"): "f6678109963f9973d5d15e38e5f5e8dca01ec5e0117598ad1ac877b841ff7961",
("app.modules.filemanager.storages.u115", "oss2"): "f7b89c8ae6dad2603f0a9e0caaa159769aef5b3581d7e728f62445b979366eae",
("app.modules.imdb.api", "requests"): "3d05532e62d2cddf3bb05495196176b76aa02216c2b0d868cf9281d2257d9627",
("app.modules.indexer.parser", "requests"): "facdb091442bacba0fc0f631637643e95e18545bd7ab63f1d49d67a9a928d016",
("app.modules.jellyfin.jellyfin", "requests"): "5c46d09ca9a4bcc0bae21ca5d554ef09baa3f901c562b27c5f5ee1439c7746b5",
("app.modules.musicbrainz", "requests"): "c54a10dd672fb0ebb32a536ae42c221ad434313466cc108d43e433dd84cac9a5",
("app.modules.plex.plex", "plexapi"): "76c1334863dc6c6623ce6ad3415bec5ff92656f51c31e4d982285898f193e57e",
("app.modules.plex.plex", "requests"): "93f20b9c01571553e9601f4b9f4ee9471073e65f938cb635320f72fa41ec3843",
("app.modules.qbittorrent.qbittorrent", "qbittorrentapi"): "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d",
("app.modules.qqbot.gateway", "websocket"): "6f6b7d61f3a95e620e67c450d544f0aa077f087a188d98f32e4501d94c6b37ae",
("app.modules.rtorrent.rtorrent", "http.client"): "6adc93b3bc479dfe81197554c7977930a60abc51714f14ef29298eb40730aaf7",
("app.modules.rtorrent.rtorrent", "socket.create_connection"): "878114d6b1bda091bd3b3d8aa819831d84381682feba78267592d2593295af90",
("app.modules.rtorrent.rtorrent", "xmlrpc.client.ServerProxy"): "2ac64b5670c930bc28bb2135cc4b73891c1cae01a74a9586f74e302878c45c85",
("app.modules.slack.slack", "slack_bolt"): "be5dcb032ece8d8627abeb243f98143aaf60f26751ab6e5e098813d4048419e6",
("app.modules.slack.slack", "slack_sdk"): "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14",
("app.modules.telegram.compat", "urllib3"): "18862bbdb252b59573f57ea776b5d64bfb775e7739603fecd23cc8f2c38a7e0d",
("app.modules.telegram.telegram", "telebot"): "78f5ab18bfd67ba4fa0f3c0fc4a1a561a7b1d9e81e335fb80de86edb480b84b8",
("app.modules.themoviedb.tmdbv3api.tmdb", "requests"): "d605eb176a203b3f4d205c5d183469b3002426682eb9cb4e7408bb6013652484",
("app.modules.transmission.transmission", "transmission_rpc"): "1652e661cb17dbadb039fdc4ab73d6d06e47eb118292ba3693313e45834959cf",
("app.modules.trimemedia.api", "requests"): "aecbe00404b2866c918c939bee5f1002045fc72aed4440c85e0811a4379913e4",
("app.modules.ugreen.api", "requests"): "a266dd2b74f216b289682e5d83c0b5a7d0d08b5447d79c165659b6d7302656d6",
("app.modules.webpush", "pywebpush"): "389c73b06150e3d5bcaf31f35a25178873d2ed38ef9a28354cb9bab691eeab76",
("app.modules.wechat.wechatbot", "websocket"): "1bae78270eadce0571e2caaa111a5c0a9065ba2da97ebb26b8a5b76d3ed5eef6",
("app.modules.zspace.zspace", "requests"): "9df3fd27b9696d45a72e7c8f67b5a9ad79a7371d1fe690bbaa17485bd1960d51",
("app.runtime.state", "docker"): "20a91ec521f7dfe6a0153dfd8ea49c4bac7f0a16b55f1c0655dfb33f54a01215",
("app.startup.lifecycle", "urllib3"): "cb6f0a314aeb1e2d3e76c240aa20460ac0c36d9f5c18c1a6ea3170f64dd3366b",
("app.testing.network_guard", "socket.getaddrinfo"): "2518de211c9ba32ccbc004cf58fe9c98837fe8ac95fe8ffb6ea82897b28d753f",
}
def _module(tmp_path: Path, name: str, content: str) -> dict[str, Path]:
"""创建供 direct egress collector 使用的独立模块。"""
path = tmp_path / f"{name.replace('.', '_')}.py"
path.write_text(content, encoding="utf-8")
return {name: path}
def _entries_by_target(value: dict[str, object]) -> dict[str, dict[str, object]]:
"""把单 source 合成事实按 target 建索引。"""
return {entry["target"]: entry for entry in value["entries"]}
def _policy_facts(groups: list[dict[str, object]]) -> list[dict[str, str]]:
"""展开按语义分组保存的人工 egress fact 引用。"""
return [fact for group in groups for fact in group["facts"]]
def _policy_differences(
facts: list[dict[str, object]],
reviewed: list[dict[str, str]],
) -> tuple[list[str], list[str]]:
"""按完整事实指纹返回未审查事实和陈旧人工 policy。"""
actual = {str(fact["fingerprint"]) for fact in facts}
expected = {fact["fingerprint"] for fact in reviewed}
return sorted(actual - expected), sorted(expected - actual)
def _classification_differences(
facts: list[dict[str, object]],
groups: list[dict[str, object]],
) -> tuple[list[tuple[str, str]], list[tuple[str, str]], list[tuple[str, str]]]:
"""返回新增边、陈旧分类和与初始语义不符的分类。"""
actual = {(str(fact["source"]), str(fact["target"])) for fact in facts}
reviewed = {
(str(fact["source"]), str(fact["target"])): str(group["reason_code"])
for group in groups
for fact in group["facts"]
}
unexpected = sorted(actual - FROZEN_EGRESS_REASON_BY_EDGE.keys())
stale = sorted(reviewed.keys() - actual)
misclassified = sorted(
edge
for edge in actual & reviewed.keys() & FROZEN_EGRESS_REASON_BY_EDGE.keys()
if reviewed[edge] != FROZEN_EGRESS_REASON_BY_EDGE[edge]
)
return unexpected, stale, misclassified
def _fingerprint_differences(
facts: list[dict[str, object]],
) -> list[tuple[str, str]]:
"""返回调用面已超过初始上界的 direct egress 边。"""
actual = {
(str(fact["source"]), str(fact["target"])): str(fact["fingerprint"])
for fact in facts
}
return sorted(
edge
for edge, fingerprint in actual.items()
if FROZEN_EGRESS_FINGERPRINT_BY_EDGE.get(edge) != fingerprint
)
def test_egress_collector_distinguishes_bindings_and_runtime_uses(
tmp_path: Path,
) -> None:
"""类型导入、真实调用、lazy import、re-export 与 SDK 必须可区分。"""
modules = _module(
tmp_path,
"app.sample",
'''
from typing import TYPE_CHECKING
import requests as req
from requests import Response
import httpx
from urllib import request as urlrequest
from app.adapters.network.http import requests as bridged
from google import genai
if TYPE_CHECKING:
import cloudscraper
EXAMPLE = "requests.get('https://docstring.invalid')"
def run():
import aiohttp
session = req.Session()
session.get("https://example.com")
client = httpx.AsyncClient()
client.stream("GET", "https://example.com")
urlrequest.urlopen("https://example.com")
bridged.Session()
genai.Client()
aiohttp.ClientSession()
unknown.get("https://ignored.example")
''',
)
value = collect_direct_egress(modules)
entries = _entries_by_target(value)
assert set(entries) == {
"aiohttp",
"google.genai",
"httpx",
"requests",
"urllib.request",
}
assert entries["requests"]["bindings"] == [
"from:requests.Response as Response",
"import:requests as req",
"reexport:app.adapters.network.http.requests as bridged",
]
assert entries["requests"]["uses"] == [
"run|call:Session",
"run|call:Session",
"run|call:get",
]
assert entries["httpx"]["uses"] == [
"run|call:AsyncClient",
"run|call:stream",
]
assert entries["urllib.request"]["uses"] == ["run|call:urlopen"]
assert entries["google.genai"]["kind"] == "network_sdk"
assert entries["aiohttp"]["uses"] == ["run|call:ClientSession"]
assert all("line" not in entry for entry in value["entries"])
def test_egress_collector_preserves_type_only_runtime_import(tmp_path: Path) -> None:
"""用于响应类型的运行期 import 必须保留，但不能伪报为外呼操作。"""
value = collect_direct_egress(
_module(tmp_path, "app.sample", "from requests import Response\n")
)
assert value["entries"] == [
{
"source": "app.sample",
"target": "requests",
"kind": "raw_transport",
"bindings": ["from:requests.Response as Response"],
"uses": [],
"fingerprint": value["entries"][0]["fingerprint"],
}
]
def test_egress_collector_tracks_aliases_contexts_and_protocol_operations(
tmp_path: Path,
) -> None:
"""Client 别名、上下文实例、socket 与 XML-RPC 获取都进入稳定 uses。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import httpx
import requests
import socket
import xmlrpc.client
from urllib.parse import urlparse
def sync_call(flag):
client_cls = requests.Session if flag else requests.Session
client = client_cls()
client.post("https://example.com")
socket.create_connection(("localhost", 1))
xmlrpc.client.ServerProxy("http://localhost")
urlparse("https://not-egress.example")
async def async_call():
async with httpx.AsyncClient() as client:
await client.get("https://example.com")
''',
)
)
entries = _entries_by_target(value)
assert entries["requests"]["uses"] == [
"sync_call|call:__call__",
"sync_call|call:post",
]
assert entries["httpx"]["uses"] == [
"async_call|call:AsyncClient",
"async_call|call:get",
]
assert entries["socket.create_connection"]["kind"] == "protocol_operation"
assert entries["xmlrpc.client.ServerProxy"]["kind"] == "protocol_operation"
assert "urllib.request" not in entries
def test_egress_collector_keeps_lexical_scopes_and_merges_branches(
tmp_path: Path,
) -> None:
"""函数 lazy import 不得泄漏，分支与 self client 必须保留可证明来源。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import requests
def first():
import httpx as client
client.get("https://example.com")
def second(client):
client.get("https://ignored.example")
def branch(flag):
if flag:
requester = requests.request
else:
requester = unknown
requester("GET", "https://example.com")
class Service:
def __init__(self):
self.client = requests.Session()
def run(self):
self.client.get("https://example.com")
''',
)
)
entries = _entries_by_target(value)
assert entries["httpx"]["uses"] == ["first|call:get"]
assert entries["requests"]["uses"] == [
"Service.__init__|call:Session",
"Service.run|call:get",
"branch|call:request",
]
def test_egress_collector_preserves_wildcard_chains_and_duplicate_calls(
tmp_path: Path,
) -> None:
"""高风险 wildcard、链式 client 和重复操作都必须改变事实 identity。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
from requests import *
import requests
def run():
requests.Session().get("https://example.com/one")
requests.get("https://example.com/two")
requests.get("https://example.com/three")
''',
)
)
entry = _entries_by_target(value)["requests"]
assert entry["bindings"] == [
"from:requests.* as *",
"import:requests as requests",
]
assert entry["uses"] == [
"run|call:Session",
"run|call:get",
"run|call:get",
"run|call:get",
]
def test_egress_collector_tracks_annotations_class_order_and_union_branches(
tmp_path: Path,
) -> None:
"""注入参数、后置初始化与双 transport 分支都必须保留 operation。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import httpx
import requests
async def injected(client: httpx.AsyncClient):
await client.get("https://example.com")
class Service:
def run(self):
self.client.post("https://example.com")
def __init__(self):
try:
self.client = requests.Session()
except Exception:
self.client = None
def branch(flag):
if flag:
client = requests.Session()
else:
client = httpx.Client()
client.get("https://example.com")
''',
)
)
entries = _entries_by_target(value)
assert entries["requests"]["uses"] == [
"Service.__init__|call:Session",
"Service.run|call:post",
"branch|call:Session",
"branch|call:get",
]
assert entries["httpx"]["uses"] == [
"branch|call:Client",
"branch|call:get",
"injected|call:get",
]
def test_egress_collector_tracks_dynamic_and_late_bound_modules(
tmp_path: Path,
) -> None:
"""注册目标的字面量动态导入与后置模块别名不能绕过门禁。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import importlib
import requests
def late_bound():
transport.post("https://example.com")
def dynamic():
importlib.import_module("httpx").get("https://example.com")
def builtin_dynamic():
__import__("httpx").post("https://example.com")
transport = requests
''',
)
)
entries = _entries_by_target(value)
assert entries["requests"]["uses"] == ["late_bound|call:post"]
assert entries["httpx"]["bindings"] == [
"dynamic:__import__(httpx)",
"dynamic:importlib.import_module(httpx)"
]
assert entries["httpx"]["uses"] == [
"builtin_dynamic|call:dynamic-import",
"builtin_dynamic|call:post",
"dynamic|call:dynamic-import",
"dynamic|call:get",
]
def test_egress_collector_respects_local_shadow_scopes(tmp_path: Path) -> None:
"""lambda、for、comprehension 与 except-as 局部名不得污染 import provenance。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import requests
def run():
for requests in []:
requests.get("https://ignored.example")
[requests.get("https://ignored.example") for requests in []]
(lambda requests: requests.get("https://ignored.example"))(object())
try:
raise RuntimeError
except RuntimeError as requests:
requests.get("https://ignored.example")
''',
)
)
assert _entries_by_target(value)["requests"]["uses"] == []
def test_egress_collector_stops_at_response_operations(tmp_path: Path) -> None:
"""直接请求进入事实，但响应 json/text 解析不是新的 egress operation。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import requests
def run():
requests.get("https://example.com").json()
''',
)
)
assert _entries_by_target(value)["requests"]["uses"] == ["run|call:get"]
def test_egress_collector_tracks_local_wrapper_injections(tmp_path: Path) -> None:
"""本地包装函数接收的可证明 client 必须传播，未知调用不得伪造来源。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import httpx
import requests
def helper(client):
client.get("https://example.com")
def run():
requests_client = requests.Session()
httpx_client = httpx.Client()
helper(requests_client)
helper(httpx_client)
helper(object())
''',
)
)
entries = _entries_by_target(value)
assert entries["requests"]["uses"] == [
"helper|call:get",
"run|call:Session",
]
assert entries["httpx"]["uses"] == [
"helper|call:get",
"run|call:Client",
]
def test_egress_collector_handles_runtime_else_and_final_module_binding(
tmp_path: Path,
) -> None:
"""TYPE_CHECKING else 是运行期路径，函数使用模块最终绑定。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import httpx as transport
else:
import requests as transport
captured = transport
transport = object()
def run():
captured.get("https://example.com")
transport.get("https://ignored.example")
''',
)
)
assert _entries_by_target(value)["requests"]["uses"] == ["run|call:get"]
def test_egress_collector_merges_class_and_zero_iteration_paths(
tmp_path: Path,
) -> None:
"""类 client 清空与零次循环都不得抹掉已证明的可达能力。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import requests
class Service:
def connect(self):
client = requests.Session()
self.client = client
def close(self):
self.client = None
def run(self):
self.client.get("https://example.com/class")
def loop():
client = requests.Session()
for client in []:
pass
client.post("https://example.com/loop")
''',
)
)
assert _entries_by_target(value)["requests"]["uses"] == [
"Service.connect|call:Session",
"Service.run|call:get",
"loop|call:Session",
"loop|call:post",
]
def test_egress_collector_preserves_annotated_optional_client(tmp_path: Path) -> None:
"""None 默认值不得抹掉注解 client 经实例属性传递的真实调用。"""
value = collect_direct_egress(
_module(
tmp_path,
"app.sample",
'''
import httpx
class Service:
def __init__(self, client: httpx.AsyncClient = None):
self.client = client
async def run(self):
await self.client.get("https://example.com")
''',
)
)
assert _entries_by_target(value)["httpx"]["uses"] == [
"Service.run|call:get"
]
def test_current_egress_facts_are_complete_and_self_consistent() -> None:
"""当前宿主 egress identity 必须完整、可收缩、排序且统计自洽。"""
value = collect_dependency_baseline()["direct_egress"]
entries = value["entries"]
application_entries = [
entry
for entry in entries
if str(entry["source"]).startswith("app.application.")
]
chain_entries = [
entry
for entry in entries
if str(entry["source"]).startswith("app.chain.")
]
assert value["count"] == len(entries)
assert sum(value["counts_by_kind"].values()) == value["count"]
assert set(value["counts_by_kind"]) == {
"raw_transport",
"network_sdk",
"protocol_operation",
}
assert value["application_chain_counts"] == {
"app.application": len(application_entries),
"app.chain": len(chain_entries),
}
assert {
(entry["source"], entry["target"])
for entry in application_entries
} <= FROZEN_EGRESS_EDGES_BY_REASON["direct_dns"]
assert chain_entries == []
assert entries == sorted(
entries,
key=lambda entry: (entry["source"], entry["target"], entry["kind"]),
)
assert all(entry["target"] != "aiohttp" for entry in entries)
assert all(len(entry["fingerprint"]) == 64 for entry in entries)
assert all(set(entry) == {
"source",
"target",
"kind",
"bindings",
"uses",
"fingerprint",
} for entry in entries)
assert all(
"*" not in binding
for entry in entries
for binding in entry["bindings"]
)
assert '"line"' not in json.dumps(value)
def test_current_egress_facts_match_exact_policy() -> None:
"""现存事实必须逐条分类，且 registry/scope 不得与 collector 漂移。"""
facts = collect_dependency_baseline()["direct_egress"]
policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8"))
egress_policy = policy["direct_egress"]
groups = egress_policy["groups"]
reviewed = _policy_facts(groups)
facts_by_fingerprint = {
fact["fingerprint"]: fact
for fact in facts["entries"]
}
assert policy["schema_version"] == 3
assert egress_policy["scope"] == facts["scope"]
assert egress_policy["registry"] == facts["registry"]
assert len(reviewed) == len({fact["fingerprint"] for fact in reviewed}) == len(
facts["entries"]
)
assert all(set(fact) == {"source", "target", "kind", "fingerprint"} for fact in reviewed)
assert all("*" not in fact["source"] + fact["target"] for fact in reviewed)
assert all(len(fact["fingerprint"]) == 64 for fact in reviewed)
assert all(
{
key: facts_by_fingerprint[fact["fingerprint"]][key]
for key in ("source", "target", "kind", "fingerprint")
}
== fact
for fact in reviewed
)
assert _policy_differences(facts["entries"], reviewed) == ([], [])
assert set(FROZEN_EGRESS_FINGERPRINT_BY_EDGE) == set(
FROZEN_EGRESS_REASON_BY_EDGE
)
assert _fingerprint_differences(facts["entries"]) == []
assert _classification_differences(facts["entries"], groups) == ([], [], [])
def test_egress_policy_enforces_debt_and_exception_schemas() -> None:
"""债务必须绑定清零叶，精确例外必须有 owner 与业务理由。"""
policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8"))
groups = policy["direct_egress"]["groups"]
debt_tracking = {
"direct_http": "S2-L7",
"requestutils_session_bridge": "S2-L7",
"direct_dns": "S2-L6",
}
exception_reasons = {
"canonical_transport",
"transport_configuration",
"sdk_transport",
"streaming_protocol",
"contained_vendor",
"local_control_plane",
"diagnostic_probe",
"type_or_compat_only",
"daemon_control_plane",
"test_network_guard",
}
for group in groups:
if group["classification"] == "temporary_debt":
assert set(group) == {
"classification",
"reason_code",
"tracking",
"target_state",
"reason",
"facts",
}
assert group["reason_code"] in debt_tracking
assert group["tracking"] == debt_tracking[group["reason_code"]]
assert group["target_state"] == "empty"
assert str(group["reason"]).strip()
else:
assert group["classification"] == "approved_exception"
assert set(group) == {
"classification",
"reason_code",
"owner",
"reason",
"facts",
}
assert group["reason_code"] in exception_reasons
assert group["owner"] == "$source"
assert str(group["reason"]).strip()
def test_egress_policy_rejects_classification_swaps() -> None:
"""事实即使仍被覆盖，也不能在债务与例外 reason 间互换。"""
facts = [
{"source": "app.agent.llm.provider", "target": "httpx"},
{"source": "app.agent.llm.capability", "target": "openai"},
]
swapped = [
{
"reason_code": "sdk_transport",
"facts": [{"source": "app.agent.llm.provider", "target": "httpx"}],
},
{
"reason_code": "direct_http",
"facts": [{"source": "app.agent.llm.capability", "target": "openai"}],
},
]
assert _classification_differences(facts, swapped) == (
[],
[],
[
("app.agent.llm.capability", "openai"),
("app.agent.llm.provider", "httpx"),
],
)
def test_egress_policy_rejects_same_edge_surface_growth() -> None:
"""同一 source/target 的调用面变化不能靠刷新 policy 绕过。"""
facts = [
{
"source": "app.agent.llm.provider",
"target": "httpx",
"fingerprint": "f" * 64,
}
]
assert _fingerprint_differences(facts) == [
("app.agent.llm.provider", "httpx")
]
def test_egress_policy_rejects_add_remove_replacement_and_stale_entries() -> None:
"""新增、事实变化和删除后未清 policy 都必须产生精确差异。"""
original = [
{
"source": "app.sample",
"target": "requests",
"kind": "raw_transport",
"fingerprint": "a" * 64,
}
]
reviewed = [
{
"source": "app.sample",
"target": "requests",
"kind": "raw_transport",
"fingerprint": "a" * 64,
}
]
added = [
*original,
{
"source": "app.other",
"target": "aiohttp",
"kind": "raw_transport",
"fingerprint": "b" * 64,
},
]
replacement = [{**original[0], "fingerprint": "c" * 64}]
assert _policy_differences(original, reviewed) == ([], [])
assert _policy_differences(added, reviewed) == (["b" * 64], [])
assert _policy_differences([], reviewed) == ([], ["a" * 64])
assert _policy_differences(replacement, reviewed) == (
["c" * 64],
["a" * 64],
)
assert _policy_differences([], []) == ([], [])
tests/test_plugin_identity_transitions.py
+28
View File
@@ -288,6 +288,34 @@ def test_bind_online_commits_legacy_and_local_first_bindings(identity_store) ->
assert local_bound.revision == 2
def test_bind_online_preserves_legacy_unknown_payload_binding(identity_store) -> None:
"""存量身份可先固化默认在线来源，且不伪造尚未应用的载荷事实。"""
legacy = identity_store.compare_and_set(
_legacy_identity("LegacyInventoryPlugin"),
expected_revision=None,
)
target = replace(
legacy,
trusted_source_type=TrustedPluginSourceType.OFFICIAL,
trusted_source_key=OFFICIAL_SOURCE,
binding_basis=PluginBindingBasis.OFFICIAL_DEFAULT,
updated_at=NOW + timedelta(seconds=1),
bound_at=NOW + timedelta(seconds=1),
)
bound = identity_store.bind_online(
target,
expected_revision=legacy.revision,
)
assert bound.trusted_source_type is TrustedPluginSourceType.OFFICIAL
assert bound.trusted_source_key == OFFICIAL_SOURCE
assert bound.binding_basis is PluginBindingBasis.OFFICIAL_DEFAULT
assert bound.payload_source_type is PluginPayloadSourceType.UNKNOWN
assert bound.payload_applied_at is None
assert bound.revision == 2
def test_bind_online_rejects_bound_identity_and_stale_revision(identity_store) -> None:
"""首次在线绑定不能覆盖已有可信来源，也不能使用失效 revision。"""
bound = identity_store.compare_and_set(_identity(), expected_revision=None)