rfc-app/backend/app/auth.py

"""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 db
from .bot import Actor
from .config import Config


@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


# 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
    if user.role in ("owner", "admin"):
        return True
    owners = _rfc_owners_set(rfc_slug)
    if not owners:
        # No owner to gate the invite-list — fall through to the
        # platform-granted contract. The first §13.1 claim engages
        # the gate; before that, anyone platform-granted can
        # contribute (mirrors the v0.5.0 / v0.6.0 contract).
        return True
    if user.gitea_login in owners:
        return True
    return is_rfc_collaborator(user, rfc_slug, role_in_rfc=None)


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
    if user.role in ("owner", "admin"):
        return True
    owners = _rfc_owners_set(rfc_slug)
    if not owners:
        # Same fall-through as can_discuss_rfc: until an owner exists,
        # the gate is open.
        return True
    if user.gitea_login in owners:
        return True
    return is_rfc_collaborator(user, rfc_slug, role_in_rfc="contributor")


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
    if user.role in ("owner", "admin"):
        return True
    return is_rfc_owner(user, rfc_slug)


def new_state() -> str:
    return secrets.token_urlsafe(16)