diff --git a/.env.example b/.env.example index 9de8ae80d..2d9225330 100644 --- a/.env.example +++ b/.env.example @@ -415,7 +415,7 @@ WEBUI_HOST=127.0.0.1 WEBUI_PORT=8000 # 启动 Web 服务前是否自动构建前端（npm install && npm run build，默认 true） WEBUI_AUTO_BUILD=true -# 反向代理下信任 X-Forwarded-For 获取真实 IP（Nginx/Cloudflare 前置时设为 true，直连公网时保持 false 防伪造） +# 单层可信反向代理（如 Nginx → App）下信任 X-Forwarded-For 获取真实 IP，取最右值用于登录限流；多级代理/CDN 场景限流 key 可能退化为边缘代理 IP，需额外评估；直连公网时保持 false 防伪造 # TRUST_X_FORWARDED_FOR=false # =================================== diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 48bd553bb..4a1e1e5b3 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ### 修复 +- 🔒 **认证限流 X-Forwarded-For 取值修复（CWE-345）**（#841 / #842）— `get_client_ip()` 从取 `X-Forwarded-For` 最左值改为最右值，防止攻击者通过伪造首部旋转限流桶绕过暴力破解保护；仅影响 `TRUST_X_FORWARDED_FOR=true` 且单层可信反向代理的部署场景，多级代理环境需按部署文档评估配置。 - 📦 **恢复 LiteLLM 官方 PyPI 安装并锁定安全上限** — `requirements.txt` 重新使用 `pip install litellm` 的官方 PyPI 安装路径，并在保留历史最低要求 `>=1.80.10` 的同时增加 `<1.82.7` 的安全上限，避免误装已被移除的 `1.82.7` / `1.82.8` 风险版本；Windows 桌面打包脚本也同步回退到标准 `pip install -r requirements.txt` 链路，减少特殊下载分支带来的维护成本。 - 📨 **Telegram Markdown 解析失败回退纯文本**（fixes #850）— `src/notification_sender/telegram_sender.py` 现在会在 Telegram 返回 `HTTP 400` 且包含 `can't parse entities` / Markdown 解析错误时，自动去掉 `parse_mode` 后重试纯文本发送，避免 `*ST` 等正文内容直接导致整条通知失败。 - 🔢 **A 股同码实时行情保留交易所提示**（fixes #852）— `DataFetcherManager` 与 `TushareFetcher` 现在会保留 `SZ000001` / `000001.SZ` 这类显式沪深提示，旧版 Tushare 实时行情降级分支不再把深市 `000001` 误判成 `sh000001` 上证指数。 diff --git a/docs/deploy-webui-cloud.md b/docs/deploy-webui-cloud.md index 071d22c4a..83e5538de 100644 --- a/docs/deploy-webui-cloud.md +++ b/docs/deploy-webui-cloud.md @@ -233,7 +233,7 @@ sudo systemctl reload nginx 配置成功后，直接用 `http://your-domain.com` 访问即可，不需要带端口号。 > **使用 Nginx 后的注意事项**： -> - 如果你开启了 Web 登录认证（`ADMIN_AUTH_ENABLED=true`），建议在 `.env` 中把 `TRUST_X_FORWARDED_FOR=true` 一并打开，否则系统可能无法正确识别真实 IP。 +> - 如果你开启了 Web 登录认证（`ADMIN_AUTH_ENABLED=true`），建议在 `.env` 中把 `TRUST_X_FORWARDED_FOR=true` 一并打开，否则系统可能无法正确识别真实 IP。该选项适用于**单层可信反向代理**（Nginx → App）部署；如果使用多级代理或 CDN（CDN → Nginx → App），登录限流的 key 可能退化为边缘代理 IP 而非真实客户端 IP，需根据实际拓扑评估。 > - 如需 HTTPS，可以用 [Certbot](https://certbot.eff.org/) 自动申请免费的 Let's Encrypt 证书。 --- diff --git a/docs/full-guide.md b/docs/full-guide.md index 2c2ea6ac1..7e6693a2c 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -291,7 +291,7 @@ daily_stock_analysis/ |--------|------|--------| | `STOCK_LIST` | 自选股代码（逗号分隔） | - | | `ADMIN_AUTH_ENABLED` | Web 登录：设为 `true` 启用密码保护；首次访问在网页设置初始密码，可在「系统设置 > 修改密码」修改；忘记密码执行 `python -m src.auth reset_password` | `false` | -| `TRUST_X_FORWARDED_FOR` | 反向代理部署时设为 `true`，从 `X-Forwarded-For` 获取真实 IP（限流等）；直连公网时保持 `false` 防伪造 | `false` | +| `TRUST_X_FORWARDED_FOR` | 单层可信反向代理部署时设为 `true`，取 `X-Forwarded-For` 最右值作为真实客户端 IP（用于登录限流等）；直连公网时保持 `false` 防伪造。多级代理/CDN 场景下限流 key 可能退化为边缘代理 IP，需额外评估 | `false` | | `MAX_WORKERS` | 并发线程数 | `3` | | `MARKET_REVIEW_ENABLED` | 启用大盘复盘 | `true` | | `MARKET_REVIEW_REGION` | 大盘复盘市场区域：cn(A股)、us(美股)、both(两者)，us 适合仅关注美股的用户 | `cn` | diff --git a/src/auth.py b/src/auth.py index 5d952c539..d6e770311 100644 --- a/src/auth.py +++ b/src/auth.py @@ -368,11 +368,17 @@ def verify_session(value: str) -> bool: def get_client_ip(request) -> str: - """Get client IP, respecting TRUST_X_FORWARDED_FOR.""" + """Get client IP, respecting TRUST_X_FORWARDED_FOR. + + When behind a single trusted reverse proxy, the proxy appends the real + client IP as the rightmost entry in X-Forwarded-For. We use [-1] instead + of [0] so that an attacker cannot spoof an arbitrary leftmost value to + rotate rate-limit buckets and bypass brute-force protection. + """ if os.getenv("TRUST_X_FORWARDED_FOR", "false").lower() == "true": forwarded = request.headers.get("X-Forwarded-For") if forwarded: - return forwarded.split(",")[0].strip() + return forwarded.split(",")[-1].strip() if request.client: return request.client.host or "127.0.0.1" return "127.0.0.1" diff --git a/tests/test_cwe345_xff_bypass.py b/tests/test_cwe345_xff_bypass.py new file mode 100644 index 000000000..2caaa0ce3 --- /dev/null +++ b/tests/test_cwe345_xff_bypass.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +"""Tests for CWE-345 fix: X-Forwarded-For IP spoofing prevention in get_client_ip.""" + +import os +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from src.auth import get_client_ip + + +def _make_request(xff_value=None, client_host=None): + """Build a minimal request-like object.""" + headers = {} + if xff_value is not None: + headers["X-Forwarded-For"] = xff_value + client = SimpleNamespace(host=client_host) if client_host else None + return SimpleNamespace(headers=headers, client=client) + + +class TestGetClientIpXffFix(unittest.TestCase): + """Verify get_client_ip uses rightmost XFF entry (proxy-appended).""" + + # --- TRUST_X_FORWARDED_FOR enabled --- + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_single_ip_returns_that_ip(self): + """Single-entry XFF should return that entry.""" + req = _make_request(xff_value="1.2.3.4") + self.assertEqual(get_client_ip(req), "1.2.3.4") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_multiple_ips_returns_rightmost(self): + """Rightmost entry is the one appended by the trusted proxy.""" + req = _make_request(xff_value="spoofed.ip, 10.0.0.1, 192.168.1.1") + self.assertEqual(get_client_ip(req), "192.168.1.1") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_attacker_cannot_control_rate_limit_bucket(self): + """Attacker-injected leftmost IP must NOT be selected (the old [0] bug).""" + req = _make_request(xff_value="evil-rotated-ip, real-client-ip") + ip = get_client_ip(req) + self.assertNotEqual(ip, "evil-rotated-ip", + "Leftmost (attacker-controlled) IP must not be used") + self.assertEqual(ip, "real-client-ip") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_whitespace_is_stripped(self): + req = _make_request(xff_value="10.0.0.1, 192.168.1.1 ") + self.assertEqual(get_client_ip(req), "192.168.1.1") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_no_xff_header_falls_back_to_client(self): + req = _make_request(client_host="172.16.0.1") + self.assertEqual(get_client_ip(req), "172.16.0.1") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_no_xff_no_client_returns_localhost(self): + req = _make_request() + self.assertEqual(get_client_ip(req), "127.0.0.1") + + # --- TRUST_X_FORWARDED_FOR disabled (default) --- + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "false"}) + def test_xff_ignored_when_trust_disabled(self): + """XFF header should be completely ignored when trust is off.""" + req = _make_request(xff_value="1.2.3.4", client_host="10.0.0.5") + self.assertEqual(get_client_ip(req), "10.0.0.5") + + @patch.dict(os.environ, {}, clear=False) + def test_xff_ignored_when_env_unset(self): + """If TRUST_X_FORWARDED_FOR is not set, default to not trusting.""" + env = os.environ.copy() + env.pop("TRUST_X_FORWARDED_FOR", None) + with patch.dict(os.environ, env, clear=True): + req = _make_request(xff_value="1.2.3.4", client_host="10.0.0.5") + self.assertEqual(get_client_ip(req), "10.0.0.5") + + # --- Edge cases --- + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) + def test_empty_xff_header(self): + """Empty XFF string should fall back to client.""" + req = SimpleNamespace(headers={"X-Forwarded-For": ""}, client=SimpleNamespace(host="10.0.0.1")) + self.assertEqual(get_client_ip(req), "10.0.0.1") + + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "TRUE"}) + def test_case_insensitive_trust_flag(self): + """TRUST_X_FORWARDED_FOR=TRUE (uppercase) should still work.""" + req = _make_request(xff_value="1.1.1.1, 2.2.2.2") + self.assertEqual(get_client_ip(req), "2.2.2.2") + + +if __name__ == "__main__": + unittest.main()