"""Gitea OAuth and user provisioning. OAuth identity is the basis for the app's user account per §18; the §6 authorization layer is built on top by reading from the users table. On first sign-in we insert a row with role='contributor' (or 'owner' if the gitea_login matches the configured OWNER_GITEA_LOGIN — bootstrapping for owner zero per §6.1). On subsequent sign-ins we refresh the display name and avatar from Gitea so a rename in Gitea propagates here. """ from __future__ import annotations import secrets from dataclasses import dataclass from typing import Any import httpx from fastapi import HTTPException, Request from . import collections as collections_mod from . import db from .bot import Actor from .config import Config from .projects import DEFAULT_PROJECT_ID @dataclass class SessionUser: user_id: int gitea_id: int gitea_login: str display_name: str email: str avatar_url: str role: str # v0.8.0 / §6.1 — admission gate. Three states: 'pending' (waiting # for an admin grant), 'granted' (active contributor), 'revoked' # (was granted, later removed). Existing rows at migration time # default to 'granted' so grandfathered users are unaffected; OTC # provisions fresh users with 'pending' (see `app/otc.py`). permission_state: str = "granted" def as_actor(self) -> Actor: return Actor( user_id=self.user_id, gitea_login=self.gitea_login, display_name=self.display_name, email=self.email, ) def authorization_url(config: Config, state: str) -> str: return ( f"{config.gitea_url}/login/oauth/authorize" f"?client_id={config.oauth_client_id}" f"&redirect_uri={config.redirect_uri}" f"&response_type=code" f"&state={state}" ) async def exchange_code(config: Config, code: str) -> dict[str, Any]: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{config.gitea_url}/login/oauth/access_token", json={ "client_id": config.oauth_client_id, "client_secret": config.oauth_client_secret, "code": code, "grant_type": "authorization_code", "redirect_uri": config.redirect_uri, }, headers={"Accept": "application/json"}, ) resp.raise_for_status() return resp.json() async def fetch_user_profile(config: Config, access_token: str) -> dict[str, Any]: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get( f"{config.gitea_url}/api/v1/user", headers={"Authorization": f"token {access_token}"}, ) resp.raise_for_status() return resp.json() def allowlist_is_active() -> bool: """The private-beta gate is on iff the `allowed_emails` table has any rows. Empty list means "open" — any successful OAuth provisions a user; first row added flips the deployment into private-beta mode. See `migrations/011_allowlist.sql` for the reasoning. """ row = db.conn().execute("SELECT 1 FROM allowed_emails LIMIT 1").fetchone() return row is not None def is_allowed_sign_in(profile: dict[str, Any]) -> bool: """Decide whether a freshly-completed OAuth profile may sign in. v0.8.0 (item #6) replaces the allowlist gate with an admin-grant flow at the OTC `/request` surface, but the Gitea OAuth callback in `main.py` still consults this helper so the fallback path keeps the v0.3.0 admission shape during the OAuth migration window. The eventual removal of the OAuth callback (§19.2) retires this function alongside it. Three accept paths: 1. The allowlist is empty (gate off). 2. The Gitea profile's email is in `allowed_emails` (case-insensitive). 3. A `users` row already exists for this `gitea_id` — grandfather per `migrations/011_allowlist.sql`. """ gitea_id = profile.get("id") if gitea_id is not None: existing = db.conn().execute( "SELECT 1 FROM users WHERE gitea_id = ? LIMIT 1", (gitea_id,) ).fetchone() if existing is not None: return True if not allowlist_is_active(): return True email = (profile.get("email") or "").strip() if not email: return False row = db.conn().execute( "SELECT 1 FROM allowed_emails WHERE email = ? LIMIT 1", (email,) ).fetchone() return row is not None def provision_user(config: Config, profile: dict[str, Any]) -> SessionUser: """Insert or update the users row for this Gitea profile. Owner zero (§6.1) is whoever's gitea_login matches OWNER_GITEA_LOGIN. The owner role is granted on first sign-in and never revoked from a later config change — once owner, always owner until an explicit role transition (which lives in §6.1 and isn't part of slice 1). """ gitea_id = profile["id"] login = profile["login"] display = profile.get("full_name") or login email = profile.get("email") or "" avatar = profile.get("avatar_url") or "" c = db.conn() existing = c.execute("SELECT * FROM users WHERE gitea_id = ?", (gitea_id,)).fetchone() if existing is None: role = "owner" if config.owner_gitea_login and login == config.owner_gitea_login else "contributor" # v0.8.0: a fresh OAuth-provisioned user is also subject to # the admin-grant flow. The OAuth fallback only fires for # users who pass `is_allowed_sign_in` (so they're already on # the legacy allowlist or are grandfathered by gitea_id); # 'granted' is the right default here since the allowlist # check is itself the admin gesture. A future release that # retires the OAuth callback (§19.2) collapses both paths # under the same gate. cur = c.execute( """ INSERT INTO users (gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state) VALUES (?, ?, ?, ?, ?, ?, 'granted') """, (gitea_id, login, email, display, avatar, role), ) user_id = cur.lastrowid permission_state = "granted" else: user_id = existing["id"] role = existing["role"] permission_state = existing["permission_state"] or "granted" c.execute( """ UPDATE users SET gitea_login = ?, email = ?, display_name = ?, avatar_url = ?, last_seen_at = datetime('now') WHERE id = ? """, (login, email, display, avatar, user_id), ) return SessionUser( user_id=user_id, gitea_id=gitea_id, gitea_login=login, display_name=display, email=email, avatar_url=avatar, role=role, permission_state=permission_state, ) # ----- Session helpers ----- SESSION_USER_KEY = "user" SESSION_STATE_KEY = "oauth_state" def store_session(request: Request, user: SessionUser) -> None: request.session[SESSION_USER_KEY] = { "user_id": user.user_id, "gitea_id": user.gitea_id, "gitea_login": user.gitea_login, "display_name": user.display_name, "email": user.email, "avatar_url": user.avatar_url, "role": user.role, # v0.8.0: persist the admission state on the cookie payload so # the post-cookie audit doesn't second-guess the row. The DB # is re-read on every `current_user` call regardless (so an # admin grant takes effect on the next request); this field # is purely structural redundancy for the cookie shape. "permission_state": user.permission_state, } def current_user(request: Request) -> SessionUser | None: raw = request.session.get(SESSION_USER_KEY) if not raw: return None # Re-read the role from the database every request so role changes # take effect on the next API call without forcing a logout. row = db.conn().execute( "SELECT id, gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state FROM users WHERE id = ?", (raw["user_id"],), ).fetchone() if row is None: return None # v0.7.0: OTC-provisioned users have NULL gitea_id / gitea_login. # Coerce nulls to the SessionUser's typed defaults so downstream # code (Actor, _on_behalf_trailer) reads a stable shape regardless # of which sign-in path the row came from. The DB remains the # source of truth for "is this an OAuth-linked user" (gitea_id IS # NOT NULL); the in-memory SessionUser is the per-request handle. # v0.8.0: permission_state comes off the row directly. A NULL # column value (shouldn't happen under the migration's # NOT NULL DEFAULT, but be defensive) reads as 'granted' so the # gate fails open for grandfathered surfaces rather than locking # everyone out on a malformed row. return SessionUser( user_id=row["id"], gitea_id=row["gitea_id"] or 0, gitea_login=row["gitea_login"] or "", display_name=row["display_name"], email=row["email"] or "", avatar_url=row["avatar_url"] or "", role=row["role"], permission_state=row["permission_state"] or "granted", ) def require_user(request: Request) -> SessionUser: user = current_user(request) if user is None: raise HTTPException(status_code=401, detail="Not authenticated") return user def require_contributor(request: Request) -> SessionUser: """§6.1: authenticated, not write-muted, and granted by an admin. v0.8.0 (item #6) widens this gate. A fresh OTC sign-in lands in `permission_state='pending'`; the user can read everything an anonymous viewer can read, but every write-shaped endpoint that funnels through this dependency now refuses with 403 until an admin grants them. The `pending` blast radius is the same as anonymous (item #4 / v0.6.0 already audited the anon-write refusal at every write site), so this widening is structurally a relabel — the same surfaces that already refused 401 to anonymous now also refuse 403 to pending. """ user = require_user(request) row = db.conn().execute("SELECT muted FROM users WHERE id = ?", (user.user_id,)).fetchone() if row and row["muted"]: raise HTTPException(status_code=403, detail="Your account is muted") if user.permission_state != "granted": # 'pending' is the post-OTC waiting state; 'revoked' is the # admin-undid-the-grant state. Both refuse with the same 403 # shape; the client distinguishes via `/api/auth/me` which # carries `permission_state` in the response. raise HTTPException( status_code=403, detail="Your beta access request is in review", ) return user def require_admin(request: Request) -> SessionUser: """§6.1: owner or admin.""" user = require_user(request) if user.role not in ("owner", "admin"): raise HTTPException(status_code=403, detail="Admin or owner role required") return user # =========================================================================== # §22.6 / §22.7 — project-scoped authorization (the multi-project middle tier). # # A deployment hosts N projects (§22). Authorization for an action on an RFC is # the *most-permissive union* of three tiers — the actor's deployment role # (§6.1), their project role (§22.6), and their per-RFC authority (§6.3/§12) — # with the §22.5 visibility gate and the §6.2 write-mute *subtractive* on top # (§22.7). The per-RFC capability helpers below (`can_discuss_rfc`, # `can_contribute_to_rfc`, `can_invite_to_rfc`) compose all three tiers, so the # ~20 endpoint call sites inherit multi-project behavior unchanged. # # Through Slice M2 the only project is the migration-seeded `default` one (the # N=1 case, §22.13); a project is resolved from an RFC slug via # `cached_rfcs.project_id` (unique per slug while N=1). M3's registry mirror # lets a deployment declare a second project; these gates already hold then. # # OPERATOR DECISIONS (M2): # * implicit-on-public — on a `public` project a granted deployment # `contributor` keeps the pre-multi-project write *baseline* (propose # freely; an owned RFC's discuss/contribute is still gated by the v0.16.0 # per-RFC invite). No project_members row is needed and no backfill runs, # so the N=1 case stays whole. Explicit project_members rows and # gated/unlisted visibility are where the new tier actually bites. # * preserve curation — the implicit-public baseline does NOT override per-RFC # owner curation; only an *explicit* project_contributor/project_admin grant # (or a deployment owner/admin) bypasses it. So §22.7's "project_contributor # ⊇ rfc_collaborators(contributor)" holds for explicit grants, while a plain # granted contributor on public behaves exactly as it did before M2. # =========================================================================== _DEPLOYMENT_SUPERUSER_ROLES = ("owner", "admin") def project_visibility(project_id: str) -> str: """The project's §22.5 visibility ('gated' | 'public' | 'unlisted'). A missing row reads as 'gated' — the safe default: an unknown project is invisible rather than open.""" row = db.conn().execute( "SELECT visibility FROM projects WHERE id = ?", (project_id,) ).fetchone() if row is None: return "gated" return row["visibility"] or "gated" def _is_default_project(project_id: str) -> bool: """True iff `project_id` owns the migration-seeded `default` collection — the deployment's primary project (§22.13), whatever its configured id. Only there do M2's role rows (which migrated to collection scope `default`) stand in for project-level authority.""" return collections_mod.project_of_collection(collections_mod.DEFAULT_COLLECTION_ID) == project_id def project_member_role(user: SessionUser | None, project_id: str) -> str | None: """The user's *project-grain* §22.6 role at this project, or None — the most-permissive of a **global** grant (inherits down to every project) and a **project**-scope grant. Mapped back to the legacy `project_admin`/`project_contributor` strings the project-grain authz speaks. Back-compat: on the deployment's *default* project only, M2's rows live at collection scope `default` (§B.3 migration), so a `default` collection-scope grant there is read as project-level too. A collection grant on any other project is NOT project authority — that is the four-layer collection resolver (`effective_scope_role`). Does not fold in the deployment tier (`is_project_superuser` adds it) or the implicit-on-public baseline.""" if user is None: return None clauses = ["scope_type = 'global'", "(scope_type = 'project' AND scope_id = ?)"] params: list = [user.user_id, project_id] if _is_default_project(project_id): clauses.append("(scope_type = 'collection' AND scope_id = ?)") params.append(collections_mod.DEFAULT_COLLECTION_ID) row = db.conn().execute( "SELECT role FROM memberships WHERE user_id = ? AND (" + " OR ".join(clauses) + ") " "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1", params, ).fetchone() if row is None: return None return "project_admin" if row["role"] == "owner" else "project_contributor" def project_of_rfc(rfc_slug: str) -> str: """The project an RFC belongs to, via its collection (`cached_rfcs.collection_id` -> `collections.project_id`, §22 three-tier). Falls back to the default project when the slug isn't cached — the same N=1 default migration 026 backfills.""" row = db.conn().execute( "SELECT c.project_id AS project_id " "FROM cached_rfcs r JOIN collections c ON c.id = r.collection_id " "WHERE r.slug = ?", (rfc_slug,), ).fetchone() if row is None: return DEFAULT_PROJECT_ID return row["project_id"] or DEFAULT_PROJECT_ID def collection_of_rfc(rfc_slug: str) -> str: """The collection an RFC belongs to (`cached_rfcs.collection_id`). Falls back to the default collection when the slug isn't cached. Mirrors `project_of_rfc`'s first-match semantics; a slug shared across collections is a known routing ambiguity (the RFC-grain helpers take a bare slug) resolved by the collection-qualified routes in later slices.""" row = db.conn().execute( "SELECT collection_id FROM cached_rfcs WHERE slug = ?", (rfc_slug,) ).fetchone() if row is None or not row["collection_id"]: return collections_mod.DEFAULT_COLLECTION_ID return row["collection_id"] def is_project_superuser(user: SessionUser | None, project_id: str) -> bool: """Maximal authority within a project: a deployment owner/admin (superuser in every project, §22.7) or an explicit `project_admin` (§22.6). Both subsume the per-RFC owners/arbiters tier.""" if user is None: return False if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return True return project_member_role(user, project_id) == "project_admin" def can_read_project(user: SessionUser | None, project_id: str) -> bool: """The §22.5 visibility gate at the project grain. `public`/`unlisted` are readable by anyone (anonymous included — `unlisted` is link-only but the link still reads); `gated` is readable only by a deployment owner/admin or a holder of any scope grant reaching the project — a global grant, a project grant, or membership at *any* collection within it (seeing a collection implies seeing its project). Used as the subtractive read gate (a gated project's entries 404 to non-members).""" vis = project_visibility(project_id) if vis in ("public", "unlisted"): return True # gated — scope-role holders + superusers only, subject to the §6 floor. if user is None or user.permission_state != "granted": return False if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return True row = db.conn().execute( "SELECT 1 FROM memberships m WHERE m.user_id = ? AND (" " m.scope_type = 'global'" " OR (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 = ?))) LIMIT 1", (user.user_id, project_id, project_id), ).fetchone() return row is not None def require_project_readable(user: SessionUser | None, project_id: str) -> None: """Raise 404 when the project is not readable by this viewer (§22.5: a gated project is invisible to non-members — indistinguishable from absent, so the shape matches an unknown slug).""" if not can_read_project(user, project_id): raise HTTPException(status_code=404, detail="RFC not found") def _has_write_baseline(user: SessionUser | None, project_id: str) -> bool: """The implicit-on-public write baseline: a granted deployment `contributor` on a `public` project carries the pre-multi-project write standing (still subject to per-RFC curation). Deployment owner/admin are handled by `is_project_superuser`; on gated/unlisted a contributor has no baseline and needs an explicit project role.""" if user is None or user.permission_state != "granted": return False return user.role == "contributor" and project_visibility(project_id) == "public" def can_contribute_in_project(user: SessionUser | None, project_id: str) -> bool: """May the user contribute *new* content to the project (propose an entry) — the project-level (not RFC-specific) contribute standing. The union of the override grants (superuser / explicit project_contributor) and the implicit-public baseline, subject to the visibility gate.""" if user is None or user.permission_state != "granted": return False if not can_read_project(user, project_id): return False if is_project_superuser(user, project_id): return True if project_member_role(user, project_id) == "project_contributor": return True return _has_write_baseline(user, project_id) def can_discuss_in_project(user: SessionUser | None, project_id: str) -> bool: """May the user participate in discussion in the project at all — the project-level discuss standing (project_viewer ⊇ discussant, §22.7). A superset of `can_contribute_in_project` (a contributor can discuss).""" if user is None or user.permission_state != "granted": return False if not can_read_project(user, project_id): return False if is_project_superuser(user, project_id): return True if project_member_role(user, project_id) in ("project_viewer", "project_contributor"): return True return _has_write_baseline(user, project_id) def visible_project_ids(user: SessionUser | None) -> list[str]: """Project ids whose entries may surface in a listing for this viewer — the §22.5 read gate applied to the catalog/idea lists. (The directory's `unlisted`-omission and the per-project routing are M3 concerns; for the M2 listing filter we include every project the viewer can read.)""" rows = db.conn().execute("SELECT id FROM projects").fetchall() return [r["id"] for r in rows if can_read_project(user, r["id"])] # =========================================================================== # §22 three-tier — S3. The four-layer scope-role resolver (§B.2) and the # collection-grain visibility gate. # # A grant attaches the unified role {owner, contributor} at a scope: global, # project, or collection (§B.1). Grants inherit downward, are additive, and # admit no negative override (§B.2). Effective authority over a *collection* is # the most-permissive union of the layers reaching it: # # global (users.role owner/admin ∪ memberships scope_type='global') # ∪ project (memberships scope_type='project' at the collection's project) # ∪ collection (memberships scope_type='collection' at the collection) # # minus the §22.5 visibility gate and §6.2 write-mute (subtractive, as today). # Per-entry authority (owners / arbiters / rfc_collaborators) is a distinct, # finer layer the RFC-grain helpers union in beneath collection. # # OPERATOR DECISIONS (S3, session 0076): # * scope-role-primary — a plain granted account (users.role 'contributor', no # membership) is a granted *account*, not a write-everywhere global role. # Write standing comes from an explicit scope grant; the lone exception is the # grandfathered implicit-public baseline below. # * grandfathered baseline — the migration-seeded `default` collection keeps the # pre-three-tier implicit-on-public write baseline (a granted deployment # contributor may propose while it is public), so the N=1 deployment (§22.13) # loses no capability. Every *explicitly-created* collection requires an # explicit scope grant to write. # * hidden-from-public — a `gated` collection is invisible to the public (404, # omitted from the directory) yet visible to any scope-role holder reaching it # (collection/project/global). A collection's visibility may be set only as # strict or stricter than its project's (the rank ordering below). # =========================================================================== # The global scope is a single tier per deployment; its grant rows use this # sentinel scope_id (migration 030). GLOBAL_SCOPE_ID = "*" # §22.5 visibility strictness on the public-exposure axis: `public` is least # strict, `gated` most strict. A collection may narrow its project's visibility # but never widen it (`rank(collection) >= rank(project)`). _VISIBILITY_RANK = {"public": 0, "unlisted": 1, "gated": 2} def visibility_rank(visibility: str | None) -> int: """The strictness rank of a §22.5 visibility (higher = stricter). An unknown value reads as the strictest (`gated`) — the safe default.""" return _VISIBILITY_RANK.get(visibility or "", _VISIBILITY_RANK["gated"]) def effective_scope_role(user: SessionUser | None, collection_id: str) -> str | None: """The most-permissive unified role ({'owner','contributor'}) the user holds over `collection_id`, folding §B.2's global → project → collection layers. Returns None when no scope grant reaches the collection. 'owner' outranks 'contributor'; there is no negative override (a parent grant is never subtracted by a child). Subject to the §6 admission floor.""" if user is None or user.permission_state != "granted": return None # Global tier — a deployment owner/admin is a global Owner (§B.1). if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return "owner" pid = collections_mod.project_of_collection(collection_id) row = db.conn().execute( "SELECT role FROM memberships " "WHERE user_id = ? AND (" " scope_type = 'global'" " OR (scope_type = 'project' AND scope_id = ?)" " OR (scope_type = 'collection' AND scope_id = ?)) " "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1", (user.user_id, pid, collection_id), ).fetchone() return row["role"] if row else None def _effective_project_role(user: SessionUser | None, project_id: str) -> str | None: """The most-permissive role the user holds *over a project* — folding the global tier (deployment owner/admin, or a `scope_type='global'` grant) and a `scope_type='project'` grant on this project. Unlike `effective_scope_role` (which keys on a collection), this answers the project grain directly, for the §22.8 request-to-join membership check. Subject to the §6 admission floor.""" if user is None or user.permission_state != "granted": return None if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return "owner" row = db.conn().execute( "SELECT role FROM memberships " "WHERE user_id = ? AND (" " scope_type = 'global'" " OR (scope_type = 'project' AND scope_id = ?)) " "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1", (user.user_id, project_id), ).fetchone() return row["role"] if row else None def effective_role_at_scope( user: SessionUser | None, scope_type: str, scope_id: str ) -> str | None: """The most-permissive scope role the user holds over a `(scope_type, scope_id)` target — the scope-grain twin of `effective_scope_role`. A `collection` target folds global → project → collection (the existing resolver); a `project` target folds global → project. Returns None when no grant reaches the scope. Drives the §22.8 "already a member?" gate.""" if scope_type == "collection": return effective_scope_role(user, scope_id) if scope_type == "project": return _effective_project_role(user, scope_id) return None def collection_visibility(collection_id: str) -> str: """The collection's own §22.5 visibility. A missing row reads as 'gated' — an unknown collection is invisible rather than open.""" row = db.conn().execute( "SELECT visibility FROM collections WHERE id = ?", (collection_id,) ).fetchone() if row is None or not row["visibility"]: return "gated" return row["visibility"] def effective_collection_visibility(collection_id: str) -> str: """The stricter of the collection's own visibility and its project's (§22.5 'both gates'). A collection is constrained to be ≥ its project in strictness, but we max() defensively so a misconfigured looser collection can never widen its project's gate.""" cvis = collection_visibility(collection_id) pid = collections_mod.project_of_collection(collection_id) pvis = project_visibility(pid) if pid else "gated" return cvis if visibility_rank(cvis) >= visibility_rank(pvis) else pvis def can_read_collection(user: SessionUser | None, collection_id: str) -> bool: """§22.5 read/existence gate at the collection grain. `public`/`unlisted` read by anyone (anonymous included — `unlisted` is link-only but the link reads); `gated` ("hidden from public existence") reads only for a scope-role holder over the collection (collection/project/global) or a deployment owner/admin. The subtractive read gate — a gated collection 404s a non-holder, indistinguishable from absent.""" vis = effective_collection_visibility(collection_id) if vis in ("public", "unlisted"): return True if user is None or user.permission_state != "granted": return False return effective_scope_role(user, collection_id) is not None def require_collection_readable(user: SessionUser | None, collection_id: str) -> None: """Raise 404 when the collection is not readable by this viewer (§22.5: a hidden/gated collection is invisible to non-holders — the shape matches an unknown collection).""" if not can_read_collection(user, collection_id): raise HTTPException(status_code=404, detail="Not found") def is_collection_superuser(user: SessionUser | None, collection_id: str) -> bool: """Maximal authority over a collection: an effective scope role of 'owner' reaching it (a collection Owner, a project Owner of its project, a global Owner, or a deployment owner/admin). Subsumes the per-entry owners/arbiters tier within the collection.""" return effective_scope_role(user, collection_id) == "owner" def _has_collection_write_baseline(user: SessionUser | None, collection_id: str) -> bool: """The grandfathered implicit-on-public write baseline, narrowed to the migration-seeded `default` collection (§22.13 N=1 case). A granted deployment `contributor` keeps its pre-three-tier write standing on the default collection while its effective visibility is public; every explicitly-created collection requires an explicit scope grant (S3 operator decision).""" if user is None or user.permission_state != "granted": return False if collection_id != collections_mod.DEFAULT_COLLECTION_ID: return False return user.role == "contributor" and effective_collection_visibility(collection_id) == "public" def can_contribute_in_collection(user: SessionUser | None, collection_id: str) -> bool: """May the user contribute *new* content to the collection (propose an entry) — the collection-level contribute standing. The union of the scope-role grant (owner/contributor reaching the collection) and the grandfathered default baseline, subject to the visibility read gate.""" if user is None or user.permission_state != "granted": return False if not can_read_collection(user, collection_id): return False if effective_scope_role(user, collection_id) is not None: return True return _has_collection_write_baseline(user, collection_id) def can_discuss_in_collection(user: SessionUser | None, collection_id: str) -> bool: """May the user participate in discussion in the collection — the collection-level discuss standing. A superset of contribute for this pass (the read-only viewer tier is deferred, §B.3), so it mirrors `can_contribute_in_collection`.""" return can_contribute_in_collection(user, collection_id) def can_create_collection(user: SessionUser | None, project_id: str) -> bool: """May the user create a new collection in this project (§B.1)? Creating a collection is a *project-level* action: a deployment owner/admin, or any holder of a project-scope or global-scope grant (Owner OR RFC Contributor — 'anyone at the project level with permission to create a collection'). A *collection*-scope grant cannot create sibling collections.""" if user is None or user.permission_state != "granted": return False if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return True row = db.conn().execute( "SELECT 1 FROM memberships " "WHERE user_id = ? AND (" " scope_type = 'global'" " OR (scope_type = 'project' AND scope_id = ?)) LIMIT 1", (user.user_id, project_id), ).fetchone() return row is not None def can_create_project(user: SessionUser | None) -> bool: """§22 S5 (§A.2 / §B.1): may the user create a new project? "+ New project" is a **global-Owner** action — a deployment owner/admin (a global Owner per §B.1) or a holder of an explicit `scope_type='global'` Owner grant. Creating a project is deployment-level, so it is not reachable by a project- or collection-scope grant nor by a global RFC Contributor (that role creates collections, not projects). Subject to the §6 admission floor.""" if user is None or user.permission_state != "granted": return False if user.role in _DEPLOYMENT_SUPERUSER_ROLES: return True row = db.conn().execute( "SELECT 1 FROM memberships " "WHERE user_id = ? AND scope_type = 'global' AND role = 'owner' LIMIT 1", (user.user_id,), ).fetchone() return row is not None def can_invite_at_project(user: SessionUser | None, project_id: str) -> bool: """§22 S4 (C.2): may the user grant scope roles at this project (or at any collection within it)? Managing membership is an *Owner* capability whose reach covers the project — a deployment owner/admin, a global Owner, or this project's Owner. An RFC Contributor does not manage membership (C.2.4); a collection Owner's reach is its own collection only (C.2.3), so it is not offered project-scope invites. Identical to `is_project_superuser` — the invite gate IS "is an Owner over this project".""" return is_project_superuser(user, project_id) def can_invite_at_collection(user: SessionUser | None, collection_id: str) -> bool: """§22 S4 (C.2): may the user grant scope roles at this collection? An Owner whose reach covers it — the collection's Owner, its project's Owner, a global Owner, or a deployment owner/admin (`is_collection_superuser`). This is the narrowest invite reach; a collection Owner who is nothing more may invite here but not at the project or globally (C.2.3).""" return is_collection_superuser(user, collection_id) # v0.16.0 (roadmap item #12): per-RFC membership helpers. # # These don't replace `require_contributor` — they layer on top of it for # endpoints that an RFC's owner can selectively open up. The "discussion" # and "PR" write surfaces consult `is_rfc_writer(...)` / `is_rfc_discussant(...)` # to admit users who are either platform-privileged (admin, RFC owner) # OR who hold an explicit invitation-accepted per-RFC role. # # The platform gate still fires first: a user whose # `permission_state != 'granted'` cannot write anywhere, invitation or # not. v0.16.0 doesn't loosen that — a per-RFC invitation is additive # *within* the granted-platform-user population. (Accepting an # invitation as a pending user surfaces in the admin-page hook per # the roadmap text; the platform grant remains the admin's decision.) def _rfc_owners_set(rfc_slug: str) -> set[str]: """The gitea_logins named in the RFC's frontmatter owners array. Read from `cached_rfcs.owners_json`. Returns an empty set if the RFC isn't cached (the caller's earlier `_require_rfc_readable` will already have rejected that case in practice). """ import json as _json row = db.conn().execute( "SELECT owners_json FROM cached_rfcs WHERE slug = ?", (rfc_slug,), ).fetchone() if row is None: return set() try: return set(_json.loads(row["owners_json"] or "[]")) except Exception: return set() def is_rfc_owner(user: SessionUser | None, rfc_slug: str) -> bool: """True iff the user is named in the RFC's frontmatter `owners` list. The platform-level admin/owner check is separate; per §6.1 an app admin/owner has all per-RFC capabilities by construction, but this predicate is intentionally narrow — it answers "is this person on the RFC's owners line?" and nothing more. """ if user is None: return False return user.gitea_login in _rfc_owners_set(rfc_slug) def is_rfc_collaborator(user: SessionUser | None, rfc_slug: str, *, role_in_rfc: str | None = None) -> bool: """True iff the user has an accepted per-RFC collaborator row. `role_in_rfc`: * None — any role qualifies (the discussion-write check uses this shape: contributor strictly includes discussant). * 'contributor' — only the contributor role qualifies (the PR-write check uses this shape). * 'discussant' — only the discussant role qualifies (not used by v0.16.0 endpoints; included for symmetry). """ if user is None: return False if role_in_rfc is None: row = db.conn().execute( "SELECT 1 FROM rfc_collaborators WHERE rfc_slug = ? AND user_id = ? LIMIT 1", (rfc_slug, user.user_id), ).fetchone() return row is not None row = db.conn().execute( "SELECT 1 FROM rfc_collaborators WHERE rfc_slug = ? AND user_id = ? AND role_in_rfc = ? LIMIT 1", (rfc_slug, user.user_id, role_in_rfc), ).fetchone() return row is not None def can_discuss_rfc(user: SessionUser | None, rfc_slug: str) -> bool: """v0.16.0 — admit to PR-less discussion writes on this RFC. True if ANY of: * platform admin/owner (the §6.1 maximal-capability path), * the RFC has no frontmatter owners yet (the gate is open until an owner exists to set it — relevant for super-drafts pre-§13.1 claim), * RFC owner (frontmatter `owners` membership), * accepted per-RFC collaborator at any role (contributor strictly includes discussant). Returns False for anonymous viewers and for users whose `permission_state != 'granted'` — the platform-level gate must hold before any per-RFC layer can apply. The platform gate is also enforced earlier in the request via `require_contributor`; the helper here is defensive so callers that compose it with `current_user` directly still respect the gate. """ if user is None: return False if user.permission_state != "granted": return False cid = collection_of_rfc(rfc_slug) # §22.5 visibility gate is subtractive (§22.7) — no capability in a # collection the viewer cannot even read. if not can_read_collection(user, cid): return False # §B.2 union, scope-role grants first — these bypass per-RFC curation (a # collection/project/global Owner or RFC Contributor ⊇ discussant). if effective_scope_role(user, cid) is not None: return True # per-RFC authority (union term). owners = _rfc_owners_set(rfc_slug) if user.gitea_login in owners: return True if is_rfc_collaborator(user, rfc_slug, role_in_rfc=None): return True # grandfathered implicit-public baseline (curation preserved): on the default # collection a granted deployment contributor may discuss only while the RFC # is unclaimed. The first §13.1 claim engages the per-RFC gate, mirroring the # pre-multi-project v0.16.0 contract. if not owners and _has_collection_write_baseline(user, cid): return True return False def can_contribute_to_rfc(user: SessionUser | None, rfc_slug: str) -> bool: """v0.16.0 — admit to PR-shaped writes on this RFC. True if ANY of: * platform admin/owner, * the RFC has no frontmatter owners yet (gate open until an owner exists), * RFC owner, * accepted per-RFC collaborator at role 'contributor' (a 'discussant' row is NOT sufficient — PRs are the higher-privilege surface). Same `permission_state` and anonymous-viewer refusals as `can_discuss_rfc`. """ if user is None: return False if user.permission_state != "granted": return False cid = collection_of_rfc(rfc_slug) if not can_read_collection(user, cid): return False # §B.2 union, scope-role grants first (a collection/project/global RFC # Contributor ⊇ rfc_collaborators(contributor); an Owner ⊇ all). if effective_scope_role(user, cid) is not None: return True # per-RFC authority (union term). A 'discussant' row is NOT sufficient — # PRs are the higher-privilege surface. owners = _rfc_owners_set(rfc_slug) if user.gitea_login in owners: return True if is_rfc_collaborator(user, rfc_slug, role_in_rfc="contributor"): return True # grandfathered implicit-public baseline (curation preserved): until an owner # exists, a granted deployment contributor on the public default collection # may contribute. if not owners and _has_collection_write_baseline(user, cid): return True return False def can_invite_to_rfc(user: SessionUser | None, rfc_slug: str) -> bool: """v0.16.0 — only RFC owners (frontmatter) and platform admin/owner can issue invitations. Per-RFC collaborators do not get the invite-others power; that stays with the RFC's owner.""" if user is None: return False if user.permission_state != "granted": return False cid = collection_of_rfc(rfc_slug) if not can_read_collection(user, cid): return False # An Owner reaching the collection (collection/project/global Owner, or a # deployment owner/admin) may invite; otherwise only the RFC's frontmatter # owner. Per-RFC collaborators and the baseline do not get the invite power. if is_collection_superuser(user, cid): return True return is_rfc_owner(user, rfc_slug) def new_state() -> str: return secrets.token_urlsafe(16)