bb774c332b
背景：命令行方案要求用户在电脑前开交互式终端，实际不可行。 改为在网页里完成 MFA，手机也能操作。 难点：garth 索取验证码走的是 *阻塞回调*，0.4.46 没有 "发起登录 -> 返回句柄 -> 稍后续接" 的接口，登录必须一直挂着。 而 gunicorn 跑多个 worker，验证码请求不一定落到挂着登录的那个 worker。 方案：登录跑在后台线程里，停在 prompt_mfa 内轮询数据库； 浏览器用另一个请求把验证码写进同一行。**汇合点是数据库而非进程内存**， 所以哪个 worker 收到验证码都能送达。 - 新增 garmin_mfa_sessions 表（不存密码，密码只活在等待线程的内存里） - services/garmin_auth.py：start_login / submit_code / cancel 状态机 starting -> awaiting_code -> finishing -> done|failed - 超时 5 分钟自动放弃，会话 1 小时后清理 - 会话按 user_id 校验，他人拿到 session id 也读不到、提交不了 接口: - POST /api/garmin/login 发起登录，202 返回 session - GET /api/garmin/login-status 轮询状态 - POST /api/garmin/mfa 提交验证码 - DELETE /api/garmin/login 取消 前端 DataSync 改为三步： - 未绑定 -> 输密码「绑定 Garmin 账号」 - 需要验证码 -> 弹出 6 位验证码输入框（inputMode=numeric、 autoComplete=one-time-code，手机可直接从短信自动填充） - 已绑定 -> 只剩「立即同步」，不再要密码 tests/test_garmin_mfa.py (20 通过): - stub 的 prompt_mfa 按 garth 的真实方式同步阻塞调用 - 关键用例：验证码直接写进数据库行也能被挂起的线程取到 （模拟验证码落到另一个 worker） - 无 MFA 的账号不经验证码直接完成 - 验证码错误 / 密码错误 / 等待超时 各自失败并给出原因 - 取消后挂起线程立即释放，不空转到超时 - 密码不出现在会话行里 - 跨用户读取和提交均被拒 全量: 271 passed Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
#!/usr/bin/env python
|
||
"""
|
||
One-time interactive Garmin login.
|
||
|
||
Garmin accounts with two-factor auth cannot be logged into by the web service:
|
||
the library asks for the MFA code on stdin, and a gunicorn worker has none —
|
||
the attempt fails with "EOFError: EOF when reading a line".
|
||
|
||
This script does that login in a terminal, where a code can actually be typed,
|
||
and stores the resulting OAuth tokens in the database. Every later sync loads
|
||
those tokens and skips the login entirely. They stay valid for roughly a year;
|
||
re-run this when a sync starts reporting an expired session.
|
||
|
||
Usage (on the NAS):
|
||
cd ~/apps/garmin-health-lab/backend
|
||
.venv/bin/python garmin_login.py
|
||
|
||
Add --email to pick an account when more than one is registered.
|
||
"""
|
||
import argparse
|
||
import getpass
|
||
import sys
|
||
|
||
import db
|
||
from services import garmin as garmin_svc
|
||
|
||
|
||
def pick_user(email=None):
|
||
if email:
|
||
row = db.query_one("SELECT id, email, garmin_email FROM users WHERE email = ?",
|
||
[email])
|
||
if not row:
|
||
sys.exit(f"找不到账号: {email}")
|
||
return row
|
||
|
||
rows = db.query_all("SELECT id, email, garmin_email FROM users ORDER BY created_at")
|
||
if not rows:
|
||
sys.exit("数据库里还没有账号，请先在网页上注册。")
|
||
if len(rows) == 1:
|
||
return rows[0]
|
||
|
||
print("有多个账号，请选择：")
|
||
for i, r in enumerate(rows, 1):
|
||
print(f" {i}) {r['email']} (Garmin: {r['garmin_email']})")
|
||
choice = input("序号: ").strip()
|
||
try:
|
||
return rows[int(choice) - 1]
|
||
except (ValueError, IndexError):
|
||
sys.exit("选择无效")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="一次性 Garmin 登录，保存令牌")
|
||
parser.add_argument("--email", help="要绑定的本站账号邮箱")
|
||
parser.add_argument("--garmin-email", help="Garmin 账号邮箱（默认用注册时填的）")
|
||
args = parser.parse_args()
|
||
|
||
db.init_db()
|
||
user = pick_user(args.email)
|
||
garmin_email = args.garmin_email or user["garmin_email"]
|
||
|
||
print(f"本站账号 : {user['email']}")
|
||
print(f"Garmin : {garmin_email}")
|
||
print()
|
||
|
||
if garmin_svc.has_token(user["id"]):
|
||
if input("已存在登录令牌，要覆盖吗？[y/N] ").strip().lower() != "y":
|
||
return
|
||
|
||
password = getpass.getpass("Garmin 密码: ")
|
||
if not password:
|
||
sys.exit("密码不能为空")
|
||
|
||
Garmin = garmin_svc._import_garmin()
|
||
client = Garmin(email=garmin_email, password=password, is_cn=garmin_svc._is_cn())
|
||
|
||
def ask_mfa():
|
||
# garth's built-in prompt is a bare English input() that is easy to
|
||
# miss in the surrounding output, so this replaces it with something
|
||
# unmistakable.
|
||
print("\n" + "=" * 52)
|
||
print(" 账号开启了两步验证，请查收短信/邮件中的验证码")
|
||
print("=" * 52)
|
||
while True:
|
||
code = input(" 验证码（6 位数字）: ").strip()
|
||
if code:
|
||
return code
|
||
print(" 验证码不能为空，请重新输入。")
|
||
|
||
print("\n正在登录……")
|
||
try:
|
||
# Call garth directly rather than Garmin.login(): only this path lets
|
||
# the MFA prompt be replaced. The two lines afterwards are what
|
||
# Garmin.login() would otherwise populate.
|
||
client.garth.login(garmin_email, password, prompt_mfa=ask_mfa)
|
||
client.display_name = client.garth.profile["displayName"]
|
||
client.full_name = client.garth.profile["fullName"]
|
||
except Exception as e:
|
||
sys.exit(f"\n登录失败: {type(e).__name__}: {e}")
|
||
|
||
garmin_svc.save_token(user["id"], client.garth.dumps(), garmin_email)
|
||
|
||
print(f"\n登录成功：{client.display_name}")
|
||
print("令牌已保存到数据库，之后网页上的同步不再需要密码或验证码。")
|
||
print("令牌大约一年后过期，届时重跑本脚本即可。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|