feat: share repos with people, and invite people who have no account yet

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).

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()