94 lines
4.1 KiB
Python
94 lines
4.1 KiB
Python
"""Canonical Gitea transport and slash-branch verification."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import Any, Callable, Mapping, Optional
|
|
from urllib.parse import quote
|
|
from urllib.request import Request, urlopen
|
|
|
|
from aios_gitops_reconciliation_core import RemoteObservation
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class GiteaPublisherConfig:
|
|
repository: str = "aios/aios"
|
|
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
|
|
remote_name: str = "gitea-http"
|
|
|
|
|
|
class GiteaPublisher:
|
|
def __init__(self, workspace: Path, *, config: GiteaPublisherConfig = GiteaPublisherConfig(), credential: Any = None, api_get: Callable[[str], Any] | None = None, git_ref: Callable[[str], Optional[str]] | None = None) -> None:
|
|
self.workspace = Path(workspace)
|
|
self.config = config
|
|
self.credential = credential
|
|
self._api_get = api_get
|
|
self._git_ref = git_ref
|
|
|
|
@staticmethod
|
|
def validate_repository(repository: str) -> None:
|
|
if repository != "aios/aios":
|
|
raise ValueError("REPOSITORY_IDENTITY_MISMATCH")
|
|
|
|
@staticmethod
|
|
def validate_branch(branch: str) -> None:
|
|
if not isinstance(branch, str) or not branch.startswith("agent/") or branch in {"main", "master"}:
|
|
raise ValueError("BRANCH_SCOPE_DENIED")
|
|
|
|
def repository_identity(self) -> str:
|
|
self.validate_repository(self.config.repository)
|
|
return self.config.repository
|
|
|
|
def _environment(self) -> dict[str, str]:
|
|
if self.credential is not None and hasattr(self.credential, "environment"):
|
|
return dict(self.credential.environment())
|
|
return {}
|
|
|
|
def read_git_ref(self, branch: str) -> Optional[str]:
|
|
self.validate_branch(branch)
|
|
if self._git_ref is not None:
|
|
return self._git_ref(branch)
|
|
result = subprocess.run(["git", "-C", str(self.workspace), "ls-remote", self.config.remote_name, f"refs/heads/{branch}"], text=True, capture_output=True, check=False, env={**os.environ, **self._environment(), "GIT_TERMINAL_PROMPT": "0"})
|
|
if result.returncode != 0 or not result.stdout.strip():
|
|
return None
|
|
return result.stdout.split()[0]
|
|
|
|
def _api(self, path: str) -> Any:
|
|
if self._api_get is not None:
|
|
return self._api_get(path)
|
|
headers = {"Accept": "application/json"}
|
|
env = self._environment()
|
|
if env.get("AIOS_GITEA_TOKEN"):
|
|
headers["Authorization"] = "token " + env["AIOS_GITEA_TOKEN"]
|
|
with urlopen(Request(self.config.api_base.rstrip("/") + path, headers=headers), timeout=20) as response:
|
|
return json.loads(response.read() or b"{}")
|
|
|
|
def read_api_branch(self, branch: str) -> Optional[str]:
|
|
self.validate_branch(branch)
|
|
for page in range(1, 101):
|
|
value = self._api(f"/repos/aios/aios/branches?page={page}&limit=50")
|
|
if not isinstance(value, list):
|
|
return None
|
|
for item in value:
|
|
if isinstance(item, Mapping) and item.get("name") == branch:
|
|
commit = item.get("commit")
|
|
return commit.get("id") if isinstance(commit, Mapping) else None
|
|
if len(value) < 50:
|
|
return None
|
|
raise RuntimeError("GITEA_BRANCH_PAGINATION_LIMIT")
|
|
|
|
def commit_visible(self, oid: str) -> bool:
|
|
if not isinstance(oid, str) or len(oid) != 40:
|
|
return False
|
|
try:
|
|
value = self._api(f"/repos/aios/aios/git/commits/{quote(oid, safe='')}")
|
|
return isinstance(value, Mapping) and value.get("sha", oid) == oid
|
|
except Exception:
|
|
return False
|
|
|
|
def observe(self, branch: str, desired_oid: str, authority_oid: Optional[str], *, expected_oid: Optional[str] = None, observed_at: str = "") -> RemoteObservation:
|
|
return RemoteObservation(self.repository_identity(), branch, desired_oid, authority_oid, self.read_git_ref(branch), self.read_api_branch(branch), self.commit_visible(desired_oid), observed_at, expected_oid)
|