From bbaec9d6b323e259bf9b99fab8e3657b38bccc83 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 23 Sep 2026 18:49:00 +0800 Subject: [PATCH] fix: preserve v3 architecture complexity budget --- app/application/security/url.py | 31 +++---------------- app/foundation/url.py | 14 ++++++++- docs/architecture-overview.md | 4 +-- docs/refactor/optimization-checklist.md | 2 +- .../architecture/dependency-baseline.json | 6 ++-- tests/test_url_utils.py | 17 +++++++++- 6 files changed, 40 insertions(+), 34 deletions(-) diff --git a/app/application/security/url.py b/app/application/security/url.py index afa7dc7f7..f3df27985 100644 --- a/app/application/security/url.py +++ b/app/application/security/url.py @@ -15,6 +15,7 @@ from anyio import Path as AsyncPath from cachetools import TTLCache from app.application.configuration import get_token_runtime_config +from app.foundation.url import url_matches_trusted_hosts from app.runtime.coalesce import ( CoalesceDecision, CoalesceSummary, @@ -827,21 +828,9 @@ class SecurityUtils: allowed_private_ranges: Optional[Iterable[str]] = None, trusted_hosts: Optional[Iterable[str]] = None, ) -> bool: - """ - 判定 URL 是否可作为图片代理请求目标。 + """先执行 DNS SSRF 校验，再允许精确匹配的受信主机或有效签名 URL。 - `trusted_hosts` 为管理员已配置的主机（如已启用的媒体服务器），URL 的 - netloc（含端口）或 hostname 精确命中时直接放行，不受私网拦截限制。 - - 校验顺序：协议 + 域名 allowlist + DNS SSRF 拦截 + 非公网放行匹配；标准 - 校验失败时再用 `verify_signed_url` 兜底，允许后端预签名的媒体服务器 - URL 跳过私网拦截。两者皆失败才视为拒绝。 - - 拒绝路径会输出结构化阻断日志：单次拦截立即打印一条 warning，同 - `(host, reason)` 的连续命中在 `_IMAGE_PROXY_BLOCK_LOG_WINDOW_SECONDS` - 窗口内合并为一条聚合摘要，避免媒体详情页一次请求把日志刷爆。日志字段 - 范围严格限定为 URL、host、reason、解析 IP 与允许网段配置；cookies、 - 签名串、token、请求头等敏感材料一律不进入日志。 + 其余目标拒绝并输出限频阻断日志，日志不记录 cookies、签名串、token 或请求头。 """ diagnosis = await SecurityUtils.evaluate_url_safety_async( url, @@ -851,7 +840,7 @@ class SecurityUtils: ) if diagnosis.allowed: return True - if trusted_hosts and SecurityUtils._is_trusted_host(url, trusted_hosts): + if trusted_hosts and url_matches_trusted_hosts(url, trusted_hosts): return True if SecurityUtils.verify_signed_url(url) is not None: return True @@ -863,18 +852,6 @@ class SecurityUtils: ) return False - @staticmethod - def _is_trusted_host(url: str, trusted_hosts: Iterable[str]) -> bool: - """URL 的 netloc（含端口）或 hostname 是否精确命中受信主机。""" - try: - parsed_url = urlparse(url) - except ValueError: - return False - if parsed_url.scheme not in {"http", "https"} or not parsed_url.hostname: - return False - trusted = {host.lower() for host in trusted_hosts} - return parsed_url.netloc.lower() in trusted or parsed_url.hostname in trusted - @staticmethod def _diagnose_resolved_addresses( url: str, diff --git a/app/foundation/url.py b/app/foundation/url.py index 13f612c05..0ebecab9c 100644 --- a/app/foundation/url.py +++ b/app/foundation/url.py @@ -1,7 +1,7 @@ import mimetypes import re from pathlib import Path -from typing import Optional, Union, Tuple +from typing import Iterable, Optional, Union, Tuple from urllib import parse from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse @@ -167,6 +167,18 @@ class UrlUtils: return None +def url_matches_trusted_hosts(url: str, trusted_hosts: Iterable[str]) -> bool: + """判断 HTTP(S) URL 的网络位置或主机名是否精确匹配受信主机。""" + try: + parsed_url = urlparse(url) + except ValueError: + return False + if parsed_url.scheme.lower() not in {"http", "https"} or not parsed_url.hostname: + return False + trusted = {host.lower() for host in trusted_hosts} + return parsed_url.netloc.lower() in trusted or parsed_url.hostname.lower() in trusted + + def split_netloc(url: str) -> Tuple[str, str]: """返回 URL 的协议与网络位置，并兼容未带协议的历史输入。""" if not url: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index ab498f43c..f86233ff9 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -760,8 +760,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 1044 | -| 内部导入边 | 8,882 | +| Python 模块 | 1045 | +| 内部导入边 | 8,884 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index a9d211c6d..60cd9e7f8 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1045 / 8,882 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档，分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与终端作用域模块的受控依赖 | +| 宿主 Python 模块 / 内部依赖边 | 1045 / 8,884 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档，分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期、终端作用域模块及图片代理主机匹配复用 foundation URL 工具的受控依赖 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index dc3df4267..1c56246a5 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1084,8 +1084,8 @@ "runtime_only": true } }, - "edge_count": 8882, - "edge_sha256": "fd866881390126c9f4a68c671c9641e27223e23a1b6c242c15042cb3c7d1a087", + "edge_count": 8884, + "edge_sha256": "c5bb4a5912d06f6e8e6a5545d78d5e043a2994f5cf85235a87397e66bb29e0ef", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3905,6 +3905,8 @@ "app.application.security.twofactor -> app.runtime.log", "app.application.security.url -> app.application", "app.application.security.url -> app.application.configuration", + "app.application.security.url -> app.foundation", + "app.application.security.url -> app.foundation.url", "app.application.security.url -> app.runtime", "app.application.security.url -> app.runtime.coalesce", "app.application.security.url -> app.runtime.log", diff --git a/tests/test_url_utils.py b/tests/test_url_utils.py index fc93472a7..78113d7a6 100644 --- a/tests/test_url_utils.py +++ b/tests/test_url_utils.py @@ -1,6 +1,6 @@ import pytest -from app.foundation.url import UrlUtils +from app.foundation.url import UrlUtils, url_matches_trusted_hosts @pytest.mark.parametrize( @@ -29,3 +29,18 @@ def test_normalize_http_url_preserves_address_except_outer_whitespace(url): def test_standardize_base_url_keeps_permissive_contract(): assert UrlUtils.standardize_base_url("example.com") == "http://example.com/" assert UrlUtils.normalize_http_url("example.com") is None + + +@pytest.mark.parametrize( + ("url", "trusted_hosts", "expected"), + [ + ("https://Media.Example:8096/image", {"media.example:8096"}, True), + ("http://media.example:8096/image", {"MEDIA.EXAMPLE"}, True), + ("http://media.example:8096/image", {"media.example:8097"}, False), + ("ftp://media.example/image", {"media.example"}, False), + ("http://[::1/image", {"::1"}, False), + ], +) +def test_url_matches_trusted_hosts_requires_exact_http_host(url, trusted_hosts, expected): + """受信主机匹配只放行 HTTP(S) 且端口或主机名精确命中的地址。""" + assert url_matches_trusted_hosts(url, trusted_hosts) is expected