"""§22 S4 (C.2) — scope-role grant operations over the `memberships` table. The membership *gates* (who may invite, who holds which role) live in `auth.py`; this module holds the *mutations* the invitation surface drives — granting, the "broader scope supersedes narrower" cleanup, listing, and revocation — mirroring how `invites.py` owns the create/claim/list of per-user invite tokens while the gate (`auth.can_invite_to_rfc`) lives in `auth.py`. The model (Part B / S3): a `memberships` row is `(scope_type ∈ {global, project, collection}, scope_id, user_id, role ∈ {owner, contributor})`, unique per `(scope_type, scope_id, user_id)`. A grant is a direct write of that row (the C.2 scenarios write the row immediately and §15-notify an existing account — there is no accept round-trip; inviting a not-yet-account email is out of S4 scope and handled by the admin-create-invite path). The "broader scope supersedes narrower" rule (C.2.6): granting a role at a broader scope removes this user's narrower rows that the new grant *subsumes* — a narrower row whose role is no more permissive than the new one. A narrower row that is *more* permissive is kept (no negative override: a child Owner grant survives a parent Contributor grant, and the §B.2 resolver still unions most-permissively). """ from __future__ import annotations from typing import Any from . import collections as collections_mod from . import db # Higher rank = more permissive. Used by the supersede rule: a narrower grant # is pruned only when its rank ≤ the new broader grant's rank. _ROLE_RANK = {"contributor": 1, "owner": 2} VALID_SCOPE_TYPES = ("global", "project", "collection") VALID_ROLES = ("owner", "contributor") GLOBAL_SCOPE_ID = "*" def user_by_email(email: str) -> dict[str, Any] | None: """The `users` row (id, display_name, email, permission_state) for an email, case-insensitively, or None. The grantee must already be an account — S4 grants a scope role to an existing user, it does not provision one.""" row = db.conn().execute( "SELECT id, display_name, email, permission_state, role " "FROM users WHERE lower(email) = lower(?) " "ORDER BY id LIMIT 1", (email.strip(),), ).fetchone() return dict(row) if row else None def grant( *, scope_type: str, scope_id: str, user_id: int, role: str, granted_by: int | None, ) -> None: """Write (or update) the membership row, then apply the C.2.6 broader-scope-supersedes cleanup. Idempotent on `(scope_type, scope_id, user_id)` — re-granting at the same scope updates the role and the grantor. The grant is recorded regardless of the grantee's deployment `permission_state`: a `pending` account's row is written (C.2.7), but the §6 admission floor in `auth.effective_scope_role` keeps it conferring no write until the account is granted at the deployment.""" db.conn().execute( "INSERT INTO memberships (scope_type, scope_id, user_id, role, granted_by) " "VALUES (?, ?, ?, ?, ?) " "ON CONFLICT (scope_type, scope_id, user_id) " "DO UPDATE SET role = excluded.role, granted_by = excluded.granted_by, " "granted_at = datetime('now')", (scope_type, scope_id, user_id, role, granted_by), ) _prune_subsumed(scope_type=scope_type, scope_id=scope_id, user_id=user_id, role=role) def _prune_subsumed(*, scope_type: str, scope_id: str, user_id: int, role: str) -> None: """Remove this user's narrower rows that the just-written broader grant subsumes (same-or-lower role rank within the broader scope's subtree). A collection grant subsumes nothing narrower (the per-entry tier is separate); a project grant subsumes its collections; a global grant subsumes every project and collection.""" rank = _ROLE_RANK[role] keep_ranks = [r for r, v in _ROLE_RANK.items() if v <= rank] if not keep_ranks: return placeholders = ",".join("?" for _ in keep_ranks) if scope_type == "project": # Narrower = collection-scope rows for collections in this project. db.conn().execute( f"DELETE FROM memberships " f"WHERE user_id = ? AND scope_type = 'collection' " f" AND role IN ({placeholders}) " f" AND scope_id IN (SELECT id FROM collections WHERE project_id = ?)", (user_id, *keep_ranks, scope_id), ) elif scope_type == "global": # Narrower = every project- and collection-scope row for this user. db.conn().execute( f"DELETE FROM memberships " f"WHERE user_id = ? AND scope_type IN ('project', 'collection') " f" AND role IN ({placeholders})", (user_id, *keep_ranks), ) def revoke(*, scope_type: str, scope_id: str, user_id: int) -> bool: """Remove a membership row at exactly this scope. Returns True if a row was removed. Revocation is scope-exact: it does not cascade to broader or narrower grants (each is its own administrative act).""" cur = db.conn().execute( "DELETE FROM memberships WHERE scope_type = ? AND scope_id = ? AND user_id = ?", (scope_type, scope_id, user_id), ) return cur.rowcount > 0 def list_for_project(project_id: str) -> list[dict[str, Any]]: """Every project-scope grant on this project plus every collection-scope grant on its collections, joined to the grantee's display fields — the data behind the project-Owner membership panel. Ordered project grants first, then by collection, then by role (Owner before Contributor).""" rows = db.conn().execute( """ SELECT m.scope_type, m.scope_id, m.user_id, m.role, m.granted_at, u.display_name, u.email, u.permission_state, c.name AS collection_name FROM memberships m JOIN users u ON u.id = m.user_id LEFT JOIN collections c ON c.id = m.scope_id AND m.scope_type = 'collection' WHERE (m.scope_type = 'project' AND m.scope_id = ?) OR (m.scope_type = 'collection' AND m.scope_id IN (SELECT id FROM collections WHERE project_id = ?)) ORDER BY (m.scope_type != 'project'), m.scope_id, CASE m.role WHEN 'owner' THEN 0 ELSE 1 END, u.display_name """, (project_id, project_id), ).fetchall() return [dict(r) for r in rows]