268 lines
9.1 KiB
Python
268 lines
9.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""来源 IP 提取与路由/机器人判定纯函数测试（离线）。
|
||
|
||
运行：uv run python tests/test_client_ip.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
from reaction_board.client_ip import ( # noqa: E402
|
||
DEFAULT_TRUSTED_PROXIES,
|
||
extract_client_ip,
|
||
is_bot,
|
||
is_trusted_proxy,
|
||
normalize_ip,
|
||
normalize_route,
|
||
parse_networks,
|
||
)
|
||
|
||
TRUSTED = parse_networks(DEFAULT_TRUSTED_PROXIES)
|
||
# 容器内看到的 Docker 网关地址，属于可信代理网段。
|
||
DOCKER_GATEWAY = "172.17.0.1"
|
||
LOOPBACK = "127.0.0.1"
|
||
PUBLIC_CLIENT = "203.0.113.7"
|
||
|
||
|
||
def test_normalize_ip():
|
||
assert normalize_ip("203.0.113.7") == "203.0.113.7"
|
||
assert normalize_ip(" 203.0.113.7 ") == "203.0.113.7"
|
||
assert normalize_ip("::ffff:203.0.113.7") == "203.0.113.7"
|
||
assert normalize_ip("2001:db8::1") == "2001:db8::1"
|
||
assert normalize_ip("2001:0db8:0000:0000:0000:0000:0000:0001") == "2001:db8::1"
|
||
# 带端口 / 方括号
|
||
assert normalize_ip("203.0.113.7:5678") == "203.0.113.7"
|
||
assert normalize_ip("[2001:db8::1]:443") == "2001:db8::1"
|
||
# 空值、非法、非字符串
|
||
assert normalize_ip("") is None
|
||
assert normalize_ip(" ") is None
|
||
assert normalize_ip("not-an-ip") is None
|
||
assert normalize_ip("203.0.113.7:abc") is None
|
||
assert normalize_ip("[2001:db8::1") is None
|
||
assert normalize_ip(None) is None
|
||
assert normalize_ip(123) is None
|
||
assert normalize_ip("999.1.1.1") is None
|
||
|
||
|
||
def test_is_trusted_proxy():
|
||
assert is_trusted_proxy(DOCKER_GATEWAY, TRUSTED)
|
||
assert is_trusted_proxy(LOOPBACK, TRUSTED)
|
||
assert is_trusted_proxy("10.10.8.5", TRUSTED)
|
||
assert is_trusted_proxy("::1", TRUSTED)
|
||
assert not is_trusted_proxy(PUBLIC_CLIENT, TRUSTED)
|
||
assert not is_trusted_proxy("2001:db8::1", TRUSTED)
|
||
assert not is_trusted_proxy(None, TRUSTED)
|
||
assert not is_trusted_proxy("garbage", TRUSTED)
|
||
|
||
|
||
def test_extract_direct_public_eip():
|
||
"""场景一：域名直接解析公网 EIP，无 CDN 头，取对端。"""
|
||
ip = extract_client_ip(
|
||
{"User-Agent": "Mozilla/5.0"},
|
||
"198.51.100.9", # 直连客户端被 nginx 记为 $remote_addr，但 app 看到的是可信对端
|
||
trusted_proxies=TRUSTED,
|
||
header_order=("X-Real-IP", "X-Forwarded-For"),
|
||
)
|
||
# 对端是可信 nginx，X-Real-IP 缺失 → 回退对端
|
||
assert ip == "198.51.100.9"
|
||
|
||
|
||
def test_extract_direct_uses_x_real_ip_from_nginx():
|
||
"""nginx 设置 X-Real-IP，app 采信之。"""
|
||
ip = extract_client_ip(
|
||
{"X-Real-IP": PUBLIC_CLIENT},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
)
|
||
assert ip == PUBLIC_CLIENT
|
||
|
||
|
||
def test_extract_cloudflare():
|
||
"""场景二：Cloudflare 中转，nginx realip 归一化到 X-Real-IP。"""
|
||
ip = extract_client_ip(
|
||
{
|
||
"X-Real-IP": PUBLIC_CLIENT,
|
||
"CF-Connecting-IP": PUBLIC_CLIENT,
|
||
"X-Forwarded-For": f"{PUBLIC_CLIENT}, 172.68.1.1",
|
||
},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
)
|
||
assert ip == PUBLIC_CLIENT
|
||
|
||
|
||
def test_extract_cloudflare_via_cf_header_only():
|
||
ip = extract_client_ip(
|
||
{"CF-Connecting-IP": PUBLIC_CLIENT},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
header_order=("CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"),
|
||
)
|
||
assert ip == PUBLIC_CLIENT
|
||
|
||
|
||
def test_extract_edgeone():
|
||
"""场景三：EdgeOne 中转，X-Real-IP 由 nginx 归一化给出。"""
|
||
ip = extract_client_ip(
|
||
{
|
||
"X-Real-IP": "198.51.100.23",
|
||
"EO-Connecting-IP": "198.51.100.23",
|
||
"X-Forwarded-For": "198.51.100.23, 43.152.0.9, 43.152.0.10",
|
||
},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
)
|
||
assert ip == "198.51.100.23"
|
||
|
||
|
||
def test_forwarded_for_takes_rightmost():
|
||
"""X-Forwarded-For 取最右侧（nginx 追加的真实对端），忽略左侧伪造值。"""
|
||
ip = extract_client_ip(
|
||
{"X-Forwarded-For": "1.2.3.4, 5.6.7.8, 203.0.113.7"},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
header_order=("X-Forwarded-For",),
|
||
)
|
||
assert ip == "203.0.113.7"
|
||
|
||
|
||
def test_forwarded_for_skips_invalid_rightmost():
|
||
ip = extract_client_ip(
|
||
{"X-Forwarded-For": "203.0.113.7, not-an-ip, "},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
header_order=("X-Forwarded-For",),
|
||
)
|
||
assert ip == "203.0.113.7"
|
||
|
||
|
||
def test_untrusted_peer_ignores_spoofed_headers():
|
||
"""不可信对端（直连容器）伪造 X-Real-IP/XFF 必须被忽略。"""
|
||
ip = extract_client_ip(
|
||
{
|
||
"X-Real-IP": "1.1.1.1",
|
||
"X-Forwarded-For": "2.2.2.2, 3.3.3.3",
|
||
"CF-Connecting-IP": "4.4.4.4",
|
||
},
|
||
PUBLIC_CLIENT,
|
||
trusted_proxies=TRUSTED,
|
||
header_order=("CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"),
|
||
)
|
||
assert ip == PUBLIC_CLIENT
|
||
|
||
|
||
def test_extract_fallbacks_and_edges():
|
||
# 可信对端但无任何头 → 对端
|
||
assert extract_client_ip({}, LOOPBACK, trusted_proxies=TRUSTED) == LOOPBACK
|
||
# 头部为空值 → 回退
|
||
assert extract_client_ip(
|
||
{"X-Real-IP": " "}, DOCKER_GATEWAY, trusted_proxies=TRUSTED
|
||
) == DOCKER_GATEWAY
|
||
# 头部非法 → 尝试下一个头
|
||
assert extract_client_ip(
|
||
{"X-Real-IP": "bad", "X-Forwarded-For": PUBLIC_CLIENT},
|
||
DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED,
|
||
) == PUBLIC_CLIENT
|
||
# 对端不可解析 → 空串
|
||
assert extract_client_ip(
|
||
{"X-Real-IP": PUBLIC_CLIENT}, "garbage", trusted_proxies=TRUSTED
|
||
) == ""
|
||
# 对端缺失 → 空串
|
||
assert extract_client_ip({"X-Real-IP": PUBLIC_CLIENT}, None, trusted_proxies=TRUSTED) == ""
|
||
# 大小写不敏感的请求头查找
|
||
assert extract_client_ip(
|
||
{"x-real-ip": PUBLIC_CLIENT}, DOCKER_GATEWAY, trusted_proxies=TRUSTED,
|
||
) == PUBLIC_CLIENT
|
||
# 多值请求头（http.server 可能给 list）
|
||
assert extract_client_ip(
|
||
{"X-Real-IP": [PUBLIC_CLIENT]}, DOCKER_GATEWAY, trusted_proxies=TRUSTED,
|
||
) == PUBLIC_CLIENT
|
||
# 空的 header_order → 回退对端
|
||
assert extract_client_ip(
|
||
{"X-Real-IP": PUBLIC_CLIENT}, DOCKER_GATEWAY,
|
||
trusted_proxies=TRUSTED, header_order=(),
|
||
) == DOCKER_GATEWAY
|
||
|
||
|
||
def test_extract_ipv6_client():
|
||
ip = extract_client_ip(
|
||
{"X-Real-IP": "2001:db8::beef"},
|
||
LOOPBACK,
|
||
trusted_proxies=TRUSTED,
|
||
)
|
||
assert ip == "2001:db8::beef"
|
||
|
||
|
||
def test_parse_networks_tolerates_junk():
|
||
networks = parse_networks(["10.0.0.0/8", "bad", "", "192.168.1.1", 42])
|
||
assert len(networks) == 2
|
||
assert is_trusted_proxy("10.1.2.3", networks)
|
||
assert is_trusted_proxy("192.168.1.1", networks)
|
||
assert not is_trusted_proxy("11.0.0.1", networks)
|
||
|
||
|
||
def test_is_bot():
|
||
assert is_bot("Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/155.0") is False
|
||
assert is_bot("") is False
|
||
assert is_bot(None) is False
|
||
assert is_bot("Googlebot/2.1 (+http://www.google.com/bot.html)") is True
|
||
assert is_bot("Mozilla/5.0 (compatible; bingbot/2.0)") is True
|
||
assert is_bot("python-requests/2.31") is True
|
||
assert is_bot("curl/8.4.0") is True
|
||
assert is_bot("Uptime-Kuma/1.23") is True
|
||
assert is_bot("Bytespider") is True
|
||
|
||
|
||
def test_normalize_route():
|
||
assert normalize_route("/") == "/home"
|
||
assert normalize_route("") == "/other"
|
||
assert normalize_route(None) == "/other"
|
||
assert normalize_route("/home") == "/home"
|
||
assert normalize_route("/videos") == "/videos"
|
||
assert normalize_route("/ups") == "/ups"
|
||
assert normalize_route("/fans") == "/fans"
|
||
assert normalize_route("/about") == "/about"
|
||
assert normalize_route("/candidates") == "/candidates"
|
||
assert normalize_route("/system") == "/system"
|
||
assert normalize_route("/rank") == "/rank"
|
||
# 带查询串 / 片段
|
||
assert normalize_route("/videos?sort=view&year=2026") == "/videos"
|
||
assert normalize_route("/video/114775988111033#top") == "/video/:aid"
|
||
# 末尾斜杠
|
||
assert normalize_route("/videos/") == "/videos"
|
||
# 动态路由折叠
|
||
assert normalize_route("/video/123") == "/video/:aid"
|
||
assert normalize_route("/up/230020975") == "/up/:mid"
|
||
# 非法 / 未知 / 超长
|
||
assert normalize_route("videos") == "/other"
|
||
assert normalize_route("/video/abc") == "/other"
|
||
assert normalize_route("/unknown") == "/other"
|
||
assert normalize_route("/" + "x" * 500) == "/other"
|
||
assert normalize_route(123) == "/other"
|
||
|
||
|
||
def main():
|
||
test_normalize_ip()
|
||
test_is_trusted_proxy()
|
||
test_extract_direct_public_eip()
|
||
test_extract_direct_uses_x_real_ip_from_nginx()
|
||
test_extract_cloudflare()
|
||
test_extract_cloudflare_via_cf_header_only()
|
||
test_extract_edgeone()
|
||
test_forwarded_for_takes_rightmost()
|
||
test_forwarded_for_skips_invalid_rightmost()
|
||
test_untrusted_peer_ignores_spoofed_headers()
|
||
test_extract_fallbacks_and_edges()
|
||
test_extract_ipv6_client()
|
||
test_parse_networks_tolerates_junk()
|
||
test_is_bot()
|
||
test_normalize_route()
|
||
print("OK: 来源 IP 提取与路由/机器人判定测试全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |