221 lines
7.5 KiB
Python
221 lines
7.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""来源 IP 提取与访问统计用纯函数。
|
||
|
||
服务经``域名 → CDN → 公网 EIP → 内网 → 宿主机 nginx → 容器``链路访问，容器内
|
||
socket 对端是 Docker 网关而非真实客户端，因此真实 IP 只能来自 nginx 注入的
|
||
请求头。安全前提：
|
||
|
||
* 只有当 socket 对端（``peer``）属于可信代理网段时才采信请求头；
|
||
* nginx 用 ``realip`` 模块按 CDN 网段校验后写入 ``X-Real-IP``（见文档）；
|
||
* 直接解析到公网 EIP 的请求没有 CDN 头，回退到 ``peer``。
|
||
|
||
本模块不依赖网络、文件或时钟，便于离线单元测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ipaddress
|
||
import re
|
||
from typing import Iterable, Mapping, Sequence
|
||
|
||
# 默认可信代理网段：回环、RFC1918、链路本地、IPv6 ULA/链路本地。
|
||
# Docker 默认桥接网段在 172.16.0.0/12 内，容器内看到的对端即网关地址。
|
||
DEFAULT_TRUSTED_PROXIES: tuple[str, ...] = (
|
||
"127.0.0.0/8",
|
||
"::1/128",
|
||
"10.0.0.0/8",
|
||
"172.16.0.0/12",
|
||
"192.168.0.0/16",
|
||
"169.254.0.0/16",
|
||
"fc00::/7",
|
||
"fe80::/10",
|
||
)
|
||
|
||
# 依次尝试的请求头：nginx 归一化后的 X-Real-IP 优先，X-Forwarded-For 兜底。
|
||
DEFAULT_CLIENT_IP_HEADERS: tuple[str, ...] = ("X-Real-IP", "X-Forwarded-For")
|
||
|
||
# 已知前端路由：未知路径统一归为 /other，避免卡片式路径污染统计。
|
||
KNOWN_ROUTES: frozenset[str] = frozenset({
|
||
"/home", "/videos", "/ups", "/fans", "/about",
|
||
"/candidates", "/system", "/rank", "/login", "/logout",
|
||
})
|
||
|
||
_VIDEO_ROUTE_RE = re.compile(r"^/video/\d+$")
|
||
_UP_ROUTE_RE = re.compile(r"^/up/\d+$")
|
||
_MAX_ROUTE_LENGTH = 200
|
||
|
||
# 常见爬虫 / 监控 UA 关键词；命中时视为机器人（是否计数由调用方决定）。
|
||
# 这些标记在 UA 中通常作为子串出现（如 Googlebot / bingbot / Baiduspider），
|
||
# 因此不使用词边界，避免漏判。
|
||
_BOT_PATTERN = re.compile(
|
||
r"(?i)("
|
||
r"bot|crawler|spider|crawl|slurp|bingpreview|facebookexternalhit|"
|
||
r"python-requests|python-urllib|httpx|aiohttp|curl|wget|libwww|"
|
||
r"monitor|uptime|healthcheck|headlesschrome|phantomjs|"
|
||
r"semrush|ahrefs|dotbot|yandex|baiduspider|sogou|bytespider"
|
||
r")"
|
||
)
|
||
|
||
|
||
def normalize_ip(value: object) -> str | None:
|
||
"""规范化 IP 字符串；非法或空值返回 ``None``。
|
||
|
||
兼容 ``[::1]:443``、``1.2.3.4:5678`` 带端口写法，并把 IPv4-mapped IPv6
|
||
折叠为 IPv4（``::ffff:1.2.3.4`` → ``1.2.3.4``），保证同一客户端去重一致。
|
||
"""
|
||
if not isinstance(value, str):
|
||
return None
|
||
text = value.strip()
|
||
if not text:
|
||
return None
|
||
if text.startswith("["):
|
||
end = text.find("]")
|
||
if end == -1:
|
||
return None
|
||
text = text[1:end]
|
||
ip = _parse_ip(text)
|
||
if ip is None:
|
||
# host:port（IPv6 已在上面的直接解析里成功，不会走到这里）
|
||
host, sep, port = text.rpartition(":")
|
||
if not sep or not port.isdigit() or not host:
|
||
return None
|
||
ip = _parse_ip(host)
|
||
if ip is None:
|
||
return None
|
||
if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
|
||
ip = ip.ipv4_mapped
|
||
return ip.compressed
|
||
|
||
|
||
def _parse_ip(text: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
|
||
try:
|
||
return ipaddress.ip_address(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def parse_networks(values: Iterable[str]) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
|
||
"""把 CIDR / 裸 IP 字符串解析为网段元组；非法项忽略。"""
|
||
networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
|
||
for raw in values:
|
||
if not isinstance(raw, str) or not raw.strip():
|
||
continue
|
||
text = raw.strip()
|
||
try:
|
||
networks.append(ipaddress.ip_network(text, strict=False))
|
||
except ValueError:
|
||
continue
|
||
return tuple(networks)
|
||
|
||
|
||
def is_trusted_proxy(
|
||
peer_ip: str | None,
|
||
networks: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
|
||
) -> bool:
|
||
ip = _parse_ip(peer_ip) if peer_ip else None
|
||
if ip is None:
|
||
return False
|
||
for network in networks:
|
||
if ip.version != network.version:
|
||
continue
|
||
if ip in network:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _header_value(headers: Mapping[str, object], name: str) -> str | None:
|
||
"""大小写不敏感地读取请求头；兼容单值与列表两种取值。"""
|
||
plain = headers.get(name)
|
||
if plain is None:
|
||
lowered = name.lower()
|
||
for key, value in headers.items():
|
||
if isinstance(key, str) and key.lower() == lowered:
|
||
plain = value
|
||
break
|
||
if plain is None:
|
||
return None
|
||
if isinstance(plain, (list, tuple)):
|
||
return ",".join(str(item) for item in plain)
|
||
return str(plain)
|
||
|
||
|
||
def _rightmost_valid_forwarded(value: str) -> str | None:
|
||
"""X-Forwarded-For 取最右侧合法 IP。
|
||
|
||
nginx 的 ``$proxy_add_x_forwarded_for`` 会把真实对端追加到最右侧，客户端
|
||
自带的伪造条目在左侧，因此最右侧最可信。
|
||
"""
|
||
for part in reversed(value.split(",")):
|
||
normalized = normalize_ip(part)
|
||
if normalized is not None:
|
||
return normalized
|
||
return None
|
||
|
||
|
||
def extract_client_ip(
|
||
headers: Mapping[str, object],
|
||
peer_ip: str | None,
|
||
*,
|
||
trusted_proxies: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
|
||
header_order: Sequence[str] = DEFAULT_CLIENT_IP_HEADERS,
|
||
) -> str:
|
||
"""返回用于统计的客户端 IP。
|
||
|
||
* 对端不可信（或无法解析）时忽略所有请求头，直接回退到对端；
|
||
* 对端可信时按 ``header_order`` 依次取第一个合法值；
|
||
* 全部缺失时回退到规范化后的对端；对端也无法解析时返回 ``""``。
|
||
"""
|
||
peer = normalize_ip(peer_ip)
|
||
if peer is None:
|
||
return ""
|
||
if not is_trusted_proxy(peer, trusted_proxies):
|
||
return peer
|
||
|
||
for name in header_order:
|
||
value = _header_value(headers, name)
|
||
if value is None:
|
||
continue
|
||
if name.lower() == "x-forwarded-for":
|
||
candidate = _rightmost_valid_forwarded(value)
|
||
else:
|
||
candidate = normalize_ip(value)
|
||
if candidate is not None:
|
||
return candidate
|
||
return peer
|
||
|
||
|
||
def is_bot(user_agent: object) -> bool:
|
||
"""UA 命中已知机器人/监控模式时返回 ``True``；空 UA 不算机器人。"""
|
||
if not isinstance(user_agent, str) or not user_agent.strip():
|
||
return False
|
||
return bool(_BOT_PATTERN.search(user_agent))
|
||
|
||
|
||
def normalize_route(path: object) -> str:
|
||
"""把上报路径归一化为有限集合中的路由标签。
|
||
|
||
``/`` → ``/home``；``/video/123`` → ``/video/:aid``；``/up/123`` →
|
||
``/up/:mid``；已知路由原样保留；其余（含空值、超长、带查询串）→ ``/other``。
|
||
"""
|
||
if not isinstance(path, str):
|
||
return "/other"
|
||
text = path.strip()
|
||
if not text:
|
||
return "/other"
|
||
# 丢弃查询串与片段，并折叠末尾斜杠
|
||
text = text.split("?", 1)[0].split("#", 1)[0]
|
||
if not text.startswith("/"):
|
||
return "/other"
|
||
if len(text) > _MAX_ROUTE_LENGTH:
|
||
return "/other"
|
||
if text == "/" or text == "":
|
||
return "/home"
|
||
if text != "/" and text.endswith("/"):
|
||
text = text.rstrip("/") or "/home"
|
||
if _VIDEO_ROUTE_RE.match(text):
|
||
return "/video/:aid"
|
||
if _UP_ROUTE_RE.match(text):
|
||
return "/up/:mid"
|
||
if text in KNOWN_ROUTES:
|
||
return text
|
||
return "/other" |