Spaces:
Sleeping
Sleeping
| """ | |
| The org/owner/role model — the foundation everything else (mailbox connections, contacts, | |
| messages, sends) is scoped by. See docs/blueprint/blueprint.md "Core entities" and | |
| docs/product/srs.md FR-1 through FR-3. | |
| Firestore layout (see backend/firestore_client.py for the local/real seam): | |
| orgs/{org_id} — {name, owner_uid, created_at} | |
| orgs/{org_id}/members/{uid} — {role, added_at, email?} | |
| orgs/{org_id}/roles/{key} — {name, permissions} (an edited built-in role, or a | |
| custom one; built-ins not stored use their defaults) | |
| user_orgs/{uid} — {org_ids: {org_id: role, ...}} (reverse index, | |
| kept in sync on every membership write, so | |
| "which orgs does this user belong to" doesn't | |
| require scanning every org) | |
| One person (one Firebase uid) can own or belong to multiple orgs — the reverse index is | |
| what makes that lookup cheap. | |
| """ | |
| from __future__ import annotations | |
| import uuid | |
| from datetime import datetime, timezone | |
| from backend import firestore_client as db | |
| from backend.core.constants import BUILTIN_ROLES, ORG_CREATOR_ROLE, PERMISSIONS, ROLE_ALIASES, ROLE_RANK | |
| class PermissionError_(Exception): | |
| """Raised when an actor's role doesn't meet the required rank for an action.""" | |
| class NotFoundError(Exception): | |
| """Raised when an org or membership doesn't exist.""" | |
| def _now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _member_path(org_id: str, uid: str) -> str: | |
| return f"orgs/{org_id}/members/{uid}" | |
| def _members_prefix(org_id: str) -> str: | |
| return f"orgs/{org_id}/members/" | |
| def _user_orgs_path(uid: str) -> str: | |
| return f"user_orgs/{uid}" | |
| def _set_reverse_index(uid: str, org_id: str, role: str) -> None: | |
| doc = db.get_doc(_user_orgs_path(uid)) or {"org_ids": {}} | |
| doc["org_ids"][org_id] = role | |
| db.set_doc(_user_orgs_path(uid), doc) | |
| def _clear_reverse_index(uid: str, org_id: str) -> None: | |
| doc = db.get_doc(_user_orgs_path(uid)) | |
| if doc and org_id in doc.get("org_ids", {}): | |
| del doc["org_ids"][org_id] | |
| db.set_doc(_user_orgs_path(uid), doc) | |
| def create_org(owner_uid: str, name: str, kind: str | None = None) -> str: | |
| """Creates a new org with `owner_uid` as its owner. Returns the new org_id. A single | |
| uid can own — and separately create — more than one org (SRS note under FR-2). | |
| `kind` is "personal" for an individual's own org (see backend/accounts.py).""" | |
| org_id = uuid.uuid4().hex | |
| doc = {"name": name, "owner_uid": owner_uid, "created_at": _now()} | |
| if kind: | |
| doc["kind"] = kind | |
| db.set_doc(f"orgs/{org_id}", doc) | |
| db.set_doc(_member_path(org_id, owner_uid), {"role": ORG_CREATOR_ROLE, "added_at": _now()}) | |
| _set_reverse_index(owner_uid, org_id, ORG_CREATOR_ROLE) | |
| return org_id | |
| def get_org(org_id: str) -> dict: | |
| org = db.get_doc(f"orgs/{org_id}") | |
| if org is None: | |
| raise NotFoundError(f"No org {org_id!r}") | |
| return org | |
| def get_role(org_id: str, uid: str) -> str | None: | |
| """The uid's role in this org, or None if they aren't a member.""" | |
| membership = db.get_doc(_member_path(org_id, uid)) | |
| return membership["role"] if membership else None | |
| def require_role(org_id: str, uid: str, minimum_role: str) -> str: | |
| """Raises PermissionError_ unless uid's role in org_id ranks at or above | |
| minimum_role. Returns the actor's actual role on success. Kept for rank checks between | |
| built-in roles; feature checks use require_permission. Custom roles rank lowest.""" | |
| role = get_role(org_id, uid) | |
| if role is None: | |
| raise PermissionError_(f"{uid!r} is not a member of org {org_id!r}") | |
| if ROLE_RANK.get(role, 0) < ROLE_RANK[minimum_role]: | |
| raise PermissionError_( | |
| f"{uid!r} has role {role!r} in org {org_id!r}, needs at least {minimum_role!r}" | |
| ) | |
| return role | |
| def _role_path(org_id: str, key: str) -> str: | |
| return f"orgs/{org_id}/roles/{key}" | |
| def list_roles(org_id: str) -> list[dict]: | |
| """Every role in the org: the built-ins (with any edits) first, then custom ones.""" | |
| prefix = f"orgs/{org_id}/roles/" | |
| stored = {path[len(prefix):]: data for path, data in db.query_prefix(prefix).items()} | |
| roles = [] | |
| for key, default in BUILTIN_ROLES.items(): | |
| data = default if key == ORG_CREATOR_ROLE else stored.get(key, default) | |
| roles.append({"key": key, "name": default["name"], "permissions": list(data["permissions"]), "builtin": True}) | |
| for key, data in sorted(stored.items(), key=lambda kv: kv[1].get("name", "").lower()): | |
| if key not in BUILTIN_ROLES: | |
| roles.append({"key": key, "name": data["name"], "permissions": list(data["permissions"]), "builtin": False}) | |
| return roles | |
| def permissions_for(org_id: str, role: str) -> set[str]: | |
| role = ROLE_ALIASES.get(role, role) | |
| if role == ORG_CREATOR_ROLE: | |
| return set(PERMISSIONS) | |
| stored = db.get_doc(_role_path(org_id, role)) | |
| if stored is not None: | |
| return set(stored["permissions"]) | |
| return set(BUILTIN_ROLES.get(role, {"permissions": []})["permissions"]) | |
| def require_permission(org_id: str, uid: str, permission: str | None = None) -> str: | |
| """Raises PermissionError_ unless uid is a member of org_id whose role allows | |
| `permission` (None: being a member is enough). Returns the actor's role.""" | |
| role = get_role(org_id, uid) | |
| if role is None: | |
| raise PermissionError_(f"{uid!r} is not a member of org {org_id!r}") | |
| if permission is not None and permission not in permissions_for(org_id, role): | |
| raise PermissionError_(f"Your role doesn't allow this ({PERMISSIONS[permission].lower()})") | |
| return role | |
| def _role_exists(org_id: str, role: str) -> bool: | |
| return role in BUILTIN_ROLES or role in ROLE_ALIASES or db.get_doc(_role_path(org_id, role)) is not None | |
| def _check_grant(org_id: str, actor_uid: str, role: str) -> str: | |
| """The checks shared by inviting, adding and changing someone's role. Nobody can hand | |
| out ownership, only an owner can hand out admin, and nobody can hand out more than | |
| they can do themselves.""" | |
| if role == ORG_CREATOR_ROLE: | |
| raise PermissionError_("Ownership can't be granted — only by creating the org") | |
| if not _role_exists(org_id, role): | |
| raise ValueError(f"Unknown role {role!r}") | |
| actor_role = require_permission(org_id, actor_uid, "manage_team") | |
| if ROLE_ALIASES.get(role, role) == "admin" and actor_role != ORG_CREATOR_ROLE: | |
| raise PermissionError_("Only an owner can make someone an admin") | |
| if not permissions_for(org_id, role) <= permissions_for(org_id, actor_role): | |
| raise PermissionError_("You can't give someone more access than you have") | |
| return actor_role | |
| def _check_manage(org_id: str, actor_uid: str, target_uid: str) -> str: | |
| """Checks before changing or removing an existing member. Returns their role.""" | |
| target_role = get_role(org_id, target_uid) | |
| if target_role is None: | |
| raise NotFoundError(f"{target_uid!r} is not a member of org {org_id!r}") | |
| if target_role == ORG_CREATOR_ROLE: | |
| raise PermissionError_("The owner can't be changed or removed") | |
| actor_role = require_permission(org_id, actor_uid, "manage_team") | |
| if target_role == "admin" and actor_role != ORG_CREATOR_ROLE: | |
| raise PermissionError_("Only an owner can change or remove an admin") | |
| return target_role | |
| def add_member(org_id: str, actor_uid: str, target_uid: str, role: str, email: str = "") -> None: | |
| """actor_uid adds target_uid to org_id with the given role (see _check_grant).""" | |
| _check_grant(org_id, actor_uid, role) | |
| doc = {"role": role, "added_at": _now()} | |
| if email: | |
| doc["email"] = email | |
| db.set_doc(_member_path(org_id, target_uid), doc) | |
| _set_reverse_index(target_uid, org_id, role) | |
| def join(org_id: str, uid: str, role: str, email: str = "") -> None: | |
| """Adds uid with an already-approved role (an accepted invite). Never lowers or | |
| replaces the role of someone who is already a member.""" | |
| if get_role(org_id, uid) is not None: | |
| return | |
| doc = {"role": role, "added_at": _now()} | |
| if email: | |
| doc["email"] = email | |
| db.set_doc(_member_path(org_id, uid), doc) | |
| _set_reverse_index(uid, org_id, role) | |
| def change_member_role(org_id: str, actor_uid: str, target_uid: str, role: str) -> None: | |
| _check_manage(org_id, actor_uid, target_uid) | |
| _check_grant(org_id, actor_uid, role) | |
| doc = db.get_doc(_member_path(org_id, target_uid)) | |
| doc["role"] = role | |
| db.set_doc(_member_path(org_id, target_uid), doc) | |
| _set_reverse_index(target_uid, org_id, role) | |
| def remove_member(org_id: str, actor_uid: str, target_uid: str) -> None: | |
| """Someone who can manage the team can remove anyone except the owner; only an | |
| owner can remove an admin.""" | |
| _check_manage(org_id, actor_uid, target_uid) | |
| db.delete_doc(_member_path(org_id, target_uid)) | |
| _clear_reverse_index(target_uid, org_id) | |
| def leave(org_id: str, uid: str) -> None: | |
| """A member leaves on their own. The owner can't leave their own org.""" | |
| role = require_permission(org_id, uid) | |
| if role == ORG_CREATOR_ROLE: | |
| raise PermissionError_("The owner can't leave their own organisation") | |
| db.delete_doc(_member_path(org_id, uid)) | |
| _clear_reverse_index(uid, org_id) | |
| def save_role(org_id: str, actor_uid: str, key: str | None, name: str, permissions: list[str]) -> str: | |
| """Creates a custom role (key None) or edits one, built-ins included except Owner. | |
| Returns the role's key.""" | |
| actor_role = require_permission(org_id, actor_uid, "manage_team") | |
| unknown = [p for p in permissions if p not in PERMISSIONS] | |
| if unknown: | |
| raise ValueError(f"Unknown permission {unknown[0]!r}") | |
| if not set(permissions) <= permissions_for(org_id, actor_role): | |
| raise PermissionError_("You can't give a role more access than you have") | |
| if key == ORG_CREATOR_ROLE: | |
| raise PermissionError_("The Owner role can't be changed") | |
| if key is None: | |
| name = (name or "").strip() | |
| if not name or len(name) > 40: | |
| raise ValueError("Give the role a name of up to 40 characters") | |
| if any(r["name"].lower() == name.lower() for r in list_roles(org_id)): | |
| raise ValueError(f"There is already a role called {name!r}") | |
| key = "custom-" + uuid.uuid4().hex[:10] | |
| elif not _role_exists(org_id, key): | |
| raise NotFoundError(f"No role {key!r}") | |
| if key in BUILTIN_ROLES: | |
| name = BUILTIN_ROLES[key]["name"] | |
| if key == "admin" and actor_role != ORG_CREATOR_ROLE: | |
| raise PermissionError_("Only an owner can change the Admin role") | |
| else: | |
| stored = db.get_doc(_role_path(org_id, key)) | |
| name = (name or "").strip() or (stored or {}).get("name", "") | |
| db.set_doc(_role_path(org_id, key), {"name": name, "permissions": sorted(set(permissions))}) | |
| return key | |
| def delete_role(org_id: str, actor_uid: str, key: str) -> None: | |
| """Removes a custom role nobody holds, or resets an edited built-in to its default.""" | |
| require_permission(org_id, actor_uid, "manage_team") | |
| if key == ORG_CREATOR_ROLE: | |
| raise PermissionError_("The Owner role can't be changed") | |
| if key not in BUILTIN_ROLES: | |
| if any(m["role"] == key for m in list_members(org_id).values()): | |
| raise ValueError("Move everyone off this role before deleting it") | |
| db.delete_doc(_role_path(org_id, key)) | |
| def list_members(org_id: str) -> dict[str, dict]: | |
| """{uid: {role, added_at}} for every member of org_id.""" | |
| prefix = _members_prefix(org_id) | |
| docs = db.query_prefix(prefix) | |
| return {path[len(prefix):]: data for path, data in docs.items()} | |
| def list_orgs_for_user(uid: str) -> dict[str, str]: | |
| """{org_id: role} for every org this uid belongs to, across all their orgs.""" | |
| doc = db.get_doc(_user_orgs_path(uid)) | |
| return dict(doc["org_ids"]) if doc else {} | |