Merge pull request 'docs: this describes the prod deployment now, not beta' (#11 ) from fix/docs-prod-forge-bases into main

Docs-only. NOTE: merged WITHOUT a completed granthi-review. The gate's webhook rejection was fixed today (item 52eea395), but reviews still fail open — diff fetch timeout, and shre-router returning prose instead of JSON (item 6d4e8819). Merged on explicit human approval, not on a green check.
docs: this describes the prod deployment now, not beta
2026-08-30 01:25:54 -04:00 · 2026-08-29 23:20:30 -04:00 · 2026-08-23 16:46:22 -04:00 · 2026-08-23 16:44:49 -04:00 · 2026-08-23 16:14:56 -04:00 · 2026-08-23 16:14:21 -04:00
7 changed files with 4481 additions and 109 deletions
@@ -1,8 +1,10 @@
-# granthi-sync v1
+# granthi-sync v1.2
 The signup → download → link-folders → cloud product spine for the Granthi
-forge, tested against the BETA forge (granthi-beta.shre.ai). Python 3 stdlib +
+forge. Developed and live-E2E-tested against the BETA forge
-git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
+(granthi-beta.shre.ai); **the deployed service now runs against the PRODUCTION
 forge** — see "Where this actually runs" below. Python 3 stdlib + git CLI only
 — same portability heritage as the estate's `gitea_sync.py` mesh.
 ```
   ┌──────────────┐   device flow    ┌─────────────────┐
@@ -13,7 +15,8 @@ git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
   │              │  POST /v1/link {zitadel_access_token, device_name}
   │              │ ───────────────▶ ┌───────────────────────────────┐
   │              │ ◀─────────────── │ granthi-link  :3042           │
-   │              │  {login, token}  │ (granthi VPS, tailnet-only)   │
+   │              │  {login, token}  │ (granthi VPS; public via      │
   │              │                  │  https://granthi-link.shre.ai)│
   │              │                  │  · userinfo validation        │
   │              │  POST /v1/repos  │  · ensure Gitea user (admin)  │
   │              │ ───────────────▶ │  · mint scoped user token     │
@@ -21,16 +24,35 @@ git CLI only — same portability heritage as the estate's `gitea_sync.py` mesh.
   │              │   git push/fetch (user token    │ admin API
   │              │   via credential helper)        ▼
   │              │ ───────────────▶ ┌───────────────────────────────┐
-   └──────────────┘                  │ BETA forge  :3041             │
+   └──────────────┘                  │ PROD forge  :3040             │
-                                     │ granthi-beta.shre.ai          │
+                                     │ granthi.shre.ai               │
                                     └───────────────────────────────┘
 ```
 ## Where this actually runs
 Verified on the granthi VPS (`[email protected]`) on 2026-08-30, because
 this file previously described the beta tier long after the deployment moved:
 | | value |
 |---|---|
 | service | `/opt/granthi-link/`, `granthi-link.service`, `systemctl is-active` → `active` |
 | public entry | `https://granthi-link.shre.ai` — `/health` → 200. `/` → 404 is **no root route**, not an outage |
 | `gitea_base` | `http://127.0.0.1:3040` (container `gitea-central-gitea-1`) |
 | `public_gitea_base` | `https://granthi.shre.ai` |
 | rate limiting | `trust_forwarded_for: true`, `trusted_proxies: ["100.107.37.98/32"]`, no `rules` key → falls back to `DEFAULT_RATE_RULES` |
 | deployed version | `/health` reports **1.1.0** while this repo is **v1.2** — the running service lags `main` |
 So a new computer that follows Quickstart lands on the **production** forge.
 Beta remains where changes are proven before they reach it.
 ## Quickstart (invited user)
 You need a shre-id account — an operator creates it; there is no open signup
-(see "Invite-only story"). You also need to be on the tailnet: the
+(see "Invite-only story"). You do **not** need to be on any private network:
-provisioning service is not publicly exposed yet.
+the provisioning service answers at `https://granthi-link.shre.ai`, which is
 the client's default server. Internal machines can still pass
 `--server http://100.111.127.127:3042` to reach it over the tailnet.
 ```sh
 git clone https://granthi.shre.ai/nirpa/granthi-sync.git
@@ -50,10 +72,17 @@ cd granthi-sync
 # 3b. ...or push a local folder up. It becomes a private repo.
 ./bin/granthi-sync add ~/work/notes
 # 3c. ...or pull down everything this account is allowed to see.
 ./bin/granthi-sync get --all --into ~/granthi
 # 4. Keep everything synced. Autocommits, ff-pulls, pushes; skips anything
 #    that has diverged rather than merging or forcing.
 ./bin/granthi-sync watch                # --once for a single pass
-./bin/granthi-sync status               # what is linked, last sync, divergence
+./bin/granthi-sync status               # what is linked, mode, last sync
 # 5. Go back to how a folder looked at some point in time.
 ./bin/granthi-sync snapshots ~/work/notes
 ./bin/granthi-sync restore  ~/work/notes --at 20260823T142530Z
 ```
 Run `watch` as a background daemon on macOS with
@@ -67,6 +96,111 @@ no account, no token, no partial state. Just run `link` again.
 to merge; when a folder shows `DIVERGED` in `status`, resolve it in git or on
 the forge web UI. The client will never force or auto-merge your work.
 ## Two modes, because two very different folders ask for this
 A folder people sync is either *their documents* or *their git project*, and
 the correct behaviour is opposite in each case. Each linked folder therefore
 carries a `mode`.
 | | `mirror` | `snapshot` |
 |---|---|---|
 | chosen for | a plain folder `add` turned into a repo | a folder that was already a git repo, and anything `get` clones |
 | commits on your behalf | yes, `sync: <ISO ts>` | **never** |
 | where work lands | the branch | `refs/granthi-backup/<device>/<ts>` |
 | a restore point is | every commit | every snapshot |
 `snapshot` mode is what "the work may not be committed, but it is still
 backed up" means in git terms. Each pass loads a scratch index from HEAD,
 stages the working tree into *that* index, writes a tree, and commits it with
 `commit-tree`. HEAD, your index, your stash and every file on disk are
 untouched — you can be mid-rebase with a dirty tree and the backup still
 records exactly what is on the disk right now. The user's history stays the
 user's.
 Why a custom ref namespace: verified on the beta forge (Gitea 1.27.2) that
 `refs/granthi-backup/...` is accepted, is readable through `ls-remote`, and
 does **not** appear in the branch list. Under `refs/heads` a machine taking a
 backup every 30 seconds would bury the branches a person actually made.
 Snapshots are parented on HEAD and deliberately **not** chained to the
 previous snapshot: chaining would keep every old snapshot reachable from the
 newest, so pruning a ref would free nothing and retention would be
 decorative.
 **Retention** (or 30-second backups become a disk leak nobody can navigate):
 everything is kept for 24 h, then thinned to hourly for 7 days, then daily.
 Pruning runs at most hourly, per device, and only over that device's own
 refs. A ref whose timestamp this version cannot parse is **kept** — deleting
 the unrecognised is how a backup system loses the one thing someone needed.
 **Restore never writes over the working tree.** `restore` materialises a
 restore point into a *new* directory and refuses a non-empty destination.
 Someone restoring a backup is already having a bad day; overwriting the files
 they still have would make the recovery tool the second disaster.
 ## The credential helper must be the ONLY helper (found by live QA)
 `credential.helper` is a list that accumulates across system, global and repo
 config, and git asks every helper in it. A stock mac already has two —
 `osxkeychain` from Xcode's gitconfig, and `store` from many people's
 `~/.gitconfig` — and they lose in both directions:
 * **reading:** a stale entry for the forge host answers before our helper, so
  pushes fail `remote: Failed to authenticate user` long after the token was
  rotated, and nothing in this tool's config explains why. This is exactly
  how the first live-QA run failed;
 * **writing:** git calls `approve` on every helper after a successful auth,
  so `store` copies the forge token into `~/.git-credentials` **in
  plaintext**. Keeping the token in a 0600 file and out of remote URLs buys
  nothing if git then hands it to a plaintext store.
 So `install_credential_helper` (and the `git clone` in `get`) sets an **empty**
 `credential.helper` first, which resets the inherited list, then adds ours.
 Exactly one helper serves this repo.
 Corollary worth remembering: a token embedded in a remote URL gets saved by
 `store` on first use. During QA a verification clone with a URL-embedded
 token re-created the very entry that had just been cleaned out. That is the
 whole reason this client passes tokens through a helper and never a URL.
 ## Device identity
 `link` mints a uuid on first run and persists it in `~/.granthi-sync/config.json`
 as `device_id`, and sends it to `/v1/link`. Hostnames are neither stable
 (people rename laptops) nor unique (every new mac is "Mac mini"), so a
 hostname cannot key a backup ref or a device registry — two machines would
 overwrite each other's snapshots. The service-side device registry is the
 next phase; the client leads so the id already exists when it lands.
 ## What "add the computer to the network" means — and does not
 The onboarding shape is: download → login → **the device is federated to the
 account** → the device can reach its repos.
 The middle step is a *device registration*, not a network membership. Those
 sound like one step and must not be built as one: this estate's tailnet is a
 single flat private network carrying the granthi VPS, aros-vps, the Shadow
 box and the Mac. Putting a customer's laptop on it to let them sync a folder
 would hand that laptop L3 reach to every piece of infrastructure we run.
 So:
 * **Our own machines** may join the tailnet — that is an operator action with
  an operator's judgement behind it.
 * **Customer devices never do.** Their transport is public HTTPS to
  granthi-link and the forge through cloudflared. **Done 2026-08-23:**
  `https://granthi-link.shre.ai` fronts `:3042` on the `pulse-granthi-edge`
  tunnel, so a new computer signs itself in with one command and never
  touches the private network.
  Two things that bit during that rollout, worth not rediscovering:
  `link.granthi.shre.ai` fails TLS — Cloudflare's Universal SSL covers
  `shre.ai` and `*.shre.ai`, **not** a third-level `*.granthi.shre.ai`, so
  the hostname has to be second-level. And exposure REQUIRES
  `trust_forwarded_for: true` with `trusted_proxies: ["100.107.37.98/32"]`
  in the same change: behind the tunnel every request otherwise looks like
  the tunnel itself, and one abuser would spend everybody's rate budget.
 ## Components
 ### `server/granthi_link.py` — provisioning service (granthi VPS)
@@ -80,16 +214,72 @@ the forge web UI. The client will never force or auto-merge your work.
  **64 KB** (413 beyond; missing `Content-Length` → 411, invalid → 400).
 * `POST /v1/repos {token, name, private}` → creates the user repo with the
  USER token; clone/html URLs are rebased onto `public_gitea_base` because the
-  container `ROOT_URL` (https://granthi-beta.shre.ai) does not resolve for
+  container `ROOT_URL` does not necessarily resolve for the client that asked
-  tailnet-only clients.
+  (it was a tailnet-only address on beta; on prod the rebase keeps clone URLs
  on `https://granthi.shre.ai` rather than the container's own view).
 Deployment: `/opt/granthi-link/{granthi_link.py,config.json,state.json}` +
 systemd unit `granthi-link.service`; binds `127.0.0.1:3042` **and**
-`100.111.127.127:3042` (tailnet). **Not publicly exposed** — see promotion
+`100.111.127.127:3042` (tailnet), and is **publicly reachable** at
-window. The service **refuses to start** (exit 2) unless `config.json` is
+`https://granthi-link.shre.ai` through the `pulse-granthi-edge` cloudflared
 tunnel (done 2026-08-23; re-verified 2026-08-30). The service **refuses to
 start** (exit 2) unless `config.json` is
 mode 0600/0400 and owned by the user it runs as — the config carries the
 forge admin password, so permissive perms fail closed, not open.
 #### Rate limiting
 Sliding-window, per `(route, client)`, **on by default** — `/v1/link` creates
 accounts and mints tokens, so unlimited has to be a deliberate config act,
 never an omission. Defaults: `/v1/link` 5/hour, `/v1/repos` 60/hour. `GET
 /health` is never limited. Over the limit → **429** with a `Retry-After`
 header, decided *before* the body is read so an abusive caller costs nothing.
 ```json
 "rate_limit": {
  "enabled": true,
  "trust_forwarded_for": false,
  "trusted_proxies": [],
  "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
 }
 ```
 * Anything malformed **refuses startup** rather than silently meaning
  unlimited: a bad rule, a non-boolean `enabled`/`trust_forwarded_for` (JSON
  `null`, `0`, or the *string* `"false"` — every non-empty string is truthy),
  or a `rate_limit` that is not an object. `[0, N]` disables an endpoint
  outright.
 * Denied requests are *not* recorded, so a client that keeps hammering cannot
  push its own window forward and lock itself out permanently.
 * State is an in-process dict behind a lock — granthi-link is one
  `ThreadingHTTPServer`, so that is the entire store. **If this ever runs
  multi-process or multi-host, the limiter must move with it.** The clock is
  read *inside* the lock; taken outside, racing threads append out of order
  and both `Retry-After` and window reclamation silently go wrong.
 * The key store is capped (`MAX_RATE_KEYS`) **per route**, not globally. At
  capacity it reclaims expired windows, and if every window is still live it
  **refuses the new key** — fail closed. Evicting a live window would let an
  attacker who can mint many distinct keys clear their *own* limit on demand.
  The per-route budget matters just as much: with one shared table, a flood
  of cheap `/v1/repos` keys would exhaust it and lock brand-new `/v1/link`
  clients out, turning the fail-closed guard into a cross-route DoS.
 * Rule values must be real integers. `bool` subclasses `int` in Python, so
  `[5, true]` would otherwise pass as a **1-second** window — an hourly limit
  quietly becoming ~5/sec.
 * `trust_forwarded_for` is **off** by default, and turning it on **requires a
  non-empty `trusted_proxies`** — the header is honored only when the socket
  peer is in that list. Without it, anyone reaching the origin directly (it
  also listens on the tailnet) could pick and rotate their own rate-limit key
  just by sending a header. Turn it on when exposing behind cloudflared,
  where every request otherwise arrives from the tunnel and one abuser would
  starve everyone. A caller can *prepend* anything to `X-Forwarded-For`; a
  trusted proxy *appends* the peer it actually saw, so the service reads the
  **last** entry, never the first, and requires it to parse as a real IP.
  `trusted_proxies` is validated at startup: it must be a list (a bare string
  would be iterated character by character), every entry a valid network, and
  wildcards (`0.0.0.0/0`, `::/0`) are refused outright — they would restore
  exactly the "trust anyone" hole the setting exists to close.
 #### Identity binding (`state.json`)
 `/v1/link` originally bound purely by `preferred_username` / email
@@ -158,13 +348,32 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
  (`authorization_pending`/`slow_down` handled), then calls `/v1/link`.
  `--token` skips the device flow with a ready Zitadel token (headless/dev).
  Result stored in `~/.granthi-sync/config.json` (0600).
-* `list` — every repo the linked token can see, with the local folder each
+* `list [pattern]` — every repo the linked token can see, with the local
-  is already synced to. Reads `GET /api/v1/user/repos` on the forge
+  folder each is already synced to. `pattern` narrows the table by name
  (substring, or a glob like `work-*`), case-insensitively, against both
  `owner/name` and the bare name. Filtering is display-only: the set already
  came from the forge under this account's token. Reads
  `GET /api/v1/user/repos` on the forge
  **directly** with the scoped user token — no granthi-link round-trip, so
  the read path needs no service change. Pagination is followed to a short
  page; if the `FORGE_MAX_PAGES` guard trips, the output says the list is
  incomplete rather than letting a bounded page read as the whole set.
-* `get <repo|owner/repo> [--into DIR]` — the download half of `add`. Clones
+* `get --all [--into DIR] [--mode M]` — clone every repo this account can
  see, skipping the ones already linked here. **"Only the repos they are
  granted" needs no client-side permission logic**: `/api/v1/user/repos` is
  evaluated by the forge against this account's own scoped token, so the
  list *is* the grant. A client-side filter would be a second opinion about
  someone else's authorisation. One repo failing does not abandon the rest,
  and a truncated listing is reported loudly — `--all` must never quietly
  mean "the first 2000". `alice/notes` and `bob/notes` both want
  `<base>/notes`; the second is cloned to `<base>/bob-notes` and the clash is
  logged, because reporting it as "already present" would leave the user
  believing they had pulled both.
 * `get <repo|owner/repo> [--into DIR] [--mode M]` — the download half of
  `add`. Defaults to `snapshot` mode unless the repo carries a
  `.granthi-sync.json` marker saying otherwise, so a plain synced folder
  behaves the same on the second machine while someone's real project is
  never autocommitted onto. Clones
  with `--origin granthi` (the remote name `watch` looks for) and
  `-c credential.helper=…` (the repo does not exist yet, so the helper
  cannot be installed first; git also persists it into the new config), then
@@ -172,22 +381,76 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
  so a cloned repo is picked up by `watch` immediately. Refuses a non-empty
  destination. Branch is read with `symbolic-ref` (an empty repo has an
  unborn HEAD) and falls back to `main`.
-* `add <folder> [--name N] [--private|--public]` — `git init -b main` if
+* `add <folder> [--name N] [--private|--public] [--mode M] [--force]` —
-  needed, creates the cloud repo via `/v1/repos`, adds remote `granthi`,
+  `git init -b main` if needed, creates the cloud repo via `/v1/repos`, adds
-  initial commit + push. The token is delivered by a **git credential
+  remote `granthi`, initial commit + push. The token is delivered by a **git
-  helper** (the client's hidden `git-credential` subcommand reading the 0600
+  credential helper** (the client's hidden `git-credential` subcommand
-  config) — never embedded in the remote URL (estate rule).
+  reading the 0600 config) — never embedded in the remote URL (estate rule).
-* `watch [--interval 30] [--once]` — per folder: autocommit
+  Pushes the folder's **current** branch, not a hardcoded `main`: an existing
-  (`sync: <ISO ts>`) → fetch → ff-pull if remote strictly ahead → push if
+  repo may sit on `master` or a feature branch, and publishing that work
-  local strictly ahead. **DIVERGED → log + record + SKIP. Never force, never
+  under the wrong name is not a cosmetic error.
-  merge** — the same policy as the mesh. SIGTERM-clean.
+  Two guards, because `add -A` takes whatever it is given: a starter
-* `status` — table of linked folders, last sync, divergence flags.
+  `.gitignore` is seeded when the folder has none (an existing one is never
  touched — it is the user's), and a folder over 20 000 files / 512 MB is
  refused unless `--force`. The seeded ignore file covers `.env`, `*.key`,
  `*.pem`, `id_rsa` and friends, and it governs snapshots too — the scratch
  index honours `.gitignore` exactly as a normal commit does.
  The size guard measures what git *would* sync, ignore rules included
  (including the machine's global excludes), because its own advice is "add
  a .gitignore for what should not sync" and advice that changes nothing is
  worse than none. It asks git through a **throwaway git dir outside the
  folder**, so a refused `add` leaves no `.git` behind in a directory the
  user never agreed to turn into a repo.
 * `watch [--interval 30] [--once]` — per folder, by mode. `mirror`:
  autocommit (`sync: <ISO ts>`) → fetch → ff-pull if remote strictly ahead →
  push if local strictly ahead. `snapshot`: fetch → push a snapshot of the
  working tree to this device's backup ref → ff-pull only when the tree is
  clean (local edits are already safe on the backup ref, so it reports and
  leaves the tree alone rather than failing) → push the user's own commits
  when they are strictly ahead. **DIVERGED → log + record + SKIP. Never
  force, never merge** — the same policy as the mesh — **but the backup
  still happens**, because divergence is when work is most at risk.
  Retention pruning runs at most hourly. SIGTERM-clean.
 * `snapshots <folder> [--limit 20]` — restore points, newest first, **across
  every device**, with the device that took each one. Read from the
  **remote**, not a local cache: the feature exists for the case where this
  machine is gone.
  The three scopes differ deliberately. Writing is device-scoped, so two
  machines never overwrite each other. Pruning is device-scoped, so machine A
  never applies its clock to machine B's refs. **Reading is not scoped** — a
  replacement laptop has a new id, and scoping the read to it would print
  "no restore points yet" while the backups sit on the forge. That defect
  was live in the first draft and is now pinned by a test that restores a
  dead machine's work from a fresh clone.
 * `restore <folder> --at <ts|sha> [--into DIR]` — materialise one restore
  point into a new directory; refuses a non-empty destination. Accepts what
  `snapshots` printed in either mode, including a mirror-mode `%cI`
  timestamp. Two commits inside the same second share that timestamp, so an
  ambiguous `--at` is **refused with the candidate ids** rather than
  resolved by guessing.
 * `status` — table of linked folders, mode, last sync, divergence flags.
 * Run as a daemon on macOS with `client/launchd/ai.granthi.sync.plist`
  (edit the script path, then `launchctl bootstrap gui/$UID <plist>`).
 ## Tests
-* `python3 -m unittest discover -s tests` — 65 tests: autocommit/ff/diverged
+* `python3 -m unittest discover -s tests` — 198 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
  up; a dirty tree blocking the ff-pull but not the backup; retention keeping
  everything recent, thinning to hourly then daily, and **keeping**
  unparseable timestamps; prune deleting only the thinned refs; `.gitignore`
  seeding never overwriting an existing one and keeping `.env` out of
  snapshots; the folder-size guard being bounded rather than walking the
  disk; mode detection; `list` filtering; `get --all` skipping what is
  already present, defaulting to snapshot mode, and shouting about
  truncation; `restore` writing a new folder, refusing a non-empty
  destination, and leaving the working tree alone; and the credential helper
  being the only one the repo consults, proven by driving
  `git credential fill` against a deliberately poisoned outer helper.
 * Earlier suite: autocommit/ff/diverged
  logic against real temp git repos (including "diverged never touches the
  remote"), config 0600 handling (including umask-proof creation and a
  no-chmod guard), credential-helper quoting/injection, mocked device-flow
@@ -214,20 +477,187 @@ 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.
 ## Workspace bootstrap — what a new computer should pull first
    granthi-sync bootstrap ~/granthi/acme --into ~/work
 A workspace is a repo holding a `workspace.json`:
 ```json
 {"name": "acme",
 "repos": ["notes", {"name": "reports", "mode": "mirror"}],
 "apps": [{"id": "dash", "repo": "nirpa/hermes-agent", "model": "aum/70b",
           "setup": "docs/SETUP.md", "needs_keys": ["anthropic"]}]}
 ```
 `bootstrap` pulls the repos it names — **still only the ones the forge grants
 this account**; a manifest asking for a repo you cannot see prints
 `NOT GRANTED` and moves on, because that is a permissions answer and not an
 error to route around.
 **It does not install applications, and that is the design, not a shortcut.**
 Dash, deck, genie and shiva each have their own repo, deploy path and
 reviewers. A sync client that installed them would create a second,
 unreviewed deploy path beside the real one — the same copy-instead-of-refer
 mistake the estate rulebook exists to end. So `apps` is *referenced*: the
 command prints each app's repo, the model it uses, its setup doc, and which
 vault keys must exist, with the `shre-cred request` line to supply them.
 Per-repo `mode` in the manifest overrides the default, so a documents folder
 can be declared `mirror` while everything else stays on the safe `snapshot`.
 ## Sharing: grants and invites
 Two commands, because there are two situations.
 **They already have an account** — share directly:
    granthi-sync share bob --repo notes                 # write by default
    granthi-sync share bob --repo notes --permission read
    granthi-sync share bob --repo notes --revoke
    granthi-sync shared --repo notes                    # who can see it
 **They do not have an account yet** — invite them:
    granthi-sync invite carol@example.com --repo notes --repo reports
 An invite creates **nothing**: no account, no token, no collaborator row. The
 grant is held against their email and applied the first time they run
 `granthi-sync link`. An invite that is never accepted leaves nothing behind.
 Three properties worth keeping:
 * **The forge decides who may share.** Before recording anything, the service
  asks Gitea whether the caller can administer that repo (listing
  collaborators requires repo admin, so a 200 is Gitea's own answer). Deciding
  it here from the repo name would be a second opinion about someone else's
  authorisation — and the wrong one the first time a repo is transferred.
 * **An unverified email collects nothing.** The address is the only thing
  tying a promise to a person, so an invite is applied only when the IdP says
  the address is verified. The invite stays pending rather than being consumed.
 * **A failed grant never blocks a sign-in.** If a promised repo has since been
  deleted, the person still links successfully and the failure is audited.
  Someone must not be locked out of their own account by somebody else's
  stale invite.
 Applying an invite uses the admin credential deliberately: the inviter
 authorised it when they issued it, and their session is long gone by the time
 it is redeemed. Everything else — sharing, unsharing, listing — runs on the
 caller's own token, which is why granting needs no elevated rights at all
 (verified against the live forge: a repo owner's scoped token adds and removes
 collaborators, HTTP 204).
 ## Superseded design note — invites and per-repo access
 Today `/v1/link` creates an account and every folder becomes a private repo
 under it. What is missing is the multi-person case: an existing account
 inviting somebody, and that person's device waking up with access to *some*
 repos and not others.
 Shape this should take, so the next session does not re-litigate it:
 * **Where grants live: granthi-link's own store, not Zitadel orgs.** The
  estate's house pattern is app-side tenancy tables with the IdP only
  providing identity (see the AROS `tenants` / `tenant_members` split).
  Grants therefore sit beside `state.json`, and **Gitea is the enforcement
  point** — a grant is materialised as a repo collaborator or an org team
  membership, so the forge itself refuses unauthorised reads. Nothing in the
  client decides access, which is why `get --all` needs no permission logic.
 * **Token scope does not change.** `write:repository,write:user` stays; per
  repo permission is collaborator/team state, not a token property.
 * **`POST /v1/invite`** (account admin → new member): creates the shre-id
  user (Zitadel admin PAT, on aros-vps at
  `/opt/shre-id/deploy/secrets/shre_id_zitadel_pat`), records the intended
  grants, and returns an invite the person redeems by running `link`. Until
  they redeem it, nothing exists on the forge.
 * **`POST /v1/grants`** (account admin): add/remove repo access for a member;
  applies the change to Gitea and records it. Removing a grant must also
  remove the collaborator — a grant store that drifts from the forge is
  worse than no store.
 * Both endpoints are account-admin-only and rate-limited like `/v1/link`.
 * The grant store inherits the same fragility already noted for the rate
  limiter: a flat JSON file behind an in-process lock, fine for one
  `ThreadingHTTPServer` and **not** fine the day this runs multi-process.
 ## Promotion window (beta → prod)
-1. **Expose :3042** behind cloudflared (granthi.shre.ai vhost or
+1. ~~**Expose :3042** behind cloudflared~~ **DONE 2026-08-23** —
-   link.granthi.shre.ai) — today it is tailnet-only by design.
+   `https://granthi-link.shre.ai` (second-level, see above), origin stays
-2. **Swap forge base URLs** in `/opt/granthi-link/config.json`:
+   tailnet-only, `trust_forwarded_for` on with the tunnel as the only
-   `gitea_base` → prod forge, `public_gitea_base` →
+   trusted proxy.
-   `https://granthi.shre.ai`; the client default server URL moves to the
+2. ~~**Swap forge base URLs** in `/opt/granthi-link/config.json`~~
-   public endpoint.
+   **DONE — verified live 2026-08-30**: the deployed config reads
   `gitea_base: http://127.0.0.1:3040` and
   `public_gitea_base: https://granthi.shre.ai`, and the client default
   server is already the public endpoint. Linking a new device therefore
   creates the account on **prod**.
 3. The `granthi-web` OIDC app already lists the prod callback; the device
   app is host-independent. Rotate the beta admin token/password out of the
   config when pointing at prod (prod forge is READ-ONLY to this estate —
   promotion is an operator action, not an agent action).
-4. Add rate limiting / abuse controls before public exposure (one token mint
+4. ~~Add rate limiting / abuse controls before public exposure.~~ **DONE** —
-   per link call today).
+   see "Rate limiting" above (5 links/hour per client by default). When
   exposing behind cloudflared, set `trust_forwarded_for: true` in the same
   change, or every request will look like the tunnel and one abuser will
   throttle everybody.
 5. **Hardening checklist (must all hold before exposing):**
   - [ ] `config.json` is 0600 (or 0400) and owned by the service user —
         the service refuses to start otherwise; verify with
@@ -0,0 +1,163 @@
 # Setting up Granthi on a new computer
 Everything here is one command at a time. You need `git` and `python3`
 (3.9 or newer). There is nothing to install — no package, no build.
 ## Read this first: what you will see when you sign in
 Signing in does **not** hand you an account you already had somewhere else.
 Granthi looks up your shre-id identity, and:
 * if it has seen you before, you get the same forge account as last time;
 * if it has not, and the login it would use is free, it makes you a **new,
  empty account**.
 So the first time you sign in on a new computer, `granthi-sync list` may show
 **nothing**. That is not a failure. It means you are a new account and nobody
 has given you access to anything yet.
 Granthi refuses to hand you an existing account just because the name lines
 up. It will only join you to one if that account's email is the same as your
 verified sign-in email. That rule is what stops somebody else claiming your
 files, so it is worth the small surprise on day one.
 **Two ways to end up with your files, and it is worth choosing on purpose:**
 | | what happens | when to pick it |
 |---|---|---|
 | **Fresh account** (recommended) | You sign in, get a new empty account, and somebody shares the repos you need with you | Your everyday laptop. The token stored on the machine can only do what that account can do. |
 | **Join an existing account** | An operator sets that account's email to your verified sign-in email first; then signing in joins you to it and you see everything it owns straight away | You want one identity everywhere and accept that the laptop then carries whatever that account can do |
 The fresh account is recommended for a plain reason: the token `link` saves on
 the machine acts as that account. If you join an admin account, every computer
 you sync from is carrying admin.
 ## 1. Get the program
    git clone https://granthi.shre.ai/nirpa/granthi-sync.git
    cd granthi-sync
 No account is needed for this step.
 ## 2. Sign in
    ./bin/granthi-sync link
 It prints a web address and a short code. Open the address, type the code, and
 approve — you have five minutes. When it finishes, your account exists and a
 key is saved at `~/.granthi-sync/config.json`, readable only by you. You never
 see or type a forge password.
 If the code runs out, **nothing was created** — no account, no key, no
 half-finished state. Run `link` again.
 You do not need to be on any private network. The client talks to
 `https://granthi-link.shre.ai`.
 ## 3. See what you have
    ./bin/granthi-sync list
    ./bin/granthi-sync list work-          # only names containing "work-"
 This list comes from the forge, so it is exactly what you are allowed to see —
 not a guess made on this computer.
 If it is empty, see the note at the top: you are a new account. Ask whoever
 owns the repos to run `granthi-sync share <your-login> --repo <name>`, or to
 add you to the right organisation.
 ## 4. Bring folders down
    ./bin/granthi-sync get notes                  # by name
    ./bin/granthi-sync get Nirlabinc/notes        # when two teams both have a "notes"
    ./bin/granthi-sync get --all --into ~/work    # everything you are allowed to see
 If a plain name matches two repos, Granthi stops and shows you both rather
 than guessing which one you meant.
 ## 5. Or send a folder up
    ./bin/granthi-sync add ~/Documents/notes
 It becomes a private repo. Two things happen for your protection:
 * a starter `.gitignore` is written if the folder has none, covering `.env`,
  `*.key`, `*.pem`, `id_rsa` and similar — those will **not** be synced;
 * a folder holding more than 20,000 files or 512 MB is refused unless you add
  `--force`, so you cannot upload your whole disk by accident.
 ## 6. Keep it syncing
 Run it in a terminal:
    ./bin/granthi-sync watch
 Or leave it running in the background on a Mac:
    cp client/launchd/ai.granthi.sync.plist ~/Library/LaunchAgents/
 Open that copied file and replace `/PATH/TO/granthi-sync` with the real path
 to this folder. Then:
    launchctl bootstrap gui/$UID ~/Library/LaunchAgents/ai.granthi.sync.plist
 Its log is `/tmp/granthi-sync.log`.
 ### What syncing does to your folder
 It depends on what the folder was:
 * **A folder that was already a git project** — Granthi **never commits for
  you**. Your history stays yours. Work you have not committed is copied to
  the cloud with a timestamp so it is not only on this laptop.
 * **A plain folder** — Granthi saves changes for you as it goes, and each save
  is a point you can go back to.
 Either way, if your copy and the cloud copy have both moved on, Granthi
 **stops and tells you**. It will never merge or overwrite your work to make
 the conflict go away.
 ## 7. Going back to an earlier version
    ./bin/granthi-sync snapshots ~/Documents/notes
    ./bin/granthi-sync restore  ~/Documents/notes --at 20260823T142530Z
 Restore writes to a **new folder**. It never overwrites what you have now —
 if you are restoring, you have enough problems already.
 You can restore from a computer that never had the folder, including one
 replacing a laptop that is gone.
 ## 8. Sharing
    ./bin/granthi-sync share bob --repo notes                 # they have an account
    ./bin/granthi-sync share bob --repo notes --revoke
    ./bin/granthi-sync shared --repo notes                    # who can see it
    ./bin/granthi-sync invite carol@example.com --repo notes  # they do not yet
 An invite creates nothing until it is accepted. If the person never signs in,
 nothing was ever made for them.
 ## 9. Keeping track of your computers
    ./bin/granthi-sync devices     # every computer signed in to your account
    ./bin/granthi-sync logout      # sign this computer out
    ./bin/granthi-sync logout --device <id>   # sign out a lost one
    ./bin/granthi-sync activity    # sign-ins and sign-outs, with times
 Signing a computer out takes effect **at the server**, immediately. The key on
 that machine stops working on its very next attempt — it is not a setting that
 machine could ignore. That is what makes it useful for a laptop you have lost.
 Your folders are left alone when you sign out; only the key is removed.
 ## If something looks wrong
 * **`list` is empty** — you are a new account, nobody has shared anything yet.
 * **"Repository not found"** — you may not have access to that repo; run
  `list` to see what you do have.
 * **"this device's access has been revoked"** — someone signed this computer
  out. Run `link` again.
 * **A folder shows DIVERGED** — your copy and the cloud copy both changed.
  Granthi is waiting for you on purpose. Nothing is lost; sort it out in git
  or on the forge web page.
@@ -1,10 +1,17 @@
 {
-  "gitea_base": "http://127.0.0.1:3041",
+  "_comment_forge": "Values below mirror the LIVE prod deployment (verified 2026-08-30). For the beta tier use gitea_base http://127.0.0.1:3041, public_gitea_base https://granthi-beta.shre.ai, container gitea-beta-gitea-1, creds /opt/gitea-beta/.admin-creds.",
-  "public_gitea_base": "http://100.111.127.127:3041",
+  "gitea_base": "http://127.0.0.1:3040",
  "public_gitea_base": "https://granthi.shre.ai",
  "zitadel_userinfo": "https://id.shre.ai/oidc/v1/userinfo",
-  "admin_token": "MINT-VIA: docker exec -u git gitea-beta-gitea-1 gitea admin user generate-access-token --username nirpa --scopes write:admin,write:user,write:repository --raw",
+  "admin_token": "MINT-VIA: docker exec -u git gitea-central-gitea-1 gitea admin user generate-access-token --username nirpa --scopes write:admin,write:user,write:repository --raw",
  "admin_login": "nirpa",
-  "admin_password": "FROM /opt/gitea-beta/.admin-creds (required: Gitea 1.27 token minting only works via basic auth + Sudo header)",
+  "admin_password": "FROM shre-cred: superadmin/granthi-link/granthi-prod-forge-admin (required: Gitea 1.27 token minting only works via basic auth + Sudo header). NEVER paste this into chat or a shell history.",
  "binds": [["127.0.0.1", 3042], ["100.111.127.127", 3042]],
-  "test_mode": false
+  "test_mode": false,
  "rate_limit": {
    "enabled": true,
    "trust_forwarded_for": true,
    "trusted_proxies": ["100.107.37.98/32"],
    "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
  }
 }
@@ -52,6 +52,7 @@ Stdlib only. Python 3.9+.
 """
 import base64
 import ipaddress
 import json
 import logging
 import os
@@ -61,7 +62,9 @@ import signal
 import string
 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
@@ -71,14 +74,166 @@ 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"
 # Sentinel: create_user hit a 409 (someone else created the login first).
 USER_CREATE_CONFLICT = object()
 # Sentinel: the address already belongs to another forge account, which is a
 # 409 the caller can act on -- not a 502 that reads like the service is down.
 EMAIL_IN_USE = object()
 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),
                      # Reads are cheap but still authenticated work.
                      "/v1/devices": (120, 3600),
                      "/v1/audit": (120, 3600),
                      # Sharing is an ordinary act; inviting reaches a person
                      # who does not exist yet, so it is the tighter of the two.
                      "/v1/grants": (120, 3600),
                      "/v1/invite": (60, 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
 # --------------------------------------------------------------------------
 # Rate limiting
 # --------------------------------------------------------------------------
 class RateLimiter:
    """Sliding-window limiter keyed by (route, client).
    In-process and lock-guarded: granthi-link is ONE ThreadingHTTPServer
    process, so a dict is the entire store -- no redis, no shared cache, and
    nothing to keep consistent across nodes. If this ever runs multi-process
    the limiter must move with it; that is why the store is behind this class
    rather than sprinkled through the handler.
    """
    def __init__(self, rules=None, max_keys=MAX_RATE_KEYS, clock=None):
        self.rules = dict(rules or DEFAULT_RATE_RULES)
        # Per-ROUTE budget, not one global table. A shared cap lets a flood of
        # cheap /v1/repos keys exhaust the table and lock brand-new /v1/link
        # clients out -- turning the fail-closed capacity guard into a
        # cross-route denial of service. Each route gets its own space.
        self.max_keys = max_keys
        self._clock = clock or time.monotonic
        self._hits = {route: {} for route in self.rules}
        self._lock = threading.Lock()
    def check(self, route, client):
        """-> (allowed: bool, retry_after: int). Records the hit when allowed.
        Denied requests are NOT recorded: a client that keeps hammering must
        not push its own window forward and lock itself out indefinitely.
        """
        rule = self.rules.get(route)
        if not rule:
            return True, 0
        limit, window = rule
        if limit <= 0:                      # 0 = endpoint disabled entirely
            return False, window
        with self._lock:
            # Clock read INSIDE the lock: taken outside, two racing threads
            # can append out of order, and both hits[0] (retry_after) and
            # v[-1] (reclamation age) assume the list is chronological.
            now = self._clock()
            table = self._hits.setdefault(route, {})
            hits = [t for t in table.get(client, ()) if now - t < window]
            if len(hits) >= limit:
                table[client] = hits
                return False, max(1, int(window - (now - hits[0])) + 1)
            if client not in table and len(table) >= self.max_keys:
                # At capacity, reclaim expired keys first...
                self._reclaim_expired(route, now)
                if len(table) >= self.max_keys:
                    # ...and if every window is still live, this is an
                    # identity flood, not organic load. Evicting here would
                    # let an attacker reset their own limit on demand, so
                    # refuse the NEW key instead. Fail closed: /v1/link is
                    # invite-only and low-volume, so hitting this cap is an
                    # attack, and turning tokens away beats minting them.
                    LOG.error("rate limiter at capacity for %s (%d keys) with "
                              "no expired windows: refusing new client %r",
                              route, self.max_keys, client)
                    return False, window
            hits.append(now)
            table[client] = hits
            return True, 0
    def _reclaim_expired(self, route, now):
        """Caller holds the lock. Drop only keys whose window has fully
        expired -- never a live one, or reclamation becomes the bypass."""
        window = self.rules[route][1]
        table = self._hits.get(route, {})
        dead = [c for c, v in table.items() if not v or now - v[-1] >= window]
        for c in dead:
            del table[c]
        if dead:
            LOG.info("rate limiter reclaimed %d expired windows on %s",
                     len(dead), route)
 def client_ip(handler, trust_forwarded_for, trusted_proxies=()):
    """The address to rate-limit on.
    Behind cloudflared every request arrives from the tunnel, so limiting on
    the socket peer would let one abuser starve everyone. X-Forwarded-For is
    client-controlled, though: a caller can prepend anything. Two guards:
    1. The header is honored ONLY when the socket peer is itself a configured
       trusted proxy. Without that, anyone who can reach the origin directly
       -- and the origin also listens on the tailnet -- picks their own
       rate-limit key and rotates it at will.
    2. A trusted proxy APPENDS the peer it actually saw, so the LAST entry is
       the only one the client did not choose. Take that, never the first.
    The value must parse as a real IP; junk falls back to the socket peer
    rather than becoming a key of its own.
    """
    peer = handler.client_address[0]
    if not trust_forwarded_for:
        return peer
    if not _ip_in_any(peer, trusted_proxies):
        LOG.warning("X-Forwarded-For ignored: peer %s is not a trusted proxy",
                    peer)
        return peer
    xff = handler.headers.get("X-Forwarded-For", "")
    parts = [p.strip() for p in xff.split(",") if p.strip()]
    if not parts:
        return peer
    try:
        return str(ipaddress.ip_address(parts[-1]))
    except ValueError:
        LOG.warning("X-Forwarded-For last hop %r is not an IP; using peer",
                    parts[-1][:60])
        return peer
 def _ip_in_any(addr, networks):
    try:
        ip = ipaddress.ip_address(addr)
    except ValueError:
        return False
    for net in networks:
        # Pre-parsed at startup by LinkService._parse_trusted_proxies, so a
        # malformed entry can never reach here as a silent per-request skip.
        if ip.version == net.version and ip in net:
            return True
    return False
 # --------------------------------------------------------------------------
 # HTTP helper (patchable in tests)
@@ -140,6 +295,14 @@ def check_config_perms(path, euid=None):
 # Identity map: zitadel sub -> gitea login (JSON, 0600, atomic writes)
 # --------------------------------------------------------------------------
 def _email_in_use_message(login, userinfo):
    email = userinfo.get("email") or "your address"
    return (f"cannot create the account '{login}': {email} already belongs to "
            f"a different account on this forge. If that other account is "
            f"yours, an operator must bind your identity to it; if it is not, "
            f"use a different address.")
 class IdentityStore:
    """Persistent map of Zitadel `sub` -> Gitea login binding records.
@@ -187,6 +350,168 @@ 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, {}
    # -- pending invites ---------------------------------------------------
    # Keyed by VERIFIED email. An invite is a promise of access made before the
    # person has a forge account; it is applied the first time they link. The
    # key must be an identity the IdP vouches for, or anyone could claim
    # someone else's invite by asserting their address.
    def add_invite(self, email, grants, invited_by):
        data = self._load()
        book = data.setdefault("invites", {})
        entry = book.setdefault(email.strip().lower(), [])
        for g in grants:
            # Re-inviting the same repo updates the permission rather than
            # stacking duplicates that would be applied twice.
            entry[:] = [e for e in entry if e["repo"] != g["repo"]]
            entry.append({"repo": g["repo"], "permission": g["permission"],
                          "invited_by": invited_by,
                          "invited_at": datetime.now(timezone.utc).isoformat(
                              timespec="seconds")})
        self._write(data)
    def take_invites(self, email):
        """Read AND clear the invites for an email, atomically under the
        caller's lock. Returns [] when there are none."""
        if not email:
            return []
        data = self._load()
        book = data.get("invites") or {}
        pending = book.pop(email.strip().lower(), [])
        if pending:
            self._write(data)
        return pending
    def consume_invite(self, email, repo):
        """Drop ONE applied grant, leaving any that failed still pending.
        The whole-list `take_invites` is what made a transient forge error
        permanent; this removes only what actually landed.
        """
        data = self._load()
        book = data.get("invites") or {}
        key = (email or "").strip().lower()
        entry = [e for e in book.get(key, []) if e.get("repo") != repo]
        if entry:
            book[key] = entry
        else:
            book.pop(key, None)
        data["invites"] = book
        self._write(data)
    def peek_invites(self, email):
        book = self._load().get("invites") or {}
        return list(book.get((email or "").strip().lower(), []))
    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)
@@ -202,6 +527,93 @@ 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
        # to be a deliberate config act rather than an omission.
        rl = config.get("rate_limit", {})
        if rl is None:
            rl = {}
        if not isinstance(rl, dict):
            raise SystemExit("config rate_limit must be an object")
        # Real JSON booleans only. `"enabled": null` or `0` must not quietly
        # turn limiting off, and the string "false" must not turn XFF trust
        # ON (every non-empty string is truthy).
        for flag, default in (("enabled", True), ("trust_forwarded_for", False)):
            if flag in rl and not isinstance(rl[flag], bool):
                raise SystemExit(
                    f"config rate_limit.{flag} must be true or false, "
                    f"got {rl[flag]!r}")
        self.trust_forwarded_for = rl.get("trust_forwarded_for", False)
        self.trusted_proxies = self._parse_trusted_proxies(
            rl.get("trusted_proxies"))
        if self.trust_forwarded_for and not self.trusted_proxies:
            raise SystemExit(
                "config rate_limit.trust_forwarded_for requires a non-empty "
                "trusted_proxies list -- trusting the header from any peer "
                "lets callers choose their own rate-limit key")
        if rl.get("enabled", True):
            rules = dict(DEFAULT_RATE_RULES)
            for route, spec in (rl.get("rules") or {}).items():
                # `type(x) is int`, NOT isinstance: bool subclasses int, so
                # isinstance lets [5, true] through as a 1-SECOND window --
                # an hourly limit silently becomes ~5/sec -- and false in the
                # limit slot disables the endpoint.
                if (not isinstance(spec, (list, tuple)) or len(spec) != 2
                        or not all(type(x) is int for x in spec)
                        or spec[1] <= 0):
                    # A malformed rule must not silently mean "unlimited".
                    raise SystemExit(
                        f"config rate_limit.rules[{route!r}] must be "
                        f"[max_requests, window_seconds] with window > 0")
                rules[route] = (spec[0], spec[1])
            self.limiter = RateLimiter(rules)
            LOG.info("rate limiting active: %s (trust_forwarded_for=%s, "
                     "trusted_proxies=%s)",
                     {k: f"{v[0]}/{v[1]}s" for k, v in rules.items()},
                     self.trust_forwarded_for, list(self.trusted_proxies))
        else:
            self.limiter = None
            LOG.warning("rate limiting DISABLED by config -- /v1/link will "
                        "mint tokens without any throttle")
    @staticmethod
    def _parse_trusted_proxies(raw):
        """Validate at STARTUP, not per request.
        This setting gates a spoofable identity, so every failure mode has to
        be loud and early: a bare string would be iterated character by
        character (each char a "network"), malformed entries would only
        surface as a per-request log line, and a wildcard like 0.0.0.0/0 or
        ::/0 quietly restores "trust X-Forwarded-For from anyone" -- the exact
        hole trusted_proxies exists to close.
        """
        if raw is None:
            return ()
        if isinstance(raw, str) or not isinstance(raw, (list, tuple)):
            raise SystemExit(
                "config rate_limit.trusted_proxies must be a list of CIDRs, "
                f"got {type(raw).__name__}")
        nets = []
        for entry in raw:
            if not isinstance(entry, str):
                raise SystemExit(
                    f"config rate_limit.trusted_proxies entry {entry!r} "
                    "must be a string")
            try:
                net = ipaddress.ip_network(entry, strict=False)
            except ValueError as e:
                raise SystemExit(
                    f"config rate_limit.trusted_proxies entry {entry!r} "
                    f"is not a valid network: {e}")
            if net.prefixlen == 0:
                raise SystemExit(
                    f"config rate_limit.trusted_proxies entry {entry!r} "
                    "matches every address, which is the same as trusting "
                    "X-Forwarded-For from any peer -- refusing")
            nets.append(net)
        return tuple(nets)
    def test_mode_enabled(self):
        """Config test_mode is honored ONLY with the env gate also set."""
@@ -276,16 +688,32 @@ class LinkService:
            headers=self._admin_hdr(), body=body)
        if status == 409:
            return USER_CREATE_CONFLICT
        if status == 422 and "e-mail already in use" in str(resp).lower():
            # The address belongs to a DIFFERENT forge account. Reported as its
            # own case because the generic path turns it into a bare 502, and
            # "502" sends the person looking for an outage when the real answer
            # is "that address is already somebody's account here". Hit live on
            # 2026-08-23: an operator moved an email onto another account and
            # the next sign-in failed with nothing but the number.
            return EMAIL_IN_USE
        if status != 201:
            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(
+        suffix = LOGIN_SAFE.sub("-", str(device_id or ""))[:12]
-            safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
+        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={
@@ -324,6 +752,8 @@ class LinkService:
                                 "was not created by this service; refusing "
                                 "to re-create"}, None
                err = self.create_user(login, userinfo)
                if err is EMAIL_IN_USE:
                    return 409, {"error": _email_in_use_message(login, userinfo)}, None
                if err and err is not USER_CREATE_CONFLICT:
                    return 502, {"error": err}, None
                LOG.info("re-created service-managed gitea user %s", login)
@@ -349,6 +779,8 @@ class LinkService:
                    return 409, {"error": "login exists and is not linked "
                                          "to this identity"}, None
                LOG.info("user %s created concurrently; continuing", login)
            elif err is EMAIL_IN_USE:
                return 409, {"error": _email_in_use_message(login, userinfo)}, None
            elif err:
                return 502, {"error": err}, None
            else:
@@ -375,7 +807,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)")
@@ -398,14 +830,300 @@ 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)
-        return 200, {"gitea_base": self.public_gitea, "login": login,
+        # A device id the client generated once and keeps. Clients older than
-                     "token": gitea_token, "token_name": token_name}
+        # 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"),
            })
        granted = self.apply_invites(login, userinfo, client_ip)
        self.audit.write("device.link", login=login, device_id=device_id,
                         device_name=device_name, token_name=token_name,
                         client_ip=client_ip)
        resp = {"gitea_base": self.public_gitea, "login": login,
                "token": gitea_token, "token_name": token_name,
                "device_id": device_id}
        if granted:
            resp["granted_repos"] = granted
        return 200, resp
-    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}"
    # -- sharing -----------------------------------------------------------
    PERMISSIONS = ("read", "write", "admin")
    def _repo_admin_check(self, token, full_name):
        """Can this caller administer that repo? Ask the FORGE.
        Listing collaborators requires repo admin, so a 200 here is Gitea's own
        answer to 'may you share this?'. Deciding it ourselves from the repo
        name would be a second opinion about someone else's authorisation --
        and the wrong one the first time a repo is transferred.
        """
        status, _ = http_json(
            "GET", f"{self.gitea}/api/v1/repos/{full_name}/collaborators",
            headers={"Authorization": f"token {token}"})
        return status == 200
    def grants(self, body, client_ip=None):
        """Share a repo with someone who already has a forge account."""
        login = self.whoami(body.get("token"))
        if not login:
            return 401, {"error": "invalid or revoked token"}
        token = body.get("token")
        action = (body.get("action") or "add").lower()
        if action not in ("add", "remove", "list"):
            return 400, {"error": "action must be add, remove or list"}
        repo = str(body.get("repo") or "").strip()
        if not repo:
            return 400, {"error": "repo required"}
        full = repo if "/" in repo else f"{login}/{repo}"
        if not self._repo_admin_check(token, full):
            self.audit.write("grant.denied", login=login, repo=full,
                             client_ip=client_ip,
                             reason="caller cannot administer this repo")
            return 403, {"error": f"you cannot administer {full}"}
        if action == "list":
            status, resp = http_json(
                "GET", f"{self.gitea}/api/v1/repos/{full}/collaborators",
                headers={"Authorization": f"token {token}"})
            people = [u.get("login") for u in resp] if isinstance(resp, list) else []
            return 200, {"repo": full, "collaborators": people}
        who = str(body.get("login") or "").strip()
        if not who:
            return 400, {"error": "login required"}
        permission = (body.get("permission") or "write").lower()
        if permission not in self.PERMISSIONS:
            return 400, {"error": f"permission must be one of "
                                  f"{', '.join(self.PERMISSIONS)}"}
        if action == "add":
            status, resp = http_json(
                "PUT",
                f"{self.gitea}/api/v1/repos/{full}/collaborators/"
                f"{urllib.parse.quote(who, safe='')}",
                headers={"Authorization": f"token {token}"},
                body={"permission": permission})
            ok = status in (200, 204)
        else:
            status, resp = http_json(
                "DELETE",
                f"{self.gitea}/api/v1/repos/{full}/collaborators/"
                f"{urllib.parse.quote(who, safe='')}",
                headers={"Authorization": f"token {token}"})
            ok = status in (200, 204, 404)  # already gone is the wanted state
        if not ok:
            self.audit.write("grant.failed", login=login, repo=full,
                             grantee=who, action=action, client_ip=client_ip,
                             reason=f"HTTP {status}")
            return 502, {"error": f"forge refused (HTTP {status}): {resp}"}
        self.audit.write(f"grant.{action}", login=login, repo=full,
                         grantee=who, permission=permission,
                         client_ip=client_ip)
        return 200, {"repo": full, "login": who, "action": action,
                     "permission": permission}
    def invite(self, body, client_ip=None):
        """Promise access to someone who has no forge account yet.
        Nothing is created for them here -- no account, no token. The grant is
        recorded against their VERIFIED email and applied the first time they
        link. If they never link, nothing ever existed.
        """
        login = self.whoami(body.get("token"))
        if not login:
            return 401, {"error": "invalid or revoked token"}
        email = str(body.get("email") or "").strip().lower()
        if "@" not in email:
            return 400, {"error": "a valid email is required"}
        repos = body.get("repos")
        if not isinstance(repos, list) or not repos:
            return 400, {"error": "repos must be a non-empty list"}
        wanted = []
        for entry in repos:
            if isinstance(entry, str):
                entry = {"name": entry}
            if not isinstance(entry, dict) or not entry.get("name"):
                return 400, {"error": "each repo needs a name"}
            name = str(entry["name"]).strip()
            full = name if "/" in name else f"{login}/{name}"
            permission = (entry.get("permission") or "write").lower()
            if permission not in self.PERMISSIONS:
                return 400, {"error": f"permission must be one of "
                                      f"{', '.join(self.PERMISSIONS)}"}
            if not self._repo_admin_check(body.get("token"), full):
                self.audit.write("invite.denied", login=login, repo=full,
                                 invitee=email, client_ip=client_ip,
                                 reason="caller cannot administer this repo")
                return 403, {"error": f"you cannot administer {full}"}
            wanted.append({"repo": full, "permission": permission})
        with self.state.lock:
            self.state.add_invite(email, wanted, login)
        self.audit.write("invite", login=login, invitee=email,
                         repos=[w["repo"] for w in wanted], client_ip=client_ip)
        return 200, {"invited": email,
                     "repos": wanted,
                     "note": "applied the first time they sign in with a "
                             "verified email that matches"}
    def apply_invites(self, login, userinfo, client_ip=None):
        """Turn recorded invites into real collaborator rows at link time.
        Uses the ADMIN credential deliberately: the inviter authorised this
        when they issued the invite, and their session is long gone by now.
        Nothing here can fail the link -- someone signing in must not be
        blocked because a repo they were promised has since been deleted.
        """
        email = (userinfo.get("email") or "").strip().lower()
        if not email or not userinfo.get("email_verified"):
            # Unverified email must never collect an invite: the address is
            # the only thing tying the promise to this person.
            return []
        with self.state.lock:
            pending = self.state.peek_invites(email)
        applied = []
        # PEEK, not take. Consuming the invite first means a transient 502 from
        # the forge destroys it: the person links successfully, gets no access,
        # and re-linking never retries because the promise is gone. Only the
        # grants that actually landed are removed, so a failure is retried on
        # the next link instead of being silently lost. (codex review, P2.)
        for grant in pending:
            status, resp = http_json(
                "PUT",
                f"{self.gitea}/api/v1/repos/{grant['repo']}/collaborators/"
                f"{urllib.parse.quote(login, safe='')}",
                headers={"Authorization": _basic(self.cfg["admin_login"],
                                                 self.cfg["admin_password"])},
                body={"permission": grant.get("permission", "write")})
            if status in (200, 204):
                applied.append(grant["repo"])
                with self.state.lock:
                    self.state.consume_invite(email, grant["repo"])
                self.audit.write("invite.applied", login=login,
                                 repo=grant["repo"],
                                 permission=grant.get("permission"),
                                 client_ip=client_ip)
            else:
                self.audit.write("invite.apply.failed", login=login,
                                 repo=grant["repo"], client_ip=client_ip,
                                 reason=f"HTTP {status}: {resp}")
                LOG.warning("invite for %s on %s could not be applied "
                            "(HTTP %s)", login, grant["repo"], status)
        return applied
    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:
