fix(auth): Quick Connect poll capability + unlinked KOSync registration

CHANGELOG.md

@@ -31,6 +31,23 @@ All notable changes to Tome are documented here. Format loosely follows
   progress bar shows a font-size-agnostic "p. 142 of 384" from the print
   edition's page count alongside the percentage.
 
+### Security
+- **Quick Connect codes can no longer be brute-forced into a login.** Polling
+  a code used to require nothing but the code itself — six characters, guessed
+  fast enough, handed out a full login token once the code was authorized. The
+  device that requests a code now also receives a long random poll secret, and
+  only that pairing can collect the login; a guessed code gets the same "not
+  found" as a nonexistent one. The short code is still all you type on the
+  already-signed-in device — nothing changes in the flow you see.
+- **A KOReader sync registration can no longer claim your account.** The KOSync
+  registration endpoint (which KOReader's "Register" button calls without any
+  authentication) used to attach the new sync credential to whatever Tome
+  account had the same username — so anyone who knew a username could register
+  it first and have their reading activity land on that account. Device-side
+  registrations now start unlinked; connecting sync to your account is done
+  signed-in via Settings → KOReader Sync, which also reclaims a squatted name
+  by overwriting its key.
+
 ### Fixed
 - **Reorganize is crash-safe and works across filesystems.** Library Health's
   one-click reorganise used to move every file first and write the database

backend/api/kosync.py

@@ -58,13 +58,16 @@ def create_kosync_user(body: dict[str, Any], db: Session = Depends(get_db)):
     if existing:
         raise HTTPException(status_code=402, detail="User already exists")
 
-    # Auto-link to Tome user with the same username if one exists
-    tome_user = db.query(User).filter(User.username == username).first()
-
+    # This endpoint is unauthenticated (KOReader's register button), so it must
+    # never attach the credential to a Tome account — matching by username would
+    # let anyone claim an account's sync identity just by knowing the username.
+    # Linking happens through the authenticated path (Settings → KOReader Sync,
+    # POST /api/auth/me/kosync), which also reclaims a squatted name by
+    # overwriting its key.
     kosync_user = KOSyncUser(
         username=username,
         userkey=password,
-        user_id=tome_user.id if tome_user else None,
+        user_id=None,
     )
     db.add(kosync_user)
     db.commit()

backend/api/quick_connect.py

@@ -1,4 +1,12 @@
-"""Quick Connect — sign in on a new device using a short code."""
+"""Quick Connect — sign in on a new device using a short code.
+
+The short code is only ever typed on the ALREADY-authenticated device
+(authorize). The new device keeps a long random poll_token from initiate and
+must present it to poll — so the 6-character code space cannot be brute-forced
+into a login JWT.
+"""
+import hmac
+import secrets
 from datetime import datetime, timedelta
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -17,13 +25,19 @@
 
 class InitiateResponse(BaseModel):
     code: str
+    poll_token: str
     expires_at: datetime
 
 
 class AuthorizeRequest(BaseModel):
     code: str
 
 
+class PollRequest(BaseModel):
+    code: str
+    poll_token: str
+
+
 @router.post("/initiate", response_model=InitiateResponse)
 def initiate(db: Session = Depends(get_db)):
     """Generate a new Quick Connect code. No authentication required."""
@@ -38,11 +52,16 @@ def initiate(db: Session = Depends(get_db)):
     while db.query(QuickConnectCode).filter(QuickConnectCode.code == code).first():
         code = generate_code()
 
-    entry = QuickConnectCode(code=code, created_at=now, expires_at=expires)
+    entry = QuickConnectCode(
+        code=code,
+        poll_token=secrets.token_urlsafe(32),
+        created_at=now,
+        expires_at=expires,
+    )
     db.add(entry)
     db.commit()
     db.refresh(entry)
-    return InitiateResponse(code=entry.code, expires_at=entry.expires_at)
+    return InitiateResponse(code=entry.code, poll_token=entry.poll_token, expires_at=entry.expires_at)
 
 
 @router.post("/authorize", status_code=200)
