fix: review round 3 — a new tab keeps its account's notices, a revoked tab account lands on the default

Deep-review round 3 over main..improvement/watch-background-audio.

Account boot
- The notification history and settings, and the hints toggle, are read
  when their modules load, which in a new tab (bookmark, typed URL) is
  before BootGate pins the account: they read as nobody's, and the tab's
  first notice replaced the account's stored history (up to 100 entries)
  with a one-item list. They now read again when the tab is pinned
  (onAccountPinned, run by setActiveAccount); a notice raised before the
  pin stays, after the history.
- A tab pinned to an account still in this browser's wallet but whose
  session ended (suspended, deleted, or revoked by "log out everywhere")
  got "signed out" from /api/me while the default account was still
  signed in, and signing in again as anyone else left it there. A refused
  per-tab account now leaves the wallet, and the probe answers as the
  default, so the tab reloads onto it.

Share page (/watch)
- The password unlock is limited per viewer (link + client IP), under a
  per-link ceiling that counts only wrong passwords. Keyed on the link
  alone, ten wrong guesses every five minutes locked out every viewer,
  the ones with the password included. client_ip moves to app.ratelimit
  so the public routes can key on it.
- Its failure messages and the load-failure screen are announced
  (role=alert, the password input described by its error). The sign-in,
  passphrase, password-change and setup form errors get role=alert too.

Comments that still described removed machinery are corrected: App's
collapse flags, SettingsPanel's Save bar, ConfigPanel, authed_format_sig.

Tests: the account's history and hints across the pin; a revoked and a
suspended pinned account; the per-viewer limit and the ceiling; the
limiter's exhausted(); the share page's announced failures; a share
link's feed filters persisted. Each mutation-checked RED for its own
reason.

backend/app/auth.py

@@ -284,7 +284,9 @@ def resolved_user_id(conn: HTTPConnection, *, strict: bool = True) -> tuple[int
     that tab would go on acting as an account it never showed, its player checkpoints and pref saves
     landing there and its Sign out signing THAT account out. Only the bootstrap /api/me probe passes
     `strict=False` and gets the default, which is how such a tab learns who it now is (the frontend
-    re-pins it and reloads).
+    re-pins it and reloads). An account whose session ended while it stayed in the wallet (deleted,
+    suspended, revoked) leaves the wallet when it is refused (`_session_user`), so a tab pinned to it
+    takes the same way back.
     """
     default_id = conn.session.get("user_id")
     wallet = conn.session.get("account_ids") or []
@@ -303,6 +305,23 @@ def resolved_user_id(conn: HTTPConnection, *, strict: bool = True) -> tuple[int
     return default_id, True
 
 
+def drop_from_wallet(session: dict, user_id: int) -> None:
+    """Take one account out of this browser's wallet (`account_ids`); the session default is untouched."""
+    session["account_ids"] = [a for a in (session.get("account_ids") or []) if a != user_id]
+
+
+def _end_account_session(request: Request, user_id: int, is_default: bool) -> None:
+    """An account whose session here is no longer valid. Only nuke the whole session when it is the
+    SESSION DEFAULT; a per-tab account just leaves the wallet (other tabs' default may still be valid).
+    Leaving is what lets a tab pinned to it recover: its next /api/me names an account no longer signed
+    in here and gets the default. Left in, it answered "signed out" — the landing page, with the default
+    still signed in, and no way off it but closing the tab."""
+    if is_default:
+        request.session.clear()
+    else:
+        drop_from_wallet(request.session, user_id)
+
+
 def current_user(request: Request, db: Session = Depends(get_db)) -> User:
     return _session_user(request, db, *resolved_user_id(request))
 
