"""The only place that talks to the git server. **Sign-in is not here, and that is the point of the file.** Members have passwords in our own database and no account on this server at all. This talks to it about one thing: the repository the site is built from. It used to do much more. Identity was delegated here over OAuth2, each member had an account, and the editor committed with a token belonging to whoever was typing. That is what made signing in leave vienalatina.com and signing out impossible to finish, so it was taken back. What remains: * **Content** — reading, writing and deleting files in the site repository, with one server-side token (`content_token()`), naming the real author on every commit so `git log` still says who wrote what. * **Accounts** — `admin_create_user` and `admin_set_password` still exist for the handful of real git users (pablo, the pipeline bots). Nothing in the members area calls them any more. """ from __future__ import annotations import base64 import secrets import string from urllib.parse import quote, urlencode import requests from flask import current_app TIMEOUT = 10 class GiteaError(RuntimeError): """Gitea refused a request. The message is safe to show a member.""" class StaleFile(GiteaError): """The file changed since it was loaded into the form. Gitea rejects a write whose `sha` no longer matches the branch, which is what makes two people editing one post a visible conflict rather than a silent overwrite of whoever saved first. """ def _base() -> str: return current_app.config["GITEA_URL"].rstrip("/") def _api(path: str) -> str: return f"{_base()}/api/v1{path}" def _repo() -> str: return current_app.config["CONTENT_REPO"].strip("/") def _branch() -> str: return current_app.config["CONTENT_BRANCH"] # --- sign-in ------------------------------------------------------------- # --- account creation (site-admin token) --------------------------------- def admin_configured() -> bool: """Whether this server holds a site-admin token. Without one it can neither create accounts nor set passwords. That is a supported configuration — server-setup.md §11.2 says why somebody might choose it — and it is why both calls below refuse before reaching Gitea. Exposed as a predicate so a screen can say so *before* asking somebody to fill in a form that cannot be saved, the way `mail.configured()` is used. """ return bool(current_app.config.get("ADMIN_TOKEN")) def generate_password() -> str: # Shown once to the admin, then changed by the member on first login. # Punctuation is left out on purpose: this gets read aloud or copied by # hand, and a password nobody can transcribe gets written on a note. alphabet = string.ascii_letters + string.digits return "".join(secrets.choice(alphabet) for _ in range(16)) def admin_create_user(login: str, email: str, full_name: str, password: str) -> None: token = current_app.config.get("ADMIN_TOKEN") if not token: raise GiteaError( "Falta GITEA_ADMIN_TOKEN: el servidor no puede crear cuentas nuevas." ) response = requests.post( _api("/admin/users"), headers={"Authorization": f"token {token}"}, json={ "username": login, "email": email, "full_name": full_name, "password": password, "must_change_password": True, "send_notify": False, }, timeout=TIMEOUT, ) if response.status_code in (201, 200): return if response.status_code == 422: raise GiteaError("Ese usuario o ese correo ya están en uso.") if response.status_code in (401, 403): raise GiteaError("El token de administración no es válido.") raise GiteaError(f"No se pudo crear la cuenta ({response.status_code}).") def admin_set_password(login: str, password: str) -> None: """Set a member's password on their behalf, after they chose it here. This is what lets the whole invitation and reset flow stay inside /comunidad/ instead of handing people to Gitea, whose own recovery page is dead without a mailer anyway. `login_name` and `source_id` are sent although nothing about them changes: Gitea's EditUserOption has historically treated them as required, and omitting them has been reported to move a local account onto a different authentication source. For a local user they are the username and 0, so sending them is a no-op that avoids the question. """ token = current_app.config.get("ADMIN_TOKEN") if not token: raise GiteaError( "Falta GITEA_ADMIN_TOKEN: el servidor no puede cambiar contraseñas." ) response = requests.patch( _api(f"/admin/users/{quote(login, safe='')}"), headers={"Authorization": f"token {token}"}, json={ "password": password, "must_change_password": False, "login_name": login, "source_id": 0, }, timeout=TIMEOUT, ) if response.status_code in (200, 201): return if response.status_code == 422: # Gitea enforces its own minimum length and complexity, and its message # is in the admin's language rather than the member's, so it is not # passed through. raise GiteaError("No se aceptó esa contraseña. Prueba con una más larga.") if response.status_code in (401, 403): raise GiteaError("El token de administración no es válido.") if response.status_code == 404: # Reachable: a member can be added here without ticking "crear también # su cuenta", and then invited. Everything works right up to this call, # which is asked to change the password of an account that was never # made. A bare "(404)" sends the admin looking at the wrong thing. raise GiteaError( f"No existe la cuenta «{login}», así que no se le puede " "poner contraseña. Pide a un administrador que la cree." ) raise GiteaError(f"No se pudo cambiar la contraseña ({response.status_code}).") # --- content (the member's own token) ------------------------------------ def _contents_url(path: str) -> str: # quote() with no safe characters: a path segment is data, not structure. return _api(f"/repos/{_repo()}/contents/{quote(path, safe='/')}") def content_token() -> str: """The token the editor commits with. One service token instead of a token per writer. Members no longer have accounts on the git server at all, so there is no per-member token to use — and attribution does not need one, because each commit names its author (see `_identity`). CONTENT_TOKEN is preferred and GITEA_ADMIN_TOKEN is the fallback, so nothing breaks on deploy. They can be the same, but they should not stay that way: this needs write access to one repository, while the admin token can create and modify every account on the instance. """ token = (current_app.config.get("CONTENT_TOKEN") or current_app.config.get("ADMIN_TOKEN") or "") if not token: raise GiteaError( "Falta CONTENT_TOKEN: el servidor no puede guardar en el repositorio." ) return token def _identity(member) -> dict: """Who a commit is by. Gitea's contents API takes `author` and `committer`, and uses the token's own owner only when neither is given. So one token can commit as whoever actually wrote the thing, and `git log` still says who to ask about a post. """ name = (member["display_name"] or member["gitea_login"]) if member else "Viena Latina" email = (member["email"] if member and member["email"] else "hola@vienalatina.com") return {"name": name, "email": email} def _content_request(method: str, url: str, token: str, **kwargs): response = requests.request( method, url, headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT, **kwargs, ) if response.status_code in (401, 403): raise PermissionError("gitea-unauthorised") # caller refreshes and retries return response def list_directory(path: str, token: str) -> list[dict]: """Names and shas only — the contents API does not return file bodies here.""" response = _content_request("GET", _contents_url(path), token, params={"ref": _branch()}) if response.status_code == 404: return [] # an empty content folder is normal, not an error if response.status_code != 200: raise GiteaError(f"No se pudo leer la lista de archivos ({response.status_code}).") payload = response.json() return [item for item in payload if item.get("type") == "file"] def read_file(path: str, token: str) -> tuple[str, str]: """(text, sha). The sha comes back so a later write can prove it is current.""" response = _content_request("GET", _contents_url(path), token, params={"ref": _branch()}) if response.status_code == 404: raise GiteaError("Ese archivo ya no existe.") if response.status_code != 200: raise GiteaError(f"No se pudo leer el archivo ({response.status_code}).") payload = response.json() text = base64.b64decode(payload.get("content", "")).decode("utf-8") return text, payload.get("sha", "") def write_file(path: str, data: bytes, message: str, token: str, sha: str | None = None, member=None) -> str: """Create when `sha` is None, update otherwise. Returns the new sha.""" body = { "content": base64.b64encode(data).decode("ascii"), "message": message, "branch": _branch(), "author": _identity(member), "committer": _identity(member), } if sha: body["sha"] = sha response = _content_request("PUT" if sha else "POST", _contents_url(path), token, json=body) if response.status_code in (200, 201): return response.json().get("content", {}).get("sha", "") if response.status_code in (409, 422): raise StaleFile( "Alguien más guardó este archivo mientras lo editabas. " "Vuelve a abrirlo para no perder su trabajo." ) raise GiteaError(f"No se pudo guardar el archivo ({response.status_code}).") def delete_file(path: str, sha: str, message: str, token: str, member=None) -> None: response = _content_request("DELETE", _contents_url(path), token, json={ "sha": sha, "message": message, "branch": _branch(), "author": _identity(member), "committer": _identity(member), }) if response.status_code in (200, 204): return if response.status_code in (409, 422): raise StaleFile("El archivo cambió desde que lo abriste. Recarga la lista.") raise GiteaError(f"No se pudo borrar el archivo ({response.status_code}).")