Gate an existing id: create(token=), set_password, set_expires

README.md

@@ -23,6 +23,7 @@ links.revoke(rec["token"])                   # now "missing"
 ## What is in it
 
 - `CapabilityLinks`: create, read, list and revoke links, and gate them on a password. Records are plain dicts in any `MutableMapping` keyed by token: the generic fields are `token`, `created`, `expires`, `password` (a hash) and `revoked`; your own fields (what the link reaches) sit beside them.
+- Gating an id you already have: `create(..., token=existing_id)` keeps the id; `set_password(token, pw_or_None)` and `set_expires(token, when)` change a live link later (a new password ends every open unlock).
 - `ScryptHasher` (stdlib scrypt) behind the `PasswordHasher` protocol (`hash`, `verify -> bool`): pass `hasher=` for argon2 or anything else.
 - `parse_expires` reads what people write (`30m`, `12h`, `7d`, `2w`, `2026-11-01` meaning the end of that day UTC, an ISO timestamp, `never`) and refuses the past; `is_expired` fails closed.
 - `SlidingWindowLimiter` behind the `RateLimiter` protocol (`allow(key) -> bool`): count password attempts per client, per link, or both. It is per process; a shared limiter of the same shape (Redis, say) fits where several workers serve the links.

caplinks/links.py

@@ -105,28 +105,32 @@ def create(
         *,
         password: str | None = None,
         expires=DFLT_EXPIRES,
+        token: str | None = None,
     ) -> dict:
         """Make a link carrying the host's ``fields``; return its record without the hash.
 
         ``expires`` is anything :func:`caplinks.expiry.parse_expires` reads, 7 days by
-        default (``"never"`` for none); ``password`` gates the link.
+        default (``"never"`` for none); ``password`` gates the link. ``token`` gates an
+        id that already exists (a host whose links must not change): it must look like
+        a token (:func:`check_token`) and be unused; leave it out for a fresh one. The
+        caller then owns the id's randomness: use 128 random bits, as :func:`new_token`.
         """
         fields = dict(fields or {})
+        if token is not None:
+            check_token(token)
+            if token in self.store:  # even a damaged record is not free to overwrite
+                raise ValueError("a link with this token already exists")
         clash = sorted(set(fields) & {*RECORD_FIELDS, *_VIEW_FIELDS})
         if clash:
             raise ValueError(f"fields {clash} are the link's own; name yours otherwise")
-        if password is not None and len(password) < self.min_password_len:
-            raise ValueError(
-                f"a link password needs at least {self.min_password_len} characters "
-                "(anyone can try it); omit it for no password"
-            )
+        hashed = self._hash_password(password)
         now = self.clock()
         record = {
-            "token": new_token(),
+            "token": token or new_token(),
             "created": now.isoformat(),
             **fields,
             "expires": parse_expires(expires, now=now),
-            "password": self.hasher.hash(password) if password else None,
+            "password": hashed,
             "revoked": None,
         }
         self.store[record["token"]] = record
@@ -169,6 +173,47 @@ def list(self, *, include_inactive: bool = False) -> list[dict]:
                 out.append(rec)
         return sorted(out, key=lambda r: r["created"], reverse=True)
 
+    def set_password(self, token: str, password: str | None) -> dict:
+        """Give an unrevoked link a new password, or ``None`` to open it.
+
+        A new password ends every open unlock at once. A revoked link stays revoked
+        (``KeyError``).
+        """
+        return self._update(token, password=self._hash_password(password))
+
+    def set_expires(self, token: str, expires) -> dict:
+        """Move an unrevoked link's expiry (anything :func:`parse_expires` reads;
+        ``"never"`` for none). An expired link may be given a later one; a revoked link
+        stays revoked (``KeyError``).
+        """
+        return self._update(token, expires=parse_expires(expires, now=self.clock()))
+
+    def _update(self, token: str, **changes) -> dict:
+        # The store has no compare-and-set: the record is read just before the write (the
+        # slow hashing is done by the caller) so a revoke that landed meanwhile is not
+        # undone. Hosts with several workers writing the
+        # same link still need to serialise writes per token.
+        rec = self._unrevoked(token)
+        if rec is None:
+            raise KeyError(token)
+        rec.update(changes)
+        self.store[token] = rec
+        return public_record(rec)
+
+    def _unrevoked(self, token: str) -> dict | None:
+        rec = self.record(token)
+        return None if rec is None or rec.get("revoked") else rec
+
+    def _hash_password(self, password: str | None) -> str | None:
+        if password is None:
+            return None
+        if len(password) < self.min_password_len:
+            raise ValueError(
+                f"a link password needs at least {self.min_password_len} characters "
+                "(anyone can try it); omit it for no password"
+            )
+        return self.hasher.hash(password)
+
     def revoke(self, token: str) -> dict:
         """End a link now (idempotent). The record stays, as a tombstone."""
         rec = self.record(token)

