Gabriel Vidal
16fa26ed86
The first project is the one the run starts in (the composer's mandatory select); the others — the project chips the guest also turned on — must each be granted chat, and join the run's write scope and system prompt. Resume and fork keep the extras the guest still holds chat on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
371 lines
20 KiB
Python
371 lines
20 KiB
Python
"""Guests (users/): the login lifecycle, the gate's caller resolution and
|
|
default-deny route table, the per-conversation filters, the server-pinned
|
|
spawn, sharing, per-user ui-state, the SSE filter and the author tag.
|
|
|
|
Imports ``main`` against throwaway paths (no startup event → no threads)."""
|
|
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
TMP = pathlib.Path(tempfile.mkdtemp())
|
|
for k, v in {"DB_PATH": TMP / "x.db", "TRANSCRIPTS_DIR": TMP / "t",
|
|
"WORKSPACE": TMP, "REPO_DIR": TMP, "ARCHIVE_DIR": TMP / "arch",
|
|
"META_PATH": TMP / "meta.json", "WORKERS_PATH": TMP / "w.json",
|
|
"WORKERS_TRANSCRIPTS_DIR": TMP / "mirror",
|
|
"CRON_PATH": TMP / "cron.json", "NOTIFY_PATH": TMP / "n.json",
|
|
"FORMS_PATH": TMP / "f.json", "UI_STATE_PATH": TMP / "ui.json",
|
|
"NOTIF_READ_PATH": TMP / "nr.json",
|
|
"MAIL_TRIGGER_SENDERS_PATH": TMP / "mt.json",
|
|
"USERS_PATH": TMP / "users.json", "AUTH_PATH": TMP / "auth.json",
|
|
"PROJECTS_DIR": TMP / "projects",
|
|
"INTERNAL_API_TOKEN": "tok", "GUEST_HOSTS": "claude.example.test",
|
|
"SIDECAR_TOKEN": "", "PI_TRANSCRIPTS_DIR": ""}.items():
|
|
os.environ[k] = str(v)
|
|
(TMP / "projects" / "film").mkdir(parents=True)
|
|
(TMP / "projects" / "other").mkdir(parents=True)
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
import main # noqa: E402
|
|
import conversations.routes as conv_routes # noqa: E402
|
|
import runs.routes as run_routes # noqa: E402
|
|
from conversations.parse import _split_author # noqa: E402
|
|
from core.routes import _guest_filter # noqa: E402
|
|
from users import access # noqa: E402
|
|
from users.caller import OWNER, SESSION_COOKIE, guest_from_record # noqa: E402
|
|
from users.runs import author_tag, check_model # noqa: E402
|
|
from users.store import DEFAULT_GUEST_SHORTCUTS, DEFAULT_SHARED_SKILLS # noqa: E402
|
|
|
|
DEPS = main.app.state.ai
|
|
OWNER_H = {"X-Internal-Auth": "tok"}
|
|
|
|
|
|
def _card(sid, **meta):
|
|
return {"sessionId": sid, "endedAt": "2026-10-10T00:00:00Z", "cost": 1.0,
|
|
"meta": {"archived": False, **meta}}
|
|
|
|
|
|
class Guests(unittest.TestCase):
|
|
def setUp(self):
|
|
for u in DEPS.users.all():
|
|
DEPS.users.delete(u["id"])
|
|
self.dom = DEPS.users.create("Dom@Example.com", "Dom",
|
|
projects={"film": "chat"},
|
|
models=["sonnet", "haiku"])
|
|
self.claire = DEPS.users.create("claire@example.com", "Claire",
|
|
projects={"film": "view"})
|
|
self.c = TestClient(main.app)
|
|
self.dom_cookie = DEPS.auth.open_session(self.dom["id"])
|
|
DEPS.users.set_defaults({"promptShortcuts": list(DEFAULT_GUEST_SHORTCUTS),
|
|
"sharedSkills": list(DEFAULT_SHARED_SKILLS)})
|
|
DEPS.ui_state.delete(f"u.{self.dom['id']}.ai-agent-settings")
|
|
|
|
def guest(self, cookie=None, host="claude.example.test"):
|
|
c = TestClient(main.app, base_url=f"https://{host}")
|
|
c.cookies.set(SESSION_COOKIE, cookie or self.dom_cookie)
|
|
return c
|
|
|
|
# ── store + login ────────────────────────────────────────────────────────
|
|
def test_login_link_is_single_use_and_opens_a_session(self):
|
|
token = DEPS.auth.mint_link(self.dom["id"])
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
r = c.post("/api/auth/login", json={"token": token})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(r.json()["id"], self.dom["id"])
|
|
self.assertIn(SESSION_COOKIE, r.cookies)
|
|
self.assertEqual(c.get("/api/me").json()["role"], "guest")
|
|
again = TestClient(main.app, base_url="https://claude.example.test")
|
|
self.assertEqual(again.post("/api/auth/login",
|
|
json={"token": token}).status_code, 401)
|
|
|
|
def test_owner_links_are_invitations(self):
|
|
# Copied / mailed by the owner: lives for days, still single-use; the
|
|
# login page's own link stays a quarter-hour one.
|
|
from users.store import INVITE_TTL_S, LINK_TTL_S
|
|
r = self.c.post(f"/api/users/{self.dom['id']}/login-link", json={"send": False},
|
|
headers=OWNER_H)
|
|
self.assertEqual(r.json()["expiresInS"], INVITE_TTL_S)
|
|
self.assertGreater(INVITE_TTL_S, LINK_TTL_S)
|
|
token = r.json()["url"].split("token=")[1]
|
|
self.assertEqual(DEPS.auth.consume_link(token), self.dom["id"])
|
|
self.assertIsNone(DEPS.auth.consume_link(token))
|
|
|
|
def test_signed_in_guest_can_still_use_sign_in_links(self):
|
|
# Dom's first sign-in worked but the page hung; every later attempt
|
|
# carried the live cookie and the gate refused /api/auth/* (403).
|
|
g = self.guest()
|
|
h = {"Origin": "https://claude.example.test"}
|
|
self.assertEqual(g.post("/api/auth/request-link", json={"email": "x@y.z"},
|
|
headers=h).status_code, 200)
|
|
token = DEPS.auth.mint_link(self.dom["id"])
|
|
r = g.post("/api/auth/login", json={"token": token}, headers=h)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
# the old session gave way to the new one
|
|
self.assertIsNone(DEPS.auth.session_user(self.dom_cookie))
|
|
self.assertEqual(g.post("/api/auth/login", json={"token": token},
|
|
headers=h).status_code, 401) # spent
|
|
|
|
def test_disabled_user_loses_the_session(self):
|
|
DEPS.users.update(self.dom["id"], {"disabled": True})
|
|
r = self.guest().get("/api/me")
|
|
self.assertEqual(r.status_code, 401) # guest surface, no valid session
|
|
|
|
def test_request_link_never_reveals_who_exists(self):
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
for e in ("dom@example.com", "nobody@example.com"):
|
|
self.assertEqual(c.post("/api/auth/request-link",
|
|
json={"email": e}).json(), {"ok": True})
|
|
|
|
# ── the gate ─────────────────────────────────────────────────────────────
|
|
def test_owner_paths_unchanged(self):
|
|
r = self.c.get("/api/me", headers=OWNER_H)
|
|
self.assertEqual(r.json()["role"], "owner")
|
|
self.assertEqual(self.c.get("/api/me").status_code, 403) # untrusted ip
|
|
|
|
def test_guest_surface_is_never_owner(self):
|
|
c = TestClient(main.app, base_url="https://claude.example.test")
|
|
r = c.get("/api/me", headers=OWNER_H)
|
|
self.assertEqual(r.status_code, 401)
|
|
self.assertEqual(r.json()["detail"]["kind"], "login_required")
|
|
|
|
def test_guest_default_deny(self):
|
|
g = self.guest()
|
|
for path in ("/api/users", "/api/cron", "/api/bundle", "/api/accounts",
|
|
"/api/workers", "/api/memories", "/api/services",
|
|
"/api/project-env?slug=film", "/api/notifications/unread"):
|
|
self.assertEqual(g.get(path).status_code, 403, path)
|
|
self.assertEqual(g.put("/api/file", json={}).status_code, 403)
|
|
|
|
def test_guest_cross_site_post_refused(self):
|
|
g = self.guest()
|
|
r = g.post("/api/interrupt", json={"sessionId": "x"},
|
|
headers={"Origin": "https://evil.example"})
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_me_lists_grants(self):
|
|
me = self.guest().get("/api/me").json()
|
|
self.assertEqual(me["projects"], [{"slug": "film", "level": "chat"}])
|
|
self.assertEqual(me["surface"], "guest")
|
|
|
|
# ── filtering ────────────────────────────────────────────────────────────
|
|
def test_conversation_list_is_own_plus_shared(self):
|
|
cards = [_card("a" * 8, owner=self.dom["id"]),
|
|
_card("b" * 8),
|
|
_card("c" * 8, participants=[self.dom["id"]]),
|
|
_card("d" * 8, owner=self.claire["id"])]
|
|
orig = conv_routes._conversation_cards
|
|
conv_routes._conversation_cards = lambda deps, full: cards
|
|
try:
|
|
got = {c["sessionId"] for c in
|
|
self.guest().get("/api/conversations").json()["conversations"]}
|
|
every = self.c.get("/api/conversations", headers=OWNER_H).json()
|
|
finally:
|
|
conv_routes._conversation_cards = orig
|
|
self.assertEqual(got, {"a" * 8, "c" * 8})
|
|
self.assertEqual(every["count"], 4)
|
|
|
|
def test_cards_carry_owner_and_participants(self):
|
|
# The list filter reads the card's meta: _conv_meta must pass both
|
|
# through (it once rebuilt meta from a fixed key list and dropped them).
|
|
from conversations.cards import _conv_meta
|
|
DEPS.meta_store.update("s-card", {"owner": "dom", "participants": ["claire"]})
|
|
m = _conv_meta(DEPS, {"sessionId": "s-card"})
|
|
self.assertEqual((m["owner"], m["participants"]), ("dom", ["claire"]))
|
|
|
|
def test_unseen_conversation_is_404(self):
|
|
DEPS.meta_store.update("s-other", {"owner": self.claire["id"]})
|
|
r = self.guest().get("/api/conversation?id=-x/s-other.jsonl")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_every_route_reaches_its_own_handler(self):
|
|
# A route decorator once slipped onto a helper above the real handler,
|
|
# which then shadowed /api/conversation-commits for everyone.
|
|
for r in main.app.routes:
|
|
ep = getattr(r, "endpoint", None)
|
|
if getattr(r, "path", "").startswith("/api/") and ep is not None:
|
|
self.assertFalse(ep.__name__.startswith("_"), r.path)
|
|
r = self.c.get("/api/conversation-commits?id=-x/nope.jsonl", headers=OWNER_H)
|
|
self.assertNotEqual(r.status_code, 422, r.text)
|
|
|
|
def test_projects_scoped_to_grants(self):
|
|
r = self.guest().get("/api/project?slug=other")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_ui_state_is_namespaced(self):
|
|
g = self.guest()
|
|
g.put("/api/ui-state/ai-agent-settings", json={"value": "guest"})
|
|
self.assertEqual(DEPS.ui_state.get(f"u.{self.dom['id']}.ai-agent-settings"),
|
|
"guest")
|
|
self.assertIsNone(self.c.get("/api/ui-state/ai-agent-settings",
|
|
headers=OWNER_H).json()["value"])
|
|
|
|
# ── guest settings + shared skills ──────────────────────────────────────
|
|
def test_fresh_guest_settings_start_from_the_guest_shortcuts(self):
|
|
import json
|
|
g = self.guest()
|
|
seed = json.loads(g.get("/api/ui-state/ai-agent-settings").json()["value"])
|
|
self.assertEqual([s["id"] for s in seed["state"]["promptShortcuts"]],
|
|
[s["id"] for s in DEFAULT_GUEST_SHORTCUTS])
|
|
self.assertEqual(seed["version"], 3)
|
|
# the owner's edit reaches a guest who never saved anything…
|
|
self.c.put("/api/guest-defaults", headers=OWNER_H, json={"promptShortcuts": [
|
|
{"id": "x", "label": "X", "guideline": "do x", "runner": "mac",
|
|
"shared": False, "bogus": 1}]})
|
|
seed = json.loads(g.get("/api/ui-state/ai-agent-settings").json()["value"])
|
|
self.assertEqual(seed["state"]["promptShortcuts"], [
|
|
{"id": "x", "label": "X", "guideline": "do x", "shared": True,
|
|
"defaultOn": False}])
|
|
self.assertEqual(g.get("/api/me").json()["defaultShortcuts"][0]["id"], "x")
|
|
# …and their own save wins from then on; the owner never gets a seed
|
|
g.put("/api/ui-state/ai-agent-settings", json={"value": "mine"})
|
|
self.assertEqual(g.get("/api/ui-state/ai-agent-settings").json()["value"], "mine")
|
|
self.assertIsNone(self.c.get("/api/ui-state/ai-agent-settings",
|
|
headers=OWNER_H).json()["value"])
|
|
|
|
def test_guest_defaults_are_the_owners(self):
|
|
self.assertEqual(self.guest().get("/api/guest-defaults").status_code, 403)
|
|
r = self.c.put("/api/guest-defaults", headers=OWNER_H, json={
|
|
"sharedSkills": ["ask-form", "../etc", "film/cut", "a/b/c", "ask-form"]})
|
|
self.assertEqual(r.json()["sharedSkills"], ["ask-form", "film/cut"])
|
|
self.assertEqual(DEPS.users.defaults()["promptShortcuts"][0]["id"], "commit")
|
|
|
|
def test_guest_run_scope_carries_the_shared_skills(self):
|
|
calls = self._capture()
|
|
DEPS.users.set_defaults({"sharedSkills": ["ask-form", "film/cut"]})
|
|
self.guest().post("/api/spawn", json={"prompt": "hi", "projects": ["film"]})
|
|
scope = calls[-1][1]["scope"]
|
|
self.assertEqual(scope["skills"], ["ask-form", "film/cut"])
|
|
self.assertIn("`shared:cut`", scope["systemPrompt"])
|
|
DEPS.users.set_defaults({"sharedSkills": []})
|
|
self.guest().post("/api/spawn", json={"prompt": "hi", "projects": ["film"]})
|
|
self.assertNotIn("skills", calls[-1][1]["scope"])
|
|
|
|
def test_guest_skills_are_shared_plus_granted_projects(self):
|
|
def skill(root, name):
|
|
d = root / ".claude" / "skills" / name
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
(d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: {name} it\n---\n")
|
|
skill(TMP, "ask-form")
|
|
skill(TMP, "kept-home")
|
|
skill(TMP / "projects" / "film", "cut")
|
|
skill(TMP / "projects" / "other", "secret")
|
|
DEPS.users.set_defaults({"sharedSkills": ["ask-form", "missing"]})
|
|
names = {s["name"]: s["source"] for s in
|
|
self.guest().get("/api/guest-skills").json()["skills"]}
|
|
self.assertEqual(names, {"shared:ask-form": "shared", "cut": "film"})
|
|
cards = {s["name"]: s for s in
|
|
self.c.get("/api/skills", headers=OWNER_H).json()["skills"]}
|
|
self.assertTrue(cards["ask-form"]["sharedWithGuests"])
|
|
self.assertFalse(cards["kept-home"]["sharedWithGuests"])
|
|
self.assertEqual(cards["cut"]["shareRef"], "film/cut")
|
|
|
|
def test_sse_filter(self):
|
|
dom = guest_from_record(self.dom)
|
|
DEPS.meta_store.update("mine", {"owner": self.dom["id"]})
|
|
flt = _guest_filter(DEPS, dom)
|
|
ev = flt({"type": "transcript", "ids": ["-p/mine.jsonl", "-p/theirs.jsonl",
|
|
"-p/mine/subagents/agent-1.jsonl"]})
|
|
self.assertEqual(ev["ids"], ["-p/mine.jsonl", "-p/mine/subagents/agent-1.jsonl"])
|
|
self.assertIsNone(flt({"type": "cron"}))
|
|
self.assertIsNone(flt({"type": "meta", "id": "theirs"}))
|
|
|
|
# ── runs ─────────────────────────────────────────────────────────────────
|
|
def _capture(self):
|
|
calls = []
|
|
orig = run_routes._send_message
|
|
run_routes._send_message = lambda deps, prompt, **kw: (
|
|
calls.append((prompt, kw)) or {"sessionId": "new", "pid": 1})
|
|
self.addCleanup(setattr, run_routes, "_send_message", orig)
|
|
return calls
|
|
|
|
def test_guest_spawn_is_pinned_server_side(self):
|
|
calls = self._capture()
|
|
r = self.guest().post("/api/spawn", json={
|
|
"prompt": "hello", "projects": ["film"], "cwd": "/",
|
|
"account": "work", "worker": "mac", "harness": "pi", "model": "sonnet"})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
prompt, kw = calls[0]
|
|
self.assertTrue(prompt.startswith('<ai-agent-from user="dom"'))
|
|
self.assertEqual(kw["cwd"], "~/projects/film")
|
|
self.assertEqual((kw["account"], kw["worker"], kw["harness"]),
|
|
("personal", "local", "claude"))
|
|
self.assertEqual(kw["stamp"], {"owner": "dom", "projects": ["film"]})
|
|
self.assertIn("~/projects/film", kw["scope"]["writeDirs"])
|
|
self.assertFalse(kw["scope"]["remoteControl"])
|
|
|
|
def test_guest_spawn_with_extra_projects(self):
|
|
DEPS.users.update(self.dom["id"], {"projects": {"film": "chat", "other": "ship"}})
|
|
calls = self._capture()
|
|
g = self.guest()
|
|
r = g.post("/api/spawn", json={"prompt": "x", "projects": ["other", "film", "other"]})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
_, kw = calls[0]
|
|
self.assertEqual(kw["cwd"], "~/projects/other") # the select's pick runs first
|
|
self.assertEqual(kw["stamp"]["projects"], ["other", "film"])
|
|
self.assertEqual(kw["scope"]["writeDirs"],
|
|
["~/projects/other", "~/projects/film", "/tmp"])
|
|
self.assertIn("project `film`", kw["scope"]["systemPrompt"])
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": []}).status_code, 400)
|
|
|
|
def test_guest_spawn_refusals(self):
|
|
self._capture()
|
|
g = self.guest()
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": ["other"]}).status_code, 403)
|
|
self.assertEqual(g.post("/api/spawn", json={ # an extra needs chat too
|
|
"prompt": "x", "projects": ["film", "other"]}).status_code, 403)
|
|
self.assertEqual(g.post("/api/spawn", json={
|
|
"prompt": "x", "projects": ["film"], "model": "opus"}).status_code, 403)
|
|
claire = self.guest(DEPS.auth.open_session(self.claire["id"]))
|
|
self.assertEqual(claire.post("/api/spawn", json={ # view only
|
|
"prompt": "x", "projects": ["film"]}).status_code, 403)
|
|
|
|
def test_resume_needs_access_and_tags_the_author(self):
|
|
calls = self._capture()
|
|
sid = "11111111-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.claire["id"], "projects": ["film"]})
|
|
g = self.guest()
|
|
self.assertEqual(g.post("/api/resume", json={
|
|
"sessionId": sid, "prompt": "hi"}).status_code, 404)
|
|
DEPS.meta_store.update(sid, {"participants": [self.dom["id"]]})
|
|
self.assertEqual(g.post("/api/resume", json={
|
|
"sessionId": sid, "prompt": "hi", "cwd": "/"}).status_code, 200)
|
|
prompt, kw = calls[-1]
|
|
self.assertIn('name="Dom"', prompt)
|
|
self.assertNotIn("cwd", kw) # never the client's
|
|
# …and the owner's own turn in that shared thread is tagged too
|
|
self.c.post("/api/resume", json={"sessionId": sid, "prompt": "me"},
|
|
headers=OWNER_H)
|
|
self.assertTrue(calls[-1][0].startswith('<ai-agent-from user="owner"'))
|
|
|
|
def test_share_by_conversation_owner(self):
|
|
sid = "22222222-2222-3333-4444-555555555555"
|
|
DEPS.meta_store.update(sid, {"owner": self.dom["id"], "projects": ["film"]})
|
|
r = self.guest().put("/api/conversation-share",
|
|
json={"id": sid, "participants": [self.claire["id"]]})
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(DEPS.meta_store.peek(sid)["participants"], ["claire"])
|
|
self.assertTrue(access.visible(DEPS, guest_from_record(self.claire), sid))
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────
|
|
def test_author_tag_round_trip(self):
|
|
dom = guest_from_record(self.dom)
|
|
author, text = _split_author(f"{author_tag(dom)}\nfix the trailer")
|
|
self.assertEqual(author, {"id": "dom", "name": "Dom", "role": "guest"})
|
|
self.assertEqual(text, "fix the trailer")
|
|
self.assertEqual(_split_author("plain")[0], None)
|
|
|
|
def test_model_families(self):
|
|
dom = guest_from_record(self.dom)
|
|
self.assertEqual(check_model(dom, None), "sonnet")
|
|
self.assertEqual(check_model(dom, "claude-haiku-4-5-20251001"),
|
|
"claude-haiku-4-5-20251001")
|
|
self.assertTrue(OWNER.can("anything", "ship"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|