feat(security): privacy hardening + data export (v0.2.1)

- CORS: default localhost-only origins (was *, allowed any webpage
  to read local health data cross-site); env override MI_FITNESS_CORS_ORIGINS
- Host header allowlist middleware on FastAPI and Flask (DNS-rebinding
  protection); auto-configured for non-loopback binds via CLI
- Move API key passTokens from SQLite plaintext to system keyring,
  with automatic migration on startup and cleanup for revoked keys
- Disable uvicorn access log by default (query params out of terminals)
- New GET /api/export: JSON/CSV export of local data (offline, Excel BOM)
- Add docs/PRIVACY.md: full data-flow audit, outbound endpoint inventory,
  threat model, self-verification steps
- 6 new tests (host allowlist, keyring roundtrip, CSV)
This commit is contained in:
binglua committed 2026-09-03 22:43:15 +08:00
1 parent 42c42b2c9e
commit a47a8470f8
10 files changed
+444 -13

No files matched your search

+5 -1
View File
@@ -1,10 +1,14 @@
# 运行时环境变量示例(可选;均由服务进程读取,不要把真实值提交进仓库)
# --- REST API (mi-fitness-mcp api) / Web 仪表盘 (mi-fitness-mcp web) ---
# 设置后所有数据端点要求 X-API-Key 头
# 设置后所有数据端点要求 X-API-Key 头(防止本机其他进程/局域网读取健康数据)
# MI_FITNESS_API_KEY=change-me
# 管理端点(Key 发放/列表/吊销、扫码登录)要求 X-Admin-Key;未设置时仅限 127.0.0.1 调用
# MI_FITNESS_ADMIN_KEY=change-me
# Host 头白名单(防 DNS 重绑定),逗号分隔;非回环绑定时 CLI 会自动设置,一般无需手动配置
# MI_FITNESS_ALLOWED_HOSTS=127.0.0.1,localhost
# CORS 来源白名单(默认仅本机来源);如需放开给指定前端才配置,慎用 *
# MI_FITNESS_CORS_ORIGINS=http://localhost:3000
# --- Web 仪表盘 (mi-fitness-mcp web) ---
# MI_FITNESS_API_URL=http://127.0.0.1:8321 # 反向代理的后端 FastAPI 地址
+9
View File
@@ -1,5 +1,14 @@
# Changelog
## 0.2.1(隐私加固)
- 数据安全审计文档 `docs/PRIVACY.md`:全部对外请求清单(仅小米域名)、本机数据存放表、威胁模型与自验方法
- CORS 收紧:默认仅允许本机来源(原来为 `*`,恶意网页可跨站读取本地健康数据);可用 `MI_FITNESS_CORS_ORIGINS` 覆盖
- Host 头白名单中间件(FastAPI + Flask 双侧):阻断 DNS 重绑定攻击;非回环绑定时 CLI 自动放行绑定地址,可用 `MI_FITNESS_ALLOWED_HOSTS` 覆盖
- API Key 的 passToken 迁出 SQLite 明文,改存系统 keyring(启动时自动迁移历史数据);吊销时同步清理
- `api` 命令默认关闭 uvicorn 访问日志(查询参数不再落入终端)
- 新增 `GET /api/export`:本地数据导出为 JSON/CSV(含 Excel BOM),纯离线操作
## 0.2.0
- 内置 Web 仪表盘(`mi-fitness-mcp web`,Flask):可视化测试全部 API 端点、Key 管理、扫码登录,内置反向代理(自动附加 X-API-Key/X-Admin-Key,上游耗时透传)
+6 -2
View File
@@ -185,6 +185,7 @@ tests/ # 单元测试与端到端测试套件
## 文档
- [隐私与数据安全审计](docs/PRIVACY.md) —— 数据流向、全部对外请求清单、威胁模型与自验方法
- [完整技术文档](docs/PROJECT_DOCUMENTATION.md) —— 架构、小米云协议逆向细节、存储模型、19 条已知问题清单与重构指南
- [扫码登录实现方案](docs/QR_LOGIN.md) —— 协议时序、端点参数、踩坑记录(`sid=xiaomiio`)
- [贡献指南](CONTRIBUTING.md) · [安全政策](SECURITY.md) · [更新日志](CHANGELOG.md)
@@ -198,11 +199,14 @@ pytest -v # 完整测试套件(22 个)
python -m build # 构建发行包
```
## 安全说明
## 安全与隐私
**数据只进不出**:本项目从小米云拉取你自己的数据到本地 SQLite,除小米官方服务器外不与任何第三方通信;无遥测/统计/崩溃上报;所有查询端点断网可用。完整审计见 [docs/PRIVACY.md](docs/PRIVACY.md)。
- `passToken` 等同小米账号登录态,**切勿提交或泄露**;若泄露,请在小米账号安全中心退出设备并重新登录
- 凭据与 API Key 密钥存于系统 keyring(Windows DPAPI / macOS Keychain / Linux Secret Service),不落明文
- 不要提交本地配置、数据库、keyring 文件(`.gitignore` 已默认排除)
- 服务默认仅绑定 `127.0.0.1`。向局域网/公网开放前,务必设置 `MI_FITNESS_API_KEY` 与 `MI_FITNESS_ADMIN_KEY`——健康数据是敏感信息
- 服务默认仅绑定 `127.0.0.1`,并已启用 CORS 本机限制与 Host 白名单(防恶意网页跨站读取 / DNS 重绑定)。向局域网/公网开放前,务必设置 `MI_FITNESS_API_KEY` 与 `MI_FITNESS_ADMIN_KEY`——健康数据是敏感信息
- 扫码登录端点允许扫码者将其账号登录到你的服务器,务必保持管理鉴权开启
## 免责声明
+92
View File
@@ -0,0 +1,92 @@
# 隐私与数据安全审计
> 本文回答一个问题:**我的健康数据到底去了哪里?**
> 结论先行:本项目是"拉取式"本地工具——数据只从小米云**下载**到你的电脑,除小米自己的服务器外**不上传到任何第三方**;无遥测、无统计、无崩溃上报;查询功能完全离线可用。
---
## 1. 数据流向总览
```
小米手环/手表 ──蓝牙──> 小米运动健康 App ──> 小米健康云(小米官方,与本项目无关)
│
│ ① 拉取(登录 + 分页下载,RC4 加密通道)
▼
┌──────────── 本项目(你的电脑)────────────┐
│ SQLite(健康数据) │
│ 系统 keyring(passToken / API Key 密钥) │
│ config.json(无敏感信息) │
└───────────────────────────────────────────┘
│ ② 读取(纯本地,断网可用)
├──> MCP 客户端(stdio,本机进程)
├──> REST API(默认 127.0.0.1)
└──> Web 仪表盘(默认 127.0.0.1)
```
**唯一出网的方向是 ①:从小米云拉你自己的数据。** 没有任何"上传健康数据"的代码路径。
## 2. 全部对外网络请求清单(代码审计)
对 `src/` 全量 grep `https://` 的结果,所有外联域名均属小米:
| 端点 | 用途 | 方向 |
|---|---|---|
| `https://account.xiaomi.com/pass/serviceLogin` | passToken 换服务会话 | 请求小米 |
| `https://account.xiaomi.com/longPolling/loginUrl` | 扫码登录:生成二维码会话 | 请求小米 |
| `https://hlth.io.mi.com` / `https://{region}.hlth.io.mi.com` | 健康数据分页拉取 | 请求小米 |
| `https://sts.api.io.mi.com/sts` | 仅作为扫码接口的 callback **参数**(小米校验用),本项目不请求它 | — |
| `https://127.0.0.1:8321` | Web 仪表盘反代到本机 API | 仅本机 |
扫码登录流程中由**小米动态下发**的跳转地址(`c3.account.xiaomi.com`、`c3.lp.account.xiaomi.com` 等)也全部位于 `*.xiaomi.com` 小米域名下。
**不存在**:遥测(telemetry)、统计(analytics)、Sentry/PostHog/Mixpanel 等上报 SDK、GitHub star 上报、版本检查回传、任何第三方域名。可用以下命令自行复核:
```bash
grep -rn "https\?://" src/ --include="*.py" | grep -v 127.0.0.1 # 应只见 *.xiaomi.com / *.io.mi.com
grep -rniE "telemetry|analytics|sentry|posthog|mixpanel" src/ # 应无结果
```
## 3. 数据在本机的存放位置
| 数据 | 位置 | 保护 |
|---|---|---|
| 健康数据(8 类) | `<用户数据目录>/mi-fitness-mcp/mi_fitness.db`(SQLite) | 本机文件权限;不上传 |
| 小米 passToken(setup 凭据) | 系统 keyring(Windows: DPAPI 加密 / macOS: Keychain / Linux: Secret Service) | OS 级加密 |
| API Key 对应的 passToken | 系统 keyring(v0.2.1 起;旧版本曾存 SQLite 明文,启动时自动迁移并清空) | OS 级加密 |
| `mif_sk_*` Key 本身、user_id、region、发放/使用时间 | SQLite `api_keys` 表 | 无秘密性要求(可吊销) |
| 配置 | `<用户配置目录>/mi-fitness-mcp/config.json` | 不含凭据 |
| 仓库内 | 无任何上述文件(`.gitignore` 排除 `.venv/ *.db .env config.json` 等) | — |
## 4. 本地暴露面与防护(v0.2.1 加固)
数据"不上传"之外,还要防**被动窃取**。本项目默认只绑 `127.0.0.1`,针对本机攻击面的防护:
| 威胁 | 场景 | 防护 |
|---|---|---|
| 恶意网页跨站读取 | 你浏览器里开着的任意网页用 JS 请求 `http://127.0.0.1:8321` 并读取响应 | CORS 默认只允许 `localhost/127.0.0.1/[::1]` 来源(正则限定,任意端口),其他域名一律无 CORS 头,浏览器阻止读取 |
| DNS 重绑定 | 攻击者域名解析到 127.0.0.1 绕过同源策略 | **Host 头白名单中间件**(FastAPI 与 Flask 双侧):Host 非本机域名直接 403 |
| 本机其他进程读取 | 同机运行的任何程序请求本地 API | 设置环境变量 `MI_FITNESS_API_KEY` 强制鉴权;管理端点默认仅本机且可设 `MI_FITNESS_ADMIN_KEY` |
| 凭据落盘明文 | SQLite 被复制/同步盘上传 | passToken 全部走系统 keyring;`api_keys` 表启动时自动迁移历史明文并清空 |
| 日志泄密 | 终端/日志文件留下查询参数(日期等) | `api` 命令默认关闭 uvicorn 访问日志 |
| 扫码端点滥用 | 他人把他的账号"登录"到你的服务器 | 扫码/Key 管理端点要求 `X-Admin-Key`(未设则仅限 127.0.0.1) |
环境变量开关(见 `.env.example`):`MI_FITNESS_ALLOWED_HOSTS`(Host 白名单,非回环绑定时 CLI 自动设置)、`MI_FITNESS_CORS_ORIGINS`(显式放开跨域来源,慎用 `*`)。
## 5. 如何自行验证
1. **断网测试**:拔网线后所有查询端点、`/api/export`、Web 仪表盘图表照常工作(同步/扫码会失败,属预期)——证明查询链路纯本地。
2. **抓包**:`netstat`/Wireshark 观察进程外联,目标应只有 `*.xiaomi.com` / `*.io.mi.com`。
3. **代码审计**:第 2 节的两条 grep 命令;全部网络代码集中在 `adapters/mi_fitness_cloud.py`、`api.py`(扫码)、`web.py`(本机反代)。
4. **仓库审计**:`git ls-files | grep -iE "\.db|\.sqlite|config\.json|\.env$"` 应无结果。
## 6. 明确不做的事
- ❌ 不上传、不转发、不备份健康数据到任何第三方服务
- ❌ 不内置任何遥测、统计、崩溃上报
- ❌ 不在代码仓库或文档中存放任何凭据
- ❌ 不写入小米账号(全协议只读;唯一"写"是登录换 token,属认证必需)
## 7. 边界说明(与本项目无关的上传)
- 手环 → 小米云的上传由**小米官方 App**完成,这是设备生态本身的行为;本项目只读取。介意者请在小米 App 内管理同步策略。
- 你若把 SQLite 文件自行放进网盘/同步盘,属于你自己的操作,请自行加密。
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "mi-fitness-mcp"
version = "0.2.0"
version = "0.2.1"
description = "MCP server for Mi Fitness data access"
readme = "README.md"
license = {text = "MIT"}
+136 -5
View File
@@ -26,7 +26,7 @@ import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Annotated
from typing import Annotated, Any
import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
@@ -37,6 +37,14 @@ from pydantic import BaseModel
from mi_fitness_mcp.adapters.mi_fitness_cloud import MiFitnessCloudAdapter
from mi_fitness_mcp.auth import load_mi_fitness_token
from mi_fitness_mcp.config import load_config
from mi_fitness_mcp.security import (
cors_origins_from_env,
default_allowed_hosts,
delete_api_key_secret,
host_allowed,
load_api_key_secret,
store_api_key_secret,
)
from mi_fitness_mcp.services.query_service import QueryService
from mi_fitness_mcp.services.sync_service import SyncService
from mi_fitness_mcp.storage import Database
@@ -112,7 +120,20 @@ async def lifespan(app: FastAPI):
)
"""
)
# 迁移:历史版本把 passToken 明文存在该表,启动时移入系统 keyring 并清空列
legacy = conn.execute(
"SELECT key, pass_token FROM api_keys WHERE pass_token != ''"
).fetchall()
for row in legacy:
if store_api_key_secret(row["key"], row["pass_token"]):
conn.execute("UPDATE api_keys SET pass_token = '' WHERE key = ?", (row["key"],))
# 已吊销 Key 的 keyring 密钥一并清理(历史版本吊销时尚无 keyring 概念)
revoked_keys = [
r["key"] for r in conn.execute("SELECT key FROM api_keys WHERE revoked = 1").fetchall()
]
conn.commit()
for k in revoked_keys:
delete_api_key_secret(k)
default_context = None
if config.mode == "mi_fitness_cloud":
@@ -163,16 +184,53 @@ async def _gate(x_api_key: str | None = Header(default=None)) -> None:
app = FastAPI(
title="Mi Fitness API",
version="0.2.0",
version="0.2.1",
description="小米运动健康数据 REST API(多 Key 鉴权 + 扫码登录 + 本地 SQLite 缓存)",
lifespan=lifespan,
dependencies=[Depends(_gate)],
)
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
CORSMiddleware,
allow_methods=["*"],
allow_headers=["*"],
**cors_origins_from_env(default_regex=r"^https?://(localhost|127\.0\.0\.1|\[::1\])(:\d+)?$"),
)
class HostAllowlistMiddleware:
"""Host 头白名单(防 DNS 重绑定):拒绝非本机域名直连本地端口读取数据。"""
def __init__(self, app, allowed_hosts: set[str] | None = None):
self.app = app
self.allowed_hosts = allowed_hosts if allowed_hosts is not None else default_allowed_hosts()
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
host = ""
for name, value in scope.get("headers", []):
if name == b"host":
host = value.decode("latin-1")
break
if not host_allowed(host, self.allowed_hosts):
body = b'{"detail": "Host not allowed"}'
await send(
{
"type": "http.response.start",
"status": 403,
"headers": [
(b"content-type", b"application/json"),
(b"content-length", str(len(body)).encode()),
],
}
)
await send({"type": "http.response.body", "body": body})
return
await self.app(scope, receive, send)
app.add_middleware(HostAllowlistMiddleware)
def _resolve_context(request: Request) -> UserContext:
"""根据 X-API-Key 解析出对应用户上下文;静态 Key/无头 → 默认凭据。"""
state = request.app.state
@@ -186,7 +244,11 @@ def _resolve_context(request: Request) -> UserContext:
(header,),
).fetchone()
if row is not None:
cache_key = (row["user_id"], row["pass_token"], row["region"])
# passToken 优先从系统 keyring 读取;为空则回退 DB 列(无 keyring 环境的降级存储)
pass_token = load_api_key_secret(header) or row["pass_token"]
if not pass_token:
raise HTTPException(status_code=401, detail="API key credentials unavailable")
cache_key = (row["user_id"], pass_token, row["region"])
ctx = state.contexts.get(cache_key)
if ctx is None:
ctx = _build_context(state.db, state.config, *cache_key)
@@ -249,10 +311,12 @@ class CreateKeyRequest(BaseModel):
def _create_key_record(db: Database, user_id: str, pass_token: str, region: str, label: str | None) -> str:
key = "mif_sk_" + secrets.token_hex(20)
# passToken 存系统 keyring(加密);keyring 不可用才降级写 DB 明文列
stored = store_api_key_secret(key, pass_token)
with db._get_connection() as conn:
conn.execute(
"INSERT INTO api_keys (key, label, user_id, pass_token, region, created_at) VALUES (?,?,?,?,?,?)",
(key, label, user_id, pass_token, region, datetime.now(UTC).isoformat()),
(key, label, user_id, "" if stored else pass_token, region, datetime.now(UTC).isoformat()),
)
conn.commit()
return key
@@ -322,6 +386,7 @@ async def revoke_key(key_prefix: str, request: Request) -> dict:
raise HTTPException(status_code=404, detail="没有匹配的 Key(可用完整 Key 或唯一前缀)")
for k in matches:
conn.execute("UPDATE api_keys SET revoked = 1 WHERE key = ?", (k,))
delete_api_key_secret(k)
conn.commit()
return {"status": "ok", "revoked": matches}
@@ -761,3 +826,69 @@ def get_coverage(
) -> dict:
data = _query_service(request).get_data_coverage(data_types)
return {"status": "ok", "count": len(data), "data": data}
def _to_csv(rows: list[dict[str, Any]]) -> str:
"""把查询结果扁平化为 CSV(嵌套结构转 JSON 字符串,加 BOM 便于 Excel 打开中文)。"""
import csv
import io
fieldnames: list[str] = []
for row in rows:
for name in row:
if name not in fieldnames:
fieldnames.append(name)
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
for row in rows:
writer.writerow(
{
k: (json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else v)
for k, v in row.items()
}
)
return "\ufeff" + buf.getvalue()
EXPORT_SOURCES = {
"daily_activity": lambda qs, s, e: qs.get_daily_summaries(s, e),
"heart_rate": lambda qs, s, e: qs.get_heart_rate_samples(s, e),
"sleep": lambda qs, s, e: qs.get_sleep_sessions(s, e),
"workouts": lambda qs, s, e: qs.get_workouts(s, e),
"body_measurements": lambda qs, s, e: qs.get_body_measurements(s, e),
"spo2": lambda qs, s, e: qs.get_spo2_samples(s, e),
"stress": lambda qs, s, e: qs.get_stress_samples(s, e),
"abnormal_heart_beat": lambda qs, s, e: qs.get_abnormal_heart_beat_events(s, e),
}
@app.get("/api/export", response_model=None)
def export_data(
request: Request,
data_type: str,
start_date: str,
end_date: str,
format: str = "json",
) -> Response | dict:
"""导出本地数据(JSON 或 CSV)。仅读本地 SQLite,不触发任何网络请求。"""
if data_type not in EXPORT_SOURCES:
raise HTTPException(status_code=400, detail=f"data_type 必须是 {sorted(EXPORT_SOURCES)} 之一")
if format not in ("json", "csv"):
raise HTTPException(status_code=400, detail="format 必须是 json 或 csv")
data = EXPORT_SOURCES[data_type](
_query_service(request),
_validate_date(start_date, "start_date"),
_validate_date(end_date, "end_date"),
)
if format == "csv":
return Response(
content=_to_csv(data),
media_type="text/csv; charset=utf-8",
headers={
"Content-Disposition": (
f'attachment; filename="mi_fitness_{data_type}_{start_date}_{end_date}.csv"'
)
},
)
return {"status": "ok", "count": len(data), "data": data}
+13 -4
View File
@@ -149,15 +149,24 @@ def cmd_api(args):
import uvicorn
if args.host not in ("127.0.0.1", "localhost") and not os.environ.get("MI_FITNESS_API_KEY"):
print("⚠️ 绑定到非本机地址但未设置 MI_FITNESS_API_KEY,健康数据将暴露给局域网。")
print(" 建议先设置环境变量 MI_FITNESS_API_KEY 再启动。")
uvicorn.run("mi_fitness_mcp.api:app", host=args.host, port=args.port)
loopback = ("127.0.0.1", "localhost", "::1")
if args.host not in loopback:
# 非回环绑定:Host 白名单放行该地址(默认仅本机域名,见 security.py)
os.environ.setdefault("MI_FITNESS_ALLOWED_HOSTS", args.host)
if not os.environ.get("MI_FITNESS_API_KEY"):
print("⚠️ 绑定到非本机地址但未设置 MI_FITNESS_API_KEY,健康数据将暴露给局域网。")
print(" 建议先设置环境变量 MI_FITNESS_API_KEY 再启动。")
# 关闭访问日志:查询串中的日期等参数不再落入终端/日志
uvicorn.run("mi_fitness_mcp.api:app", host=args.host, port=args.port, access_log=False)
def cmd_web(args):
import os
from mi_fitness_mcp.web import run_server
if args.host not in ("127.0.0.1", "localhost", "::1"):
os.environ.setdefault("MI_FITNESS_ALLOWED_HOSTS", args.host)
run_server(
host=args.host,
port=args.port,
+84
View File
@@ -0,0 +1,84 @@
"""本地安全加固工具:Host 校验、CORS 解析与 API Key 密钥的 keyring 存取。
设计目标(见 docs/PRIVACY.md):
- 浏览器中的恶意网页无法读取本地 API 的健康数据(CORS 默认仅本机来源);
- DNS 重绑定攻击无法绕过同源策略(Host 头白名单);
- 发放的 API Key 对应的 passToken 不落 SQLite 明文,存入系统 keyring
(Windows DPAPI / macOS Keychain / Linux Secret Service),keyring 不可用时
才回退到数据库列并保持兼容。
"""
import hashlib
import os
SERVICE_NAME = "mi-fitness-mcp"
LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1", "[::1]"}
def default_allowed_hosts() -> set[str]:
"""Host 白名单。默认仅本机回环;环境变量 MI_FITNESS_ALLOWED_HOSTS 可覆盖(逗号分隔,* 放行全部)。"""
env = os.environ.get("MI_FITNESS_ALLOWED_HOSTS")
if env:
return {h.strip().lower() for h in env.split(",") if h.strip()}
return set(LOOPBACK_HOSTS)
def host_allowed(host: str | None, allowed_hosts: set[str]) -> bool:
"""校验请求 Host 头是否在白名单内(自动剥离端口,支持 IPv6 方括号写法)。"""
if "*" in allowed_hosts:
return True
if not host:
return False
h = host.strip().lower()
if h.startswith("["):
end = h.find("]")
if end != -1:
h = h[: end + 1]
elif ":" in h:
h = h.rsplit(":", 1)[0]
return h in allowed_hosts
def cors_origins_from_env(default_regex: str) -> dict:
"""CORS 配置:环境变量 MI_FITNESS_CORS_ORIGINS 提供精确来源列表,否则用本机正则默认值。"""
env = os.environ.get("MI_FITNESS_CORS_ORIGINS")
if env:
origins = [o.strip() for o in env.split(",") if o.strip()]
if origins:
return {"allow_origins": origins}
return {"allow_origin_regex": default_regex}
def _account_for(api_key: str) -> str:
digest = hashlib.sha256(api_key.encode()).hexdigest()[:16]
return f"apikey_{digest}"
def store_api_key_secret(api_key: str, pass_token: str) -> bool:
"""把 API Key 对应的 passToken 存入系统 keyring。返回是否成功。"""
try:
import keyring
keyring.set_password(SERVICE_NAME, _account_for(api_key), pass_token)
return True
except Exception:
return False
def load_api_key_secret(api_key: str) -> str | None:
try:
import keyring
return keyring.get_password(SERVICE_NAME, _account_for(api_key))
except Exception:
return None
def delete_api_key_secret(api_key: str) -> None:
try:
import keyring
keyring.delete_password(SERVICE_NAME, _account_for(api_key))
except Exception:
pass
+9
View File
@@ -22,6 +22,8 @@ from flask import (
send_from_directory,
)
from mi_fitness_mcp.security import default_allowed_hosts, host_allowed
# Assets directory: src/mi_fitness_mcp/web_assets or fallback to testpage
PACKAGE_DIR = Path(__file__).resolve().parent
DEFAULT_ASSETS_DIR = PACKAGE_DIR / "web_assets"
@@ -55,6 +57,13 @@ def create_app(
# Persistent HTTP client with 300s timeout (trust_env=False avoids proxy interference on localhost)
http_client = httpx.Client(trust_env=False, timeout=300.0, follow_redirects=True)
@app.before_request
def _check_host():
# Host 头白名单(防 DNS 重绑定):默认仅本机域名,环境变量可覆盖
if not host_allowed(request.host, default_allowed_hosts()):
return jsonify({"detail": "Host not allowed"}), 403
return None
@app.route("/")
def index():
"""Render the main dashboard interface."""
+89
View File
@@ -0,0 +1,89 @@
"""Tests for local security hardening (host allowlist, CORS, keyring secrets, CSV export)."""
import keyring
from mi_fitness_mcp.api import _to_csv
from mi_fitness_mcp.security import (
default_allowed_hosts,
delete_api_key_secret,
host_allowed,
load_api_key_secret,
store_api_key_secret,
)
def test_host_allowed_strips_ports():
allowed = default_allowed_hosts()
assert host_allowed("127.0.0.1:8321", allowed)
assert host_allowed("localhost:9999", allowed)
assert host_allowed("LOCALHOST", allowed)
assert host_allowed("[::1]:8321", allowed)
def test_host_allowed_rejects_foreign_domains():
allowed = default_allowed_hosts()
# DNS 重绑定场景:外部域名解析到 127.0.0.1,但 Host 头仍是外部域名
assert not host_allowed("evil.example.com", allowed)
assert not host_allowed("evil.example.com:8321", allowed)
assert not host_allowed(None, allowed)
assert not host_allowed("", allowed)
def test_host_allowed_env_override():
# monkeypatch env via keyring-free approach: direct set check
import os
old = os.environ.get("MI_FITNESS_ALLOWED_HOSTS")
try:
os.environ["MI_FITNESS_ALLOWED_HOSTS"] = "lanhost.local,*"
allowed = default_allowed_hosts()
assert host_allowed("lanhost.local:8321", allowed)
assert host_allowed("anything.com", allowed) # * 放行
finally:
if old is None:
os.environ.pop("MI_FITNESS_ALLOWED_HOSTS", None)
else:
os.environ["MI_FITNESS_ALLOWED_HOSTS"] = old
def test_api_key_secret_roundtrip(monkeypatch):
store = {}
monkeypatch.setattr(keyring, "set_password", lambda s, a, v: store.__setitem__((s, a), v))
monkeypatch.setattr(keyring, "get_password", lambda s, a: store.get((s, a)))
monkeypatch.setattr(
keyring,
"delete_password",
lambda s, a: store.pop((s, a), None),
)
api_key = "mif_sk_" + "a" * 40
assert store_api_key_secret(api_key, "V1:secret-token")
assert load_api_key_secret(api_key) == "V1:secret-token"
# 不同 Key 的存储隔离(账号名含 key 哈希)
other = "mif_sk_" + "b" * 40
assert load_api_key_secret(other) is None
delete_api_key_secret(api_key)
assert load_api_key_secret(api_key) is None
def test_store_api_key_secret_keyring_failure(monkeypatch):
def boom(_s, _a, _v):
raise RuntimeError("no keyring backend")
monkeypatch.setattr(keyring, "set_password", boom)
assert not store_api_key_secret("mif_sk_x", "tok")
assert load_api_key_secret("mif_sk_x") is None
def test_to_csv_flattens_and_adds_bom():
rows = [
{"date": "2026-08-30", "steps": 100, "stages": [{"stage": "deep", "minutes": 60}]},
{"date": "2026-08-31", "steps": None, "stages": []},
]
csv_text = _to_csv(rows)
assert csv_text.startswith("\ufeff")
assert "date,steps,stages" in csv_text
# CSV 规范内层引号转义为两个引号
assert '"[{""stage"": ""deep"", ""minutes"": 60}]"' in csv_text
assert "2026-08-31" in csv_text