mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
883 lines
28 KiB
Python
883 lines
28 KiB
Python
"""
|
||
验证 URL 签名、地址安全判断及 DNS 缓存和并发解析。
|
||
"""
|
||
|
||
import socket
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
|
||
from app.application.security.url import (
|
||
SecurityUtils,
|
||
UrlSafetyDiagnosis,
|
||
UrlSafetyReason,
|
||
_dns_inflight_locks,
|
||
_dns_negative_cache,
|
||
_dns_positive_cache,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def _isolate_dns_state():
|
||
"""
|
||
在空缓存下运行 DNS 校验，并在用例结束时恢复原有缓存和锁映射。
|
||
"""
|
||
caches = (_dns_positive_cache, _dns_negative_cache, _dns_inflight_locks)
|
||
originals = [dict(cache) for cache in caches]
|
||
for cache in caches:
|
||
cache.clear()
|
||
try:
|
||
yield
|
||
finally:
|
||
for cache, original in zip(caches, originals):
|
||
cache.clear()
|
||
cache.update(original)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_signed_url_roundtrip_returns_clean_url() -> None:
|
||
"""
|
||
URL 签名验证成功后返回不含签名片段的真实请求地址。
|
||
"""
|
||
url = "http://192.168.1.50:8096/Items/abc/Images/Primary?api_key=demo"
|
||
|
||
signed_url = SecurityUtils.sign_url(url)
|
||
|
||
assert "#mp_sig=" in signed_url
|
||
assert "mp_purpose=image-proxy" in signed_url
|
||
assert SecurityUtils.verify_signed_url(signed_url) == url
|
||
assert SecurityUtils.strip_url_signature(signed_url) == url
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_signed_url_rejects_tampered_url() -> None:
|
||
"""
|
||
签名绑定完整 URL，签名后修改路径必须校验失败。
|
||
"""
|
||
signed_url = SecurityUtils.sign_url("http://192.168.1.50:8096/Items/abc/Images/Primary")
|
||
tampered_url = signed_url.replace(
|
||
"/Items/abc/Images/Primary",
|
||
"/System/Info/Public",
|
||
)
|
||
|
||
assert SecurityUtils.verify_signed_url(tampered_url) is None
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_signed_url_is_deterministic_for_same_inputs() -> None:
|
||
"""
|
||
相同 URL 与 RESOURCE_SECRET_KEY 多次签名结果必须完全一致，
|
||
保证浏览器 / Service Worker 缓存能稳定命中。
|
||
"""
|
||
url = "http://192.168.1.50:8096/Items/abc/Images/Primary"
|
||
|
||
first = SecurityUtils.sign_url(url)
|
||
second = SecurityUtils.sign_url(url)
|
||
|
||
assert first == second
|
||
assert SecurityUtils.verify_signed_url(first) == url
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_signed_url_invalidated_after_secret_rotation() -> None:
|
||
"""
|
||
`RESOURCE_SECRET_KEY` 变更（进程重启或运维显式轮换）后旧签名必须作废，
|
||
作为签名长期有效模型的失效兜底。
|
||
"""
|
||
url = "http://192.168.1.50:8096/Items/abc/Images/Primary"
|
||
|
||
with patch(
|
||
"app.runtime.config.settings.RESOURCE_SECRET_KEY",
|
||
"old-secret-value-aaaaaaaaaaaaaaaaaaaaaaaa",
|
||
):
|
||
signed_url = SecurityUtils.sign_url(url)
|
||
assert SecurityUtils.verify_signed_url(signed_url) == url
|
||
|
||
with patch(
|
||
"app.runtime.config.settings.RESOURCE_SECRET_KEY",
|
||
"new-secret-value-bbbbbbbbbbbbbbbbbbbbbbbb",
|
||
):
|
||
assert SecurityUtils.verify_signed_url(signed_url) is None
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_signed_url_rejects_other_purpose() -> None:
|
||
"""
|
||
签名绑定 `purpose`，挪用到其它签名用途必须被拒绝。
|
||
"""
|
||
url = "http://192.168.1.50:8096/Items/abc/Images/Primary"
|
||
signed_url = SecurityUtils.sign_url(url)
|
||
|
||
assert SecurityUtils.verify_signed_url(signed_url, purpose="other-purpose") is None
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_keeps_default_allowlist_behavior() -> None:
|
||
"""
|
||
默认 URL 校验保持历史 allowlist 行为，避免影响非代理调用方。
|
||
"""
|
||
assert SecurityUtils.is_safe_url("http://192.168.1.50:8096/secret.png", {"http://192.168.1.50:8096"})
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_private_literal_ip_when_enabled() -> None:
|
||
"""
|
||
启用 SSRF 防护时，即使内网 IP 命中 allowlist 也不能放行。
|
||
"""
|
||
assert not SecurityUtils.is_safe_url(
|
||
"http://192.168.1.50:8096/secret.png", {"http://192.168.1.50:8096"}, block_private=True
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_loopback_dns_result_when_enabled() -> None:
|
||
"""
|
||
主机名解析到回环地址时必须拒绝，防止通过域名绕过内网地址拦截。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("127.0.0.1", 0),
|
||
)
|
||
],
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"http://internal.example.com/secret.png", {"example.com"}, block_private=True
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_mixed_public_and_private_dns_results() -> None:
|
||
"""
|
||
同一域名只要存在任一非公网解析结果，就不能作为图片代理目标。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
),
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("10.0.0.8", 0),
|
||
),
|
||
],
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg", {"example.com"}, block_private=True
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_allows_public_dns_result_when_enabled() -> None:
|
||
"""
|
||
域名解析结果全部为公网地址且命中 allowlist 时继续允许访问。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
],
|
||
):
|
||
assert SecurityUtils.is_safe_url("https://assets.example.com/poster.jpg", {"example.com"}, block_private=True)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_rejects_dns_resolution_failure_when_enabled() -> None:
|
||
"""
|
||
SSRF 防护无法确认目标地址时按失败处理，避免解析异常时继续请求。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
side_effect=socket.gaierror,
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg", {"example.com"}, block_private=True
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_allows_configured_private_range_after_domain_match() -> None:
|
||
"""
|
||
图片域名命中 allowlist 后，可通过配置允许 TUN fake-ip 等特定非公网网段。
|
||
"""
|
||
with (
|
||
patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("198.18.16.96", 0),
|
||
)
|
||
],
|
||
),
|
||
patch("app.application.security.url.logger.debug") as debug_log,
|
||
):
|
||
assert SecurityUtils.is_safe_url(
|
||
"https://img1.doubanio.com/poster.webp",
|
||
{"doubanio.com"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
debug_message = debug_log.call_args.args[0]
|
||
assert "ips=198.18.16.96" in debug_message
|
||
assert "ranges=198.18.0.0/15" in debug_message
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_configured_private_range_without_domain_match() -> None:
|
||
"""
|
||
非公网网段例外必须依附域名白名单，不能单独放行任意用户 URL。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("198.18.16.96", 0),
|
||
)
|
||
],
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://attacker.example.com/poster.webp",
|
||
{"doubanio.com"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_private_result_outside_configured_range() -> None:
|
||
"""
|
||
仅允许显式配置的非公网网段，其它内网解析结果仍按 SSRF 风险拦截。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("10.0.0.8", 0),
|
||
)
|
||
],
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_blocks_mixed_allowed_and_disallowed_private_results() -> None:
|
||
"""
|
||
同一域名的解析结果必须全部落在允许网段内，避免部分安全结果掩盖风险地址。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("198.18.16.96", 0),
|
||
),
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("10.0.0.8", 0),
|
||
),
|
||
],
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_async_uses_event_loop_resolver() -> None:
|
||
"""
|
||
异步版本通过事件循环的非阻塞 getaddrinfo 完成 SSRF 校验，
|
||
且语义与同步版本保持一致：解析到非公网地址时仍然拒绝。
|
||
"""
|
||
import asyncio
|
||
|
||
async def fake_getaddrinfo(host, *_args, **_kwargs):
|
||
"""
|
||
返回预设 DNS 地址，隔离真实域名解析并记录调用。
|
||
"""
|
||
assert host == "internal.example.com"
|
||
return [
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("127.0.0.1", 0),
|
||
)
|
||
]
|
||
|
||
async def run() -> bool:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
with patch.object(loop, "getaddrinfo", side_effect=fake_getaddrinfo):
|
||
return await SecurityUtils.is_safe_url_async(
|
||
"http://internal.example.com/secret.png",
|
||
{"example.com"},
|
||
block_private=True,
|
||
)
|
||
|
||
assert not asyncio.run(run())
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_async_hits_dns_cache() -> None:
|
||
"""
|
||
异步与同步版本共享 DNS TTL 缓存：同步预热后，异步版本不应再发起 DNS 查询。
|
||
"""
|
||
import asyncio
|
||
|
||
# 先用同步路径预热缓存
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
],
|
||
):
|
||
assert SecurityUtils.is_safe_url("https://assets.example.com/poster.jpg", {"example.com"}, block_private=True)
|
||
|
||
async def run() -> bool:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
with patch.object(
|
||
loop,
|
||
"getaddrinfo",
|
||
side_effect=AssertionError("缓存命中后不应再次发起 DNS 查询"),
|
||
):
|
||
return await SecurityUtils.is_safe_url_async(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
)
|
||
|
||
assert asyncio.run(run())
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_async_allows_public_dns_result() -> None:
|
||
"""
|
||
异步版本对全公网解析结果且命中 allowlist 时放行。
|
||
"""
|
||
import asyncio
|
||
|
||
async def fake_getaddrinfo(host, *_args, **_kwargs):
|
||
"""
|
||
返回预设 DNS 地址，隔离真实域名解析并记录调用。
|
||
"""
|
||
assert host == "assets.example.com"
|
||
return [
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
]
|
||
|
||
async def run() -> bool:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
with patch.object(loop, "getaddrinfo", side_effect=fake_getaddrinfo):
|
||
return await SecurityUtils.is_safe_url_async(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
)
|
||
|
||
assert asyncio.run(run())
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_dns_resolution_failure_populates_negative_cache() -> None:
|
||
"""
|
||
DNS 解析失败应回填负向缓存，避免短期内对同一目标反复触发 `getaddrinfo`。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
side_effect=socket.gaierror,
|
||
) as mock_resolve:
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg", {"example.com"}, block_private=True
|
||
)
|
||
assert mock_resolve.call_count == 1
|
||
assert "assets.example.com" in _dns_negative_cache
|
||
|
||
assert not SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/another.jpg", {"example.com"}, block_private=True
|
||
)
|
||
assert mock_resolve.call_count == 1, "命中负向缓存后不应再次调用 getaddrinfo"
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_literal_ip_skips_dns_cache() -> None:
|
||
"""
|
||
URL 中的字面量 IP 走快路径，不应进入 DNS 缓存或触发 `getaddrinfo`。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
side_effect=AssertionError("字面量 IP 不应触发 getaddrinfo"),
|
||
):
|
||
assert not SecurityUtils.is_safe_url(
|
||
"http://10.0.0.5:8080/secret.png", {"http://10.0.0.5:8080"}, block_private=True
|
||
)
|
||
assert "10.0.0.5" not in _dns_positive_cache
|
||
assert "10.0.0.5" not in _dns_negative_cache
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_literal_ipv6_in_brackets_is_recognized() -> None:
|
||
"""
|
||
`urlparse` 已为 IPv6 字面量脱壳，`_literal_ip` 兼容直接传入带方括号的形式。
|
||
"""
|
||
assert str(SecurityUtils._literal_ip("[::1]")) == "::1"
|
||
assert str(SecurityUtils._literal_ip("::1")) == "::1"
|
||
assert SecurityUtils._literal_ip("not-an-ip") is None
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_is_safe_url_async_dedupes_concurrent_inflight_queries() -> None:
|
||
"""
|
||
同 hostname 的并发未命中请求应通过 in-flight 锁去重，只触发一次 DNS 查询。
|
||
"""
|
||
import asyncio
|
||
|
||
call_count = 0
|
||
|
||
async def run() -> None:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
nonlocal call_count
|
||
loop = asyncio.get_running_loop()
|
||
release = asyncio.Event()
|
||
|
||
async def slow_getaddrinfo(host, *_args, **_kwargs):
|
||
"""
|
||
等待测试信号后返回 DNS 地址，用于观察并发合并和取消行为。
|
||
"""
|
||
nonlocal call_count
|
||
call_count += 1
|
||
await release.wait()
|
||
return [
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
]
|
||
|
||
with patch.object(loop, "getaddrinfo", side_effect=slow_getaddrinfo):
|
||
tasks = [
|
||
asyncio.create_task(
|
||
SecurityUtils.is_safe_url_async(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
)
|
||
)
|
||
for _ in range(5)
|
||
]
|
||
# 让所有任务都进入 in-flight 等待状态
|
||
await asyncio.sleep(0)
|
||
await asyncio.sleep(0)
|
||
release.set()
|
||
results = await asyncio.gather(*tasks)
|
||
|
||
assert all(results)
|
||
assert call_count == 1, "并发未命中应去重为单次 DNS 查询"
|
||
|
||
asyncio.run(run())
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_sync_cache_access_is_thread_safe() -> None:
|
||
"""
|
||
同步路径下并发线程访问 DNS 缓存不应触发异常或拿到不一致结果。
|
||
TTLCache 自身非线程安全，依赖模块级 `_dns_cache_lock` 串行化读写。
|
||
"""
|
||
import threading
|
||
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
],
|
||
):
|
||
results: list = []
|
||
errors: list = []
|
||
|
||
def worker() -> None:
|
||
"""
|
||
在线程中执行本次地址解析并记录结果。
|
||
"""
|
||
try:
|
||
for _ in range(50):
|
||
results.append(
|
||
SecurityUtils.is_safe_url(
|
||
"https://assets.example.com/poster.jpg",
|
||
{"example.com"},
|
||
block_private=True,
|
||
)
|
||
)
|
||
# 捕获解析线程传播的所有异常，验证取消路径不向调用方泄漏失败。
|
||
except Exception as exc: # noqa: BLE001
|
||
errors.append(exc)
|
||
|
||
threads = [threading.Thread(target=worker) for _ in range(8)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
assert errors == []
|
||
assert all(results)
|
||
assert len(results) == 8 * 50
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_async_dns_resolution_failure_releases_inflight_lock() -> None:
|
||
"""
|
||
DNS 解析失败后 in-flight 锁字典中必须被清理，避免每个解析失败的 hostname
|
||
都在 `_dns_inflight_locks` 里残留一把 `asyncio.Lock`。
|
||
"""
|
||
import asyncio
|
||
|
||
async def fail_getaddrinfo(*_args, **_kwargs):
|
||
"""
|
||
模拟 DNS 解析失败，验证异常路径释放资源。
|
||
"""
|
||
raise socket.gaierror()
|
||
|
||
async def run() -> None:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
with patch.object(loop, "getaddrinfo", side_effect=fail_getaddrinfo):
|
||
result = await SecurityUtils._hostname_addresses_async("bad-host.example")
|
||
assert result is None
|
||
|
||
asyncio.run(run())
|
||
assert not any((key[1] == "bad-host.example" for key in _dns_inflight_locks)), (
|
||
"解析失败路径必须释放 in-flight 锁字典条目"
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_async_dns_resolution_success_releases_inflight_lock() -> None:
|
||
"""
|
||
正常解析完成后 in-flight 锁字典也必须被清理，避免 hostname 累积。
|
||
"""
|
||
import asyncio
|
||
|
||
async def fake_getaddrinfo(*_args, **_kwargs):
|
||
"""
|
||
返回预设 DNS 地址，隔离真实域名解析并记录调用。
|
||
"""
|
||
return [
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
]
|
||
|
||
async def run() -> None:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
with patch.object(loop, "getaddrinfo", side_effect=fake_getaddrinfo):
|
||
result = await SecurityUtils._hostname_addresses_async("ok-host.example")
|
||
assert result is not None
|
||
|
||
asyncio.run(run())
|
||
assert not any((key[1] == "ok-host.example" for key in _dns_inflight_locks)), (
|
||
"正常解析路径必须释放 in-flight 锁字典条目"
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_async_dns_concurrent_waiters_release_inflight_lock() -> None:
|
||
"""
|
||
并发未命中场景下，所有等待者完成后 in-flight 锁字典也必须被清理，
|
||
覆盖"等到锁但缓存已被前一个协程回填"的二次返回路径。
|
||
"""
|
||
import asyncio
|
||
|
||
async def run() -> None:
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
release = asyncio.Event()
|
||
|
||
async def slow_getaddrinfo(*_args, **_kwargs):
|
||
"""
|
||
等待测试信号后返回 DNS 地址，用于观察并发合并和取消行为。
|
||
"""
|
||
await release.wait()
|
||
return [
|
||
(
|
||
socket.AF_INET,
|
||
socket.SOCK_STREAM,
|
||
0,
|
||
"",
|
||
("93.184.216.34", 0),
|
||
)
|
||
]
|
||
|
||
with patch.object(loop, "getaddrinfo", side_effect=slow_getaddrinfo):
|
||
tasks = [
|
||
asyncio.create_task(SecurityUtils._hostname_addresses_async("multi-host.example")) for _ in range(5)
|
||
]
|
||
await asyncio.sleep(0)
|
||
await asyncio.sleep(0)
|
||
release.set()
|
||
await asyncio.gather(*tasks)
|
||
|
||
asyncio.run(run())
|
||
assert not any((key[1] == "multi-host.example" for key in _dns_inflight_locks)), (
|
||
"并发等待者全部退出后必须释放 in-flight 锁字典条目"
|
||
)
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_domain_not_allowed_returns_reason_and_no_host() -> None:
|
||
"""
|
||
协议或 allowlist 校验未通过时，诊断返回 DOMAIN_NOT_ALLOWED，
|
||
且不暴露 host/ips 字段。
|
||
"""
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://attacker.example.com/x.jpg",
|
||
{"image.tmdb.org"},
|
||
)
|
||
|
||
assert isinstance(diag, UrlSafetyDiagnosis)
|
||
assert not diag.allowed
|
||
assert diag.reason is UrlSafetyReason.DOMAIN_NOT_ALLOWED
|
||
assert diag.host is None
|
||
assert diag.ips == []
|
||
assert diag.matched_private_ranges == []
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_allowed_without_block_private_skips_dns() -> None:
|
||
"""
|
||
未启用 block_private 时直接放行，不发起 DNS 解析，ips 保持为空。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
side_effect=AssertionError("不应触发 DNS 解析"),
|
||
):
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://image.tmdb.org/t/p/w500/x.jpg",
|
||
{"image.tmdb.org"},
|
||
)
|
||
|
||
assert diag.allowed
|
||
assert diag.reason is UrlSafetyReason.ALLOWED
|
||
assert diag.host == "image.tmdb.org"
|
||
assert diag.ips == []
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_dns_resolution_failed_carries_host_without_ips() -> None:
|
||
"""
|
||
`block_private=True` 下 DNS 抛错时返回 DNS_RESOLUTION_FAILED，
|
||
附带 host 便于排查但不携带 ips。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
side_effect=socket.gaierror,
|
||
):
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://image.tmdb.org/t/p/w500/x.jpg",
|
||
{"image.tmdb.org"},
|
||
block_private=True,
|
||
)
|
||
|
||
assert not diag.allowed
|
||
assert diag.reason is UrlSafetyReason.DNS_RESOLUTION_FAILED
|
||
assert diag.host == "image.tmdb.org"
|
||
assert diag.ips == []
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_non_global_dns_result_lists_resolved_ips() -> None:
|
||
"""
|
||
命中 allowlist 但 DNS 解析到非公网且未配置允许网段时，诊断标记
|
||
NON_GLOBAL_DNS_RESULT 并把解析到的 IP 列出来，供日志附带 fake-ip 提示。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("198.18.16.96", 0)),
|
||
],
|
||
):
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://image.tmdb.org/t/p/w500/x.jpg",
|
||
{"image.tmdb.org"},
|
||
block_private=True,
|
||
)
|
||
|
||
assert not diag.allowed
|
||
assert diag.reason is UrlSafetyReason.NON_GLOBAL_DNS_RESULT
|
||
assert diag.host == "image.tmdb.org"
|
||
assert diag.ips == ["198.18.16.96"]
|
||
assert diag.matched_private_ranges == []
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_mixed_private_and_public_with_ranges_reports_mixed_reason() -> None:
|
||
"""
|
||
配置了 allowed_private_ranges 但解析结果存在公网或不在允许网段内的私网
|
||
地址时，诊断必须标记 MIXED_OR_DISALLOWED_PRIVATE_RESULT，避免与"未配置
|
||
允许网段"场景混淆。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("198.18.16.96", 0)),
|
||
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.8", 0)),
|
||
],
|
||
):
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://image.tmdb.org/t/p/w500/x.jpg",
|
||
{"image.tmdb.org"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
|
||
assert not diag.allowed
|
||
assert diag.reason is UrlSafetyReason.MIXED_OR_DISALLOWED_PRIVATE_RESULT
|
||
assert diag.ips == ["198.18.16.96", "10.0.0.8"]
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_allowed_via_configured_private_range_reports_matched_networks() -> None:
|
||
"""
|
||
通过 allowed_private_ranges 放行时返回 ALLOWED，同时把命中的 IP 与
|
||
网段填入诊断对象，便于排查日志确认放行依据。
|
||
"""
|
||
with patch(
|
||
"app.adapters.network.resolver.socket.getaddrinfo",
|
||
return_value=[
|
||
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("198.18.16.96", 0)),
|
||
],
|
||
):
|
||
diag = SecurityUtils.evaluate_url_safety(
|
||
"https://image.tmdb.org/t/p/w500/x.jpg",
|
||
{"image.tmdb.org"},
|
||
block_private=True,
|
||
allowed_private_ranges=["198.18.0.0/15"],
|
||
)
|
||
|
||
assert diag.allowed
|
||
assert diag.reason is UrlSafetyReason.ALLOWED
|
||
assert diag.ips == ["198.18.16.96"]
|
||
assert diag.matched_private_ranges == ["198.18.0.0/15"]
|
||
|
||
|
||
@pytest.mark.usefixtures("_isolate_dns_state")
|
||
def test_async_evaluation_returns_same_diagnosis() -> None:
|
||
"""
|
||
异步版本走事件循环线程池但应保持与同步版本一致的诊断结果。
|
||
"""
|
||
import asyncio
|
||
|
||
async def fake_getaddrinfo(host, *_args, **_kwargs):
|
||
"""
|
||
返回预设 DNS 地址，隔离真实域名解析并记录调用。
|
||
"""
|
||
return [
|
||
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("198.18.16.96", 0)),
|
||
]
|
||
|
||
async def run():
|
||
"""
|
||
在独立事件循环中执行本用例的异步地址校验，并验证缓存或锁状态。
|
||
"""
|
||
with patch.object(
|
||
asyncio.get_running_loop(),
|
||
"getaddrinfo",
|
||
side_effect=fake_getaddrinfo,
|
||
create=True,
|
||
):
|
||
return await SecurityUtils.evaluate_url_safety_async(
|
||
"https://image.tmdb.org/x.jpg",
|
||
{"image.tmdb.org"},
|
||
block_private=True,
|
||
)
|
||
|
||
diag = asyncio.run(run())
|
||
assert not diag.allowed
|
||
assert diag.reason is UrlSafetyReason.NON_GLOBAL_DNS_RESULT
|
||
assert diag.ips == ["198.18.16.96"]
|