feat(auth): 登录限流 IP 取 XFF 末段(防首段轮换伪造)
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -46,6 +46,21 @@ class ChangePasswordRequest(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
def _client_ip(request: Request) -> str:
|
def _client_ip(request: Request) -> str:
|
||||||
|
"""限流键 IP(spec 2026-10-19 §6): XFF **末段**。
|
||||||
|
|
||||||
|
反代模板 proxy_add_x_forwarded_for 语义 = 客户端自带 XFF 在前、反代把
|
||||||
|
真实来源 $remote_addr 追加在尾——末段不可伪造(单跳拓扑);取首段会被
|
||||||
|
轮换伪造绕过限流(比现状更糟)。将来加 CDN 时末段退化为 CDN IP →
|
||||||
|
键退化为按用户名 = 现状,故障安全。无 XFF/空串/空白回退 client.host。
|
||||||
|
|
||||||
|
与引擎 _server_ip(取首段,"CDN 在前"拓扑假设,数据质量用途)刻意
|
||||||
|
不对称——两者用途与威胁模型不同,不许互相"统一"(spec §6 双向注记)。
|
||||||
|
"""
|
||||||
|
xff = request.headers.get("x-forwarded-for", "")
|
||||||
|
if xff.strip():
|
||||||
|
last = xff.split(",")[-1].strip()
|
||||||
|
if last:
|
||||||
|
return last
|
||||||
return request.client.host if request.client else ""
|
return request.client.host if request.client else ""
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -70,6 +70,55 @@ def test_login_lockout_after_5_fails(app_client):
|
|||||||
assert _login(app_client).status_code == 200
|
assert _login(app_client).status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
class TestClientIpXffLast:
|
||||||
|
"""spec 2026-10-19 §6:限流 IP 取 XFF 末段(反代追加语义,不可伪造)。"""
|
||||||
|
|
||||||
|
def test_末段入键_伪造首段轮换不绕过(self, app_client):
|
||||||
|
# 伪造首段轮换 + 固定真末段 → 同键计数,第 5 次失败后第 6 次 429
|
||||||
|
for i in range(5):
|
||||||
|
app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
r = app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": "9.9.9.9, 1.2.3.4"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
assert r.status_code == 429
|
||||||
|
|
||||||
|
def test_末段不同即独立键_判别用例(self, app_client):
|
||||||
|
# RED 判别用例:现状(无 XFF 解析)所有请求同键(client.host 恒
|
||||||
|
# testclient)→ 也会 429;新行为下 8.8.8.8 是独立键 → 401
|
||||||
|
for i in range(5):
|
||||||
|
app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
r = app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": "7.7.7.7, 8.8.8.8"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
assert r.status_code == 401 # 独立键:未锁,密码错 → 401(现状此断言为 429 → RED)
|
||||||
|
|
||||||
|
def test_无XFF回退client_host(self, app_client):
|
||||||
|
r = app_client.post(LOGIN,
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
assert r.status_code == 401 # 正常走通(键为 client.host)
|
||||||
|
|
||||||
|
def test_空串XFF回退(self, app_client):
|
||||||
|
r = app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": " "},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
assert r.status_code == 401 # 空白视同无 XFF,回退 client.host
|
||||||
|
|
||||||
|
def test_单段直连形态(self, app_client):
|
||||||
|
# XFF "1.2.3.4"(无反代追加)→ 末段即唯一段,入键
|
||||||
|
for _ in range(5):
|
||||||
|
app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": "1.2.3.4"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
r = app_client.post(LOGIN,
|
||||||
|
headers={"X-Forwarded-For": "1.2.3.4"},
|
||||||
|
json={"username": "admin", "password": "wrong"})
|
||||||
|
assert r.status_code == 429
|
||||||
|
|
||||||
|
|
||||||
# ---- token 生命周期 ------------------------------------------------------------
|
# ---- token 生命周期 ------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user