tests/test_caplinks.py

@@ -266,3 +266,81 @@ def test_parse_expires_far_future_is_a_value_error(value):
 def test_limiter_refuses_settings_that_disable_it(kw):
     with pytest.raises(ValueError):
         SlidingWindowLimiter(**kw)
+
+
+def test_create_gates_an_existing_id():
+    links = CapabilityLinks()
+    tid = "A" * 22
+    rec = links.create({"x": 1}, password=PW, expires="never", token=tid)
+    assert rec["token"] == tid and rec["protected"]
+    assert links.status(tid)[0] == "ok"
+    with pytest.raises(ValueError, match="already exists"):
+        links.create(token=tid)
+    with pytest.raises(ValueError, match="not a link token"):
+        links.create(token="../x")
+
+
+def test_set_password_ends_open_unlocks():
+    links = CapabilityLinks()
+    tok = links.create(password=PW)["token"]
+    cookie = links.unlock_value(links.record(tok))
+    assert links.unlocked(links.record(tok), cookie, max_age_s=3600)
+    assert links.set_password(tok, "another pass")["protected"]
+    assert not links.unlocked(links.record(tok), cookie, max_age_s=3600)
+    assert not links.set_password(tok, None)["protected"]
+    with pytest.raises(ValueError):
+        links.set_password(tok, "short")
+
+
+def test_set_expires_revives_an_expired_link_but_not_a_revoked_one():
+    clock = Clock()
+    links = CapabilityLinks(clock=clock)
+    tok = links.create(expires="1d")["token"]
+    clock.t = T0 + timedelta(days=2)
+    assert links.status(tok)[0] == "expired"
+    links.set_expires(tok, "3d")
+    assert links.status(tok)[0] == "ok"
+    assert links.set_expires(tok, "never")["expires"] is None
+    links.revoke(tok)
+    for call in (lambda: links.set_expires(tok, "3d"), lambda: links.set_password(tok, PW)):
+        with pytest.raises(KeyError):
+            call()
+
+
+def test_set_password_new_one_opens_old_one_does_not():
+    links = CapabilityLinks(hasher=ScryptHasher(n=2**4))
+    tok = links.create(password=PW)["token"]
+    links.set_password(tok, "another pass")
+    rec = links.record(tok)
+    assert links.check_password(rec, "another pass") and not links.check_password(rec, PW)
+    links.set_password(tok, None)
+    assert links.unlocked(links.record(tok), None, max_age_s=60)
+
+
+def test_setters_refuse_the_past_and_the_missing():
+    links = CapabilityLinks()
+    tok = links.create()["token"]
+    with pytest.raises(ValueError, match="past"):
+        links.set_expires(tok, "2000-01-01")
+    for call in (lambda: links.set_expires("B" * 22, "1d"), lambda: links.set_password("B" * 22, PW)):
+        with pytest.raises(KeyError):
+            call()
+
+
+def test_a_revoked_id_cannot_be_gated_again():
+    links = CapabilityLinks()
+    tid = "C" * 22
+    links.create(token=tid)
+    links.revoke(tid)
+    with pytest.raises(ValueError, match="already exists"):
+        links.create(token=tid)
+    assert links.status(tid)[0] == "missing"
+
+
+def test_create_token_does_not_overwrite_a_damaged_record():
+    store = {"D" * 22: {"junk": 1}}
+    links = CapabilityLinks(store)
+    with pytest.raises(ValueError, match="already exists"):
+        links.create(token="D" * 22)
+    assert store["D" * 22] == {"junk": 1}
+