AIOS Mission Convergence
03da23a1e3
autonomy-unit-tests Mission source convergence checks passed at exact head
autonomy-smoke-tests Mission source convergence checks passed at exact head
git-diff-check Mission source convergence checks passed at exact head
py-compile Mission source convergence checks passed at exact head
reconciliation-contract Mission source convergence checks passed at exact head
release-state-machine Mission source convergence checks passed at exact head
secret-scan Mission source convergence checks passed at exact head
signer-validator-contract Mission source convergence checks passed at exact head
owner-approval-gate Mission source convergence checks passed at exact head
campaign-approval-gate Mission source convergence checks passed at exact head
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
"""Fail-closed Authority/Gitea ref reconciliation after protected merge."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
|
|
class RefReconciliationError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _oid(value: str, field: str) -> str:
|
|
if not isinstance(value, str) or len(value) != 40 or any(char not in "0123456789abcdef" for char in value):
|
|
raise RefReconciliationError(field + "_INVALID")
|
|
return value
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RefReconciler:
|
|
git_dir: Path
|
|
|
|
def read(self, ref: str) -> str:
|
|
result = subprocess.run(
|
|
["git", f"--git-dir={self.git_dir}", "rev-parse", "--verify", ref],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RefReconciliationError("AUTHORITY_REF_UNAVAILABLE")
|
|
return result.stdout.strip()
|
|
|
|
def is_ancestor(self, ancestor: str, descendant: str) -> bool:
|
|
_oid(ancestor, "ancestor")
|
|
_oid(descendant, "descendant")
|
|
result = subprocess.run(
|
|
["git", f"--git-dir={self.git_dir}", "merge-base", "--is-ancestor", ancestor, descendant],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0:
|
|
return True
|
|
if result.returncode == 1:
|
|
return False
|
|
raise RefReconciliationError("AUTHORITY_ANCESTRY_CHECK_FAILED")
|
|
|
|
def reconcile_main(self, *, expected_oid: str, merged_oid: str, gitea_main_oid: str) -> str:
|
|
_oid(expected_oid, "expected_oid")
|
|
_oid(merged_oid, "merged_oid")
|
|
_oid(gitea_main_oid, "gitea_main_oid")
|
|
if gitea_main_oid != merged_oid:
|
|
raise RefReconciliationError("GITEA_MAIN_EXPECTATION_MISMATCH")
|
|
if not self.is_ancestor(expected_oid, merged_oid):
|
|
raise RefReconciliationError("MAIN_FF_REQUIRED")
|
|
current = self.read("refs/heads/main")
|
|
if current == merged_oid:
|
|
return merged_oid
|
|
if current != expected_oid:
|
|
raise RefReconciliationError("AUTHORITY_MAIN_EXPECTATION_MISMATCH")
|
|
result = subprocess.run(
|
|
["git", f"--git-dir={self.git_dir}", "update-ref", "refs/heads/main", merged_oid, expected_oid],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RefReconciliationError("AUTHORITY_MAIN_CAS_FAILED")
|
|
if self.read("refs/heads/main") != merged_oid:
|
|
raise RefReconciliationError("AUTHORITY_MAIN_CAS_READBACK_MISMATCH")
|
|
return merged_oid
|