Merge pull request 'feat: share repos with people, and invite people who have no account yet' (#5) from feat/invites-and-grants into main

Closes the last piece of the product picture: give one person access to some
of your repos and not others.

- POST /v1/grants add|remove|list -- runs on the CALLER'S OWN token. Verified
  against the live forge that a repo owner's scoped token adds and removes
  collaborators (204), so sharing needs no elevated rights anywhere.
- POST /v1/invite -- records a promise against a VERIFIED email and creates
  nothing until it is redeemed. Applied on first link.
- client: share / shared / invite.

Three properties the tests pin:
  * the FORGE decides who may share (listing collaborators requires repo
    admin, so its 200 is the authorisation answer, not ours);
  * an unverified email collects nothing, and its invite stays pending rather
    than being consumed;
  * a failed grant never blocks a sign-in -- nobody is locked out of their own
    account because a repo they were promised has since been deleted.

Applying an invite uses the admin credential deliberately: the inviter
authorised it at invite time and their session is long gone by redemption.

198 tests (was 184).

README.md

@@ -412,7 +412,7 @@ deleted it again, `DELETE …/tokens/{id}` returning 204 under basic auth):
 
 ## Tests
 
-* `python3 -m unittest discover -s tests` — 184 tests. The v1.2 additions
+* `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
@@ -541,7 +541,48 @@ 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`.
 
-## Next phase — invites and per-repo access (designed, not built)
+## 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

client/granthi_sync_client.py

@@ -1339,6 +1339,76 @@ def cmd_bootstrap(args):
     return 0
 
 
+def _post(cfg, path, body):
+    status, resp = http_json("POST", f"{cfg['server']}{path}", body=body)
+    if status == 401:
+        raise SystemExit("this device's access has been revoked -- "
+                         "run: granthi-sync link")
+    return status, resp
+
+
+def cmd_share(args):
+    """Give someone who already has an account access to one of your repos."""
+    cfg = require_linked(load_config())
+    status, resp = _post(cfg, "/v1/grants",
+                         {"token": cfg["token"],
+                          "action": "remove" if args.revoke else "add",
+                          "repo": args.repo, "login": args.who,
+                          "permission": args.permission})
+    if status == 403:
+        raise SystemExit(f"you cannot administer {args.repo} -- only the "
+                         f"owner can share it")
+    if status != 200:
+        raise SystemExit(f"share failed (HTTP {status}): {resp}")
+    verb = "no longer shared with" if args.revoke else \
+        f"shared with ({resp.get('permission')})"
+    log(f"{resp.get('repo')} {verb} {resp.get('login')}")
+    return 0
+
+
+def cmd_shared(args):
+    """Who can see one of your repos."""
+    cfg = require_linked(load_config())
+    status, resp = _post(cfg, "/v1/grants",
+                         {"token": cfg["token"], "action": "list",
+                          "repo": args.repo})
+    if status == 403:
+        raise SystemExit(f"you cannot administer {args.repo}")
+    if status != 200:
+        raise SystemExit(f"could not list access (HTTP {status}): {resp}")
+    people = resp.get("collaborators") or []
+    if not people:
+        print(f"{resp.get('repo')}: nobody else has access")
+        return 0
+    print(f"{resp.get('repo')} is shared with:")
+    for p in people:
+        print(f"  {p}")
+    return 0
+
+
+def cmd_invite(args):
+    """Promise access to someone who has no account yet.
+
+    Nothing is created for them now. The grant is held against their email and
+    applied the first time they sign in -- so an invite that is never accepted
+    leaves nothing behind.
+    """
+    cfg = require_linked(load_config())
+    repos = [{"name": r, "permission": args.permission} for r in args.repo]
+    status, resp = _post(cfg, "/v1/invite",
+                         {"token": cfg["token"], "email": args.email,
+                          "repos": repos})
+    if status == 403:
+        raise SystemExit(str(resp.get("error") or "you cannot share that repo"))
+    if status != 200:
+        raise SystemExit(f"invite failed (HTTP {status}): {resp}")
+    log(f"invited {resp['invited']} to "
+        f"{', '.join(g['repo'] for g in resp['repos'])}")
+    log("they get access the first time they run `granthi-sync link` with "
+        "that email -- until then nothing exists for them")
+    return 0
+
+
 def _folder_meta(cfg, folder):
     path = os.path.abspath(folder)
     meta = cfg.get("folders", {}).get(path)
@@ -1552,6 +1622,28 @@ def main(argv=None):
     sp.add_argument("--dry-run", action="store_true")
     sp.set_defaults(fn=cmd_bootstrap)
 
+    sp = sub.add_parser("share", help="share one of your repos with someone")
+    sp.add_argument("who", help="their forge login")
+    sp.add_argument("--repo", required=True)
+    sp.add_argument("--permission", default="write",
+                    choices=("read", "write", "admin"))
+    sp.add_argument("--revoke", action="store_true",
+                    help="take the access away again")
+    sp.set_defaults(fn=cmd_share)
+
+    sp = sub.add_parser("shared", help="who can see one of your repos")
+    sp.add_argument("--repo", required=True)
+    sp.set_defaults(fn=cmd_shared)
+
+    sp = sub.add_parser("invite",
+                        help="invite someone who has no account yet")
+    sp.add_argument("email")
+    sp.add_argument("--repo", action="append", required=True,
+                    help="repeatable")
+    sp.add_argument("--permission", default="write",
+                    choices=("read", "write", "admin"))
+    sp.set_defaults(fn=cmd_invite)
+
     sp = sub.add_parser("status", help="show linked folders")
     sp.set_defaults(fn=cmd_status)
 

server/granthi_link.py

@@ -90,6 +90,10 @@ 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
@@ -368,6 +372,42 @@ class IdentityStore:
                 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 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 {}
@@ -767,12 +807,16 @@ class LinkService:
                 "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)
-        return 200, {"gitea_base": self.public_gitea, "login": login,
+        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
 
     # -- device registry endpoints -----------------------------------------
 
@@ -862,6 +906,162 @@ class LinkService:
             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.take_invites(email)
+        applied = []
+        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"])
+                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:
@@ -982,6 +1182,10 @@ class Handler(BaseHTTPRequestHandler):
             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:

tests/test_server.py

@@ -51,6 +51,14 @@ class StubUpstream(BaseHTTPRequestHandler):
             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"]:
@@ -96,8 +104,36 @@ 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
@@ -125,7 +161,8 @@ class ServiceTestBase(unittest.TestCase):
         StubUpstream.state = {"users": {}, "created": [], "repos": set(),
                               "token_reqs": [], "hide_once": set(),
                               "tokens": {}, "tokens_by_sha": {},
-                              "revoke_fails": False}
+                              "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)
@@ -784,6 +821,145 @@ class TestDeviceEndpointsOverHttp(HandlerTestBase):
         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"]])
+
 if __name__ == "__main__":
     unittest.main()