Profeto Agent
6c24672e87
安全加固: - /api/v1/predict 限流改用 get_client_ip 防 X-Forwarded-For 伪造 - 新增 TRUST_PROXY_HEADERS/REQUIRE_ADMIN_AUTH 配置,默认 fail-closed - 生产环境管理接口未配置鉴权时拒绝(503),不再放行 数据管线修复: - injuries IntegrityError 改用 begin_nested(SAVEPOINT)隔离批次 - injuries 空名单 vs 未配置语义区分(has_data 精确标记) - bzzoiro 统计字段映射(shots/possession/cards) + 入库条件放宽 - available_at 语义收紧(回测防泄漏) - team_names NFKD 去变音双重查找 + 补齐变体键 预测系统改进: - multi-agent 全专家失败时 status=degraded 跳过终裁 - agent_weights 独立持久化到 predictions 表 新增迁移: - 0014_predictions_agent_weights.py 新增测试(8个文件): - test_ip_spoofing.py: 限流防伪造 - test_require_admin_fail_closed.py: 生产 fail-closed - test_injuries_integrity_rollback.py: SAVEPOINT 隔离 - test_injuries_slice_semantic.py: 空名单 vs 未配置 - test_available_at.py: 回测防泄漏 - test_bzzoirot_stats.py: 统计字段映射 - test_team_names_normalize.py: NFKD 变体 - test_multi_agent_degraded.py: 全失败 degraded - test_agent_weights_persist.py: 权重持久化
89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""回归测试: 限流与登录防爆破在不可信 X-Forwarded-For 下的 IP 伪造问题。
|
|
|
|
验证:
|
|
1. TRUST_PROXY_HEADERS=False(默认)时忽略伪造的 X-Forwarded-For
|
|
2. TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For
|
|
3. 限流与登录共用同一套 IP 提取逻辑
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.api.deps import get_client_ip
|
|
from src.core.config import Settings
|
|
|
|
|
|
class TestGetClientIP:
|
|
"""get_client_ip 防伪造逻辑。"""
|
|
|
|
def _make_request(self, client_host: str | None, xff: str | None = None):
|
|
req = MagicMock()
|
|
req.client = MagicMock(host=client_host) if client_host else None
|
|
req.headers = {}
|
|
if xff is not None:
|
|
req.headers["X-Forwarded-For"] = xff
|
|
return req
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False))
|
|
def test_untrusted_proxy_ignores_xff(self):
|
|
"""TRUST_PROXY_HEADERS=False 时忽略伪造的 X-Forwarded-For。"""
|
|
# 客户端伪造 X-Forwarded-For,但 TRUST_PROXY_HEADERS=False
|
|
req = self._make_request("1.2.3.4", xff="10.0.0.1, 192.168.1.1")
|
|
ip = get_client_ip(req)
|
|
assert ip == "1.2.3.4", f"应使用连接层 IP,实际 {ip}"
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True))
|
|
def test_trusted_proxy_parses_xff(self):
|
|
"""TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For 第一个 IP。"""
|
|
req = self._make_request("127.0.0.1", xff="10.0.0.1, 192.168.1.1")
|
|
ip = get_client_ip(req)
|
|
assert ip == "10.0.0.1", f"应使用 XFF 第一个 IP,实际 {ip}"
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True))
|
|
def test_trusted_proxy_without_xff(self):
|
|
"""TRUST_PROXY_HEADERS=True 但无 XFF 头时回退到 client.host。"""
|
|
req = self._make_request("1.2.3.4", xff=None)
|
|
ip = get_client_ip(req)
|
|
assert ip == "1.2.3.4", f"应回退到连接层 IP,实际 {ip}"
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False))
|
|
def test_untrusted_proxy_no_client(self):
|
|
"""TRUST_PROXY_HEADERS=False 且无 client 时返回 unknown。"""
|
|
req = self._make_request(None, xff="10.0.0.1")
|
|
ip = get_client_ip(req)
|
|
assert ip == "unknown", f"应返回 unknown,实际 {ip}"
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True))
|
|
def test_trusted_proxy_single_ip(self):
|
|
"""TRUST_PROXY_HEADERS=True 且 XFF 只有一个 IP。"""
|
|
req = self._make_request("127.0.0.1", xff="10.0.0.1")
|
|
ip = get_client_ip(req)
|
|
assert ip == "10.0.0.1", f"应返回 10.0.0.1,实际 {ip}"
|
|
|
|
|
|
class TestRateLimitIPSpoofing:
|
|
"""验证限流使用 get_client_ip 防伪造。"""
|
|
|
|
@patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False))
|
|
def test_rate_limit_ignores_spoofed_xff(self):
|
|
"""限流在 TRUST_PROXY_HEADERS=False 时不受 XFF 伪造影响。"""
|
|
from src.api.deps import _RateLimiter, get_client_ip
|
|
|
|
limiter = _RateLimiter(max_requests=10, window_seconds=60)
|
|
|
|
# 模拟不同伪造 XFF,但真实 IP 相同
|
|
def make_request(spoofed_xff):
|
|
req = MagicMock()
|
|
req.client = MagicMock(host="1.2.3.4")
|
|
req.headers = {"X-Forwarded-For": spoofed_xff}
|
|
return req
|
|
|
|
# 伪造不同 XFF,但真实 IP 都是 1.2.3.4
|
|
for i in range(10):
|
|
req = make_request(f"10.0.0.{i}")
|
|
ip = get_client_ip(req)
|
|
assert ip == "1.2.3.4", f"迭代 {i}: 应返回 1.2.3.4,实际 {ip}"
|
|
assert limiter.is_allowed(ip), f"迭代 {i}: 应允许"
|