fix(link): revoke endpoint was disabled by its own rate-limit rule

Live QA: every call to /v1/devices/revoke returned 429 retry_after=3600.
The rule was written (0, 3600) with a comment saying 'never throttle someone
out of signing a lost laptop out' -- but in this limiter a limit of 0
DISABLES the endpoint outright. The comment said unlimited; the code said
never. Set to 600/hour instead.

Every unit test passed while the endpoint was 100% dead over HTTP, because
they called svc.revoke_device() directly and never went through the handler.
Added 4 tests that speak HTTP, including one that fails if ANY route in
DEFAULT_RATE_RULES is configured to 0.

176 tests (was 172).

README.md

@@ -403,7 +403,7 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
 
 ## Tests
 
-* `python3 -m unittest discover -s tests` — 153 tests. The v1.2 additions
+* `python3 -m unittest discover -s tests` — 172 tests. The v1.2 additions
   cover: a snapshot capturing uncommitted work while HEAD, the index and the
   working tree stay byte-identical; snapshots landing outside `refs/heads`;
   an unchanged tree not being re-pushed; a diverged folder still being backed
@@ -445,6 +445,64 @@ the provisioning service creates their forge account + scoped token on the
 fly — the forge never sees a password and the user never sees the forge admin.
 Every linked folder becomes a private repo under their account.
 
+## Devices, sign-out, and the audit trail
+
+**`granthi-sync devices`** lists every computer signed in to the account —
+name, id, when it linked, and whether it is still active. The registry hangs
+off the identity that owns it in `state.json`, so "which computers can reach
+my files" has one answer that cannot drift from the identity map.
+
+**`granthi-sync logout [--device ID]`** signs a computer out. Revocation
+happens **at the forge**: granthi-link deletes that device's Gitea token
+(admin basic auth + `Sudo`, the only mechanism Gitea 1.27 accepts — verified
+204, after which the token returns 401 immediately). It is not a flag a
+client could ignore, which is the only kind of sign-out worth having for a
+laptop somebody lost. Signing out the current computer also deletes the local
+token; linked folders are left on disk untouched.
+
+Two properties worth keeping:
+
+* The device id is part of the **token name**, because revocation deletes by
+  name. Without it, one user linking two machines called "macbook" in the
+  same second would collide, and signing one out would kill the other.
+* A failed forge deletion is **not** recorded as revoked. A registry that
+  says "revoked" while the token still works is worse than an honest error.
+
+**`granthi-sync activity`** shows the security events for the account:
+`device.link`, `device.revoke`, `device.revoke.denied`, with timestamp,
+device, and client IP. The log is append-only JSONL (0600, rotated at 64 MB),
+kept **separate** from `state.json` on purpose — state is rewritten
+atomically on every change, and an audit trail the audited thing can rewrite
+is not an audit trail. A failed audit write is logged loudly and never breaks
+the request it was auditing.
+
+**What the audit log cannot see, and this matters.** Every event in it is one
+granthi-link handled. **Git pushes and pulls do not pass through this
+service** — they go straight to the forge — so:
+
+| you want to know | where it actually lives |
+|---|---|
+| who linked/revoked a computer, from what IP | `granthi-sync activity` (this log) |
+| who pushed what, and when | Gitea: the repo's activity feed and commit history |
+| who *pulled* or cloned | **nowhere by default** — Gitea does not record fetches unless its router access log is enabled |
+| last time a device used its token | Gitea `access_token.updated_unix` |
+
+Reading the audit log and believing it lists file activity would be a real
+mistake, so `/v1/audit` returns that caveat in its own response.
+
+### Endpoints added
+
+* `POST /v1/devices {token}` → the caller's devices.
+* `POST /v1/devices/revoke {token, device_id}` → kills that device's forge
+  token. **Never rate-limited** — nobody should be throttled out of signing
+  a lost laptop out.
+* `POST /v1/audit {token, limit}` → the caller's own events.
+
+Authorisation on all three is the same: the caller proves who they are by
+holding a working forge token, and **the forge decides** whose it is
+(`GET /api/v1/user`). No login is ever read from the request body, so a body
+claiming another account changes nothing.
+
 ## Next phase — invites and per-repo access (designed, not built)
 
 Today `/v1/link` creates an account and every folder becomes a private repo

client/granthi_sync_client.py

@@ -632,6 +632,10 @@ def cmd_link(args):
               "device_id": dev})
     if status != 200:
         raise SystemExit(f"link failed (HTTP {status}): {resp}")
