fix: preserve v3 architecture complexity budget

This commit is contained in:
jxxghp
2026-09-23 18:49:00 +08:00
parent 58bb4961af
commit bbaec9d6b3
6 changed files with 40 additions and 34 deletions

View File

@@ -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,

View File

@@ -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:

View File

@@ -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) |

View File

@@ -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 等基线均为零 |

View File

@@ -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",

View File

@@ -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