# IP/UA 解析 :::tip 说明 提供客户端 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(真实客户端 IP) │ │ │ └─ 否 → request.client.host │ │ └─ client 为 None → 回退 127.0.0.1 │ └──────────┬───────────────────────────────────────────┘ │ ┌─────┴─────┐ ▼ ▼ ┌─────────┐ ┌──────────────────────────────────┐ │ 限流 │ │ get_ip_region(ip) │ │ 登录锁定 │ │ ├─ 内网/保留 IP → "内网IP" │ │ 验证码防刷│ │ ├─ ip2region.xdb 离线查询 │ └─────────┘ │ └─ 异常降级 → "未知" │ └──────────┬───────────────────────┘ │ ▼ ┌──────────────────────┐ │ parse_user_agent(ua) │ │ ├─ 操作系统(正则) │ │ └─ 浏览器(正则) │ └──────────┬───────────┘ │ ▼ ┌──────────────────────┐ │ 写入日志表 │ │ ip / location / │ │ os / browser │ └──────────────────────┘ ``` ## 一、IP 获取 位于 `src/utils/ip.py`,从 FastAPI Request 获取客户端真实 IP。 ### get_client_ip(request) ```python def get_client_ip(request) -> str: """ 获取客户端真实IP地址 优先从 X-Forwarded-For 头取最左侧IP(即真实客户端IP), 无代理头时回退到 request.client.host。 """ ``` **返回值**:客户端 IP 地址字符串(如 `"1.2.3.4"`)。 ### 代理链解析原理 当应用部署在 Nginx/CDN 等反向代理之后,`request.client.host` 拿到的是代理 IP 而非真实客户端 IP。此时从 `X-Forwarded-For`(XFF)头取最左侧 IP 即为真实客户端 IP。 ``` 客户端 (1.2.3.4) │ ▼ CDN (10.0.0.1) ← 追加 XFF: 1.2.3.4 │ ▼ Nginx ← 透传 XFF: 1.2.3.4(不追加) │ ▼ FastAPI 应用 X-Forwarded-For = "1.2.3.4, 10.0.0.1" get_client_ip → 取最左侧 → "1.2.3.4" ``` **取值逻辑**:XFF 最左侧由 CDN/反代追加的真实客户端 IP,右侧为代理节点 IP。直接取最左侧即可。 :::warning Nginx 配置注意 Nginx 代理配置应使用 `$http_x_forwarded_for`(透传上游值)而非 `$proxy_add_x_forwarded_for`(会追加本机 IP),避免 XFF 条目数不可控: ```nginx proxy_set_header X-Forwarded-For $http_x_forwarded_for; ``` ::: ### 降级策略 | 场景 | 行为 | |------|------| | X-Forwarded-For 存在 | 取最左侧 IP | | X-Forwarded-For 为空 | 返回 `request.client.host` | | `request.client` 为 None | 回退到 `127.0.0.1` | ## 二、IP 地理位置查询 位于 `src/utils/ip2region.py`,基于 [ip2region](https://github.com/lionsoul2014/ip2region) 离线数据库,无需网络请求。 ### 核心接口 #### get_ip_region(ip) → dict 获取完整区域信息字典,是最底层的查询接口。 ```python 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": ""}` | #### get_ip_location(ip) → str 获取地理位置文本描述,与原 `utils/ip.py` 的返回格式兼容。 ```python 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' """ ``` :::tip 去重逻辑 当城市与省份相同时,城市不会重复拼接。例如 `province="北京", city="北京"` 时返回 `"中国 北京"` 而非 `"中国 北京 北京"`。 ::: #### 便捷函数 | 函数 | 返回值 | 示例 | |:-----|:-------|:-----| | `get_ip_country(ip)` | 国家 | `"中国"` | | `get_ip_province(ip)` | 省份 | `"浙江省"` | | `get_ip_city(ip)` | 城市 | `"杭州市"` | | `get_ip_isp(ip)` | 运营商 | `"阿里"` | ### 离线数据库 ip2region.xdb 文件位于 `src/utils/ip2region.xdb`,采用懒加载单例模式: ```python _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 _searcher ``` - **首次查询时**加载 xdb 文件到内存(约 10MB),后续查询零 IO 延迟 - **xdb 文件不存在**时返回 `None`,查询降级为 `"未知"` - xdb 数据定期更新可从 ip2region 官方仓库获取 ### 内网 IP 判断 内网/保留 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` | :::warning 畸形 IP 防御 畸形输入(如 `172.abc`)不抛异常,按非内网处理交由查询侧兜底返回 `"未知"`。这避免了 `int(parts[1])` 抛出 `ValueError` 冒泡到 `record_login_log` / `record_operation_log`,导致日志整段丢失。 ::: ### 异常降级策略 所有查询接口均采用 **fail-open** 降级策略——查询失败绝不向上抛异常,保证日志落库不受影响: ```python except Exception as e: logger.warning(f"IP地理位置查询失败 [{ip}]: {e}") return {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""} ``` | 异常场景 | 降级行为 | |:---------|:---------| | xdb 文件不存在 | `get_ip_region` 返回 `"未知"` | | xdb 加载失败 | 同上 | | 畸形 IP 地址 | 同上 | | 查询运行时异常 | 同上 + warning 日志 | ## 三、UA 解析 位于 `src/utils/ua.py`,通过正则匹配从 User-Agent 字符串提取操作系统和浏览器信息。 ### parse_user_agent(ua_string) ```python 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` | :::warning 注意 Safari 的匹配模式 `Safari/(\d+\.?\d*)` 也会匹配 Chrome UA 中的 Safari 版本号。但由于 Chrome 的匹配在 Safari 之前,实际不会误判。如果需要支持更多浏览器(如 Opera、微信浏览器等),可在 `_BROWSER_PATTERNS` 列表头部追加。 ::: ## 四、应用场景 ### 1. 登录日志 登录/退出时自动记录 IP、地理位置、操作系统、浏览器: ```python # 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` | ### 2. 操作日志 写操作(add/update/delete/status)自动记录操作者 IP 和设备信息: ```python # 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)` | 浏览器 | ### 3. 请求限流 基于 IP 的 Redis 滑动窗口限流,防止接口被刷: ```python # 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 前缀 | ### 4. 登录失败锁定 按 IP + 用户名维度统计登录失败次数,超过阈值自动锁定: ```python # 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 | ### 5. 验证码防刷 按 IP 统计验证码请求频率,超过阈值拒绝生成新验证码: ```python # 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 秒后再试") ``` ## 五、配置参考 ### Nginx 代理配置 ```nginx location /api/ { proxy_pass http://127.0.0.1:8031/api/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; # 透传上游X-Forwarded-For(不追加本机IP);无上游头时回退到直连IP proxy_set_header X-Forwarded-For $http_x_forwarded_for; } ``` ### 请求限流配置(.env) ```bash # ============================================================ # 请求限流配置 # ============================================================ # 是否启用请求限流(默认关闭) RATE_LIMIT_ENABLED=False # 单窗口内每个 IP 允许的最大请求数 RATE_LIMIT_LIMIT=100 # 统计窗口时长(秒) RATE_LIMIT_WINDOW_SECONDS=60 # Redis 计数 key 前缀 RATE_LIMIT_KEY_PREFIX=rl: ``` ## 六、常见问题 ### 开发环境日志中 IP 显示为 127.0.0.1 **原因**:本地直连无代理,`request.client.host` 就是 `127.0.0.1`。 **说明**:这是正常行为。生产环境部署在 Nginx/CDN 后会从 `X-Forwarded-For` 获取真实客户端 IP。 ### 地理位置显示为"未知" **可能原因**: 1. ip2region.xdb 文件不存在或加载失败(检查启动日志) 2. 查询的 IP 是 IPv6(当前仅支持 IPv4) 3. 畸形 IP 地址 **解决**:检查 `src/utils/ip2region.xdb` 文件是否存在,启动日志中是否有加载成功提示。 ### 地理位置显示为"内网IP" **原因**:客户端 IP 在内网/保留地址段内(`10.x`、`172.16-31.x`、`192.168.x`、`127.0.0.1`)。 **说明**:常见于开发环境或内网部署。生产环境通过反代获取真实外网 IP 即可。 ### 限流不生效 **排查步骤**: 1. 检查 `.env` 中 `RATE_LIMIT_ENABLED=True` 2. 检查 Redis 连接是否正常 3. 检查 Nginx 是否正确透传 `X-Forwarded-For`——如果限流基于代理 IP 而非真实客户端 IP,所有请求会共享同一 IP,导致限流提前触发或形同虚设 ### 如何更新 ip2region 数据库 从 [ip2region 官方仓库](https://github.com/lionsoul2014/ip2region) 下载最新的 `ip2region.xdb` 文件,替换 `src/utils/ip2region.xdb`,重启应用即可。首次查询时自动加载新数据。 ## 七、总结 | 模块 | 文件 | 核心能力 | 降级策略 | |:-----|:-----|:---------|:---------| | IP 获取 | `src/utils/ip.py` | 从 Request 提取真实客户端 IP(优先 X-Forwarded-For 最左侧) | client 为 None 时回退 `127.0.0.1` | | 地理位置 | `src/utils/ip2region.py` | 基于 ip2region.xdb 离线查询,零网络依赖 | 异常/畸形/内网统一降级返回 `"未知"` 或 `"内网IP"` | | UA 解析 | `src/utils/ua.py` | 正则匹配提取操作系统和浏览器 | UA 为空时返回 `('', '')` | 三者组合为登录日志、操作日志、限流、登录锁定、验证码防刷等场景提供完整的客户端身份信息链路。所有接口均采用降级策略,保证日志落库等核心流程不受异常影响。