"""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 expected_merge_head: str | None = None expected_merge_base: str | None = None 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_branch_fast_forward(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_branch_fast_forward(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") # The preflight read binds the expected old tip. The Git receive-pack # then enforces the actual fast-forward CAS while transferring the new # objects; no force/lease override or arbitrary ref is permitted. self._push_branch_fast_forward(branch, new_oid) 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, context: str = "aios/durable-release") -> 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": context, "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") head_oid = (pr.get("head") or {}).get("sha") base_oid = (pr.get("base") or {}).get("sha") if self.expected_merge_head is not None and head_oid != self.expected_merge_head: raise GiteaTransportError("PR_HEAD_EXPECTATION_MISMATCH") if self.expected_merge_base is not None and base_oid != self.expected_merge_base: raise GiteaTransportError("PR_BASE_EXPECTATION_MISMATCH") body = {"do": "fast-forward-only", "force_merge": False, "merge_when_checks_succeed": False, "head_commit_id": head_oid} return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}/merge", body)[1]