@@ -452,6 +1170,29 @@ class Handler(BaseHTTPRequestHandler):
            self._send(404, {"error": "not found"})
    def do_POST(self):
        # Rate-limit BEFORE reading the body or doing any work -- the point is
        # to spend nothing on an abusive caller. Same close-the-connection
        # treatment the 413 path uses, for the same reason: we are not going
        # to drain a body we already decided to reject.
        limiter = getattr(self.service, "limiter", None)
        if limiter is not None:
            who = client_ip(self, self.service.trust_forwarded_for,
                            self.service.trusted_proxies)
            allowed, retry_after = limiter.check(self.path, who)
            if not allowed:
                LOG.warning("rate limited %s %s (retry after %ss)",
                            who, self.path, retry_after)
                self.close_connection = True
                payload = json.dumps({
                    "error": "rate limit exceeded",
                    "retry_after": retry_after}).encode()
                self.send_response(429)
                self.send_header("Content-Type", "application/json")
                self.send_header("Retry-After", str(retry_after))
                self.send_header("Content-Length", str(len(payload)))
                self.end_headers()
                self.wfile.write(payload)
                return
        cl = self.headers.get("Content-Length")
        if cl is None:
            self.close_connection = True
@@ -475,10 +1216,25 @@ 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/grants":
            status, resp = self.service.grants(body, peer)
        elif self.path == "/v1/invite":
            status, resp = self.service.invite(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)