@@ -312,19 +331,15 @@ def _session_user(request: Request, db: Session, user_id: int | None, is_default
         raise HTTPException(status_code=401, detail="Not authenticated")
     user = db.get(User, user_id)
     if user is None or not user.is_active or user.is_suspended:
-        # Suspended/deactivated mid-session → drop the session so the block takes effect at once.
-        # But only nuke the whole session when the SESSION DEFAULT identity is gone; a stale
-        # per-tab header account just 401s that one tab (other tabs' default may still be valid).
-        if is_default:
-            request.session.clear()
+        # Suspended/deactivated mid-session → end it so the block takes effect at once.
+        _end_account_session(request, user_id, is_default)
         raise HTTPException(status_code=401, detail="Not authenticated")
     # SA4: reject a cookie whose recorded epoch is behind the account's current one (revoked by a
     # password reset/change or "log out everywhere"). A cookie with no recorded epoch (pre-SA4, or an
     # account signed in before this shipped) is treated as epoch 0 — so the FIRST bump revokes it too.
     epochs = request.session.get("epochs") or {}
     if int(epochs.get(str(user_id), 0)) != (user.session_epoch or 0):
-        if is_default:
-            request.session.clear()
+        _end_account_session(request, user_id, is_default)
         raise HTTPException(status_code=401, detail="Not authenticated")
     # Always keep the session default in the switchable list — covers sessions created before
     # multi-session existed (their account_ids was never seeded), so the switcher isn't empty.
@@ -341,7 +356,14 @@ def optional_current_user(request: Request, db: Session = Depends(get_db)) -> Us
     Used by the bootstrap /api/me probe so the logged-out landing doesn't log a failed request to
     the browser console (a non-2xx fetch is a console error no matter how the app handles it).
     The one lenient resolution (`strict=False`): a tab naming an account no longer signed in here gets
-    the default account, which is how it learns who it now is — see `resolved_user_id`."""
+    the default account, which is how it learns who it now is — see `resolved_user_id`. That includes an
+    account refused right here, which `_session_user` has just taken out of the wallet."""
+    user_id, is_default = resolved_user_id(request, strict=False)
+    try:
+        return _session_user(request, db, user_id, is_default)
+    except HTTPException:
+        if is_default:
+            return None
     try:
         return _session_user(request, db, *resolved_user_id(request, strict=False))
     except HTTPException:

backend/app/config.py

@@ -57,7 +57,7 @@ class Settings(BaseSettings):
     # VPS Caddy front door. A request from any other peer (someone hitting the app port directly)
     # is NOT trusted and its raw socket IP is used, so it can't spoof its rate-limit identity via
     # XFF. Empty = trust no proxy (use the direct peer) — correct for a directly-exposed / local-dev
-    # deploy. Comma-separated. See routes.auth._client_ip + the deploy docs.
+    # deploy. Comma-separated. See ratelimit.client_ip + the deploy docs.
     trusted_proxy_ips: str = ""
 
     # --- Outbound email (onboarding: access-request + approval notices) ---

backend/app/downloads/credentials.py

@@ -11,7 +11,7 @@ in prose):
     is downloading into is that private, salted identity — so a cookie can never poison a shared
     asset (the load-bearing cross-user privacy fix).
 
-Both go through `cookie_salt` / `authed_format_sig` / `host_key_for_url` here, so they can't drift.
+Both key the private identity on `cookie_salt` / `host_key_for_url` here, so they can't drift.
 
 Security controls enforced here:
   * **host scope (SEC-12)** — `host_key_for_url` matches the download URL's netloc against a curated
@@ -101,8 +101,9 @@ def cookie_salt(user_id: int, host_key: str) -> str:
 
 
 def authed_format_sig(spec: dict, user_id: int, host_key: str) -> str:
-    """The salted `format_sig` an authed download for (user, host) resolves to — the private asset
-    identity. Both enqueue (to create it) and the worker (to recognise it) compute it HERE."""
+    """The salted `format_sig` an authed download for (user, host) resolves to under the current
+    signature version — the private asset enqueue creates, and one of the identities `is_authed_asset`
+    recognises. Neither calls it; the tests use it to hold the two to each other."""
     return formats.format_sig(spec, salt=cookie_salt(user_id, host_key))
 
 

