1c80b16325
用户问：限流了，重新输账号密码验证码换个新令牌行不行。
不行，而且是最糟的一种试法。`garth.login()` 和 `refresh_oauth2()` 打的是
同一个 SSO 端点，流程还更重；限流按**账号**计（不是按 IP、按 UA），换设备
换网络都绕不开；而窗口内每次尝试都会把窗口往后推。
而这正是被卡住时第一个会去试的操作，代码里却只有 `_connect` 的刷新有闸门，
重新绑定那条路照发不误。
- start_login 在 sso 冷却窗口内直接拒绝，不建会话行、不碰网络
- 错误信息说清三件事：为什么现在不试、什么时候恢复、换设备没用
- 路由返 429（请求本身没毛病，是该晚点再来）并带 retryAfterSeconds
- 数据端点的 429 不参与拦截，force 可以推翻
前端补上 UI：报错文案早先承诺了「同步页选择强制重试」，但那个按钮不存在。
现在只在被冷却拒绝之后才出现，样式刻意做得不像第二个「开始同步」——它是给
估算失准时的出口，不是随手可点的第二选择。
顺带修一个正要被我引入的 bug：`onClick={syncHistory}` 会把 MouseEvent 当成
force 传进去，等于每次点开始同步都跳过冷却。
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""
|
|
Unit tests for the web-driven two-factor Garmin login.
|
|
|
|
No network and no real library: a stub Garmin/garth pair stands in, and its
|
|
`prompt_mfa` callback is invoked exactly the way garth invokes it — blocking,
|
|
mid-login — because that blocking behaviour is the whole reason this flow
|
|
needs a background thread and a database rendezvous.
|
|
"""
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from services import garmin as garmin_svc
|
|
from services import garmin_auth
|
|
|
|
|
|
def wait_for(predicate, timeout=8.0, interval=0.05):
|
|
"""Poll until the background thread reaches the expected state."""
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
value = predicate()
|
|
if value:
|
|
return value
|
|
time.sleep(interval)
|
|
return None
|
|
|
|
|
|
def status_of(session_id):
|
|
row = garmin_auth.get_session(session_id)
|
|
return row["status"] if row else None
|
|
|
|
|
|
def wait_status(session_id, *wanted, timeout=8.0):
|
|
return wait_for(lambda: status_of(session_id) in wanted and status_of(session_id),
|
|
timeout=timeout)
|
|
|
|
|
|
class StubGarth:
|
|
profile = {"displayName": "Tester", "fullName": "Test User"}
|
|
|
|
def __init__(self, needs_mfa=True, accept_code="123456", fail_login=False):
|
|
self.needs_mfa = needs_mfa
|
|
self.accept_code = accept_code
|
|
self.fail_login = fail_login
|
|
self.seen_code = None
|
|
self.logged_in_with = None
|
|
|
|
def login(self, email, password, prompt_mfa=None):
|
|
self.logged_in_with = (email, password)
|
|
if self.fail_login:
|
|
raise RuntimeError("401 Unauthorized")
|
|
if self.needs_mfa:
|
|
# garth calls this synchronously, in the middle of the login.
|
|
self.seen_code = prompt_mfa()
|
|
if self.seen_code != self.accept_code:
|
|
raise RuntimeError("验证码错误")
|
|
|
|
def dumps(self):
|
|
return "token-blob"
|
|
|
|
|
|
class StubGarmin:
|
|
"""Class factory: `make()` returns something usable as `Garmin`."""
|
|
|
|
last = None
|
|
|
|
@classmethod
|
|
def make(cls, **garth_kwargs):
|
|
def factory(is_cn=False, **_):
|
|
instance = cls()
|
|
instance.garth = StubGarth(**garth_kwargs)
|
|
cls.last = instance
|
|
return instance
|
|
return lambda: factory
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fast_polling(monkeypatch):
|
|
"""Keep the rendezvous poll short so tests stay quick."""
|
|
monkeypatch.setattr(garmin_auth, "POLL_INTERVAL_SECONDS", 0.05)
|
|
monkeypatch.setattr(garmin_auth, "CODE_WAIT_SECONDS", 5)
|
|
|
|
|
|
class TestMfaFlow:
|
|
def test_login_parks_waiting_for_a_code(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
assert wait_status(sid, "awaiting_code") == "awaiting_code"
|
|
|
|
def test_submitting_the_code_completes_the_login(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
|
|
ok, _ = garmin_auth.submit_code(sid, user["id"], "123456")
|
|
assert ok is True
|
|
assert wait_status(sid, "done", "failed") == "done"
|
|
|
|
def test_token_is_saved_on_success(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
garmin_auth.submit_code(sid, user["id"], "123456")
|
|
wait_status(sid, "done", "failed")
|
|
|
|
assert garmin_svc.has_token(user["id"]) is True
|
|
assert garmin_svc.load_token(user["id"]) == "token-blob"
|
|
|
|
def test_the_code_reaches_garth(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
garmin_auth.submit_code(sid, user["id"], "123456")
|
|
wait_status(sid, "done", "failed")
|
|
|
|
assert StubGarmin.last.garth.seen_code == "123456"
|
|
|
|
def test_account_without_mfa_completes_without_a_code(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw",
|
|
import_garmin=StubGarmin.make(needs_mfa=False),
|
|
)
|
|
assert wait_status(sid, "done", "failed") == "done"
|
|
assert garmin_svc.has_token(user["id"]) is True
|
|
|
|
def test_wrong_code_fails_with_a_reason(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
garmin_auth.submit_code(sid, user["id"], "000000")
|
|
assert wait_status(sid, "done", "failed") == "failed"
|
|
|
|
assert "验证码错误" in garmin_auth.get_session(sid)["error"]
|
|
assert garmin_svc.has_token(user["id"]) is False
|
|
|
|
def test_bad_password_fails_before_any_code(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "wrong",
|
|
import_garmin=StubGarmin.make(fail_login=True),
|
|
)
|
|
assert wait_status(sid, "failed") == "failed"
|
|
assert "401" in garmin_auth.get_session(sid)["error"]
|
|
|
|
def test_timeout_when_no_code_arrives(self, db, user, monkeypatch):
|
|
monkeypatch.setattr(garmin_auth, "CODE_WAIT_SECONDS", 0.2)
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
assert wait_status(sid, "failed") == "failed"
|
|
assert "超时" in garmin_auth.get_session(sid)["error"]
|
|
|
|
|
|
class TestRendezvousIsNotInMemory:
|
|
"""The handoff must survive the code arriving on a different worker, so it
|
|
goes through the database rather than process memory."""
|
|
|
|
def test_code_written_directly_to_the_row_is_picked_up(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
|
|
# Exactly what another worker's request would do: write the row.
|
|
db.execute(
|
|
"UPDATE garmin_mfa_sessions SET code = ? WHERE id = ?", ["123456", sid]
|
|
)
|
|
assert wait_status(sid, "done", "failed") == "done"
|
|
|
|
def test_cancelling_releases_the_parked_thread(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
garmin_auth.cancel(sid, user["id"])
|
|
|
|
# The row is gone, so the waiting thread must stop rather than spin
|
|
# until its timeout.
|
|
assert wait_for(lambda: garmin_auth.get_session(sid) is None)
|
|
|
|
|
|
class TestSessionIsolation:
|
|
def test_another_users_session_is_not_readable(self, db, user, make_user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
other = make_user("o@example.com")
|
|
|
|
assert garmin_auth.get_session(sid, other["id"]) is None
|
|
|
|
def test_another_user_cannot_submit_a_code(self, db, user, make_user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "g@example.com", "pw", import_garmin=StubGarmin.make()
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
other = make_user("o2@example.com")
|
|
|
|
ok, _ = garmin_auth.submit_code(sid, other["id"], "123456")
|
|
assert ok is False
|
|
|
|
def test_unknown_session_is_refused(self, db, user):
|
|
ok, msg = garmin_auth.submit_code("no-such-session", user["id"], "123456")
|
|
assert ok is False
|
|
assert "不存在" in msg
|
|
|
|
|
|
class TestPasswordHandling:
|
|
def test_password_is_never_written_to_the_session_row(self, db, user):
|
|
sid = garmin_auth.start_login(
|
|
user["id"], "hunter2@example.com", "SuperSecret123",
|
|
import_garmin=StubGarmin.make(),
|
|
)
|
|
wait_status(sid, "awaiting_code")
|
|
|
|
row = garmin_auth.get_session(sid)
|
|
assert "SuperSecret123" not in str(dict(row))
|
|
|
|
|
|
class TestEndpoints:
|
|
def test_all_require_auth(self, client):
|
|
assert client.post("/api/garmin/login", json={}).status_code == 401
|
|
assert client.get("/api/garmin/login-status?session=x").status_code == 401
|
|
assert client.post("/api/garmin/mfa", json={}).status_code == 401
|
|
|
|
def test_login_requires_a_password(self, client, auth):
|
|
r = client.post("/api/garmin/login", headers=auth, json={})
|
|
assert r.status_code == 400
|
|
|
|
def test_login_returns_a_session(self, client, auth, monkeypatch):
|
|
monkeypatch.setattr(
|
|
garmin_auth, "start_login", lambda *a, **k: "session-123"
|
|
)
|
|
r = client.post(
|
|
"/api/garmin/login",
|
|
headers=auth,
|
|
json={"garminEmail": "g@example.com", "garminPassword": "pw"},
|
|
)
|
|
assert r.status_code == 202
|
|
assert r.get_json()["session"] == "session-123"
|
|
|
|
def test_status_of_unknown_session_is_404(self, client, auth):
|
|
assert client.get(
|
|
"/api/garmin/login-status?session=nope", headers=auth
|
|
).status_code == 404
|
|
|
|
def test_mfa_requires_both_fields(self, client, auth):
|
|
assert client.post(
|
|
"/api/garmin/mfa", headers=auth, json={"session": "x"}
|
|
).status_code == 400
|
|
|
|
def test_full_flow_through_http(self, client, auth, user, db, monkeypatch):
|
|
monkeypatch.setattr(
|
|
garmin_svc, "_import_garmin", StubGarmin.make()
|
|
)
|
|
r = client.post(
|
|
"/api/garmin/login",
|
|
headers=auth,
|
|
json={"garminEmail": "g@example.com", "garminPassword": "pw"},
|
|
)
|
|
sid = r.get_json()["session"]
|
|
|
|
assert wait_status(sid, "awaiting_code") == "awaiting_code"
|
|
assert client.get(
|
|
f"/api/garmin/login-status?session={sid}", headers=auth
|
|
).get_json()["status"] == "awaiting_code"
|
|
|
|
r = client.post(
|
|
"/api/garmin/mfa", headers=auth, json={"session": sid, "code": "123456"}
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
assert wait_status(sid, "done", "failed") == "done"
|
|
assert client.get("/api/garmin/auth-status", headers=auth).get_json()[
|
|
"hasToken"] is True
|
|
|
|
|
|
class TestRebindingIsGatedToo:
|
|
""""Just re-enter the password and get a fresh token" is the obvious thing
|
|
to try when syncing is blocked — and it is the worst thing to try.
|
|
|
|
`garth.login()` is the same SSO endpoint that is doing the blocking, by a
|
|
heavier path than the token refresh, and the limit is keyed to the account
|
|
so a new device or network reaches the same wall.
|
|
"""
|
|
|
|
def test_login_is_refused_inside_the_window(self, db, user):
|
|
garmin_svc._note_rate_limit(user["id"], "sso")
|
|
with pytest.raises(garmin_auth.LoginRateLimited):
|
|
garmin_auth.start_login(user["id"], "a@example.com", "pw")
|
|
|
|
def test_no_session_row_is_created_by_a_refusal(self, db, user):
|
|
garmin_svc._note_rate_limit(user["id"], "sso")
|
|
with pytest.raises(garmin_auth.LoginRateLimited):
|
|
garmin_auth.start_login(user["id"], "a@example.com", "pw")
|
|
rows = db.query_all(
|
|
"SELECT id FROM garmin_mfa_sessions WHERE user_id = ?", [user["id"]])
|
|
assert rows == []
|
|
|
|
def test_a_data_429_does_not_block_rebinding(self, db, user, monkeypatch):
|
|
started = []
|
|
monkeypatch.setattr(garmin_auth.threading, "Thread",
|
|
lambda **kw: type("T", (), {"start": lambda s: started.append(1)})())
|
|
garmin_svc._note_rate_limit(user["id"], "data")
|
|
garmin_auth.start_login(user["id"], "a@example.com", "pw")
|
|
assert started == [1]
|
|
|
|
def test_force_overrules_the_estimate(self, db, user, monkeypatch):
|
|
started = []
|
|
monkeypatch.setattr(garmin_auth.threading, "Thread",
|
|
lambda **kw: type("T", (), {"start": lambda s: started.append(1)})())
|
|
garmin_svc._note_rate_limit(user["id"], "sso")
|
|
garmin_auth.start_login(user["id"], "a@example.com", "pw", force=True)
|
|
assert started == [1]
|
|
|
|
def test_the_endpoint_answers_429_with_a_retry_hint(self, client, auth, db, user):
|
|
garmin_svc._note_rate_limit(user["id"], "sso")
|
|
resp = client.post("/api/garmin/login",
|
|
json={"garminPassword": "pw", "garminEmail": "a@example.com"},
|
|
headers=auth)
|
|
assert resp.status_code == 429
|
|
body = resp.get_json()
|
|
assert body["retryAfterSeconds"] > 0
|
|
assert "延长封锁" in body["error"]
|