@@ -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,21 @@ 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.endswith("/collaborators") and self.path.startswith("/api/v1/repos/"):
            full = self.path.split("/api/v1/repos/")[1].rsplit("/collaborators", 1)[0]
            tok = self.headers.get("Authorization", "").replace("token ", "")
            caller = st["tokens_by_sha"].get(tok)
            if st["repo_owner"].get(full) != caller:
                return self._json(403, {"message": "must have admin rights"})
            return self._json(200, [{"login": w}
                                    for w in st["collab"].get(full, {})])
        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"]:
@@ -61,6 +77,10 @@ class StubUpstream(BaseHTTPRequestHandler):
        if self.path == "/api/v1/admin/users":
            if body["username"] in st["users"]:
                return self._json(409, {"message": "user already exists"})
            if body.get("email") in st["users"].values():
                # real Gitea: emails are unique across accounts
                return self._json(422, {"message":
                    f"e-mail already in use [email: {body['email']}]"})
            st["users"][body["username"]] = body["email"]
            st["created"].append(body)
            return self._json(201, {"login": body["username"]})
@@ -71,7 +91,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 +108,54 @@ class StubUpstream(BaseHTTPRequestHandler):
                                    f"alice/{body['name']}"})
        self._json(404, {})
    def do_PUT(self):
        """Gitea adds a collaborator with PUT, and the admin path uses basic
        auth because the inviter's session is long gone by then."""
        st = self.state
        length = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(length) or b"{}")
        if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"):
            full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0]
            who = self.path.rsplit("/", 1)[1]
            if full not in st["repo_owner"]:
                return self._json(404, {"message": "no such repo"})
            tok = self.headers.get("Authorization", "")
            caller = st["tokens_by_sha"].get(tok.replace("token ", ""))
            if not tok.startswith("Basic ") and st["repo_owner"].get(full) != caller:
                return self._json(403, {"message": "forbidden"})
            st["collab"].setdefault(full, {})[who] = body.get("permission", "write")
            return self._json(204, {})
        self._json(404, {})
    def do_DELETE(self):
        st = self.state
        if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"):
            full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0]
            who = self.path.rsplit("/", 1)[1]
            tok = self.headers.get("Authorization", "").replace("token ", "")
            caller = st["tokens_by_sha"].get(tok)
            if st["repo_owner"].get(full) != caller:
                return self._json(403, {"message": "forbidden"})
            st["collab"].get(full, {}).pop(who, None)
            return self._json(204, {})
        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 +163,10 @@ 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,
                              "repo_owner": {}, "collab": {}}
        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 +180,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 +190,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 +229,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,5 +564,843 @@ 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")
 class TestSharing(ServiceTestBase):
    """Give a person access to one repo -- and to nothing else."""
    def setUp(self):
        super().setUp()
        self.enable_test_mode()
        self.alice = self.stub_link("s1", "alice", email="[email protected]",
                                    verified=True, device_id="dev-a")[1]
        self.bob = self.stub_link("s2", "bob", email="[email protected]",
                                  verified=True, device_id="dev-b")[1]
        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
        StubUpstream.state["repo_owner"]["bob/private"] = "bob"
    def test_owner_can_share_and_unshare(self):
        status, resp = self.svc.grants({"token": self.alice["token"],
                                        "repo": "notes", "login": "bob"})
        self.assertEqual(status, 200, resp)
        self.assertIn("bob", StubUpstream.state["collab"]["alice/notes"])
        status, _ = self.svc.grants({"token": self.alice["token"],
                                     "action": "remove",
                                     "repo": "notes", "login": "bob"})
        self.assertEqual(status, 200)
        self.assertNotIn("bob", StubUpstream.state["collab"]["alice/notes"])
    def test_a_non_owner_cannot_share_someone_elses_repo(self):
        """The forge decides who may share, not this service."""
        status, resp = self.svc.grants({"token": self.bob["token"],
                                        "repo": "alice/notes",
                                        "login": "bob"})
        self.assertEqual(status, 403)
        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
    def test_permission_is_validated(self):
        status, _ = self.svc.grants({"token": self.alice["token"],
                                     "repo": "notes", "login": "bob",
                                     "permission": "owner"})
        self.assertEqual(status, 400)
    def test_listing_shows_who_has_access(self):
        self.svc.grants({"token": self.alice["token"], "repo": "notes",
                         "login": "bob"})
        status, resp = self.svc.grants({"token": self.alice["token"],
                                        "action": "list", "repo": "notes"})
        self.assertEqual((status, resp["collaborators"]), (200, ["bob"]))
    def test_removing_someone_who_never_had_access_is_not_an_error(self):
        status, _ = self.svc.grants({"token": self.alice["token"],
                                     "action": "remove", "repo": "notes",
                                     "login": "nobody"})
        self.assertEqual(status, 200)
 class TestInvites(ServiceTestBase):
    """A promise of access for someone with no account yet."""
    def setUp(self):
        super().setUp()
        self.enable_test_mode()
        self.alice = self.stub_link("s1", "alice", email="[email protected]",
                                    verified=True, device_id="dev-a")[1]
        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
    def invite(self, email, repos=("notes",), token=None):
        return self.svc.invite({"token": token or self.alice["token"],
                                "email": email, "repos": list(repos)})
    def test_invite_creates_nothing_until_they_sign_in(self):
        status, resp = self.invite("[email protected]")
        self.assertEqual(status, 200, resp)
        # no account, no collaborator row yet
        self.assertNotIn("carol", StubUpstream.state["users"])
        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
    def test_access_lands_on_first_link_with_a_verified_email(self):
        self.invite("[email protected]")
        status, resp = self.stub_link("s9", "carol", email="[email protected]",
                                      verified=True, device_id="dev-c")
        self.assertEqual(status, 200, resp)
        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
        self.assertIn("carol", StubUpstream.state["collab"]["alice/notes"])
    def test_an_unverified_email_collects_nothing(self):
        """The address is the only thing tying the promise to this person."""
        self.invite("[email protected]")
        status, resp = self.stub_link("s10", "dave", email="[email protected]",
                                      verified=False, device_id="dev-d")
        self.assertEqual(status, 200)
        self.assertIsNone(resp.get("granted_repos"))
        self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {})
        # and the invite is still waiting, not consumed
        self.assertTrue(self.svc.state.peek_invites("[email protected]"))
    def test_an_invite_is_applied_once(self):
        self.invite("[email protected]")
        self.stub_link("s11", "erin", email="[email protected]", verified=True,
                       device_id="dev-e")
        status, resp = self.stub_link("s11", "erin", email="[email protected]",
                                      verified=True, device_id="dev-e2")
        self.assertIsNone(resp.get("granted_repos"))
    def test_cannot_invite_to_a_repo_you_do_not_administer(self):
        StubUpstream.state["repo_owner"]["bob/secret"] = "bob"
        status, _ = self.invite("[email protected]", repos=("bob/secret",))
        self.assertEqual(status, 403)
        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])
    def test_reinviting_updates_the_permission_instead_of_stacking(self):
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": [{"name": "notes", "permission": "read"}]})
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": [{"name": "notes", "permission": "write"}]})
        pending = self.svc.state.peek_invites("[email protected]")
        self.assertEqual(len(pending), 1)
        self.assertEqual(pending[0]["permission"], "write")
    def test_a_bad_email_is_refused(self):
        status, _ = self.svc.invite({"token": self.alice["token"],
                                     "email": "not-an-email",
                                     "repos": ["notes"]})
        self.assertEqual(status, 400)
    def test_a_failed_grant_does_not_block_the_person_signing_in(self):
        """Someone must not be locked out because a repo they were promised
        has since been deleted."""
        self.invite("[email protected]", repos=("notes",))
        StubUpstream.state["repo_owner"].pop("alice/notes")
        status, resp = self.stub_link("s12", "gina", email="[email protected]",
                                      verified=True, device_id="dev-g")
        self.assertEqual(status, 200)
        self.assertIn("token", resp)
    def test_invite_and_application_are_both_audited(self):
        self.invite("[email protected]")
        self.stub_link("s13", "hana", email="[email protected]", verified=True,
                       device_id="dev-h")
        _, mine = self.svc.audit_read({"token": self.alice["token"]})
        self.assertIn("invite", [e["event"] for e in mine["events"]])
 class TestInviteSurvivesAFailedGrant(ServiceTestBase):
    """[P2, codex] A transient forge error must not destroy the promise."""
    def setUp(self):
        super().setUp()
        self.enable_test_mode()
        self.alice = self.stub_link("s1", "alice", email="[email protected]",
                                    verified=True, device_id="dev-a")[1]
        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
        StubUpstream.state["repo_owner"]["alice/reports"] = "alice"
    def test_a_failed_grant_leaves_the_invite_pending_for_next_time(self):
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": ["notes"]})
        # the forge loses the repo mid-flight -> the PUT 404s
        StubUpstream.state["repo_owner"].pop("alice/notes")
        status, resp = self.stub_link("s9", "carol", email="[email protected]",
                                      verified=True, device_id="dev-c")
        self.assertEqual(status, 200)                 # sign-in still succeeds
        self.assertIsNone(resp.get("granted_repos"))
        # the promise is STILL THERE rather than silently consumed
        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
                         ["alice/notes"])
        # and it lands on the next link, once the repo is back
        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
        status, resp = self.stub_link("s9", "carol", email="[email protected]",
                                      verified=True, device_id="dev-c2")
        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])
    def test_a_partial_failure_only_consumes_what_landed(self):
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": ["notes", "reports"]})
        StubUpstream.state["repo_owner"].pop("alice/reports")   # one of two fails
        _, resp = self.stub_link("s10", "dan", email="[email protected]",
                                 verified=True, device_id="dev-d")
        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
                         ["alice/reports"])
 class TestDuplicateEmailIsExplained(ServiceTestBase):
    """A 502 sends someone looking for an outage. The real answer is that the
    address already belongs to another account here -- say so. (Hit live on
    2026-08-23 when an operator moved an email onto a different account.)"""
    def setUp(self):
        super().setUp()
        self.enable_test_mode()
        # an existing account already holds the address
        StubUpstream.state["users"]["existing"] = "[email protected]"
    def test_it_is_a_409_that_names_the_problem(self):
        status, resp = self.stub_link("s-new", "brandnew", email="[email protected]",
                                      verified=True, device_id="dev-x")
        self.assertEqual(status, 409, resp)
        msg = resp["error"]
        self.assertIn("[email protected]", msg)
        self.assertIn("already belongs to a different account", msg)
        self.assertIn("brandnew", msg)          # names the login it tried
        self.assertNotIn("502", msg)
    def test_a_normal_create_is_unaffected(self):
        status, resp = self.stub_link("s-ok", "fresh", email="[email protected]",
                                      verified=True, device_id="dev-y")
        self.assertEqual(status, 200, resp)
        self.assertIn("token", resp)
 if __name__ == "__main__":
    unittest.main()
 class TestRateLimiterUnit(unittest.TestCase):
    """The limiter in isolation, on a fake clock -- no sleeping in tests."""
    def setUp(self):
        self.now = 1000.0
        self.rl = granthi_link.RateLimiter(
            rules={"/v1/link": (3, 60)}, clock=lambda: self.now)
    def test_allows_up_to_the_limit_then_denies(self):
        for i in range(3):
            allowed, _ = self.rl.check("/v1/link", "1.1.1.1")
            self.assertTrue(allowed, f"request {i} should pass")
        allowed, retry = self.rl.check("/v1/link", "1.1.1.1")
        self.assertFalse(allowed)
        self.assertGreater(retry, 0)
        self.assertLessEqual(retry, 61)
    def test_window_slides(self):
        for _ in range(3):
            self.rl.check("/v1/link", "1.1.1.1")
        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
        self.now += 61
        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
    def test_denied_requests_do_not_extend_the_window(self):
        """A client that keeps hammering must not push its own window
        forward and lock itself out forever."""
        for _ in range(3):
            self.rl.check("/v1/link", "1.1.1.1")
        for _ in range(20):          # hammer while denied
            self.now += 1
            self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
        self.now = 1000.0 + 61       # just past the ORIGINAL window
        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
    def test_clients_are_isolated(self):
        for _ in range(3):
            self.rl.check("/v1/link", "1.1.1.1")
        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
        self.assertTrue(self.rl.check("/v1/link", "2.2.2.2")[0])
    def test_routes_are_isolated_and_unknown_routes_pass(self):
        for _ in range(3):
            self.rl.check("/v1/link", "1.1.1.1")
        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
        self.assertTrue(self.rl.check("/v1/repos", "1.1.1.1")[0])
        for _ in range(50):
            self.assertTrue(self.rl.check("/health", "1.1.1.1")[0])
    def test_zero_limit_disables_the_endpoint(self):
        rl = granthi_link.RateLimiter(rules={"/v1/link": (0, 60)},
                                      clock=lambda: self.now)
        self.assertFalse(rl.check("/v1/link", "1.1.1.1")[0])
    def test_key_store_stays_bounded(self):
        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
                                      max_keys=50, clock=lambda: self.now)
        for i in range(500):
            self.now += 0.001
            rl.check("/v1/link", f"10.0.{i // 256}.{i % 256}")
        self.assertLessEqual(len(rl._hits["/v1/link"]), 50 + 1)
    def test_expired_keys_are_reclaimed(self):
        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
                                      max_keys=10, clock=lambda: self.now)
        for i in range(10):
            rl.check("/v1/link", f"10.0.0.{i}")
        self.now += 120                      # everything expires
        for i in range(10, 25):
            rl.check("/v1/link", f"10.0.0.{i}")
        self.assertLessEqual(len(rl._hits["/v1/link"]), 11)
    def test_concurrent_checks_never_exceed_the_limit(self):
        """The lock has to actually hold under threads: 40 racing callers
        against a limit of 10 must yield exactly 10 allows."""
        rl = granthi_link.RateLimiter(rules={"/v1/link": (10, 60)})
        results, lock = [], threading.Lock()
        def hit():
            a, _ = rl.check("/v1/link", "9.9.9.9")
            with lock:
                results.append(a)
        ts = [threading.Thread(target=hit) for _ in range(40)]
        for t in ts:
            t.start()
        for t in ts:
            t.join()
        self.assertEqual(sum(results), 10)
 class _FakeHandler:
    def __init__(self, peer, xff=None):
        self.client_address = (peer, 12345)
        self.headers = {} if xff is None else {"X-Forwarded-For": xff}
        if xff is not None:
            self.headers = type("H", (), {"get": lambda s, k, d="":
                                          xff if k == "X-Forwarded-For" else d})()
 class TestClientIp(unittest.TestCase):
    # Pre-parsed, exactly as LinkService._parse_trusted_proxies hands them
    # over -- validation happens once at startup, never per request.
    PROXY = granthi_link.LinkService._parse_trusted_proxies(["10.0.0.0/8"])
    def test_socket_peer_by_default(self):
        h = _FakeHandler("5.5.5.5", xff="1.2.3.4")
        self.assertEqual(granthi_link.client_ip(h, False, self.PROXY), "5.5.5.5")
    def test_trusted_proxy_uses_the_last_hop_not_the_client_supplied_first(self):
        """A caller can PREPEND anything to X-Forwarded-For; a trusted proxy
        appends the peer it really saw. Only the last entry is trustworthy."""
        h = _FakeHandler("10.0.0.1", xff="1.2.3.4, 203.0.113.9")
        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
                         "203.0.113.9")
    def test_untrusted_peer_cannot_choose_its_own_key(self):
        """The origin also listens on the tailnet. Anyone reaching it
        directly must not be able to pick (and rotate) their rate-limit key
        just by sending a header."""
        h = _FakeHandler("203.0.113.50", xff="9.9.9.9")
        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
                         "203.0.113.50")
    def test_non_ip_last_hop_falls_back_to_peer(self):
        for junk in ("not-an-ip", "', OR 1=1", "x" * 500, "10.0.0.1:8080"):
            with self.subTest(junk=junk):
                h = _FakeHandler("10.0.0.1", xff=f"1.2.3.4, {junk}")
                self.assertEqual(
                    granthi_link.client_ip(h, True, self.PROXY), "10.0.0.1")
    def test_trusted_proxy_falls_back_when_header_absent(self):
        h = _FakeHandler("10.0.0.1")
        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
                         "10.0.0.1")
    def test_malformed_trusted_proxy_entry_is_rejected_at_startup(self):
        """It never reaches a request: a bad CIDR kills the process rather
        than degrading to a per-request log line."""
        with self.assertRaises(SystemExit):
            granthi_link.LinkService._parse_trusted_proxies(["not-a-cidr"])
    def test_v4_peer_does_not_match_a_v6_trusted_net(self):
        nets = granthi_link.LinkService._parse_trusted_proxies(["fd00::/8"])
        h = _FakeHandler("10.0.0.1", xff="9.9.9.9")
        self.assertEqual(granthi_link.client_ip(h, True, nets), "10.0.0.1")
 class TestRateLimitConfig(ServiceTestBase):
    def _svc(self, rl):
        cfg = dict(self.svc.cfg)
        cfg["rate_limit"] = rl
        return granthi_link.LinkService(cfg)
    def test_enabled_by_default_when_key_absent(self):
        self.assertIsNotNone(self.svc.limiter)
    def test_explicit_disable_is_honored(self):
        self.assertIsNone(self._svc({"enabled": False}).limiter)
    def test_custom_rule_overrides_default(self):
        svc = self._svc({"rules": {"/v1/link": [99, 120]}})
        self.assertEqual(svc.limiter.rules["/v1/link"], (99, 120))
    def test_malformed_rule_refuses_startup_rather_than_meaning_unlimited(self):
        for bad in ({"rules": {"/v1/link": [5]}},
                    {"rules": {"/v1/link": "5/hour"}},
                    {"rules": {"/v1/link": [5, 0]}},
                    {"rules": {"/v1/link": [5, -1]}},
                    {"rules": {"/v1/link": ["5", "60"]}}):
            with self.subTest(cfg=bad):
                with self.assertRaises(SystemExit):
                    self._svc(bad)
 class TestRateLimitOverHttp(HandlerTestBase):
    """Proven on the wire: real 429, real Retry-After."""
    def setUp(self):
        super().setUp()
        self.svc.limiter = granthi_link.RateLimiter(rules={"/v1/link": (2, 60)})
    def test_third_request_gets_429_with_retry_after(self):
        body = json.dumps({"zitadel_access_token": "x"}).encode()
        hdrs = {"Content-Length": str(len(body)),
                "Content-Type": "application/json"}
        for _ in range(2):
            st, _ = self.raw_post("/v1/link", body, hdrs)
            self.assertNotEqual(st, 429)
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        self.addCleanup(conn.close)
        conn.request("POST", "/v1/link", body=body, headers=hdrs)
        resp = conn.getresponse()
        self.assertEqual(resp.status, 429)
        self.assertTrue(resp.getheader("Retry-After"))
        self.assertIn("rate limit", json.loads(resp.read())["error"])
    def test_health_is_never_rate_limited(self):
        for _ in range(30):
            conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
            conn.request("GET", "/health")
            self.assertEqual(conn.getresponse().status, 200)
            conn.close()
 class TestRateLimitHardening(unittest.TestCase):
    """The four codex [P2] findings, each pinned by a test."""
    def test_capacity_fails_closed_instead_of_resetting_a_live_window(self):
        """Evicting a live window would let an attacker who can mint many
        distinct keys clear their OWN limit on demand. New keys are refused
        instead while every window is still live."""
        now = [1000.0]
        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 3600)},
                                      max_keys=5, clock=lambda: now[0])
        for i in range(5):
            self.assertTrue(rl.check("/v1/link", f"10.0.0.{i}")[0])
        victim_hits = list(rl._hits["/v1/link"]["10.0.0.0"])
        for i in range(100, 140):                 # identity flood
            allowed, retry = rl.check("/v1/link", f"10.0.1.{i}")
            self.assertFalse(allowed)
            self.assertGreater(retry, 0)
        # the earlier client's window survived the flood untouched
        self.assertEqual(rl._hits["/v1/link"]["10.0.0.0"], victim_hits)
        self.assertLessEqual(len(rl._hits["/v1/link"]), 5)
    def test_capacity_recovers_once_windows_expire(self):
        now = [1000.0]
        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 60)},
                                      max_keys=3, clock=lambda: now[0])
        for i in range(3):
            rl.check("/v1/link", f"10.0.0.{i}")
        self.assertFalse(rl.check("/v1/link", "10.0.9.9")[0])
        now[0] += 61
        self.assertTrue(rl.check("/v1/link", "10.0.9.9")[0])
    def test_hits_stay_chronological_under_thread_contention(self):
        """retry_after uses hits[0] and reclamation uses v[-1]; both assume
        the list is ordered. Reading the clock outside the lock let racing
        threads append out of order."""
        rl = granthi_link.RateLimiter(rules={"/v1/link": (500, 3600)})
        ts = [threading.Thread(target=lambda: rl.check("/v1/link", "7.7.7.7"))
              for _ in range(200)]
        for t in ts:
            t.start()
        for t in ts:
            t.join()
        hits = rl._hits["/v1/link"]["7.7.7.7"]
        self.assertEqual(len(hits), 200)
        self.assertEqual(hits, sorted(hits), "timestamps out of order")
 class TestRateLimitConfigTypes(ServiceTestBase):
    def _svc(self, rl):
        cfg = dict(self.svc.cfg)
        cfg["rate_limit"] = rl
        return granthi_link.LinkService(cfg)
    def test_non_boolean_enabled_refuses_startup(self):
        """`"enabled": null` or `0` must not quietly mean unlimited."""
        for bad in (None, 0, "", "false", "no", []):
            with self.subTest(enabled=bad):
                with self.assertRaises(SystemExit):
                    self._svc({"enabled": bad})
    def test_string_false_does_not_enable_forwarded_trust(self):
        """Every non-empty string is truthy -- "false" used to mean True."""
        with self.assertRaises(SystemExit):
            self._svc({"trust_forwarded_for": "false",
                       "trusted_proxies": ["10.0.0.0/8"]})
    def test_rate_limit_must_be_an_object(self):
        for bad in ("yes", 5, ["/v1/link"]):
            with self.subTest(rl=bad):
                with self.assertRaises(SystemExit):
                    self._svc(bad)
    def test_null_rate_limit_means_defaults_not_disabled(self):
        svc = self._svc(None)
        self.assertIsNotNone(svc.limiter)
    def test_forwarded_trust_without_trusted_proxies_refuses_startup(self):
        """Trusting the header from ANY peer lets callers choose their own
        rate-limit key -- that must not be reachable by omission."""
        with self.assertRaises(SystemExit):
            self._svc({"trust_forwarded_for": True})
        with self.assertRaises(SystemExit):
            self._svc({"trust_forwarded_for": True, "trusted_proxies": []})
    def test_forwarded_trust_with_proxies_is_accepted(self):
        svc = self._svc({"trust_forwarded_for": True,
                         "trusted_proxies": ["10.0.0.0/8", "127.0.0.1/32"]})
        self.assertTrue(svc.trust_forwarded_for)
        self.assertEqual(len(svc.trusted_proxies), 2)
 class TestRateLimitRound2(unittest.TestCase):
    """The three findings from the codex re-review."""
    def test_flooding_one_route_cannot_lock_out_another(self):
        """A shared key table turned the fail-closed capacity guard into a
        cross-route DoS: cheap /v1/repos keys could exhaust it and shut new
        /v1/link clients out. Budgets are per route."""
        now = [1000.0]
        rl = granthi_link.RateLimiter(
            rules={"/v1/link": (5, 3600), "/v1/repos": (60, 3600)},
            max_keys=20, clock=lambda: now[0])
        for i in range(300):                       # flood the cheap route
            rl.check("/v1/repos", f"10.1.{i // 256}.{i % 256}")
        allowed, _ = rl.check("/v1/link", "203.0.113.7")   # never-seen client
        self.assertTrue(allowed, "/v1/link locked out by a /v1/repos flood")
    def test_reclaim_uses_the_right_window_per_route(self):
        now = [1000.0]
        rl = granthi_link.RateLimiter(
            rules={"/v1/link": (1, 3600), "/v1/repos": (1, 10)},
            max_keys=2, clock=lambda: now[0])
        rl.check("/v1/repos", "a")
        rl.check("/v1/repos", "b")
        self.assertFalse(rl.check("/v1/repos", "c")[0])
        now[0] += 11                    # past the /v1/repos window only
        self.assertTrue(rl.check("/v1/repos", "c")[0])
 class TestRateLimitConfigRound2(ServiceTestBase):
    def _svc(self, rl):
        cfg = dict(self.svc.cfg)
        cfg["rate_limit"] = rl
        return granthi_link.LinkService(cfg)
    def test_booleans_are_rejected_in_rule_slots(self):
        """bool subclasses int, so isinstance let [5, True] through as a
        1-SECOND window -- 5/hour silently became ~5/sec."""
        for bad in ([5, True], [True, 3600], [False, 3600], [5, False]):
            with self.subTest(rule=bad):
                with self.assertRaises(SystemExit):
                    self._svc({"rules": {"/v1/link": bad}})
    def test_valid_int_rule_still_accepted(self):
        svc = self._svc({"rules": {"/v1/link": [5, 3600]}})
        self.assertEqual(svc.limiter.rules["/v1/link"], (5, 3600))
    def test_trusted_proxies_as_bare_string_refuses_startup(self):
        """A string would be iterated character by character, each char
        treated as a network."""
        with self.assertRaises(SystemExit):
            self._svc({"trust_forwarded_for": True,
                       "trusted_proxies": "10.0.0.0/8"})
    def test_wildcard_trusted_proxy_refuses_startup(self):
        """0.0.0.0/0 or ::/0 restores 'trust XFF from anyone' -- the exact
        hole trusted_proxies exists to close."""
        for wild in ("0.0.0.0/0", "::/0"):
            with self.subTest(cidr=wild):
                with self.assertRaises(SystemExit):
                    self._svc({"trust_forwarded_for": True,
                               "trusted_proxies": [wild]})
    def test_non_string_proxy_entry_refuses_startup(self):
        with self.assertRaises(SystemExit):
            self._svc({"trust_forwarded_for": True,
                       "trusted_proxies": [10, "10.0.0.0/8"]})
    def test_host_address_without_prefix_is_accepted(self):
        svc = self._svc({"trust_forwarded_for": True,
                         "trusted_proxies": ["127.0.0.1", "10.0.0.0/8"]})
        self.assertEqual(len(svc.trusted_proxies), 2)