@@ -75,11 +94,15 @@ def authorize(
     return {"status": "authorized"}
 
 
-@router.get("/poll/{code}")
-def poll(code: str, request: Request, db: Session = Depends(get_db)):
-    """Poll for code authorization status. Returns JWT once authorized."""
-    entry = db.query(QuickConnectCode).filter(QuickConnectCode.code == code.upper().strip()).first()
-    if not entry:
+@router.post("/poll")
+def poll(body: PollRequest, request: Request, db: Session = Depends(get_db)):
+    """Poll for code authorization status. Returns JWT once authorized.
+
+    Requires the poll_token issued by initiate — a guessed code alone gets the
+    same 410 as an unknown one, so there is no oracle and nothing to steal.
+    """
+    entry = db.query(QuickConnectCode).filter(QuickConnectCode.code == body.code.upper().strip()).first()
+    if not entry or not entry.poll_token or not hmac.compare_digest(entry.poll_token, body.poll_token):
         raise HTTPException(status_code=status.HTTP_410_GONE, detail="Code not found or expired")
     if entry.expires_at < datetime.utcnow():
         db.delete(entry)

backend/main.py

@@ -139,6 +139,12 @@ async def lifespan(app: FastAPI):
         if "scope" not in at_cols:
             conn.execute(text("ALTER TABLE api_tokens ADD COLUMN scope VARCHAR(16) NOT NULL DEFAULT 'full'"))
             conn.commit()
+        # Quick Connect poll capability. Pre-existing rows keep NULL — they can
+        # never be polled again, which is fine: codes live for five minutes.
+        qc_cols = {r[1] for r in conn.execute(text("PRAGMA table_info(quick_connect_codes)")).fetchall()}
+        if qc_cols and "poll_token" not in qc_cols:
+            conn.execute(text("ALTER TABLE quick_connect_codes ADD COLUMN poll_token VARCHAR(64)"))
+            conn.commit()
         # Per-user ratings/reviews on user_book_status (nullable — existing rows stay unrated).
         ubs_cols = {r[1] for r in conn.execute(text("PRAGMA table_info(user_book_status)")).fetchall()}
         if ubs_cols and "rating" not in ubs_cols:

backend/models/quick_connect.py

@@ -22,6 +22,9 @@ class QuickConnectCode(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     code: Mapped[str] = mapped_column(String(8), unique=True, nullable=False, index=True)
+    # Capability held by the device that initiated the code: polling requires it,
+    # so guessing the short display code is never enough to steal the login JWT.
+    poll_token: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
     user_id: Mapped[Optional[int]] = mapped_column(
         Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=True, index=True
     )

frontend/src/pages/LoginPage.tsx

@@ -93,7 +93,7 @@ export function LoginPage() {
     setQcError(null)
     setQcLoading(true)
     try {
-      const data = await apiFetch<{ code: string; expires_at: string }>('/auth/quick-connect/initiate', {
+      const data = await apiFetch<{ code: string; poll_token: string; expires_at: string }>('/auth/quick-connect/initiate', {
         method: 'POST',
         headers: { 'Content-Type': 'application/json' },
       })
@@ -116,7 +116,12 @@ export function LoginPage() {
       pollRef.current = setInterval(async () => {
         try {
           const res = await apiFetch<{ status: string; access_token?: string; token_type?: string }>(
-            `/auth/quick-connect/poll/${data.code}`
+            '/auth/quick-connect/poll',
+            {
+              method: 'POST',
+              headers: { 'Content-Type': 'application/json' },
+              body: JSON.stringify({ code: data.code, poll_token: data.poll_token }),
+            }
           )
           if (res.status === 'authorized' && res.access_token) {
             if (pollRef.current) clearInterval(pollRef.current)

tests/test_auth_hardening.py

@@ -0,0 +1,145 @@
+"""Auth hardening: Quick Connect poll capability + KOSync create binding.
+
+Covers the fixes from the 2026-07-03 auth surface review:
+- Quick Connect: polling requires the poll_token issued at initiate, so the
+  6-character display code can no longer be brute-forced into a login JWT.
+- KOSync /users/create (unauthenticated) never binds the new sync credential
+  to an existing Tome account; linking is the authenticated Settings path.
+"""
+import hashlib
+from datetime import datetime, timedelta
+
+from starlette.testclient import TestClient
+
+from backend.models.kosync import KOSyncUser
+from backend.models.quick_connect import QuickConnectCode
+
+
+# ---------------------------------------------------------------------------
+# Quick Connect
+# ---------------------------------------------------------------------------
+
+class TestQuickConnectPollToken:
+    def test_initiate_returns_poll_token(self, client: TestClient):
+        resp = client.post("/api/auth/quick-connect/initiate")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data["code"]) == 6
+        assert len(data["poll_token"]) >= 32
+
+    def test_full_flow_with_token(self, client: TestClient, admin_user):
+        user, _ = admin_user
+        init = client.post("/api/auth/quick-connect/initiate").json()
+
+        # Pending until authorized
+        pending = client.post("/api/auth/quick-connect/poll",
+                              json={"code": init["code"], "poll_token": init["poll_token"]})
+        assert pending.status_code == 200
+        assert pending.json()["status"] == "pending"
+
+        # Authorize from the logged-in device (client carries the admin JWT)
+        auth = client.post("/api/auth/quick-connect/authorize", json={"code": init["code"]})
+        assert auth.status_code == 200
+
+        done = client.post("/api/auth/quick-connect/poll",
+                           json={"code": init["code"], "poll_token": init["poll_token"]})
+        assert done.status_code == 200
+        body = done.json()
+        assert body["status"] == "authorized"
+        assert body["access_token"]
+
+    def test_poll_with_wrong_token_is_indistinguishable_from_unknown_code(
+        self, client: TestClient
+    ):
+        init = client.post("/api/auth/quick-connect/initiate").json()
+        client.post("/api/auth/quick-connect/authorize", json={"code": init["code"]})
+
+        # Correct code, wrong token: the brute-forcer's position. Must get the
+        # same 410 as a nonexistent code — no oracle, no JWT.
+        wrong = client.post("/api/auth/quick-connect/poll",
+                            json={"code": init["code"], "poll_token": "A" * 43})
+        unknown = client.post("/api/auth/quick-connect/poll",
+                              json={"code": "ZZZZZZ", "poll_token": "A" * 43})
+        assert wrong.status_code == 410
+        assert unknown.status_code == 410
+        assert wrong.json() == unknown.json()
+
+        # The code is still consumable by the legitimate holder afterwards.
+        ok = client.post("/api/auth/quick-connect/poll",
+                         json={"code": init["code"], "poll_token": init["poll_token"]})
+        assert ok.status_code == 200
+        assert ok.json()["status"] == "authorized"
+
+    def test_legacy_tokenless_rows_cannot_be_polled(self, client: TestClient, db, admin_user):
+        user, _ = admin_user
+        # A pre-migration row (poll_token NULL) that is already authorized.
+        entry = QuickConnectCode(
+            code="ABCDEF", poll_token=None, user_id=user.id,
+            created_at=datetime.utcnow(),
+            expires_at=datetime.utcnow() + timedelta(minutes=5),
+            authorized_at=datetime.utcnow(),
+        )
+        db.add(entry)
+        db.commit()
+
+        resp = client.post("/api/auth/quick-connect/poll",
+                           json={"code": "ABCDEF", "poll_token": ""})
+        assert resp.status_code == 410
+
+    def test_old_get_poll_route_yields_no_jwt(self, client: TestClient):
+        # The tokenless GET route is removed; unmatched GETs fall through to
+        # the SPA catch-all. Whatever comes back, it must never carry a JWT.
+        init = client.post("/api/auth/quick-connect/initiate").json()
+        client.post("/api/auth/quick-connect/authorize", json={"code": init["code"]})
+        resp = client.get(f"/api/auth/quick-connect/poll/{init['code']}")
+        assert "access_token" not in resp.text
+
+    def test_expired_code_polls_410(self, client: TestClient, db):
+        init = client.post("/api/auth/quick-connect/initiate").json()
+        entry = db.query(QuickConnectCode).filter(QuickConnectCode.code == init["code"]).first()
+        entry.expires_at = datetime.utcnow() - timedelta(seconds=1)
+        db.commit()
+
+        resp = client.post("/api/auth/quick-connect/poll",
+                           json={"code": init["code"], "poll_token": init["poll_token"]})
+        assert resp.status_code == 410
+
+
+# ---------------------------------------------------------------------------
+# KOSync registration
+# ---------------------------------------------------------------------------
+
+class TestKOSyncCreateBinding:
+    def test_create_does_not_bind_to_existing_tome_account(
+        self, client: TestClient, db, admin_user
+    ):
+        user, _ = admin_user
+        # Attacker registers a KOSync credential under the victim's username.
+        resp = client.post("/api/v1/users/create",
+                           json={"username": user.username, "password": "attacker-md5"})
+        assert resp.status_code == 201
+
+        ks = db.query(KOSyncUser).filter(KOSyncUser.username == user.username).first()
+        assert ks is not None
+        assert ks.user_id is None  # NOT linked to the Tome account
+
+    def test_settings_register_reclaims_squatted_username(
+        self, client: TestClient, db, admin_user
+    ):
+        user, _ = admin_user
+        client.post("/api/v1/users/create",
+                    json={"username": user.username, "password": "attacker-md5"})
+
+        # The real user links from Settings (authenticated): the credential is
+        # re-keyed to their password and bound to their account.
+        resp = client.post("/api/auth/me/kosync", json={"password": "devicepass"})
+        assert resp.status_code == 201
+
+        ks = db.query(KOSyncUser).filter(KOSyncUser.username == user.username).first()
+        assert ks.user_id == user.id
+        assert ks.userkey == hashlib.md5(b"devicepass").hexdigest()
+
+    def test_duplicate_create_still_402(self, client: TestClient):
+        client.post("/api/v1/users/create", json={"username": "reader", "password": "k1"})
+        dup = client.post("/api/v1/users/create", json={"username": "reader", "password": "k2"})
+        assert dup.status_code == 402