backend/app/ratelimit.py

@@ -1,15 +1,20 @@
 """A tiny in-process sliding-window rate limiter.
 
-Generic groundwork: keyed by an arbitrary string (e.g. a client IP), so it can throttle any
+Generic groundwork: keyed by an arbitrary string (e.g. a client IP — `client_ip`), so it can throttle any
 endpoint. Single-worker uvicorn → in-memory state is sufficient; it resets on restart, which
 is fine for abuse throttling (not for anything that must survive a deploy). Not shared across
 processes — if we ever run multiple workers, swap the backing store for Redis behind the same
 ``allow()`` interface.
 """
 
+import ipaddress
 import threading
 import time
 
+from fastapi import Request
+
+from app.config import settings
+
 
 class RateLimiter:
     def __init__(self, max_events: int, window_seconds: float):
@@ -25,6 +30,14 @@ class RateLimiter:
         with self._lock:
             self._hits.clear()
 
+    def exhausted(self, key: str) -> bool:
+        """Whether ``key``'s window is already full (``allow`` would refuse it), WITHOUT counting an
+        attempt — for a limit that counts only some outcomes: ask it first, ``allow`` the ones that
+        count."""
+        cutoff = time.monotonic() - self.window
+        with self._lock:
+            return sum(1 for t in self._hits.get(key, []) if t > cutoff) >= self.max_events
+
     def allow(self, key: str) -> bool:
         """Record an attempt for ``key`` and return whether it is within the limit.
 
@@ -49,3 +62,34 @@ class RateLimiter:
                     else:
                         del self._hits[k]
             return True
+
+
+def _norm_ip(s: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    """Parse an IP for comparison, unwrapping IPv4-mapped IPv6 (`::ffff:10.0.0.1` → `10.0.0.1`) so a
+    plain-IPv4 allowlist still matches a peer that surfaces in mapped form. None for a non-IP string."""
+    try:
+        ip = ipaddress.ip_address(s.strip())
+    except ValueError:
+        return None
+    return ip.ipv4_mapped if getattr(ip, "ipv4_mapped", None) else ip
+
+
+def client_ip(request: Request) -> str:
+    """Best-effort client IP for rate limiting. X-Forwarded-For is client-controlled, so we trust it
+    ONLY when the request actually arrived from a configured reverse proxy (settings.trusted_proxy_ips
+    — the address the proxy connects FROM, e.g. the VPS Caddy's WireGuard peer IP). Our proxy APPENDS
+    the client it saw to XFF, so the RIGHTMOST entry is the real client and can't be forged by a
+    client pre-seeding the header. Any other peer (someone hitting the app port directly, bypassing the
+    proxy) is untrusted — we use its real socket IP, so it can't spoof its rate-limit identity.
+
+    SAFETY: this relies on `request.client` being the TRUE socket peer. uvicorn is started WITHOUT
+    `--proxy-headers` (see entrypoint.sh) precisely so it never rewrites `request.client` from XFF —
+    do NOT add that flag, or the trust check below would compare against a forgeable value."""
+    peer = request.client.host if request.client else "unknown"
+    trusted = {_norm_ip(p) for p in settings.trusted_proxy_ip_set} - {None}
+    if _norm_ip(peer) in trusted:
+        xff = request.headers.get("x-forwarded-for")
+        parts = [p.strip() for p in (xff or "").split(",") if p.strip()]
+        if parts:
+            return parts[-1]
+    return peer

backend/app/routes/auth.py

@@ -3,7 +3,6 @@
 Split out of app/auth.py (R12 S2); the core auth surface (current_user, scopes, token
 helpers) stays in app/auth.py, imported below."""
 
-import ipaddress
 import secrets
 from datetime import datetime, timedelta, timezone
 from typing import NamedTuple
@@ -40,7 +39,7 @@ from app.auth import (
 from app.config import settings
 from app.db import SessionLocal, get_db
 from app.models import AuthToken, DemoWhitelist, Invite, OAuthToken, User
-from app.ratelimit import RateLimiter
+from app.ratelimit import RateLimiter, client_ip
 from app.security import encrypt, hash_password, hash_token, verify_password
 from app.utils import canonical_email, valid_email
 
@@ -172,37 +171,6 @@ def is_demo_allowed(db: Session, email: str) -> bool:
     )
 
 
-def _norm_ip(s: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
-    """Parse an IP for comparison, unwrapping IPv4-mapped IPv6 (`::ffff:10.0.0.1` → `10.0.0.1`) so a
-    plain-IPv4 allowlist still matches a peer that surfaces in mapped form. None for a non-IP string."""
-    try:
-        ip = ipaddress.ip_address(s.strip())
-    except ValueError:
-        return None
-    return ip.ipv4_mapped if getattr(ip, "ipv4_mapped", None) else ip
-
-
-def _client_ip(request: Request) -> str:
-    """Best-effort client IP for rate limiting. X-Forwarded-For is client-controlled, so we trust it
-    ONLY when the request actually arrived from a configured reverse proxy (settings.trusted_proxy_ips
-    — the address the proxy connects FROM, e.g. the VPS Caddy's WireGuard peer IP). Our proxy APPENDS
-    the client it saw to XFF, so the RIGHTMOST entry is the real client and can't be forged by a
-    client pre-seeding the header. Any other peer (someone hitting the app port directly, bypassing the
-    proxy) is untrusted — we use its real socket IP, so it can't spoof its rate-limit identity.
-
-    SAFETY: this relies on `request.client` being the TRUE socket peer. uvicorn is started WITHOUT
-    `--proxy-headers` (see entrypoint.sh) precisely so it never rewrites `request.client` from XFF —
-    do NOT add that flag, or the trust check below would compare against a forgeable value."""
-    peer = request.client.host if request.client else "unknown"
-    trusted = {_norm_ip(p) for p in settings.trusted_proxy_ip_set} - {None}
-    if _norm_ip(peer) in trusted:
-        xff = request.headers.get("x-forwarded-for")
-        parts = [p.strip() for p in (xff or "").split(",") if p.strip()]
-        if parts:
-            return parts[-1]
-    return peer
-
-
 class InviteRequest(NamedTuple):
     """What an access request settled on.
 
@@ -674,7 +642,7 @@ async def request_access(
     email = (body.email or "").strip().lower()
     if not valid_email(email):
         raise HTTPException(status_code=400, detail="Enter a valid email address.")
-    if not _request_access_limiter.allow(_client_ip(request)):
+    if not _request_access_limiter.allow(client_ip(request)):
         return {"status": "pending"}
     if is_allowed(db, email):
         return {"status": "pending"}  # already allowed — don't reveal it
@@ -691,7 +659,7 @@ def demo_login(body: EmailIn, request: Request, db: Session = Depends(get_db)) -
     so there is no visible 'demo login' button. Rate-limited per IP; when blocked it answers
     exactly like a non-match (authenticated=false) so the field can't be hammered as an
     enumeration oracle. On a match it sets the session and the client reloads into the app."""
-    if not _demo_limiter.allow(_client_ip(request)):
+    if not _demo_limiter.allow(client_ip(request)):
         return {"authenticated": False}
     email = (body.email or "").strip().lower()
     if valid_email(email) and is_demo_allowed(db, email):
@@ -837,7 +805,7 @@ def register(
         )
     if not sysconfig.get_bool(db, "allow_registration"):
         raise HTTPException(status_code=403, detail="Registration is currently closed.")
-    if not _register_limiter.allow(_client_ip(request)):
+    if not _register_limiter.allow(client_ip(request)):
         return {"status": "ok"}  # silently throttle; uniform response
 
     # Do the existence check + account creation OFF the response path (a background task with its own
@@ -1005,7 +973,7 @@ def verify_email_resend(
     and re-registering an already-registered email is a silent no-op, so a user who lost the first
     mail had no way back. Uniform response + timing regardless of whether the email has a pending
     account (the lookup runs off the response path), so it can't probe for registered emails."""
-    if not _verify_resend_limiter.allow(_client_ip(request)):
+    if not _verify_resend_limiter.allow(client_ip(request)):
         return {"status": "ok"}  # silently throttle; uniform response
     email = (body.email or "").strip().lower()
     if valid_email(email):
@@ -1064,7 +1032,7 @@ def password_login(
     """Email+password sign-in. Wrong email/password give a single uniform 401 (no enumeration).
     Once the password is proven correct, the owner gets a specific reason if their account is
     still pending or suspended — that's not an enumeration leak (they already hold the password)."""
-    if not _login_limiter.allow(_client_ip(request)):
+    if not _login_limiter.allow(client_ip(request)):
         raise _login_refused(429, "rate_limited")
     email = (body.email or "").strip().lower()
     password = body.password or ""
@@ -1095,7 +1063,7 @@ def password_reset_request(
 ) -> dict:
     """Request a password-reset link. Uniform response + timing regardless of whether the email has a
     password account (the lookup runs off the response path), so it can't probe for registered emails."""
-    if not _reset_limiter.allow(_client_ip(request)):
+    if not _reset_limiter.allow(client_ip(request)):
         return {"status": "ok"}
     email = (body.email or "").strip().lower()
     # Do the account lookup + token issue + email OFF the response path (a background task with its
@@ -1144,7 +1112,7 @@ def password_reset_check(body: TokenIn, request: Request, db: Session = Depends(
     told `true`, never `false`: this check may only ever save a user a wasted form, so when it cannot
     answer it must degrade to the old behaviour instead of declaring a live link dead.
     """
-    if not _reset_check_limiter.allow(_client_ip(request)):
+    if not _reset_check_limiter.allow(client_ip(request)):
         return {"ok": True}
     return {"ok": any(_peek_token(db, body.token, k) is not None for k in _PASSWORD_TOKEN_KINDS)}
 

backend/app/routes/me.py

@@ -7,6 +7,7 @@ from app.auth import (
     count_admins,
     count_pending_activations,
     current_user,
+    drop_from_wallet,
     google_enabled,
     has_read_scope,
     has_write_scope,
@@ -80,9 +81,7 @@ def switch_account(
     # switch list and refuse, rather than logging everyone out. (B-core-9)
     epochs = request.session.get("epochs") or {}
     if int(epochs.get(str(target), 0)) != (u.session_epoch or 0):
-        request.session["account_ids"] = [
-            a for a in (request.session.get("account_ids") or []) if a != target
-        ]
+        drop_from_wallet(request.session, target)
         raise HTTPException(
             status_code=403, detail="That account was signed out elsewhere — sign in again."
         )

backend/app/routes/public.py

@@ -6,7 +6,7 @@ the app or any account data. Password-protected links exchange the password once
 a short-lived signed grant that the media URL carries (a `<video src>` can't send a header).
 """
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 from sqlalchemy import select
@@ -17,13 +17,18 @@ from app.db import get_db
 from app.downloads import links as linksmod
 from app.downloads import storage
 from app.models import DownloadJob, DownloadLink, MediaAsset
-from app.ratelimit import RateLimiter
+from app.ratelimit import RateLimiter, client_ip
 from app.security import verify_password
 
 router = APIRouter(prefix="/api/public", tags=["public"])
 
-# Slow password guessing per link (the token itself is 192-bit, so it needs no throttle).
+# Slow password guessing on a link (the token itself is 192-bit, so it needs no throttle). Counted per
+# viewer — the link AND the client — so someone who has the link but not its password can't lock out the
+# viewers who do, and a busy link's own unlocks don't count against its next viewer.
 _unlock_limit = RateLimiter(max_events=10, window_seconds=300)
+# ...under a ceiling on WRONG passwords for the link from every client together, so guessing spread over
+# many addresses stays slow too. Viewers who have the password never add to it.
+_unlock_failures = RateLimiter(max_events=50, window_seconds=300)
 
 
 def _link_or_404(db: Session, token: str) -> DownloadLink:
@@ -83,9 +88,12 @@ class WatchUnlockIn(BaseModel):
 
 
 @router.post("/watch/{token}/unlock")
-def watch_unlock(token: str, body: WatchUnlockIn, db: Session = Depends(get_db)) -> dict:
+def watch_unlock(
+    token: str, body: WatchUnlockIn, request: Request, db: Session = Depends(get_db)
+) -> dict:
     """Exchange the link password for a signed grant carried by the media URL."""
-    if not _unlock_limit.allow(token):
+    viewer = f"{token}|{client_ip(request)}"
+    if _unlock_failures.exhausted(token) or not _unlock_limit.allow(viewer):
         raise HTTPException(status_code=429, detail="Too many attempts. Try again later.")
     link = _link_or_404(db, token)
     if not link.password_hash:
@@ -96,6 +104,7 @@ def watch_unlock(token: str, body: WatchUnlockIn, db: Session = Depends(get_db))
             **linksmod.public_meta(link, job, asset, _file_url(token), _poster_url(token, asset)),
         }
     if not verify_password(body.password, link.password_hash):
+        _unlock_failures.allow(token)
         raise HTTPException(status_code=403, detail="Wrong password.")
     job, asset = _ready_target(db, link)
     link.view_count += 1

backend/entrypoint.sh

@@ -11,7 +11,7 @@ echo "Starting Siftlode API..."
 # on the next request -> 502. Non-idempotent POSTs (the player's progress/state writes,
 # fired every 5s) can't be auto-retried by the proxy, so those 502s reached the user.
 #
-# SECURITY: do NOT add --proxy-headers / --forwarded-allow-ips here. auth._client_ip trusts
+# SECURITY: do NOT add --proxy-headers / --forwarded-allow-ips here. ratelimit.client_ip trusts
 # X-Forwarded-For only when the TRUE socket peer is a configured proxy (TRUSTED_PROXY_IPS); those
 # flags would make uvicorn overwrite request.client from the forgeable XFF, defeating that check and
 # reopening the rate-limit bypass (SA3).

backend/tests/test_auth_session.py

@@ -12,6 +12,7 @@ Note: `/api/me` is the bootstrap probe — it returns 200 `{authenticated: False
 
 from datetime import timedelta
 
+import pytest
 from fastapi.testclient import TestClient
 
 from app.auth import ACTIVE_ACCOUNT_HEADER
@@ -76,6 +77,31 @@ def test_a_tab_pinned_to_an_account_signed_out_elsewhere_acts_as_nobody(client,
     assert client.get("/api/me", headers=tab).json()["email"] == "bob3@example.com"
 
 
+@pytest.mark.parametrize("ended", ["revoked", "suspended"])
+def test_a_tab_pinned_to_an_account_whose_session_ended_comes_back_as_the_default(
+    client, make_user, db, ended
+):
+    # A tab pinned to Alice, Bob the default. Alice's session then ends while she is still in this
+    # browser's wallet ("log out everywhere" on another device, or an admin suspends her). The tab's
+    # bootstrap probe used to answer "signed out": the landing page with Bob still signed in, and no way
+    # off it but closing the tab (signing in as anyone else there kept the tab pinned to Alice).
+    alice = make_user(email="alice4@example.com", password=_PW)
+    make_user(email="bob4@example.com", password=_PW)
+    assert _login(client, "alice4@example.com", _PW).status_code == 200
+    assert _login(client, "bob4@example.com", _PW).status_code == 200
+    tab = {ACTIVE_ACCOUNT_HEADER: str(alice.id)}
+    assert client.get("/api/feed", headers=tab).status_code == 200
+    if ended == "revoked":
+        alice.session_epoch = (alice.session_epoch or 0) + 1
+    else:
+        alice.is_suspended = True
+    db.commit()
+
+    assert client.get("/api/me", headers=tab).json()["email"] == "bob4@example.com"
+    assert client.get("/api/feed", headers=tab).status_code == 401  # still nobody, never Bob
+    assert client.get("/api/feed").status_code == 200  # Bob's session is intact
+
+
 def test_logout_others_revokes_the_other_session_but_keeps_the_caller(client, make_user):
     # Two independent cookie jars (browsers) signed into the same account. `logout-others` from one
     # bumps the account epoch so the OTHER's cookie is rejected on its next request, while the caller

backend/tests/test_public_unlock_limit.py

@@ -0,0 +1,59 @@
+"""The password unlock of a shared /watch link (routes/public.py), DB-lane.
+
+Guesses are slowed per viewer (the link AND the client), under a per-link ceiling on wrong passwords, so
+someone who holds a password link without its password can't lock out the viewers who have it.
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.downloads import probe as probemod
+from app.main import app
+from app.ratelimit import RateLimiter
+from app.routes import public
+
+_PW = "open sesame"
+
+
+@pytest.fixture
+def unlock_url(as_user, make_user, tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(probemod, "probe_media", lambda _p: ("h264", "aac", 320, 240, 5))
+    owner = as_user(make_user())
+    job = owner.post(
+        "/api/downloads/upload", files={"file": ("clip.mp4", b"fake-video", "video/mp4")}
+    )
+    assert job.status_code == 200, job.text
+    link = owner.post(f"/api/downloads/{job.json()['id']}/links", json={"password": _PW})
+    assert link.status_code == 200, link.text
+    return f"/api/public/watch/{link.json()['token']}/unlock"
+
+
+def _from(ip: str) -> TestClient:
+    # Its own socket peer: with no trusted proxy configured, that is the client's IP.
+    return TestClient(app, client=(ip, 50000))
+
+
+def test_a_link_holder_without_the_password_cannot_lock_out_its_viewers(unlock_url):
+    # The limit used to be keyed on the link alone: ten wrong guesses every five minutes kept every
+    # viewer out, the ones with the password included.
+    guesser = _from("203.0.113.5")
+    for _ in range(10):
+        assert guesser.post(unlock_url, json={"password": "guess"}).status_code == 403
+    assert guesser.post(unlock_url, json={"password": "guess"}).status_code == 429
+
+    assert _from("198.51.100.7").post(unlock_url, json={"password": _PW}).status_code == 200
+
+
+def test_wrong_passwords_from_many_clients_still_reach_the_links_ceiling(unlock_url, monkeypatch):
+    monkeypatch.setattr(public, "_unlock_failures", RateLimiter(max_events=3, window_seconds=300))
+    viewer = _from("198.51.100.7")
+    for _ in range(5):  # unlocking with the password never adds to the ceiling
+        assert viewer.post(unlock_url, json={"password": _PW}).status_code == 200
+
+    for n in range(3):
+        wrong = _from(f"203.0.113.{n}").post(unlock_url, json={"password": "guess"})
+        assert wrong.status_code == 403
+    late = _from("203.0.113.9").post(unlock_url, json={"password": "guess"})
+    assert late.status_code == 429

backend/tests/test_ratelimit.py

@@ -57,6 +57,19 @@ def test_reset_clears_every_key():
     assert lim.allow("b") is True
 
 
+def test_exhausted_reports_a_full_window_without_counting_an_attempt():
+    # The asking half of a limit that counts only some outcomes (a share link's unlock counts only
+    # wrong passwords toward its ceiling): asking must never use up the window it asks about.
+    lim = RateLimiter(max_events=2, window_seconds=300)
+
+    for _ in range(5):
+        assert lim.exhausted("link") is False
+    assert lim.allow("link") is True
+    assert lim.exhausted("link") is False
+    assert lim.allow("link") is True
+    assert lim.exhausted("link") is True
+
+
 def test_expired_attempts_fall_out_of_the_window():
     # A zero-length window means every previous attempt is already outside it.
     lim = RateLimiter(max_events=1, window_seconds=0)