Files
ai-agent/backend/users/store.py
Gabriel Vidal deb01427d5 feat(backend): New projects — owner and guests start a project from a template
newproject/: the policy (guestDefaults.newProjects — quota, templates, the
ones the owner approves first, visibility, level, domain — and a per-guest
createProjects + address bucket), the New project form built from it, and the
flow: POST /api/new-project/form (the run's guesses as defaults) → submit →
shape/policy and host checks (a problem sends a corrected form back) → an
owner ask for a backend template → sidecar /scaffold → grant → /relocate +
resume the same session inside the project with an app-authored turn.
SpawnBody.newProject starts a guest's run in ~/projects/.new-project/<uid>
(staging_scope: /tmp only); /api/me reports newProjects; the thread follows
ConversationDetail.continuedIn.

Fix: a guest's turn into an existing conversation resumed with no cwd, so the
sidecar ran it in the lab checkout (its CLAUDE.md loaded) — it now resumes
where the session lives.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 23:42:07 +02:00

497 lines
21 KiB
Python

"""
The people other than the owner who can use the app — **guests** — and the
login state that proves who a browser is.
Two JSON files under ``/data``, both whole-file rewrites (temp file + rename,
like ``meta.py``):
* ``users.json`` — ``{"users": {<id>: {...}}}``. The owner is *not* in it: the
owner is whoever reaches the admin host (Authelia) or the trusted loopback,
exactly as before this file existed. A user record:
``id`` (slug), ``email``, ``name``, ``role`` (``guest``), ``projects``
(``{slug: "view"|"chat"|"ship"}``), ``models`` (aliases a run may use),
``budgetUsd`` (monthly cap), ``notify`` (``{"channels": [...]}`` — the
channels this user wants to be reached on), ``createProjects`` (may start
projects from the owner's templates — ``newproject/``), ``bucket`` (their
address bucket, ``<name>.<bucket>.<domain>``; blank ⇒ their id),
``disabled``, ``createdAt``, ``lastLoginAt``.
Beside ``users``, ``guestDefaults`` holds what the owner sets for every
guest at once: ``promptShortcuts`` (the list a new guest's composer starts
with — copied into their own settings on first load, theirs to edit after)
and ``sharedSkills`` (the owner's skills every guest run gets, as a
generated Claude Code plugin — ``name`` for one of the lab repo's skills,
``<project>/<name>`` for a project's), and ``newProjects`` — what a guest
allowed to create projects may pick (:data:`DEFAULT_NEW_PROJECTS`).
* ``auth.json`` — ``{"links": {...}, "sessions": {...}}``. One-time login
links (sent by mail) and browser sessions. Only **sha256 hashes** of the
tokens are stored: a leaked file can't be replayed.
"""
from __future__ import annotations
import datetime
import hashlib
import json
import os
import pathlib
import re
import secrets
import threading
import time
LEVELS = ("view", "chat", "ship")
DEFAULT_MODELS = ("opus", "sonnet", "haiku")
DEFAULT_BUDGET_USD = float(os.environ.get("GUEST_DEFAULT_BUDGET_USD", "500"))
NOTIFY_CHANNELS = ("in_app", "email", "webpush")
LINK_TTL_S = int(os.environ.get("LOGIN_LINK_TTL_S", str(15 * 60)))
# A link the owner hands out (Settings → Users: copy or mail it) is an
# *invitation*: read whenever the guest gets to their inbox, so it lives for
# days, not the quarter-hour of a link the guest asked for themselves.
INVITE_TTL_S = int(os.environ.get("INVITE_LINK_TTL_S", str(3 * 24 * 3600)))
SESSION_TTL_S = int(os.environ.get("SESSION_TTL_S", str(30 * 24 * 3600)))
# The owner's "view as this guest" session (Settings → Users → Impersonate):
# long enough to test a flow, short enough not to leave the admin stuck as a
# guest on a forgotten tab.
IMPERSONATE_TTL_S = int(os.environ.get("IMPERSONATE_TTL_S", str(4 * 3600)))
# A session's expiry slides forward when used, but its file write is batched:
# only once this much of the TTL has elapsed since the last bump.
_SESSION_BUMP_S = 6 * 3600
# What a fresh install shares with guests: the ones a guest run is told to use
# (ask-form, notify-done; new-frontend-project — the New project form), and
# two read-only helpers. Missing ones are skipped.
DEFAULT_SHARED_SKILLS = ("ask-form", "notify-done", "screenshot", "page-scrape",
"new-frontend-project")
# A new guest's composer chips — generic wrap-ups, nothing lab-specific (no
# worktree script, no deploy target). The owner rewrites them in Settings → Users.
DEFAULT_GUEST_SHORTCUTS = (
{"id": "commit", "label": "Commit", "icon": "GitCommitHorizontal",
"guideline": "Commit your changes when done (and push them if you can ship this project).",
"defaultOn": True, "shared": True},
{"id": "ask", "label": "Ask me", "icon": "MessageCircleQuestion",
"guideline": "Ask me clarifying questions with the ask-form skill before starting if anything is unclear.",
"defaultOn": False, "shared": True},
{"id": "notify", "label": "Notify", "icon": "Bell",
"guideline": "Notify me (notify-done) when you finish.",
"defaultOn": True, "shared": True},
)
# What a guest allowed to create projects gets (Settings → Users → Every guest
# → New projects): ``quota`` projects all told; the ``templates`` offered, of
# which ``approve`` wait for the owner's OK (a backend on the zipgo host); the
# site ``visibility`` choices; the creator's grant ``level``; and the
# ``domain`` their address bucket sits under.
NEW_PROJECT_VISIBILITY = ("public", "password", "geo")
DEFAULT_NEW_PROJECTS = {
"quota": 10,
"templates": ["react-vite", "vite-react-pocketbase"],
"approve": ["vite-react-pocketbase"],
"visibility": list(NEW_PROJECT_VISIBILITY),
"visibilityDefault": "public",
"level": "ship",
"domain": "gabvdl.xyz",
}
_TEMPLATE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
_BUCKET_RE = re.compile(r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)?$")
_DOMAIN_RE = re.compile(r"^[a-z0-9-]+(?:\.[a-z0-9-]+)+$")
_SKILL_REF_RE = re.compile(r"^(?:[A-Za-z0-9][A-Za-z0-9._-]{0,63}/)?[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
_SHORTCUT_KEYS = {"id", "label", "icon", "guideline", "defaultOn", "labelOff",
"guidelineOff", "children", "triggers", "shared", "runner"}
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,39}$")
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
def _now_iso() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _hash(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()
def norm_email(email: str) -> str:
return (email or "").strip().lower()
def valid_email(email: str) -> bool:
return bool(_EMAIL_RE.match(norm_email(email)))
def slug_for(email: str, name: str | None, taken: set[str]) -> str:
"""A readable id: the name (or the mail's local part), slugged, deduped."""
base = re.sub(r"[^a-z0-9]+", "-", (name or norm_email(email).split("@")[0])
.lower()).strip("-")[:32] or "guest"
cand, n = base, 2
while cand in taken or cand == "owner":
cand, n = f"{base}-{n}", n + 1
return cand
def clean_projects(raw) -> dict[str, str]:
"""``{slug: level}`` with unknown levels dropped (a typo grants nothing)."""
out: dict[str, str] = {}
if isinstance(raw, dict):
for slug, level in raw.items():
if isinstance(slug, str) and slug.strip() and level in LEVELS:
out[slug.strip()] = level
return out
def clean_skill_refs(raw) -> list[str]:
"""``name`` / ``<project>/<name>`` refs, deduped; anything path-like dropped."""
if not isinstance(raw, (list, tuple)):
return []
return list(dict.fromkeys(r.strip() for r in raw
if isinstance(r, str) and _SKILL_REF_RE.match(r.strip())))[:64]
def clean_shortcuts(raw) -> list[dict]:
"""Prompt shortcuts in the frontend's ``PromptShortcut`` shape — known keys
only, each one with an id, a label and a guideline. Always shared: a
guest's runs only ever land on the lab."""
out: list[dict] = []
seen: set[str] = set()
for sc in raw if isinstance(raw, (list, tuple)) else []:
if not isinstance(sc, dict):
continue
sid = str(sc.get("id") or "").strip()[:40]
if not sid or sid in seen or not str(sc.get("label") or "").strip():
continue
seen.add(sid)
clean = {k: v for k, v in sc.items() if k in _SHORTCUT_KEYS}
for k in ("label", "icon", "labelOff"):
if k in clean:
clean[k] = str(clean[k])[:80]
for k in ("guideline", "guidelineOff"):
if k in clean:
clean[k] = str(clean[k])[:2000]
for k in ("children", "triggers"):
if k in clean:
clean[k] = [str(x)[:80] for x in clean[k] if isinstance(x, str)][:32] \
if isinstance(clean[k], list) else []
clean["id"] = sid
clean["defaultOn"] = bool(clean.get("defaultOn"))
clean["shared"] = True
clean.pop("runner", None)
out.append(clean)
return out[:40]
def clean_new_projects(raw, base: dict | None = None) -> dict:
"""The New projects policy, each key validated; a key that's missing or
malformed keeps ``base``'s value (the defaults)."""
out = json.loads(json.dumps(base or DEFAULT_NEW_PROJECTS))
raw = raw if isinstance(raw, dict) else {}
try:
if "quota" in raw:
out["quota"] = max(0, min(1000, int(raw["quota"])))
except (TypeError, ValueError):
pass
for key in ("templates", "approve"):
if isinstance(raw.get(key), (list, tuple)):
out[key] = list(dict.fromkeys(
t.strip() for t in raw[key]
if isinstance(t, str) and _TEMPLATE_NAME_RE.match(t.strip())))
out["approve"] = [t for t in out["approve"] if t in out["templates"]]
if isinstance(raw.get("visibility"), (list, tuple)):
vis = [v for v in raw["visibility"] if v in NEW_PROJECT_VISIBILITY]
out["visibility"] = list(dict.fromkeys(vis)) or ["public"]
if raw.get("visibilityDefault") in NEW_PROJECT_VISIBILITY:
out["visibilityDefault"] = raw["visibilityDefault"]
if out["visibilityDefault"] not in out["visibility"]:
out["visibilityDefault"] = out["visibility"][0]
if raw.get("level") in ("chat", "ship"):
out["level"] = raw["level"]
dom = str(raw.get("domain") or "").strip().lower()
if dom and _DOMAIN_RE.match(dom):
out["domain"] = dom
return out
def clean_bucket(raw) -> str:
"""A guest's address bucket: one DNS label, or blank (⇒ their id)."""
b = str(raw or "").strip().lower()
if not _BUCKET_RE.match(b):
raise ValueError(f"{b!r} isn't a valid address bucket (one DNS label)")
return b
def clean_channels(raw) -> list[str]:
if not isinstance(raw, (list, tuple)):
return ["in_app"]
out = [c for c in raw if c in NOTIFY_CHANNELS]
return list(dict.fromkeys(out)) or ["in_app"]
class _JsonFile:
def __init__(self, path: str, empty: dict):
self.path = pathlib.Path(path)
self._empty = empty
self._lock = threading.RLock()
self._data = self._load()
def _load(self) -> dict:
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
if isinstance(data, dict):
return {**json.loads(json.dumps(self._empty)), **data}
except (OSError, ValueError):
pass
return json.loads(json.dumps(self._empty))
def _save(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(self.path.suffix + ".tmp")
tmp.write_text(json.dumps(self._data, indent=2, sort_keys=True),
encoding="utf-8")
os.chmod(tmp, 0o600)
os.replace(tmp, self.path)
class UsersStore(_JsonFile):
def __init__(self, path: str):
super().__init__(path, {"users": {}, "guestDefaults": {
"promptShortcuts": [dict(s) for s in DEFAULT_GUEST_SHORTCUTS],
"sharedSkills": list(DEFAULT_SHARED_SKILLS),
"newProjects": dict(DEFAULT_NEW_PROJECTS)}})
# ── what every guest gets ────────────────────────────────────────────────
def defaults(self) -> dict:
with self._lock:
d = self._data.get("guestDefaults") or {}
return {"promptShortcuts": [dict(s) for s in d.get("promptShortcuts") or []],
"sharedSkills": list(d.get("sharedSkills") or []),
# An install from before New projects has no key: defaults.
"newProjects": clean_new_projects(d.get("newProjects") or {})}
def set_defaults(self, fields: dict) -> dict:
with self._lock:
d = self._data.setdefault("guestDefaults", {})
if "promptShortcuts" in fields:
d["promptShortcuts"] = clean_shortcuts(fields["promptShortcuts"])
if "sharedSkills" in fields:
d["sharedSkills"] = clean_skill_refs(fields["sharedSkills"])
if "newProjects" in fields:
d["newProjects"] = clean_new_projects(
fields["newProjects"], clean_new_projects(d.get("newProjects") or {}))
self._save()
return self.defaults()
# ── reads ────────────────────────────────────────────────────────────────
def all(self) -> list[dict]:
with self._lock:
return [dict(u) for u in self._data["users"].values()]
def get(self, uid: str) -> dict | None:
with self._lock:
u = self._data["users"].get(uid)
return dict(u) if u else None
def by_email(self, email: str) -> dict | None:
e = norm_email(email)
with self._lock:
for u in self._data["users"].values():
if u.get("email") == e:
return dict(u)
return None
def active(self, uid: str) -> dict | None:
u = self.get(uid)
return u if u and not u.get("disabled") else None
def emails(self) -> list[str]:
"""Every enabled user's address (the mail trigger's half of the list)."""
return [u["email"] for u in self.all() if not u.get("disabled")]
# ── writes ───────────────────────────────────────────────────────────────
def create(self, email: str, name: str | None = None, **fields) -> dict:
e = norm_email(email)
if not valid_email(e):
raise ValueError("invalid email")
with self._lock:
if self.by_email(e):
raise ValueError("a user with this email already exists")
uid = slug_for(e, name, set(self._data["users"]))
u = {"id": uid, "email": e, "name": (name or "").strip() or uid,
"role": "guest", "projects": {}, "models": list(DEFAULT_MODELS),
"budgetUsd": DEFAULT_BUDGET_USD,
"notify": {"channels": ["in_app", "email"]},
"createProjects": False, "bucket": "",
"disabled": False, "createdAt": _now_iso(), "lastLoginAt": None}
u.update(self._clean(fields))
self._data["users"][uid] = u
self._save()
return dict(u)
def update(self, uid: str, fields: dict) -> dict:
with self._lock:
u = self._data["users"].get(uid)
if u is None:
raise KeyError(uid)
clean = self._clean(fields)
if "email" in fields:
e = norm_email(fields["email"])
if not valid_email(e):
raise ValueError("invalid email")
other = self.by_email(e)
if other and other["id"] != uid:
raise ValueError("a user with this email already exists")
clean["email"] = e
u.update(clean)
self._save()
return dict(u)
def delete(self, uid: str) -> bool:
with self._lock:
if self._data["users"].pop(uid, None) is None:
return False
self._save()
return True
def touch_login(self, uid: str) -> None:
with self._lock:
u = self._data["users"].get(uid)
if u is not None:
u["lastLoginAt"] = _now_iso()
self._save()
@staticmethod
def _clean(fields: dict) -> dict:
out: dict = {}
if "name" in fields and isinstance(fields["name"], str):
out["name"] = fields["name"].strip()[:80]
if "projects" in fields:
out["projects"] = clean_projects(fields["projects"])
if "models" in fields and isinstance(fields["models"], (list, tuple)):
out["models"] = [m for m in fields["models"]
if isinstance(m, str) and m.strip()][:12]
if "budgetUsd" in fields:
try:
out["budgetUsd"] = max(0.0, float(fields["budgetUsd"]))
except (TypeError, ValueError):
pass
if "notify" in fields and isinstance(fields["notify"], dict):
out["notify"] = {"channels": clean_channels(
fields["notify"].get("channels"))}
if "disabled" in fields:
out["disabled"] = bool(fields["disabled"])
if "createProjects" in fields and fields["createProjects"] is not None:
out["createProjects"] = bool(fields["createProjects"])
if "bucket" in fields and fields["bucket"] is not None:
out["bucket"] = clean_bucket(fields["bucket"])
return out
class AuthStore(_JsonFile):
"""One-time login links and browser sessions, keyed by token hash."""
def __init__(self, path: str):
super().__init__(path, {"links": {}, "sessions": {}})
def _prune(self, now: float) -> None:
for kind in ("links", "sessions"):
dead = [h for h, r in self._data[kind].items()
if r.get("exp", 0) < now]
for h in dead:
del self._data[kind][h]
# ── login links ──────────────────────────────────────────────────────────
def mint_link(self, uid: str, ttl_s: int = LINK_TTL_S) -> str:
token = secrets.token_urlsafe(32)
now = time.time()
with self._lock:
self._prune(now)
self._data["links"][_hash(token)] = {"user": uid, "exp": now + ttl_s,
"created": now}
self._save()
return token
def last_link_at(self, uid: str) -> float:
"""When the newest live link for ``uid`` was minted (0 = none)."""
with self._lock:
made = [r.get("created") or (r["exp"] - LINK_TTL_S)
for r in self._data["links"].values() if r.get("user") == uid]
return max(made) if made else 0.0
def consume_link(self, token: str) -> str | None:
"""The link's user id, once — the link is gone after this call."""
now = time.time()
with self._lock:
rec = self._data["links"].pop(_hash(token or ""), None)
self._prune(now)
self._save()
if not rec or rec.get("exp", 0) < now:
return None
return rec.get("user")
# ── sessions ─────────────────────────────────────────────────────────────
def open_session(self, uid: str, ua: str = "", impersonator: str | None = None) -> str:
"""A browser session for ``uid``. With ``impersonator`` it is the
owner looking through the guest's eyes: short-lived, never extended,
and not counted among the guest's own sessions."""
token = secrets.token_urlsafe(32)
now = time.time()
ttl = IMPERSONATE_TTL_S if impersonator else SESSION_TTL_S
rec = {"user": uid, "exp": now + ttl, "created": now, "seen": now,
"ua": (ua or "")[:160]}
if impersonator:
rec["imp"] = impersonator
with self._lock:
self._prune(now)
self._data["sessions"][_hash(token)] = rec
self._save()
return token
def session(self, token: str) -> dict | None:
"""The live session record behind ``token`` (``user``, ``imp``…)."""
if not token:
return None
h = _hash(token)
now = time.time()
with self._lock:
rec = self._data["sessions"].get(h)
if not rec:
return None
if rec.get("exp", 0) < now:
del self._data["sessions"][h]
self._save()
return None
if not rec.get("imp") and now - rec.get("seen", 0) > _SESSION_BUMP_S:
rec["seen"] = now
rec["exp"] = now + SESSION_TTL_S
self._save()
return dict(rec)
def session_user(self, token: str) -> str | None:
rec = self.session(token)
return rec.get("user") if rec else None
def close_session(self, token: str) -> None:
with self._lock:
if self._data["sessions"].pop(_hash(token or ""), None) is not None:
self._save()
def revoke_user(self, uid: str) -> int:
"""Drop every session and pending link of ``uid``."""
with self._lock:
n = 0
for kind in ("links", "sessions"):
for h in [h for h, r in self._data[kind].items()
if r.get("user") == uid]:
del self._data[kind][h]
n += kind == "sessions"
self._save()
return n
def sessions_of(self, uid: str) -> list[dict]:
"""The guest's own sessions (the owner's impersonations aren't)."""
with self._lock:
return [{"created": r.get("created"), "seen": r.get("seen"),
"ua": r.get("ua")}
for r in self._data["sessions"].values()
if r.get("user") == uid and not r.get("imp")]