fix(auth): use rightmost X-Forwarded-For entry to prevent rate-limit bypass (CWE-345) (#841)

* fix(auth): use rightmost X-Forwarded-For entry to prevent rate-limit bypass (CWE-345)

When TRUST_X_FORWARDED_FOR=true, get_client_ip() previously took the
leftmost (index [0]) entry from the X-Forwarded-For header. An attacker
can inject arbitrary leftmost values to rotate rate-limit buckets and
bypass brute-force protection on the login endpoint.

Change to [-1] (rightmost entry) which is the IP appended by the trusted
reverse proxy and cannot be controlled by the attacker.

* fix: address review feedback — add CHANGELOG entry, update deployment docs for XFF trust scope

* fix: move CHANGELOG entry from [3.10.1] to [Unreleased] section

* fix: align .env.example TRUST_X_FORWARDED_FOR comment with single-proxy security semantics

---------

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

Author: Sebastion

.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
 
 # ===================================

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` 上证指数。

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 证书。
 
 ---

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

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"

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