294 lines
12 KiB
Python
294 lines
12 KiB
Python
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", {})
|