+    if resp.get("device_id"):
+        # The service is authoritative for the id it filed the device under
+        # (an older client that sent none gets one back).
+        cfg["device_id"] = resp["device_id"]
     cfg.update({"server": args.server.rstrip("/"),
                 "gitea_base": resp["gitea_base"],
                 "login": resp["login"],
@@ -1137,6 +1141,88 @@ def cmd_watch(args):
     return 0
 
 
+def cmd_devices(args):
+    """Every computer signed in to this account."""
+    cfg = require_linked(load_config())
+    status, resp = http_json("POST", f"{cfg['server']}/v1/devices",
+                             body={"token": cfg["token"]})
+    if status == 401:
+        raise SystemExit("this device's access has been revoked -- "
+                         "run: granthi-sync link")
+    if status != 200:
+        raise SystemExit(f"could not list devices (HTTP {status}): {resp}")
+    devices = resp.get("devices") or []
+    if not devices:
+        print("no devices recorded for this account")
+        return 0
+    this = cfg.get("device_id")
+    rows = [("DEVICE", "ID", "LINKED", "STATUS")]
+    for d in devices:
+        rows.append((d.get("name") or "?", (d.get("device_id") or "")[:12],
+                     (d.get("linked_at") or "")[:19],
+                     "REVOKED" if d.get("revoked_at") else
+                     ("active (this computer)" if d.get("device_id") == this
+                      else "active")))
+    widths = [max(len(r[i]) for r in rows) for i in range(len(rows[0]))]
+    for r in rows:
+        print("  ".join(c.ljust(w) for c, w in zip(r, widths)))
+    return 0
+
+
+def cmd_logout(args):
+    """Sign a computer out. Revocation happens at the FORGE -- the token is
+    deleted, so that machine's next fetch or push fails at the server. It is
+    not a flag a client could ignore, which is the only kind of logout worth
+    having for a lost laptop."""
+    cfg = require_linked(load_config())
+    target = args.device or cfg.get("device_id")
+    if not target:
+        raise SystemExit("no device id recorded; pass --device (see: "
+                         "granthi-sync devices)")
+    is_self = target == cfg.get("device_id")
+    status, resp = http_json("POST", f"{cfg['server']}/v1/devices/revoke",
+                             body={"token": cfg["token"], "device_id": target})
+    if status == 404:
+        raise SystemExit(f"no device {target} on this account "
+                         f"(see: granthi-sync devices)")
+    if status != 200:
+        raise SystemExit(f"revoke failed (HTTP {status}): {resp}")
+    log(f"revoked {target} at the forge ({resp.get('revoked_at')})")
+    if is_self:
+        # Drop the local token too. The forge already refuses it, but leaving
+        # a dead secret on disk is pointless risk -- and `status` should say
+        # "not linked" rather than pretend.
+        for key in ("token", "token_name"):
+            cfg.pop(key, None)
+        cfg["logged_out_at"] = datetime.now(timezone.utc).isoformat(
+            timespec="seconds")
+        save_config(cfg)
+        log("local token deleted; syncing stops at the next pass. "
+            "Linked folders are left on disk untouched.")
+        log("run `granthi-sync link` to sign back in")
+    return 0
+
+
+def cmd_activity(args):
+    """Security events for this account, newest first."""
+    cfg = require_linked(load_config())
+    status, resp = http_json("POST", f"{cfg['server']}/v1/audit",
+                             body={"token": cfg["token"],
+                                   "limit": args.limit})
+    if status != 200:
+        raise SystemExit(f"could not read activity (HTTP {status}): {resp}")
+    events = resp.get("events") or []
+    if not events:
+        print("no recorded events for this account")
+    for e in events:
+        extra = " ".join(f"{k}={v}" for k, v in sorted(e.items())
+                         if k not in ("ts", "event", "login"))
+        print(f"{e.get('ts')}  {e.get('event'):<22} {extra}")
+    if resp.get("note"):
+        print(f"\nnote: {resp['note']}")
+    return 0
+
+
 def _folder_meta(cfg, folder):
     path = os.path.abspath(folder)
     meta = cfg.get("folders", {}).get(path)
@@ -1328,6 +1414,21 @@ def main(argv=None):
     sp.add_argument("--once", action="store_true", help="single pass then exit")
     sp.set_defaults(fn=cmd_watch)
 
+    sp = sub.add_parser("devices",
+                        help="list the computers signed in to this account")
+    sp.set_defaults(fn=cmd_devices)
+
+    sp = sub.add_parser("logout",
+                        help="sign a computer out (revokes it at the forge)")
+    sp.add_argument("--device",
+                    help="device id to revoke (default: this computer)")
+    sp.set_defaults(fn=cmd_logout)
+
+    sp = sub.add_parser("activity",
+                        help="security events for this account")
+    sp.add_argument("--limit", type=int, default=50)
+    sp.set_defaults(fn=cmd_activity)
+
     sp = sub.add_parser("status", help="show linked folders")
     sp.set_defaults(fn=cmd_status)
 

server/granthi_link.py

@@ -64,6 +64,7 @@ import sys
 import threading
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -73,6 +74,7 @@ LOG = logging.getLogger("granthi-link")
 
 DEFAULT_CONFIG = "/opt/granthi-link/config.json"
 DEFAULT_STATE = "/opt/granthi-link/state.json"
+DEFAULT_AUDIT = "/opt/granthi-link/audit.jsonl"
 MAX_BODY_BYTES = 64 * 1024
 TEST_MODE_ENV = "GRANTHI_LINK_ALLOW_TEST_MODE"
 
@@ -84,7 +86,18 @@ LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+")
 # Rate limiting. /v1/link is the expensive endpoint -- it round-trips Zitadel,
 # can CREATE a forge account and always mints a token -- so its default is
 # deliberately tight. /v1/repos only spends the caller's own token.
-DEFAULT_RATE_RULES = {"/v1/link": (5, 3600), "/v1/repos": (60, 3600)}
+DEFAULT_RATE_RULES = {"/v1/link": (5, 3600), "/v1/repos": (60, 3600),
+                      # Reads are cheap but still authenticated work.
+                      "/v1/devices": (120, 3600),
+                      "/v1/audit": (120, 3600),
+                      # Revocation is a safety action, so its limit is set
+                      # high rather than tight. It is NOT 0: in this limiter
+                      # a limit of 0 DISABLES the endpoint outright (see
+                      # RateLimiter.check), which would mean nobody could
+                      # ever sign a lost laptop out. Caught in live QA --
+                      # unit tests called the service method directly and so
+                      # never went through the limiter at all.
+                      "/v1/devices/revoke": (600, 3600)}
 MAX_RATE_KEYS = 10000
 
 
@@ -322,6 +335,115 @@ class IdentityStore:
         data["identities"][str(sub)] = record
         self._write(data)
 
+    # -- device registry ---------------------------------------------------
+    # Devices hang off the identity that owns them, so "which computers can
+    # reach my files" has exactly one answer and it cannot drift from the
+    # identity map. Callers hold self.lock.
+
+    def record_device(self, sub, device_id, record):
+        data = self._load()
+        ident = data["identities"].get(str(sub))
+        if ident is None:
+            return False
+        devices = ident.setdefault("devices", {})
+        existing = devices.get(device_id, {})
+        # Re-linking the same device REPLACES its token name and clears any
+        # previous revocation: the user just proved identity again.
+        existing.update(record)
+        existing["revoked_at"] = None
+        devices[device_id] = existing
+        data["identities"][str(sub)] = ident
+        self._write(data)
+        return True
+
+    def devices_for_login(self, login):
+        """(sub, devices) for a Gitea login, or (None, {}).
+
+        Keyed on login because the caller authenticates with a forge token,
+        which proves a login -- not a Zitadel sub.
+        """
+        data = self._load()
+        for sub, ident in data["identities"].items():
+            if ident.get("login") == login:
+                return sub, dict(ident.get("devices") or {})
+        return None, {}
+
+    def mark_revoked(self, sub, device_id, when):
+        data = self._load()
+        ident = data["identities"].get(str(sub)) or {}
+        device = (ident.get("devices") or {}).get(device_id)
+        if device is None:
+            return False
+        device["revoked_at"] = when
+        self._write(data)
+        return True
+
+
+# --------------------------------------------------------------------------
+# Audit log: append-only JSONL, one line per security-relevant event
+# --------------------------------------------------------------------------
+
+class AuditLog:
+    """Who linked a computer, from where, when, and what was refused.
+
+    Append-only and separate from the identity map on purpose: state.json is
+    rewritten atomically on every change, so anything kept only there is one
+    bad write away from gone. An audit trail that can be rewritten by the
+    thing it audits is not an audit trail.
+
+    NOTE what this can and cannot see. Every event here is one granthi-link
+    handled. Git pushes and pulls do NOT pass through this service -- they go
+    straight to the forge -- so they are recorded by Gitea, not here. Reading
+    this file and believing it lists file activity would be a real mistake.
+    """
+
+    def __init__(self, path, keep_bytes=64 * 1024 * 1024):
+        self.path = path
+        self.keep_bytes = keep_bytes
+        self.lock = threading.Lock()
+
+    def write(self, event, **fields):
+        line = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+                "event": event}
+        line.update({k: v for k, v in fields.items() if v is not None})
+        try:
+            with self.lock:
+                self._rotate_if_needed()
+                fd = os.open(self.path,
+                             os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
+                with os.fdopen(fd, "a") as f:
+                    f.write(json.dumps(line, sort_keys=True) + "\n")
+        except OSError as e:
+            # Never let auditing break the request it is auditing, but say so
+            # loudly -- a silently dead audit log is worse than none.
+            LOG.error("AUDIT WRITE FAILED (%s): %s", e, line)
+
+    def _rotate_if_needed(self):
+        try:
+            if os.path.getsize(self.path) < self.keep_bytes:
+                return
+        except OSError:
+            return
+        os.replace(self.path, self.path + ".1")
+
+    def read_for(self, login, limit=200):
+        """Events belonging to one login, newest first."""
+        out = []
+        for path in (self.path, self.path + ".1"):
+            try:
+                with open(path) as f:
+                    for raw in f:
+                        try:
+                            rec = json.loads(raw)
+                        except ValueError:
+                            continue
+                        if rec.get("login") == login:
+                            out.append(rec)
+            except OSError:
+                continue
+        out.sort(key=lambda r: r.get("ts") or "", reverse=True)
+        return out[:limit]
+
 
 # --------------------------------------------------------------------------
 # Core logic (class so tests can instantiate with a stub config)
@@ -337,6 +459,7 @@ class LinkService:
         self.userinfo_url = config.get(
             "zitadel_userinfo", "https://id.shre.ai/oidc/v1/userinfo")
         self.state = IdentityStore(config.get("state_path", DEFAULT_STATE))
+        self.audit = AuditLog(config.get("audit_path", DEFAULT_AUDIT))
 
         # Rate limiting is ON by default: this endpoint creates accounts and
         # mints tokens, so the safe default is limited, and disabling it has
@@ -501,12 +624,20 @@ class LinkService:
             return f"gitea admin user create failed (HTTP {status}): {resp}"
         return None
 
-    def mint_token(self, login, device_name):
+    def mint_token(self, login, device_name, device_id=None):
         """Admin basic auth + Sudo header. Verified on Gitea 1.27.1:
-        token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201."""
+        token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201.
+
+        The device id is part of the token NAME because revocation deletes by
+        name. With only device_name + a one-second timestamp, one user linking
+        two machines called "macbook" in the same second would collide -- and
+        a collision means signing out one laptop kills the other's access.
+        """
         safe_dev = LOGIN_SAFE.sub("-", device_name or "device")[:40]
-        token_name = "granthi-sync-{}-{}".format(
-            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
+        suffix = LOGIN_SAFE.sub("-", str(device_id or ""))[:12]
+        token_name = "granthi-sync-{}-{}{}".format(
+            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
+            f"-{suffix}" if suffix else "")
         status, resp = http_json(
             "POST", f"{self.gitea}/api/v1/users/{login}/tokens",
             headers={
@@ -596,7 +727,7 @@ class LinkService:
                               "identity"}, None
 
     # -- endpoints ---------------------------------------------------------
-    def link(self, body):
+    def link(self, body, client_ip=None):
         device_name = body.get("device_name") or "device"
         if self.test_mode_enabled() and isinstance(body.get("test_userinfo"), dict):
             LOG.warning("TEST-MODE link request (stubbed userinfo)")
@@ -619,14 +750,133 @@ class LinkService:
             return 500, {"error": "identity state unavailable"}
         if login is None:
             return status, err_resp
-        gitea_token, token_name, err = self.mint_token(login, device_name)
+        wanted_device = str(body.get("device_id") or "").strip() or None
+        gitea_token, token_name, err = self.mint_token(login, device_name,
+                                                       wanted_device)
         if err:
             return 502, {"error": err}
         LOG.info("minted token %s for %s (sub %s)", token_name, login, sub)
+        # A device id the client generated once and keeps. Clients older than
+        # v1.2 do not send one; fall back to the token name, which is unique
+        # per link, so every device still lands in the registry.
+        device_id = wanted_device or token_name
+        with self.state.lock:
+            self.state.record_device(sub, device_id, {
+                "name": device_name,
+                "token_name": token_name,
+                "linked_at": datetime.now(timezone.utc).isoformat(
+                    timespec="seconds"),
+            })
+        self.audit.write("device.link", login=login, device_id=device_id,
+                         device_name=device_name, token_name=token_name,
+                         client_ip=client_ip)
         return 200, {"gitea_base": self.public_gitea, "login": login,
-                     "token": gitea_token, "token_name": token_name}
+                     "token": gitea_token, "token_name": token_name,
+                     "device_id": device_id}
 
-    def repos(self, body):
+    # -- device registry endpoints -----------------------------------------
+
+    def whoami(self, token):
+        """The login a forge token belongs to, or None.
+
+        Authorisation for every device endpoint rests on this: the caller
+        proves who they are by holding a working token for that account, and
+        the forge is the one that decides. No login is ever taken from the
+        request body.
+        """
+        if not token or not isinstance(token, str):
+            return None
+        status, resp = http_json(
+            "GET", f"{self.gitea}/api/v1/user",
+            headers={"Authorization": f"token {token}"})
+        if status == 200 and isinstance(resp, dict):
+            return resp.get("login")
+        return None
+
+    def devices(self, body, client_ip=None):
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        with self.state.lock:
+            _, devices = self.state.devices_for_login(login)
+        out = []
+        for device_id, rec in sorted(
+                devices.items(), key=lambda kv: kv[1].get("linked_at") or ""):
+            out.append({"device_id": device_id,
+                        "name": rec.get("name") or "device",
+                        "linked_at": rec.get("linked_at"),
+                        "token_name": rec.get("token_name"),
+                        "revoked_at": rec.get("revoked_at")})
+        return 200, {"login": login, "devices": out}
+
+    def revoke_device(self, body, client_ip=None):
+        """Sign one computer out. Immediate: the forge token is deleted, so
+        the next fetch or push from that machine fails at the server. There
+        is no 'stop syncing' flag for a client to honour or ignore."""
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        device_id = str(body.get("device_id") or "").strip()
+        if not device_id:
+            return 400, {"error": "device_id required"}
+        with self.state.lock:
+            sub, devices = self.state.devices_for_login(login)
+            record = devices.get(device_id)
+            if sub is None or record is None:
+                # Do not confirm the existence of ids on other accounts.
+                self.audit.write("device.revoke.denied", login=login,
+                                 device_id=device_id,
+                                 client_ip=client_ip,
+                                 reason="not a device of this account")
+                return 404, {"error": "no such device on this account"}
+            token_name = record.get("token_name")
+            err = self.delete_forge_token(login, token_name)
+            if err:
+                self.audit.write("device.revoke.failed", login=login,
+                                 device_id=device_id, token_name=token_name,
+                                 client_ip=client_ip, reason=err)
+                # Do NOT mark it revoked: the token still works, and a
+                # registry claiming otherwise is worse than an honest error.
+                return 502, {"error": err}
+            when = datetime.now(timezone.utc).isoformat(timespec="seconds")
+            self.state.mark_revoked(sub, device_id, when)
+        self.audit.write("device.revoke", login=login, device_id=device_id,
+                         token_name=token_name, client_ip=client_ip)
+        LOG.info("revoked device %s (%s) for %s", device_id, token_name, login)
+        return 200, {"revoked": device_id, "revoked_at": when}
+
+    def delete_forge_token(self, login, token_name):
+        """Gitea refuses token-auth deletion of tokens (403); admin basic
+        auth + Sudo works (verified 204 on 1.27.2)."""
+        if not token_name:
+            return "device has no recorded token to revoke"
+        status, resp = http_json(
+            "DELETE",
+            f"{self.gitea}/api/v1/users/{login}/tokens/"
+            f"{urllib.parse.quote(token_name, safe='')}",
+            headers={"Authorization": _basic(self.cfg["admin_login"],
+                                             self.cfg["admin_password"]),
+                     "Sudo": login})
+        if status in (204, 404):
+            # 404 = already gone. The caller wanted it dead; it is dead.
+            return None
+        return f"forge refused token deletion (HTTP {status}): {resp}"
+
+    def audit_read(self, body, client_ip=None):
+        login = self.whoami(body.get("token"))
+        if not login:
+            return 401, {"error": "invalid or revoked token"}
+        try:
+            limit = min(int(body.get("limit") or 100), 500)
+        except (TypeError, ValueError):
+            limit = 100
+        return 200, {"login": login,
+                     "events": self.audit.read_for(login, limit=limit),
+                     "note": "granthi-link events only; git pushes and pulls "
+                             "go straight to the forge and are recorded "
+                             "there, not here"}
+
+    def repos(self, body, client_ip=None):
         token = body.get("token")
         name = body.get("name")
         if not token or not name:
@@ -719,10 +969,21 @@ class Handler(BaseHTTPRequestHandler):
             return self._send(400, {"error": "invalid JSON body"})
         if not isinstance(body, dict):
             return self._send(400, {"error": "body must be a JSON object"})
+        # Recomputed rather than reused from the rate-limit block above:
+        # that block is skipped entirely when no limiter is configured, and
+        # an audit trail must not have holes because limiting was off.
+        peer = client_ip(self, self.service.trust_forwarded_for,
+                         self.service.trusted_proxies)
         if self.path == "/v1/link":
-            status, resp = self.service.link(body)
+            status, resp = self.service.link(body, peer)
         elif self.path == "/v1/repos":
-            status, resp = self.service.repos(body)
+            status, resp = self.service.repos(body, peer)
+        elif self.path == "/v1/devices":
+            status, resp = self.service.devices(body, peer)
+        elif self.path == "/v1/devices/revoke":
+            status, resp = self.service.revoke_device(body, peer)
+        elif self.path == "/v1/audit":
+            status, resp = self.service.audit_read(body, peer)
         else:
             return self._send(404, {"error": "not found"})
         self._send(status, resp)

tests/test_server.py

@@ -9,6 +9,7 @@ import sys
 import tempfile
 import threading
 import unittest
+import urllib.parse
 import urllib.request
 from unittest import mock
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -43,6 +44,13 @@ class StubUpstream(BaseHTTPRequestHandler):
                                         "[email protected]", "email":
                                         "[email protected]", "name": "Alice"})
             return self._json(401, {"error": "invalid token"})
+        if self.path == "/api/v1/user":
+            # whoami: the forge decides who a token belongs to
+            tok = self.headers.get("Authorization", "").replace("token ", "")
+            login = st["tokens_by_sha"].get(tok)
+            if not login:
+                return self._json(401, {"message": "unauthorized"})
+            return self._json(200, {"login": login})
         if self.path.startswith("/api/v1/users/") and not self.path.endswith("/tokens"):
             login = self.path.rsplit("/", 1)[1]
             if login in st["hide_once"]:
@@ -71,7 +79,14 @@ class StubUpstream(BaseHTTPRequestHandler):
                 "sudo": self.headers.get("Sudo", ""), "body": body})
             if not self.headers.get("Authorization", "").startswith("Basic "):
                 return self._json(401, {"message": "auth required"})
-            return self._json(201, {"sha1": "MINTED", "name": body["name"]})
+            login = self.path.split("/")[4]
+            # unique per mint, like a real forge -- two users may legitimately
+            # hold the same token NAME
+            st["sha_seq"] = st.get("sha_seq", 0) + 1
+            sha = f"SHA-{login}-{st['sha_seq']}-{body['name']}"
+            st["tokens_by_sha"][sha] = login
+            st["tokens"].setdefault(login, set()).add(body["name"])
+            return self._json(201, {"sha1": sha, "name": body["name"]})
         if self.path == "/api/v1/user/repos":
             if body["name"] in st["repos"]:
                 return self._json(409, {"message": "exists"})
@@ -81,6 +96,26 @@ class StubUpstream(BaseHTTPRequestHandler):
                                     f"alice/{body['name']}"})
         self._json(404, {})
 
+    def do_DELETE(self):
+        st = self.state
+        if self.path.startswith("/api/v1/users/") and "/tokens/" in self.path:
+            if not self.headers.get("Authorization", "").startswith("Basic "):
+                # matches real Gitea: token auth cannot delete tokens
+                return self._json(403, {"message": "basic auth required"})
+            login, name = self.path.split("/tokens/")
+            login = login.rsplit("/", 1)[1]
+            name = urllib.parse.unquote(name)
+            if st.get("revoke_fails"):
+                return self._json(500, {"message": "forge exploded"})
+            if name not in st["tokens"].get(login, set()):
+                return self._json(404, {"message": "not found"})
+            st["tokens"][login].discard(name)
+            for sha, owner in list(st["tokens_by_sha"].items()):
+                if owner == login and sha.endswith(f"-{name}"):
+                    del st["tokens_by_sha"][sha]
+            return self._json(204, {})
+        self._json(404, {})
+
     def log_message(self, *a):
         pass
 
@@ -88,7 +123,9 @@ class StubUpstream(BaseHTTPRequestHandler):
 class ServiceTestBase(unittest.TestCase):
     def setUp(self):
         StubUpstream.state = {"users": {}, "created": [], "repos": set(),
-                              "token_reqs": [], "hide_once": set()}
+                              "token_reqs": [], "hide_once": set(),
+                              "tokens": {}, "tokens_by_sha": {},
+                              "revoke_fails": False}
         self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), StubUpstream)
         threading.Thread(target=self.upstream.serve_forever, daemon=True).start()
         self.addCleanup(self.upstream.shutdown)
@@ -102,6 +139,7 @@ class ServiceTestBase(unittest.TestCase):
             "admin_token": "ADMTOK", "admin_login": "root",
             "admin_password": "rootpw", "test_mode": False,
             "state_path": self.state_path,
+            "audit_path": os.path.join(self.state_dir, "audit.jsonl"),
         })
 
     def enable_test_mode(self):
@@ -111,13 +149,17 @@ class ServiceTestBase(unittest.TestCase):
         patcher.start()
         self.addCleanup(patcher.stop)
 
-    def stub_link(self, sub, username, email=None, verified=None, device="d"):
+    def stub_link(self, sub, username, email=None, verified=None, device="d",
+                  device_id=None, client_ip=None):
         ui = {"sub": sub, "preferred_username": username}
         if email is not None:
             ui["email"] = email
         if verified is not None:
             ui["email_verified"] = verified
-        return self.svc.link({"test_userinfo": ui, "device_name": device})
+        body = {"test_userinfo": ui, "device_name": device}
+        if device_id:
+            body["device_id"] = device_id
+        return self.svc.link(body, client_ip)
 
     def read_state(self):
         with open(self.state_path) as f:
@@ -146,7 +188,8 @@ class TestLink(ServiceTestBase):
                                       "device_name": "mac studio"})
         self.assertEqual(status, 200)
         self.assertEqual(resp["login"], "alice.smith")
-        self.assertEqual(resp["token"], "MINTED")
+        # the stub now issues a per-token sha so revocation can be tested
+        self.assertIn("granthi-sync-", resp["token"])
         self.assertEqual(resp["gitea_base"], "http://public.example:3041")
         st = StubUpstream.state
         self.assertEqual(len(st["created"]), 1)
@@ -480,6 +523,267 @@ class TestHealth(HandlerTestBase):
         self.assertEqual(body["service"], "granthi-link")
 
 
+
+class TestDeviceRegistry(ServiceTestBase):
+    """Which computers can reach my files, and can I cut one off."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def link_device(self, device_id, name="laptop"):
+        status, resp = self.stub_link("s1", "alice", device=name,
+                                      device_id=device_id)
+        self.assertEqual(status, 200, resp)
+        return resp
+
+    def test_devices_are_recorded_and_listed_per_account(self):
+        self.link_device("dev-a", "work-laptop")
+        second = self.link_device("dev-b", "home-mac")
+        status, resp = self.svc.devices({"token": second["token"]})
+        self.assertEqual(status, 200)
+        self.assertEqual(resp["login"], "alice")
+        ids = [d["device_id"] for d in resp["devices"]]
+        self.assertEqual(sorted(ids), ["dev-a", "dev-b"])
+        names = {d["device_id"]: d["name"] for d in resp["devices"]}
+        self.assertEqual(names["dev-a"], "work-laptop")
+
+    def test_listing_needs_a_working_token(self):
+        self.link_device("dev-a")
+        status, _ = self.svc.devices({"token": "not-a-real-token"})
+        self.assertEqual(status, 401)
+        status, _ = self.svc.devices({})
+        self.assertEqual(status, 401)
+
+    def test_login_is_never_taken_from_the_request_body(self):
+        """Authorisation comes from the forge's answer about the token, so a
+        body claiming another account changes nothing."""
+        first = self.link_device("dev-a")
+        status, resp = self.svc.devices({"token": first["token"],
+                                         "login": "somebody-else"})
+        self.assertEqual(resp["login"], "alice")
+
+    def test_relinking_the_same_device_does_not_duplicate_it(self):
+        self.link_device("dev-a", "laptop")
+        self.link_device("dev-a", "laptop-renamed")
+        status, resp = self.svc.devices({"token": self.link_device("dev-a")["token"]})
+        self.assertEqual(len(resp["devices"]), 1)
+
+    def test_old_clients_without_a_device_id_still_register(self):
+        resp = self.stub_link("s1", "alice", device="ancient")[1]
+        self.assertEqual(resp["device_id"], resp["token_name"])
+        status, listed = self.svc.devices({"token": resp["token"]})
+        self.assertEqual([d["device_id"] for d in listed["devices"]],
+                         [resp["token_name"]])
+
+
+class TestRevoke(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.a = self.stub_link("s1", "alice", device="laptop-a",
+                                device_id="dev-a")[1]
+        self.b = self.stub_link("s1", "alice", device="laptop-b",
+                                device_id="dev-b")[1]
+
+    def test_revoking_a_device_kills_its_token_at_the_forge(self):
+        status, resp = self.svc.revoke_device({"token": self.b["token"],
+                                               "device_id": "dev-a"})
+        self.assertEqual(status, 200, resp)
+        # the revoked device's token no longer authenticates ANYTHING
+        self.assertEqual(self.svc.whoami(self.a["token"]), None)
+        # the device that did the revoking still works
+        self.assertEqual(self.svc.whoami(self.b["token"]), "alice")
+
+    def test_a_device_can_revoke_itself(self):
+        status, _ = self.svc.revoke_device({"token": self.a["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 200)
+        self.assertEqual(self.svc.whoami(self.a["token"]), None)
+
+    def test_revoked_device_is_marked_not_deleted(self):
+        self.svc.revoke_device({"token": self.b["token"],
+                                "device_id": "dev-a"})
+        status, resp = self.svc.devices({"token": self.b["token"]})
+        by_id = {d["device_id"]: d for d in resp["devices"]}
+        self.assertIsNotNone(by_id["dev-a"]["revoked_at"])
+        self.assertIsNone(by_id["dev-b"]["revoked_at"])
+
+    def test_cannot_revoke_a_device_on_another_account(self):
+        other = self.stub_link("s2", "mallory", device="theirs",
+                               device_id="dev-x")[1]
+        status, resp = self.svc.revoke_device({"token": other["token"],
+                                               "device_id": "dev-a"})
+        self.assertEqual(status, 404)
+        # alice's device is untouched
+        self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
+
+    def test_a_failed_forge_delete_is_not_recorded_as_revoked(self):
+        """A registry that says 'revoked' while the token still works is
+        worse than an honest error."""
+        StubUpstream.state["revoke_fails"] = True
+        status, _ = self.svc.revoke_device({"token": self.b["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 502)
+        StubUpstream.state["revoke_fails"] = False
+        _, listed = self.svc.devices({"token": self.b["token"]})
+        by_id = {d["device_id"]: d for d in listed["devices"]}
+        self.assertIsNone(by_id["dev-a"]["revoked_at"])
+        self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
+
+    def test_revoking_twice_is_not_an_error(self):
+        self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"})
+        status, _ = self.svc.revoke_device({"token": self.b["token"],
+                                            "device_id": "dev-a"})
+        self.assertEqual(status, 200)  # forge 404 = already gone = success
+
+
+class TestAuditLog(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def test_link_and_revoke_are_recorded_with_who_and_where(self):
+        first = self.stub_link("s1", "alice", device="laptop",
+                               device_id="dev-a", client_ip="203.0.113.9")[1]
+        self.stub_link("s1", "alice", device="mac", device_id="dev-b")
+        second = self.svc.devices({"token": first["token"]})[1]
+        self.assertEqual(len(second["devices"]), 2)
+        self.svc.revoke_device({"token": first["token"],
+                                "device_id": "dev-b"}, client_ip="198.51.100.4")
+
+        status, resp = self.svc.audit_read({"token": first["token"]})
+        self.assertEqual(status, 200)
+        events = resp["events"]
+        kinds = [e["event"] for e in events]
+        self.assertIn("device.link", kinds)
+        self.assertIn("device.revoke", kinds)
+        link_ev = [e for e in events if e["event"] == "device.link"
+                   and e["device_id"] == "dev-a"][0]
+        self.assertEqual(link_ev["client_ip"], "203.0.113.9")
+        self.assertEqual(link_ev["login"], "alice")
+        revoke_ev = [e for e in events if e["event"] == "device.revoke"][0]
+        self.assertEqual(revoke_ev["client_ip"], "198.51.100.4")
+        self.assertEqual(revoke_ev["device_id"], "dev-b")
+
+    def test_events_are_newest_first_and_scoped_to_the_caller(self):
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        self.stub_link("s2", "mallory", device_id="dev-x")
+        _, resp = self.svc.audit_read({"token": mine["token"]})
+        self.assertTrue(resp["events"])
+        self.assertTrue(all(e["login"] == "alice" for e in resp["events"]))
+        stamps = [e["ts"] for e in resp["events"]]
+        self.assertEqual(stamps, sorted(stamps, reverse=True))
+
+    def test_reading_the_log_needs_a_working_token(self):
+        self.stub_link("s1", "alice", device_id="dev-a")
+        status, _ = self.svc.audit_read({"token": "nope"})
+        self.assertEqual(status, 401)
+
+    def test_a_refused_revoke_is_recorded_too(self):
+        self.stub_link("s1", "alice", device_id="dev-a")
+        other = self.stub_link("s2", "mallory", device_id="dev-x")[1]
+        self.svc.revoke_device({"token": other["token"],
+                                "device_id": "dev-a"}, client_ip="192.0.2.7")
+        _, resp = self.svc.audit_read({"token": other["token"]})
+        denied = [e for e in resp["events"]
+                  if e["event"] == "device.revoke.denied"]
+        self.assertEqual(len(denied), 1)
+        self.assertEqual(denied[0]["client_ip"], "192.0.2.7")
+
+    def test_the_log_says_what_it_cannot_see(self):
+        """Reading this and believing it lists file activity would be a real
+        mistake: git traffic never passes through this service."""
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        _, resp = self.svc.audit_read({"token": mine["token"]})
+        self.assertIn("pushes and pulls", resp["note"])
+
+    def test_a_broken_audit_file_does_not_break_the_request(self):
+        self.svc.audit.path = "/nonexistent-dir/audit.jsonl"
+        status, _ = self.stub_link("s1", "alice", device_id="dev-a")
+        self.assertEqual(status, 200)  # linking still works
+
+    def test_rotation_keeps_the_previous_file_readable(self):
+        self.svc.audit.keep_bytes = 1500  # one rotation, not many
+        mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
+        for i in range(20):
+            self.svc.audit.write("noise", login="alice", n=i)
+        _, resp = self.svc.audit_read({"token": mine["token"]}, )
+        self.assertGreater(len(resp["events"]), 15)  # spans both files
+        self.assertTrue(os.path.exists(self.svc.audit.path + ".1"))
+
+
+class TestTokenNameUniqueness(ServiceTestBase):
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+
+    def test_two_devices_named_the_same_get_different_token_names(self):
+        """Revocation deletes by token name, so a collision would mean
+        signing out one laptop kills the other."""
+        a = self.stub_link("s1", "alice", device="macbook", device_id="dev-a")[1]
+        b = self.stub_link("s1", "alice", device="macbook", device_id="dev-b")[1]
+        self.assertNotEqual(a["token_name"], b["token_name"])
+        self.svc.revoke_device({"token": b["token"], "device_id": "dev-a"})
+        self.assertIsNone(self.svc.whoami(a["token"]))
+        self.assertEqual(self.svc.whoami(b["token"]), "alice")
+
+
+class TestDeviceEndpointsOverHttp(HandlerTestBase):
+    """Through the real handler, not the service method.
+
+    The first live run of the revoke endpoint returned 429 on every call
+    while every unit test passed: the tests called svc.revoke_device()
+    directly, so nothing ever went through the rate limiter. Any route whose
+    limit is a policy decision needs at least one test that speaks HTTP.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.a = self.stub_link("s1", "alice", device="laptop",
+                                device_id="dev-a")[1]
+        self.b = self.stub_link("s1", "alice", device="desktop",
+                                device_id="dev-b")[1]
+
+    def post(self, path, obj):
+        body = json.dumps(obj).encode()
+        return self.raw_post(path, body,
+                             {"Content-Type": "application/json",
+                              "Content-Length": str(len(body))})
+
+    def test_revoke_is_reachable_and_not_rate_limited_away(self):
+        status, resp = self.post("/v1/devices/revoke",
+                                 {"token": self.b["token"],
+                                  "device_id": "dev-a"})
+        self.assertEqual(status, 200, resp)
+        self.assertIsNone(self.svc.whoami(self.a["token"]))
+
+    def test_repeated_revokes_keep_working(self):
+        """A person signing several lost machines out in one sitting must not
+        be locked out partway through."""
+        for i in range(12):
+            self.stub_link("s1", "alice", device=f"d{i}", device_id=f"gone-{i}")
+        for i in range(12):
+            status, resp = self.post("/v1/devices/revoke",
+                                     {"token": self.b["token"],
+                                      "device_id": f"gone-{i}"})
+            self.assertEqual(status, 200, f"revoke {i}: {resp}")
+
+    def test_devices_and_audit_are_reachable_over_http(self):
+        status, resp = self.post("/v1/devices", {"token": self.a["token"]})
+        self.assertEqual(status, 200)
+        self.assertEqual(len(resp["devices"]), 2)
+        status, resp = self.post("/v1/audit", {"token": self.a["token"]})
+        self.assertEqual(status, 200)
+        self.assertTrue(resp["events"])
+
+    def test_no_route_a_client_uses_is_configured_to_zero(self):
+        """0 means DISABLED in this limiter, so a zero on a live route is a
+        dead endpoint, not an unlimited one."""
+        for route, (limit, _window) in granthi_link.DEFAULT_RATE_RULES.items():
+            self.assertGreater(limit, 0, f"{route} is disabled by its rule")
+
 if __name__ == "__main__":
     unittest.main()