fix(security): consume magic-link tokens atomically

Validation and marking-used were two separate steps, so two concurrent
requests for the same registration token could both pass validation before
either marked it used — a replay window.

Add an atomic consume() at the repository (conditional UPDATE ... WHERE
used = 0 AND not expired, gated on rowcount) and service layers, and switch
the /register handler to consume() instead of validate()+mark_used().

Refs: porchlight-ur7

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/porchlight/authn/routes.py

@@ -77,7 +77,8 @@ async def register_magic_link(request: Request, token: str) -> Response:
     magic_link_service = request.app.state.magic_link_service
     user_repo = request.app.state.user_repo
 
-    link = await magic_link_service.validate(token)
+    # Atomically validate and consume the token (single-use, no replay race).
+    link = await magic_link_service.consume(token)
     if link is None:
         return HTMLResponse("<p>Invalid or expired registration link.</p>", status_code=400)
 
@@ -91,8 +92,6 @@ async def register_magic_link(request: Request, token: str) -> Response:
         user = User(userid=userid, username=link.username, groups=["users"])
         await user_repo.create(user)
 
-    await magic_link_service.mark_used(token)
-
     request.session["userid"] = user.userid
     request.session["username"] = user.username
 

src/porchlight/invite/service.py

@@ -66,6 +66,19 @@ class MagicLinkService:
         """Mark a magic link as used. Returns True if found and marked."""
         return await self._repo.mark_used(self._hash_token(token))
 
+    async def consume(self, token: str) -> MagicLink | None:
+        """Atomically validate and consume a token in one step.
+
+        Prefer this over validate()+mark_used(): it closes the replay race
+        where two concurrent requests could both pass validation before either
+        marks the token used. Returns the link (raw token re-attached) on
+        success, else None.
+        """
+        link = await self._repo.consume(self._hash_token(token))
+        if link is None:
+            return None
+        return link.model_copy(update={"token": token})
+
     async def cleanup_expired(self) -> int:
         """Delete expired unused links. Returns count deleted."""
         return await self._repo.delete_expired()

src/porchlight/store/protocols.py

@@ -55,6 +55,8 @@ class MagicLinkRepository(Protocol):
 
     async def mark_used(self, token: str) -> bool: ...
 
+    async def consume(self, token: str) -> MagicLink | None: ...
+
     async def delete_expired(self) -> int: ...
 
 

src/porchlight/store/sqlite/repositories.py

@@ -311,6 +311,25 @@ class SQLiteMagicLinkRepository:
         await self._db.commit()
         return cursor.rowcount > 0
 
+    async def consume(self, token: str) -> MagicLink | None:
+        """Atomically validate-and-mark a token as used.
+
+        The conditional UPDATE is the single point of decision, so two
+        concurrent requests cannot both consume the same token. Returns the
+        link if this call won the race (unused and unexpired), else None.
+        """
+        now = datetime.now(UTC).isoformat()
+        cursor = await self._db.execute(
+            "UPDATE magic_links SET used = 1 WHERE token = ? AND used = 0 AND expires_at > ?",
+            (token, now),
+        )
+        await self._db.commit()
+        if cursor.rowcount == 0:
+            return None
+        async with self._db.execute("SELECT * FROM magic_links WHERE token = ?", (token,)) as c:
+            row = await c.fetchone()
+        return self._row_to_magic_link(row) if row is not None else None
+
     async def delete_expired(self) -> int:
         now = datetime.now(UTC).isoformat()
         cursor = await self._db.execute("DELETE FROM magic_links WHERE expires_at < ? AND used = 0", (now,))