Become a sponsor

说明
提供客户端 IP 获取、IP 地理位置查询和 User-Agent 解析三大能力。IP 获取位于 src/utils/ip.py,地理位置查询基于 ip2region 离线数据库(src/utils/ip2region.py),UA 解析位于 src/utils/ua.py。三者组合使用,为登录日志、操作日志、限流、登录锁定等场景提供客户端身份信息。
请求进入
│
▼
┌──────────────────────────────────────────────────────┐
│ get_client_ip(request) │
│ ├─ 直连对端命中可信代理白名单? │
│ │ ├─ 是 → 从 X-Forwarded-For 取真实客户端 IP │
│ │ └─ 否 → request.client.host(fail-closed) │
│ └─ client 为 None → 回退 127.0.0.1 │
└──────────┬───────────────────────────────────────────┘
│
┌─────┴─────┐
▼ ▼
┌─────────┐ ┌──────────────────────────────────┐
│ 限流 │ │ get_ip_region(ip) │
│ 登录锁定 │ │ ├─ 内网/保留 IP → "内网IP" │
│ 验证码防刷│ │ ├─ ip2region.xdb 离线查询 │
└─────────┘ │ └─ 异常降级 → "未知" │
└──────────┬───────────────────────┘
│
▼
┌──────────────────────┐
│ parse_user_agent(ua) │
│ ├─ 操作系统(正则) │
│ └─ 浏览器(正则) │
└──────────┬───────────┘
│
▼
┌──────────────────────┐
│ 写入日志表 │
│ ip / location / │
│ os / browser │
└──────────────────────┘位于 src/utils/ip.py,从 FastAPI Request 获取客户端真实 IP。
def get_client_ip(request) -> str:
"""
获取客户端真实IP地址
仅当直连对端命中 TRUSTED_PROXY_IPS 白名单且 TRUSTED_PROXY_COUNT>0 时才信任
X-Forwarded-For 头:
- TRUSTED_PROXY_COUNT==0 或 TRUSTED_PROXY_IPS 未配置 / 对端不命中:不信任代理头,
直接使用 request.client.host(fail-closed,防止被直接暴露时伪造 XFF)
- 命中白名单:从 X-Forwarded-For 右侧取倒数第 N 个 IP 作为真实客户端IP
"""返回值:客户端 IP 地址字符串(如 "1.2.3.4")。
当应用部署在 Nginx/HAProxy 等反向代理之后,request.client.host 拿到的是代理 IP 而非真实客户端 IP。此时需要从 X-Forwarded-For(XFF)头提取真实 IP。
客户端 (1.2.3.4)
│
▼
外层代理 (10.0.0.1) ← 追加 XFF: 1.2.3.4
│
▼
内层代理 (10.0.0.2) ← 追加 XFF: 1.2.3.4, 10.0.0.1
│
▼
FastAPI 应用
request.client.host = 10.0.0.2
X-Forwarded-For = "1.2.3.4, 10.0.0.1"取值逻辑:XFF 最右侧由受信代理追加,客户端可控的伪造条目只可能出现在左侧。跳过最右侧 TRUSTED_PROXY_COUNT 个代理条目,取真实客户端 IP。
| 场景 | 行为 |
|---|---|
TRUSTED_PROXY_COUNT=0 | 不信任任何代理头,直接返回 request.client.host |
TRUSTED_PROXY_IPS 未配置或为空 | 同上,fail-closed |
| 直连对端不在白名单中 | 同上(防止应用被直接暴露时客户端伪造 XFF) |
| 对端命中白名单 + XFF 条目少于代理层数 | 回退到 XFF 最右侧条目(避免拿到代理 IP) |
request.client 为 None | 回退到 127.0.0.1 |
在 .env 中配置可信代理:
# 信任的代理层数(0 = 不信任代理头)
TRUSTED_PROXY_COUNT=1
# 可信代理IP白名单(逗号分隔)
TRUSTED_PROXY_IPS=127.0.0.1常见部署配置:
| 部署方式 | TRUSTED_PROXY_COUNT | TRUSTED_PROXY_IPS |
|---|---|---|
| 直连(无代理) | 0 | 留空 |
| 单层 Nginx | 1 | 127.0.0.1(本机)或 Nginx 内网 IP |
| Nginx + CDN | 2 | Nginx 内网 IP |
| Docker bridge 映射端口 | 0 | 留空(容器外为客户端直连) |
安全提示
如果应用被直接暴露在公网(如 Docker bridge 映射端口、未走 Nginx),务必保持 TRUSTED_PROXY_COUNT=0 或 TRUSTED_PROXY_IPS 留空。否则客户端可随意伪造 XFF 首部,绕过基于 IP 的登录锁定和请求限流。
位于 src/utils/ip2region.py,基于 ip2region 离线数据库,无需网络请求。
获取完整区域信息字典,是最底层的查询接口。
def get_ip_region(ip) -> dict:
"""
根据IP获取完整区域信息字典
Returns:
dict: {
"country": "中国", # 国家
"province": "浙江省", # 省份(无数据时为空字符串)
"city": "杭州市", # 城市(无数据时为空字符串)
"isp": "阿里", # 运营商(无数据时为空字符串)
"country_code": "CN" # 国家代码(无数据时为空字符串)
}
"""返回值示例:
| IP | 返回值 |
|---|---|
8.8.8.8 | {"country": "United States", "province": "California", "city": "", "isp": "Google LLC", "country_code": "US"} |
114.114.114.114 | {"country": "中国", "province": "江苏省", "city": "南京市", "isp": "南京信风网络科技", "country_code": "CN"} |
10.0.0.1 | {"country": "内网IP", "province": "", "city": "", "isp": "", "country_code": ""} |
172.abc | {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""} |
获取地理位置文本描述,与原 utils/ip.py 的返回格式兼容。
def get_ip_location(ip) -> str:
"""
格式: "国家 省份 城市"
内网IP返回 "内网IP",查询失败返回 "未知"
示例:
>>> get_ip_location('114.114.114.114')
'中国 江苏省 南京市'
>>> get_ip_location('8.8.8.8')
'United States California'
>>> get_ip_location('10.0.0.1')
'内网IP'
"""去重逻辑
当城市与省份相同时,城市不会重复拼接。例如 province="北京", city="北京" 时返回 "中国 北京" 而非 "中国 北京 北京"。
| 函数 | 返回值 | 示例 |
|---|---|---|
get_ip_country(ip) | 国家 | "中国" |
get_ip_province(ip) | 省份 | "浙江省" |
get_ip_city(ip) | 城市 | "杭州市" |
get_ip_isp(ip) | 运营商 | "阿里" |
ip2region.xdb 文件位于 src/utils/ip2region.xdb,采用懒加载单例模式:
_XDB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ip2region.xdb")
_searcher = None
def _get_searcher():
"""获取搜索器单例(懒加载,首次调用时初始化)"""
global _searcher
if _searcher is not None:
return _searcher
c_buff = util.load_content_from_file(_XDB_PATH)
_searcher = searcher.new_with_buffer(util.IPv4, c_buff)
return _searcherNone,查询降级为 "未知"内网/保留 IP 直接返回 "内网IP",不查询数据库:
| 类型 | 匹配规则 | 示例 |
|---|---|---|
| 回环地址 | 127.0.0.1、localhost、0.0.0.0 | 127.0.0.1 |
| A 类私有 | 10.x.x.x | 10.0.0.1 |
| B 类私有 | 172.16.x.x ~ 172.31.x.x | 172.16.5.5 |
| C 类私有 | 192.168.x.x | 192.168.1.1 |
畸形 IP 防御
畸形输入(如 172.abc)不抛异常,按非内网处理交由查询侧兜底返回 "未知"。这避免了 int(parts[1]) 抛出 ValueError 冒泡到 record_login_log / record_operation_log,导致日志整段丢失。
所有查询接口均采用 fail-open 降级策略——查询失败绝不向上抛异常,保证日志落库不受影响:
except Exception as e:
logger.warning(f"IP地理位置查询失败 [{ip}]: {e}")
return {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""}| 异常场景 | 降级行为 |
|---|---|
| xdb 文件不存在 | get_ip_region 返回 "未知" |
| xdb 加载失败 | 同上 |
| 畸形 IP 地址 | 同上 |
| 查询运行时异常 | 同上 + warning 日志 |
位于 src/utils/ua.py,通过正则匹配从 User-Agent 字符串提取操作系统和浏览器信息。
def parse_user_agent(ua_string):
"""
从 User-Agent 字符串提取 OS 和浏览器信息
Args:
ua_string: HTTP User-Agent 头部字符串
Returns:
tuple: (os_name, browser_name)
示例:
>>> parse_user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
('Windows 10.0', 'Chrome 120.0')
>>> parse_user_agent("Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Mobile/15E148 Safari/604.1")
('iOS 17.2', 'Safari 605.1')
>>> parse_user_agent("")
('', '')
"""| 操作系统 | UA 关键字 | 输出示例 |
|---|---|---|
| Windows | Windows NT 10.0 | Windows 10.0 |
| macOS | Mac OS X 14_2_1 | Mac OS X 14.2.1 |
| Android | Android 14 | Android 14 |
| iPhone | iPhone OS 17_2 | iOS 17.2 |
| iPad | iPad; CPU OS 17_2 | iOS 17.2 |
| Linux | Linux | Linux |
| Ubuntu | Ubuntu | Ubuntu |
按匹配优先级排列(Edge 优先于 Chrome,避免 Edg/ 被误判为 Chrome):
| 浏览器 | UA 关键字 | 输出示例 |
|---|---|---|
| Edge | Edg/120.0.0.0 | Edge 120.0 |
| Chrome | Chrome/120.0.0.0 | Chrome 120.0 |
| Firefox | Firefox/121.0 | Firefox 121.0 |
| Safari | Safari/605.1.15 | Safari 605.1 |
| IE 11 | Trident/...rv:11.0 | IE 11.0 |
| IE 10- | MSIE 10.0 | IE 10.0 |
注意
Safari 的匹配模式 Safari/(\d+\.?\d*) 也会匹配 Chrome UA 中的 Safari 版本号。但由于 Chrome 的匹配在 Safari 之前,实际不会误判。如果需要支持更多浏览器(如 Opera、微信浏览器等),可在 _BROWSER_PATTERNS 列表头部追加。
登录/退出时自动记录 IP、地理位置、操作系统、浏览器:
# src/modules/login_log/service.py
ip = get_client_ip(request)
location = get_ip_location(ip)
ua = request.headers.get('User-Agent', '')
os_name, browser_name = parse_user_agent(ua)
log = LoginLog(
username=username,
ip=ip, # "1.2.3.4"
location=location, # "中国 浙江省 杭州市"
os=os_name, # "Windows 10.0"
browser=browser_name, # "Chrome 120.0"
...
)数据库字段(fastapi_login_log 表):
| 字段 | 类型 | 说明 | 示例 |
|---|---|---|---|
ip | VARCHAR(50) | 客户端 IP | 1.2.3.4 |
location | VARCHAR(100) | IP 归属地 | 中国 浙江省 杭州市 |
os | VARCHAR(100) | 操作系统 | Windows 10.0 |
browser | VARCHAR(100) | 浏览器 | Chrome 120.0 |
写操作(add/update/delete/status)自动记录操作者 IP 和设备信息:
# src/modules/operation_log/service.py
ip = get_client_ip(request)
location = get_ip_location(ip)
ua = request.headers.get('User-Agent', '')
os_name, browser_name = parse_user_agent(ua)数据库字段(fastapi_operation_log 表):
| 字段 | 类型 | 说明 |
|---|---|---|
ip | VARCHAR(100) | 请求 IP |
location | VARCHAR(255) | 请求所在区域 |
os | VARCHAR(100) | 操作系统 |
browser | VARCHAR(100) | 浏览器 |
基于 IP 的 Redis 滑动窗口限流,防止接口被刷:
# src/middleware/rate_limit.py
client_ip = get_client_ip(request)
current_key = f"rl:{client_ip}:{window_index}"
count = await sliding_window_incr(redis, current_key, ...)
if count > RATE_LIMIT_LIMIT:
return JSONResponse({"code": 1, "msg": "请求过于频繁,请 N 秒后再试"})配置项(.env):
| 变量 | 默认值 | 说明 |
|---|---|---|
RATE_LIMIT_ENABLED | False | 是否启用限流 |
RATE_LIMIT_LIMIT | 100 | 单窗口内每 IP 最大请求数 |
RATE_LIMIT_WINDOW_SECONDS | 60 | 窗口时长(秒) |
RATE_LIMIT_KEY_PREFIX | rl: | Redis key 前缀 |
按 IP + 用户名维度统计登录失败次数,超过阈值自动锁定:
# src/modules/auth/service.py
ip = get_client_ip(request)
locked, remaining = await _is_locked(redis, ip, data.username)
if locked:
return R.failed(f"登录失败次数过多,请{remaining}秒后再试")Redis Key 设计:
| Key | 格式 | 说明 |
|---|---|---|
| 失败计数 | login:fail:{ip}:{username} | 累计失败次数 |
| 锁定状态 | login:lock:{ip}:{username} | 锁定时自动设置 TTL |
按 IP 统计验证码请求频率,超过阈值拒绝生成新验证码:
# src/utils/captcha.py
client_ip = get_client_ip(request)
rate_key = f"captcha:rate:{client_ip}"
count = await redis.get(rate_key)
if count and int(count) >= CAPTCHA_RATE_LIMIT:
raise BusinessException(msg="请求过于频繁,请等待 N 秒后再试")# ============================================================
# 可信代理配置
# ============================================================
# 信任的代理层数(0 = 不信任代理头,直接使用 request.client.host)
TRUSTED_PROXY_COUNT=1
# 可信代理IP白名单(逗号分隔,仅白名单内的直连对端才信任 XFF)
TRUSTED_PROXY_IPS=127.0.0.1
# ============================================================
# 请求限流配置
# ============================================================
# 是否启用请求限流(默认关闭)
RATE_LIMIT_ENABLED=False
# 单窗口内每个 IP 允许的最大请求数
RATE_LIMIT_LIMIT=100
# 统计窗口时长(秒)
RATE_LIMIT_WINDOW_SECONDS=60
# Redis 计数 key 前缀
RATE_LIMIT_KEY_PREFIX=rl:原因:本地直连无代理,request.client.host 就是 127.0.0.1。
说明:这是正常行为。生产环境配置 TRUSTED_PROXY_COUNT 和 TRUSTED_PROXY_IPS 后即可获取真实客户端 IP。
可能原因:
解决:检查 src/utils/ip2region.xdb 文件是否存在,启动日志中是否有加载成功提示。
原因:客户端 IP 在内网/保留地址段内(10.x、172.16-31.x、192.168.x、127.0.0.1)。
说明:常见于开发环境或内网部署。生产环境通过反代获取真实外网 IP 即可。
排查步骤:
.env 中 RATE_LIMIT_ENABLED=TrueTRUSTED_PROXY_COUNT 配置——如果限流基于 request.client.host(代理 IP),所有请求会共享同一 IP,导致限流提前触发或形同虚设从 ip2region 官方仓库 下载最新的 ip2region.xdb 文件,替换 src/utils/ip2region.xdb,重启应用即可。首次查询时自动加载新数据。
| 模块 | 文件 | 核心能力 | 降级策略 |
|---|---|---|---|
| IP 获取 | src/utils/ip.py | 从 Request 提取真实客户端 IP,支持可信代理白名单 | client 为 None 时回退 127.0.0.1 |
| 地理位置 | src/utils/ip2region.py | 基于 ip2region.xdb 离线查询,零网络依赖 | 异常/畸形/内网统一降级返回 "未知" 或 "内网IP" |
| UA 解析 | src/utils/ua.py | 正则匹配提取操作系统和浏览器 | UA 为空时返回 ('', '') |
三者组合为登录日志、操作日志、限流、登录锁定、验证码防刷等场景提供完整的客户端身份信息链路。所有接口均采用降级策略,保证日志落库等核心流程不受异常影响。