205 lines
9.5 KiB
Python
205 lines
9.5 KiB
Python
"""Gitea HTTP/Git transport using only a projected systemd credential."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
from typing import Any, Mapping
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote, urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
class GiteaTransportError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _safe_oid(value: str, field: str) -> str:
|
|
if not isinstance(value, str) or len(value) != 40 or any(ch not in "0123456789abcdef" for ch in value):
|
|
raise GiteaTransportError(f"{field}_INVALID")
|
|
return value
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GiteaMachineTransport:
|
|
api_base: str
|
|
repository_id: str
|
|
remote_url: str
|
|
authority_git_dir: Path
|
|
credential_name: str = "gitea-machine-token"
|
|
actor: str = "aios-autonomy-agent"
|
|
timeout: int = 20
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.credential_name != "gitea-machine-token":
|
|
raise GiteaTransportError("CREDENTIAL_NAME_INVALID")
|
|
if "/" not in self.repository_id or self.repository_id.startswith("/"):
|
|
raise GiteaTransportError("REPOSITORY_INVALID")
|
|
if not self.api_base.startswith("https://") and not self.api_base.startswith("http://127.0.0.1"):
|
|
raise GiteaTransportError("GITEA_API_SCOPE_INVALID")
|
|
if not self.remote_url.startswith("https://"):
|
|
raise GiteaTransportError("GITEA_REMOTE_SCOPE_INVALID")
|
|
|
|
@property
|
|
def _credential_path(self) -> Path:
|
|
directory = os.environ.get("CREDENTIALS_DIRECTORY", "")
|
|
if not directory:
|
|
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
|
|
path = Path(directory) / self.credential_name
|
|
if path.is_symlink() or not path.is_file():
|
|
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
|
|
return path
|
|
|
|
def _token(self) -> str:
|
|
value = self._credential_path.read_text(encoding="utf-8").strip()
|
|
if not value:
|
|
raise GiteaTransportError("CREDENTIAL_EMPTY")
|
|
return value
|
|
|
|
def _request(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> tuple[int, Any]:
|
|
data = None if body is None else json.dumps(dict(body), separators=(",", ":")).encode()
|
|
headers = {"Accept": "application/json", "Authorization": f"token {self._token()}"}
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
request = Request(self.api_base.rstrip("/") + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urlopen(request, timeout=self.timeout) as response:
|
|
raw = response.read()
|
|
return response.status, json.loads(raw.decode()) if raw else {}
|
|
except HTTPError as exc:
|
|
# Never include response bodies: Gitea may echo sensitive request context.
|
|
raise GiteaTransportError(f"GITEA_HTTP_{exc.code}") from exc
|
|
except (URLError, TimeoutError, OSError) as exc:
|
|
raise GiteaTransportError("GITEA_NETWORK_ERROR") from exc
|
|
|
|
def whoami(self) -> Mapping[str, Any]:
|
|
return self._request("GET", "/user")[1]
|
|
|
|
def repository_info(self) -> Mapping[str, Any]:
|
|
owner, name = self.repository_id.split("/", 1)
|
|
return self._request("GET", f"/repos/{quote(owner)}/{quote(name)}")[1]
|
|
|
|
def repository(self, owner: str, name: str) -> Mapping[str, Any]:
|
|
if f"{owner}/{name}" != self.repository_id:
|
|
raise GiteaTransportError("REPOSITORY_SCOPE_DENIED")
|
|
return self.repository_info()
|
|
|
|
def read_branch(self, branch: str) -> str | None:
|
|
owner, name = self.repository_id.split("/", 1)
|
|
try:
|
|
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
|
|
except GiteaTransportError as exc:
|
|
if str(exc) == "GITEA_HTTP_404":
|
|
return None
|
|
raise
|
|
sha = value.get("commit", {}).get("id") if isinstance(value, dict) else None
|
|
return _safe_oid(sha, "remote_branch_oid")
|
|
|
|
def _askpass(self) -> tuple[Path, dict[str, str]]:
|
|
credential = self._credential_path
|
|
fd, name = tempfile.mkstemp(prefix="aios-gitea-askpass-", dir="/tmp")
|
|
path = Path(name)
|
|
os.fchmod(fd, 0o700)
|
|
script = "#!/bin/sh\ncase \"$1\" in *Username*) printf '%s\\n' \"$GITEA_MACHINE_ACTOR\" ;; *) cat \"$CREDENTIAL_FILE\" ;; esac\n"
|
|
with os.fdopen(fd, "w", encoding="utf-8") as stream:
|
|
stream.write(script)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
return path, {
|
|
"GIT_ASKPASS": str(path),
|
|
"GIT_TERMINAL_PROMPT": "0",
|
|
"GITEA_MACHINE_ACTOR": self.actor,
|
|
"CREDENTIAL_FILE": str(credential),
|
|
}
|
|
|
|
def _push_new_branch(self, branch: str, new_oid: str) -> None:
|
|
askpass, additions = self._askpass()
|
|
environment = os.environ.copy()
|
|
environment.update(additions)
|
|
try:
|
|
result = subprocess.run(
|
|
["git", f"--git-dir={self.authority_git_dir}", "push", self.remote_url, f"{new_oid}:refs/heads/{branch}"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
env=environment,
|
|
timeout=self.timeout,
|
|
)
|
|
if result.returncode != 0:
|
|
raise GiteaTransportError("GITEA_BRANCH_CREATE_FAILED")
|
|
finally:
|
|
askpass.unlink(missing_ok=True)
|
|
|
|
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
|
|
_safe_oid(expected_oid, "expected_oid")
|
|
_safe_oid(new_oid, "new_oid")
|
|
current = self.read_branch(branch)
|
|
if current == new_oid:
|
|
return new_oid
|
|
if current is None:
|
|
self._push_new_branch(branch, new_oid)
|
|
readback = self.read_branch(branch)
|
|
if readback != new_oid:
|
|
raise GiteaTransportError("GITEA_BRANCH_CREATE_READBACK_MISMATCH")
|
|
return readback
|
|
if current != expected_oid:
|
|
raise GiteaTransportError("GITEA_BRANCH_EXPECTATION_MISMATCH")
|
|
owner, name = self.repository_id.split("/", 1)
|
|
status, value = self._request(
|
|
"PUT",
|
|
f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}",
|
|
{"new_commit_id": new_oid, "old_commit_id": expected_oid, "force": False},
|
|
)
|
|
if status not in (200, 201, 204):
|
|
raise GiteaTransportError("GITEA_BRANCH_CAS_FAILED")
|
|
readback = self.read_branch(branch)
|
|
if readback != new_oid:
|
|
raise GiteaTransportError("GITEA_BRANCH_CAS_READBACK_MISMATCH")
|
|
return readback
|
|
|
|
def _pulls(self) -> list[Mapping[str, Any]]:
|
|
owner, name = self.repository_id.split("/", 1)
|
|
query = urlencode({"state": "all", "base_branch": "main", "limit": 100})
|
|
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls?{query}")[1]
|
|
return value if isinstance(value, list) else []
|
|
|
|
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]:
|
|
if target != "main":
|
|
raise GiteaTransportError("PR_BASE_SCOPE_DENIED")
|
|
head = self.read_branch(branch)
|
|
if head is None:
|
|
raise GiteaTransportError("PR_HEAD_MISSING")
|
|
matches = [p for p in self._pulls() if (p.get("head") or {}).get("ref") == branch and (p.get("base") or {}).get("ref") == target]
|
|
open_matches = [p for p in matches if p.get("state") == "open"]
|
|
if len(open_matches) > 1:
|
|
raise GiteaTransportError("MULTIPLE_MATCHING_PRS")
|
|
if open_matches:
|
|
if (open_matches[0].get("head") or {}).get("sha") != head:
|
|
raise GiteaTransportError("PR_HEAD_CONFLICT")
|
|
return open_matches[0]
|
|
owner, name = self.repository_id.split("/", 1)
|
|
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls", {"head": branch, "base": target, "title": title, "body": body, "allow_maintainer_edit": False})[1]
|
|
|
|
def publish_status(self, branch: str, state: str, description: str) -> Mapping[str, Any]:
|
|
oid = self.read_branch(branch)
|
|
if oid is None:
|
|
raise GiteaTransportError("STATUS_BRANCH_MISSING")
|
|
owner, name = self.repository_id.split("/", 1)
|
|
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}", {"state": state, "context": "aios/durable-release", "description": description})[1]
|
|
|
|
def protected_merge(self, pr_number: int) -> Mapping[str, Any]:
|
|
if not isinstance(pr_number, int) or pr_number < 1:
|
|
raise GiteaTransportError("PR_NUMBER_INVALID")
|
|
owner, name = self.repository_id.split("/", 1)
|
|
pr = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}")[1]
|
|
if pr.get("state") != "open" or (pr.get("base") or {}).get("ref") != "main":
|
|
raise GiteaTransportError("PR_MERGE_BINDING_INVALID")
|
|
if pr.get("mergeable") is False:
|
|
raise GiteaTransportError("PR_NOT_MERGEABLE")
|
|
body = {"do": "fast-forward-only", "force_merge": False, "merge_when_checks_succeed": False, "head_commit_id": (pr.get("head") or {}).get("sha")}
|
|
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}/merge", body)[1]
|