Ben Stull
ca8ba69acb
Replaces the v0.3.0 / v0.7.0 allowed_emails admission gate with an admin-grant flow (roadmap item #6, SPEC §6.1 / §6.2 / §14.1 / §17). Any valid email can sign in via OTC; a fresh user lands in permission_state='pending' with a captured first/last/why profile, and an admin grant flips them to 'granted' before write endpoints accept them. Grandfathered users pass through the migration with the column default 'granted' so existing contributors are unaffected. The allowed_emails table stays in the schema as a fast-path bypass pending v0.9.0's admin user-management page (item #7). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""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
|
|
|
|
|
|
def new_state() -> str:
|
|
return secrets.token_urlsafe(16)
|