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", {})