From 6d58cbf7c7cc9ca94a9c385e5d51883c42f571fc Mon Sep 17 00:00:00 2001 From: matcha Date: Sun, 4 Oct 2026 02:57:59 +0000 Subject: [PATCH] Consolidate Persimmons shared backend and tests --- backend/shared/persimmons_web.py | 293 +++++++++++++++++++++++++++++++ 1 file changed, 293 insertions(+) create mode 100644 backend/shared/persimmons_web.py diff --git a/backend/shared/persimmons_web.py b/backend/shared/persimmons_web.py new file mode 100644 index 0000000..9897817 --- /dev/null +++ b/backend/shared/persimmons_web.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +import asyncio +import html +import re +from typing import Any + +import httpx + + +_SECRET_NAME_RE = re.compile(r'data-modal-secret-name\.value="([^"]+)"|data-modal-dialog-secret-name="([^"]+)"', re.I) +_SECRET_DELETE_RE = re.compile(r'data-url="([^"]*/delete\?id=([0-9]+)[^"]*)"', re.I) +_VAR_RE = re.compile( + r'data-modal-form\.(?:url|action)="([^"]*/([0-9]+)/edit)"' + r'[\s\S]{0,1800}?data-modal-dialog-variable-name="([^"]*)"' + r'[\s\S]{0,1800}?data-modal-dialog-variable-data="([^"]*)"' + r'[\s\S]{0,1800}?data-modal-dialog-variable-description="([^"]*)"', + re.I, +) +_REPO_NAME_RE = re.compile(r'^[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}$') +_SECRET_SAFE_RE = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$') + + +def _unescape(value: str | None) -> str: + return html.unescape(value or "").strip() + + +def parse_secret_page(page: str) -> list[dict[str, Any]]: + matches = list(_SECRET_NAME_RE.finditer(page)) + result: list[dict[str, Any]] = [] + seen: set[str] = set() + for index, match in enumerate(matches): + name = _unescape(match.group(1) or match.group(2)) + if not name or name in seen: + continue + seen.add(name) + end = matches[index + 1].start() if index + 1 < len(matches) else min(len(page), match.start() + 5000) + segment = page[match.start():end] + delete_match = _SECRET_DELETE_RE.search(segment) + result.append({ + "name": name, + "id": int(delete_match.group(2)) if delete_match else None, + "description": "", + }) + return result + + +def parse_variable_page(page: str) -> list[dict[str, Any]]: + result: list[dict[str, Any]] = [] + seen: set[str] = set() + for match in _VAR_RE.finditer(page): + name = _unescape(match.group(3)) + if not name or name in seen: + continue + seen.add(name) + result.append({ + "id": int(match.group(2)), + "name": name, + "data": _unescape(match.group(4)), + "description": _unescape(match.group(5)), + }) + return result + + +def _repo_results(payload: Any) -> list[dict[str, Any]]: + if isinstance(payload, list): + raw = payload + elif isinstance(payload, dict): + if payload.get("ok") is False: + raise ValueError("Gitea repository search reported failure") + raw = None + for key in ("data", "results", "repositories"): + if key in payload: + value = payload.get(key) + if not isinstance(value, list): + raise ValueError("Gitea repository search returned invalid results") + raw = value + break + if raw is None: + raise ValueError("Gitea repository search returned unexpected JSON") + else: + raise ValueError("Gitea repository search returned unexpected JSON") + + result: list[dict[str, Any]] = [] + for item in raw: + if not isinstance(item, dict): + continue + repository = item.get("repository") + if isinstance(repository, dict): + result.append(repository) + else: + result.append(item) + return result + + +class PersimmonsWebRequester: + """Translate the Persimmons storage contract onto the signed-in Gitea web session.""" + + def __init__(self, base_url: str, cookie: str, actor_name: str) -> None: + self.base_url = base_url.rstrip("/") + self.cookie = cookie + self.actor_name = actor_name + self.headers = { + "Cookie": cookie, + "Accept": "text/html,application/xhtml+xml,application/json", + "User-Agent": "Tea Persimmons browser broker/1.0", + } + + async def __call__(self, method: str, path: str, *, params=None, json=None) -> Any: + method = method.upper() + path = path.lstrip("/") + if method == "GET" and path == "user/repos": + return await self._repositories(params or {}) + if method == "GET" and path.startswith("repos/") and "/actions/" not in path: + repository = path[len("repos/"):] + await self._assert_repository(repository) + owner = repository.split("/", 1)[0] + return { + "full_name": repository, + "owner": {"login": owner}, + "private": True, + "permissions": {"admin": True, "push": True}, + } + + match = re.fullmatch(r"(user|repos/[^/]+/[^/]+)/actions/(secrets|variables)(?:/([^/]+))?", path) + if not match: + raise RuntimeError("unsupported caller-authorized Gitea web operation") + scope, kind, item_name = match.groups() + repository = scope[len("repos/"):] if scope.startswith("repos/") else None + if repository: + await self._assert_repository(repository) + base = self._settings_base(repository) + + if kind == "secrets": + if method == "GET" and item_name is None: + return await self._secret_list(base) + if method == "PUT" and item_name: + body = json or {} + await self._secret_put(base, item_name, str(body.get("data") or ""), str(body.get("description") or "")) + return None + if method == "DELETE" and item_name: + await self._secret_delete(base, item_name) + return None + else: + if method == "GET" and item_name is None: + return await self._variable_list(base) + if method == "GET" and item_name: + item = next((v for v in await self._variable_list(base) if v.get("name") == item_name), None) + if item is None: + raise RuntimeError("caller-authorized Gitea request failed (HTTP 404)") + return item + if method in {"POST", "PUT"} and item_name: + body = json or {} + await self._variable_put(base, item_name, str(body.get("value") or ""), str(body.get("description") or "")) + return None + if method == "DELETE" and item_name: + await self._variable_delete(base, item_name) + return None + raise RuntimeError("unsupported caller-authorized Gitea web operation") + + def _settings_base(self, repository: str | None) -> str: + if repository: + return f"/{repository}/settings/actions" + return "/user/settings/actions" + + async def _get(self, path: str, *, params=None, accept_json: bool = False) -> httpx.Response: + headers = dict(self.headers) + if accept_json: + headers["Accept"] = "application/json" + async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client: + response = await client.get(self.base_url + path, params=params, headers=headers) + if response.status_code in {301, 302, 303, 307, 308, 401, 403}: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + return response + + async def _post(self, path: str, data: dict[str, str], *, params=None) -> httpx.Response: + # Gitea 1.27 uses net/http CrossOriginProtection instead of hidden CSRF + # form tokens. Preserve the caller's signed-in cookie and make the + # delegated write explicitly same-origin. + async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client: + response = await client.post( + self.base_url + path, + params=params, + data=data, + headers={ + **self.headers, + "Origin": self.base_url, + "Referer": self.base_url + path, + "Sec-Fetch-Site": "same-origin", + }, + ) + if response.status_code not in {200, 201, 204, 302, 303}: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + return response + + async def _assert_repository(self, repository: str) -> None: + if not _REPO_NAME_RE.fullmatch(repository): + raise ValueError("invalid repository") + response = await self._get(f"/{repository}/settings/actions/secrets") + if response.status_code != 200: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + + async def _repositories(self, params: dict[str, Any]) -> list[dict[str, Any]]: + candidates: dict[str, dict[str, Any]] = {} + for page in range(1, 6): + response = await self._get( + "/repo/search", + params={ + "sort": "updated", "order": "desc", "q": "", "page": page, + "limit": 50, "mode": "", "archived": "false", + }, + accept_json=True, + ) + if response.status_code != 200: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + try: + batch = _repo_results(response.json()) + except ValueError as exc: + raise RuntimeError(str(exc)) from exc + for item in batch: + full_name = str(item.get("full_name") or item.get("fullName") or "").strip() + if _REPO_NAME_RE.fullmatch(full_name): + candidates[full_name] = item + if len(batch) < 50 or len(candidates) >= 200: + break + + semaphore = asyncio.Semaphore(8) + + async def writable(full_name: str) -> dict[str, Any] | None: + async with semaphore: + try: + await self._assert_repository(full_name) + except (RuntimeError, ValueError): + return None + owner = full_name.split("/", 1)[0] + source = candidates[full_name] + return { + "full_name": full_name, + "private": bool(source.get("private")), + "owner": {"login": owner}, + "permissions": {"admin": True, "push": True}, + } + + found = await asyncio.gather(*(writable(name) for name in candidates)) + return [item for item in found if item is not None] + + async def _secret_list(self, base: str) -> list[dict[str, Any]]: + response = await self._get(base + "/secrets") + if response.status_code != 200: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + return parse_secret_page(response.text) + + async def _secret_put(self, base: str, name: str, value: str, description: str) -> None: + if not _SECRET_SAFE_RE.fullmatch(name): + raise ValueError("invalid secret name") + page = await self._get(base + "/secrets") + await self._post(base + "/secrets", { + "name": name, + "data": value, + "description": description, + }) + + async def _secret_delete(self, base: str, name: str) -> None: + page = await self._get(base + "/secrets") + item = next((s for s in parse_secret_page(page.text) if s.get("name") == name), None) + if item is None: + raise RuntimeError("caller-authorized Gitea request failed (HTTP 404)") + if not item.get("id"): + raise RuntimeError("Gitea secret id was not found") + await self._post(base + "/secrets/delete", {"id": str(item["id"])}, params={"id": item["id"]}) + + async def _variable_list(self, base: str) -> list[dict[str, Any]]: + response = await self._get(base + "/variables") + if response.status_code != 200: + raise RuntimeError(f"caller-authorized Gitea request failed (HTTP {response.status_code})") + return parse_variable_page(response.text) + + async def _variable_put(self, base: str, name: str, value: str, description: str) -> None: + page = await self._get(base + "/variables") + existing = next((v for v in parse_variable_page(page.text) if v.get("name") == name), None) + path = f"{base}/variables/{existing['id']}/edit" if existing else f"{base}/variables/new" + await self._post(path, { + "name": name, + "data": value, + "description": description, + }) + + async def _variable_delete(self, base: str, name: str) -> None: + page = await self._get(base + "/variables") + existing = next((v for v in parse_variable_page(page.text) if v.get("name") == name), None) + if existing is None: + raise RuntimeError("caller-authorized Gitea request failed (HTTP 404)") + await self._post(f"{base}/variables/{existing['id']}/delete", {})