Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f7ab821186 |
@@ -1,125 +1,136 @@
|
||||
"""Non-Owner Gitea machine capability and branch reconciliation boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Mapping, Protocol
|
||||
|
||||
|
||||
class MachineBrokerError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class GiteaTransport(Protocol):
|
||||
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str: ...
|
||||
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]: ...
|
||||
def publish_status(self, branch: str, state: str, description: str) -> Mapping[str, Any]: ...
|
||||
def protected_merge(self, pr_number: int) -> Mapping[str, Any]: ...
|
||||
def read_branch(self, branch: str) -> str | None: ...
|
||||
def whoami(self) -> Mapping[str, Any]: ...
|
||||
def repository(self, owner: str, name: str) -> Mapping[str, Any]: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MachineCredentialProjection:
|
||||
"""Opaque systemd credential reference; the credential value is never accepted."""
|
||||
|
||||
reference: str
|
||||
account: str
|
||||
owner_account: str = "aiman"
|
||||
|
||||
@classmethod
|
||||
def from_mapping(cls, value: Mapping[str, Any]) -> "MachineCredentialProjection":
|
||||
if set(value) != {"scheme", "name", "account"}:
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
if value["scheme"] != "systemd-credential" or value["name"] != "gitea-machine-token":
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
return cls(f"{value['scheme']}:{value['name']}", str(value["account"]))
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.reference != "systemd-credential:gitea-machine-token":
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
if not self.account or self.account == self.owner_account:
|
||||
raise MachineBrokerError("OWNER_MACHINE_IDENTITY_REJECTED")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CampaignCapabilityReceipt:
|
||||
campaign_id: str
|
||||
task_id: str
|
||||
repository: str
|
||||
branch: str
|
||||
expected_oid: str
|
||||
capabilities: frozenset[str]
|
||||
denied: frozenset[str]
|
||||
|
||||
def require(self, capability: str) -> None:
|
||||
if capability in self.denied or capability not in self.capabilities:
|
||||
raise MachineBrokerError("CAPABILITY_DENIED")
|
||||
if self.branch in {"main", "master"}:
|
||||
raise MachineBrokerError("PROTECTED_BRANCH_DENIED")
|
||||
|
||||
|
||||
class GiteaMachineBroker:
|
||||
def __init__(self, transport: GiteaTransport, credential: MachineCredentialProjection, receipt: CampaignCapabilityReceipt) -> None:
|
||||
self.transport = transport
|
||||
self.credential = credential
|
||||
self.receipt = receipt
|
||||
|
||||
def verify_identity(self) -> Mapping[str, Any]:
|
||||
identity = dict(self.transport.whoami())
|
||||
login = identity.get("login") or identity.get("username")
|
||||
if not isinstance(login, str) or not login or login == self.credential.owner_account or login != self.credential.account:
|
||||
raise MachineBrokerError("GITEA_MACHINE_IDENTITY_INVALID")
|
||||
owner, name = self.receipt.repository.split("/", 1)
|
||||
repo = dict(self.transport.repository(owner, name))
|
||||
if repo.get("full_name") != self.receipt.repository:
|
||||
raise MachineBrokerError("GITEA_REPOSITORY_MISMATCH")
|
||||
return {"login": login, "repository": self.receipt.repository, "owner_account": self.credential.owner_account}
|
||||
|
||||
def push_branch(self, new_oid: str) -> str:
|
||||
self.receipt.require("exact_task_branch_push")
|
||||
return self.transport.compare_and_swap_branch(self.receipt.branch, self.receipt.expected_oid, new_oid)
|
||||
|
||||
def create_pr(self, title: str, body: str) -> Mapping[str, Any]:
|
||||
self.receipt.require("pr_create_update")
|
||||
return self.transport.create_or_update_pr(self.receipt.branch, "main", title, body)
|
||||
|
||||
def status(self, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
|
||||
self.receipt.require("status_publish")
|
||||
return self.transport.publish_status(self.receipt.branch, state, description, context)
|
||||
|
||||
def merge(self, pr_number: int) -> Mapping[str, Any]:
|
||||
self.receipt.require("protected_merge")
|
||||
return self.transport.protected_merge(pr_number)
|
||||
|
||||
def readback(self) -> str | None:
|
||||
self.receipt.require("remote_readback")
|
||||
return self.transport.read_branch(self.receipt.branch)
|
||||
|
||||
|
||||
def reconcile_exact_branch(authority_oid: str, gitea_oid: str | None, expected_oid: str) -> str:
|
||||
if authority_oid != expected_oid:
|
||||
return "AUTHORITY_EXPECTATION_MISMATCH"
|
||||
if gitea_oid is None:
|
||||
return "CREATE_REQUIRED"
|
||||
if gitea_oid == expected_oid:
|
||||
return "ALREADY_IN_SYNC"
|
||||
return "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
|
||||
|
||||
def reconcile_fast_forward(
|
||||
remote_oid: str | None,
|
||||
expected_old_oid: str,
|
||||
new_oid: str,
|
||||
*,
|
||||
old_is_ancestor_of_new: bool,
|
||||
) -> str:
|
||||
"""Return a bounded action for an active Job/lease branch scope."""
|
||||
if remote_oid is None:
|
||||
return "CREATE_REQUIRED"
|
||||
if remote_oid == new_oid:
|
||||
return "ALREADY_IN_SYNC"
|
||||
if remote_oid == expected_old_oid and old_is_ancestor_of_new:
|
||||
return "FAST_FORWARD_REQUIRED"
|
||||
return "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
"""Non-Owner Gitea machine capability and branch reconciliation boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Mapping, Protocol
|
||||
|
||||
|
||||
class MachineBrokerError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class GiteaTransport(Protocol):
|
||||
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str: ...
|
||||
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]: ...
|
||||
def publish_status(self, branch: str, state: str, description: str, context: str) -> Mapping[str, Any]: ...
|
||||
def publish_status_for_sha(self, sha: str, state: str, description: str, context: str) -> Mapping[str, Any]: ...
|
||||
def read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]: ...
|
||||
def protected_merge(self, pr_number: int) -> Mapping[str, Any]: ...
|
||||
def read_branch(self, branch: str) -> str | None: ...
|
||||
def whoami(self) -> Mapping[str, Any]: ...
|
||||
def repository(self, owner: str, name: str) -> Mapping[str, Any]: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MachineCredentialProjection:
|
||||
"""Opaque systemd credential reference; the credential value is never accepted."""
|
||||
|
||||
reference: str
|
||||
account: str
|
||||
owner_account: str = "aiman"
|
||||
|
||||
@classmethod
|
||||
def from_mapping(cls, value: Mapping[str, Any]) -> "MachineCredentialProjection":
|
||||
if set(value) != {"scheme", "name", "account"}:
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
if value["scheme"] != "systemd-credential" or value["name"] != "gitea-machine-token":
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
return cls(f"{value['scheme']}:{value['name']}", str(value["account"]))
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.reference != "systemd-credential:gitea-machine-token":
|
||||
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
|
||||
if not self.account or self.account == self.owner_account:
|
||||
raise MachineBrokerError("OWNER_MACHINE_IDENTITY_REJECTED")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CampaignCapabilityReceipt:
|
||||
campaign_id: str
|
||||
task_id: str
|
||||
repository: str
|
||||
branch: str
|
||||
expected_oid: str
|
||||
capabilities: frozenset[str]
|
||||
denied: frozenset[str]
|
||||
|
||||
def require(self, capability: str) -> None:
|
||||
if capability in self.denied or capability not in self.capabilities:
|
||||
raise MachineBrokerError("CAPABILITY_DENIED")
|
||||
if self.branch in {"main", "master"}:
|
||||
raise MachineBrokerError("PROTECTED_BRANCH_DENIED")
|
||||
|
||||
|
||||
class GiteaMachineBroker:
|
||||
def __init__(self, transport: GiteaTransport, credential: MachineCredentialProjection, receipt: CampaignCapabilityReceipt) -> None:
|
||||
self.transport = transport
|
||||
self.credential = credential
|
||||
self.receipt = receipt
|
||||
|
||||
def verify_identity(self) -> Mapping[str, Any]:
|
||||
identity = dict(self.transport.whoami())
|
||||
login = identity.get("login") or identity.get("username")
|
||||
if not isinstance(login, str) or not login or login == self.credential.owner_account or login != self.credential.account:
|
||||
raise MachineBrokerError("GITEA_MACHINE_IDENTITY_INVALID")
|
||||
owner, name = self.receipt.repository.split("/", 1)
|
||||
repo = dict(self.transport.repository(owner, name))
|
||||
if repo.get("full_name") != self.receipt.repository:
|
||||
raise MachineBrokerError("GITEA_REPOSITORY_MISMATCH")
|
||||
return {"login": login, "repository": self.receipt.repository, "owner_account": self.credential.owner_account}
|
||||
|
||||
def push_branch(self, new_oid: str) -> str:
|
||||
self.receipt.require("exact_task_branch_push")
|
||||
return self.transport.compare_and_swap_branch(self.receipt.branch, self.receipt.expected_oid, new_oid)
|
||||
|
||||
def create_pr(self, title: str, body: str) -> Mapping[str, Any]:
|
||||
self.receipt.require("pr_create_update")
|
||||
return self.transport.create_or_update_pr(self.receipt.branch, "main", title, body)
|
||||
|
||||
def status(self, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
|
||||
self.receipt.require("status_publish")
|
||||
return self.transport.publish_status(self.receipt.branch, state, description, context)
|
||||
|
||||
def status_for_sha(self, sha: str, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
|
||||
"""Publish a status against an explicit commit SHA (protected-PR exact binding)."""
|
||||
self.receipt.require("status_publish")
|
||||
return self.transport.publish_status_for_sha(sha, state, description, context)
|
||||
|
||||
def read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]:
|
||||
self.receipt.require("status_publish")
|
||||
return self.transport.read_statuses(sha, context=context)
|
||||
|
||||
def merge(self, pr_number: int) -> Mapping[str, Any]:
|
||||
self.receipt.require("protected_merge")
|
||||
return self.transport.protected_merge(pr_number)
|
||||
|
||||
def readback(self) -> str | None:
|
||||
self.receipt.require("remote_readback")
|
||||
return self.transport.read_branch(self.receipt.branch)
|
||||
|
||||
|
||||
def reconcile_exact_branch(authority_oid: str, gitea_oid: str | None, expected_oid: str) -> str:
|
||||
if authority_oid != expected_oid:
|
||||
return "AUTHORITY_EXPECTATION_MISMATCH"
|
||||
if gitea_oid is None:
|
||||
return "CREATE_REQUIRED"
|
||||
if gitea_oid == expected_oid:
|
||||
return "ALREADY_IN_SYNC"
|
||||
return "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
|
||||
|
||||
def reconcile_fast_forward(
|
||||
remote_oid: str | None,
|
||||
expected_old_oid: str,
|
||||
new_oid: str,
|
||||
*,
|
||||
old_is_ancestor_of_new: bool,
|
||||
) -> str:
|
||||
"""Return a bounded action for an active Job/lease branch scope."""
|
||||
if remote_oid is None:
|
||||
return "CREATE_REQUIRED"
|
||||
if remote_oid == new_oid:
|
||||
return "ALREADY_IN_SYNC"
|
||||
if remote_oid == expected_old_oid and old_is_ancestor_of_new:
|
||||
return "FAST_FORWARD_REQUIRED"
|
||||
return "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
|
||||
@@ -1,29 +1,29 @@
|
||||
"""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
|
||||
|
||||
|
||||
"""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
|
||||
|
||||
|
||||
def _safe_branch(value: str) -> str:
|
||||
branch = str(value or "")
|
||||
if (
|
||||
@@ -43,119 +43,119 @@ def _safe_branch(value: str) -> str:
|
||||
return branch
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
@dataclass(slots=True)
|
||||
class GiteaMachineTransport:
|
||||
api_base: str
|
||||
repository_id: str
|
||||
remote_url: str
|
||||
authority_git_dir: Path
|
||||
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")
|
||||
|
||||
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:
|
||||
|
||||
def read_branch(self, branch: str) -> str | None:
|
||||
branch = _safe_branch(branch)
|
||||
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),
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
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 fetch_branch_exact(self, branch: str, expected_oid: str) -> str:
|
||||
"""Fetch one exact remote branch tip into a quarantine ref.
|
||||
|
||||
@@ -249,64 +249,86 @@ class GiteaMachineTransport:
|
||||
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
|
||||
return expected_oid
|
||||
|
||||
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:
|
||||
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")
|
||||
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]
|
||||
|
||||
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)
|
||||
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]
|
||||
|
||||
def publish_status_for_sha(self, sha: str, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
|
||||
"""Publish a commit status against an explicit 40-hex commit SHA.
|
||||
|
||||
Unlike :meth:`publish_status` (which binds to the current tip of a branch),
|
||||
this variant binds immutably to the supplied object ID. It is the correct
|
||||
primitive for protected-PR reporting where the PR head SHA is authoritative
|
||||
and must not drift with a moving branch tip.
|
||||
"""
|
||||
if state not in {"pending", "success", "error", "failure", "warning"}:
|
||||
raise GiteaTransportError("STATUS_STATE_INVALID")
|
||||
oid = _safe_oid(sha, "status_oid")
|
||||
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 read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]:
|
||||
"""Read commit statuses for an explicit SHA (used for RECONCILE)."""
|
||||
oid = _safe_oid(sha, "status_read_oid")
|
||||
owner, name = self.repository_id.split("/", 1)
|
||||
query = f"?context={quote(context)}" if context else ""
|
||||
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}{query}")[1]
|
||||
return value if isinstance(value, list) else []
|
||||
|
||||
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:
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
"""Pure scope classifier for N/A-capable status contexts.
|
||||
|
||||
Architecture §2.6: campaign-approval-gate and signer-validator-contract may be
|
||||
FORMALLY_NOT_APPLICABLE for a given PR only when an explicit applicability
|
||||
contract reference exists. This module is pure (no I/O) so it can be tested on
|
||||
any platform.
|
||||
|
||||
Policy:
|
||||
- UNKNOWN/ambiguous scope facts -> NEVER success. The classifier raises or
|
||||
returns a FAIL/UNKNOWN verdict rather than guessing APPLICABLE and later
|
||||
publishing PASS, or guessing N/A without a contract ref.
|
||||
- FORMALLY_NOT_APPLICABLE always carries a na_contract_ref so the applicability
|
||||
decision is auditable and traceable (architecture §2.7 evidence field).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Mapping
|
||||
|
||||
# Campaign N/A contract references (governance clause defining N/A as a legal pass).
|
||||
CAMPAIGN_NA_CONTRACT_REF = "aios.campaign-approval-gate.v1#formally-not-applicable"
|
||||
SIGNER_NA_CONTRACT_REF = "aios.signer-validator-contract.v1#formally-not-applicable"
|
||||
|
||||
|
||||
class ScopeClassificationError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ScopeVerdict:
|
||||
context: str
|
||||
scope: str # "APPLICABLE" | "FORMALLY_NOT_APPLICABLE"
|
||||
na_contract_ref: str | None
|
||||
reason: str
|
||||
facts: Mapping[str, Any]
|
||||
|
||||
@property
|
||||
def applicable(self) -> bool:
|
||||
return self.scope == "APPLICABLE"
|
||||
|
||||
@property
|
||||
def formally_not_applicable(self) -> bool:
|
||||
return self.scope == "FORMALLY_NOT_APPLICABLE"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Context-specific applicability rules
|
||||
# ---------------------------------------------------------------------------
|
||||
def _campaign_append_tuples():
|
||||
return ("campaign_authorization_present", "campaign_active", "in_campaign_lineage")
|
||||
|
||||
|
||||
def _classify_campaign(facts: Mapping[str, Any]) -> ScopeVerdict:
|
||||
authorization_present = facts.get("campaign_authorization_present")
|
||||
campaign_active = facts.get("campaign_active")
|
||||
in_lineage = facts.get("in_campaign_lineage")
|
||||
if authorization_present is None or campaign_active is None or in_lineage is None:
|
||||
raise ScopeClassificationError("CAMPAIGN_SCOPE_FACTS_INCOMPLETE")
|
||||
if authorization_present is True and campaign_active is True and in_lineage is True:
|
||||
return ScopeVerdict(
|
||||
"campaign-approval-gate",
|
||||
"APPLICABLE",
|
||||
None,
|
||||
"active_campaign_governs_pr",
|
||||
dict(facts),
|
||||
)
|
||||
if authorization_present is False and in_lineage is False:
|
||||
return ScopeVerdict(
|
||||
"campaign-approval-gate",
|
||||
"FORMALLY_NOT_APPLICABLE",
|
||||
CAMPAIGN_NA_CONTRACT_REF,
|
||||
"no_active_campaign_lineage",
|
||||
dict(facts),
|
||||
)
|
||||
# Ambiguous (e.g. authorization present but inactive, or lineage unknown):
|
||||
# an N/A verdict without a clean contract basis is forbidden - fail closed.
|
||||
raise ScopeClassificationError("CAMPAIGN_SCOPE_AMBIGUOUS_FAIL_CLOSED")
|
||||
|
||||
|
||||
def _classify_signer(facts: Mapping[str, Any]) -> ScopeVerdict:
|
||||
touches_signer = facts.get("touches_signer_bound_change")
|
||||
if touches_signer is None:
|
||||
raise ScopeClassificationError("SIGNER_SCOPE_FACTS_INCOMPLETE")
|
||||
if touches_signer is True:
|
||||
return ScopeVerdict(
|
||||
"signer-validator-contract",
|
||||
"APPLICABLE",
|
||||
None,
|
||||
"signer_bound_change_present",
|
||||
dict(facts),
|
||||
)
|
||||
return ScopeVerdict(
|
||||
"signer-validator-contract",
|
||||
"FORMALLY_NOT_APPLICABLE",
|
||||
SIGNER_NA_CONTRACT_REF,
|
||||
"no_signer_bound_change",
|
||||
dict(facts),
|
||||
)
|
||||
|
||||
|
||||
def classify_na(context: str, pr_scope_facts: Mapping[str, Any]) -> ScopeVerdict:
|
||||
"""Classify whether a context is APPLICABLE or FORMALLY_NOT_APPLICABLE.
|
||||
|
||||
Raises ScopeClassificationError (never returns a usable success) when scope
|
||||
facts are incomplete or ambiguous. UNKNOWN never becomes APPLICABLE_PASS
|
||||
or a bare FORMALLY_NOT_APPLICABLE success without a contract ref.
|
||||
"""
|
||||
if context == "campaign-approval-gate":
|
||||
return _classify_campaign(pr_scope_facts)
|
||||
if context == "signer-validator-contract":
|
||||
return _classify_signer(pr_scope_facts)
|
||||
# Any other context has no N/A semantics in this pipeline.
|
||||
raise ScopeClassificationError(f"NO_NA_SEMANTICS_FOR_CONTEXT:{context}")
|
||||
@@ -0,0 +1,200 @@
|
||||
"""Protected-main status evidence contract (pure, no external side effects).
|
||||
|
||||
This module defines the canonical evidence record that a status producer emits
|
||||
and the Status Publisher persists. It deliberately has zero I/O: everything
|
||||
here is deterministic derivation from an in-memory record, so the schema and
|
||||
identity rules can be unit-tested on any platform.
|
||||
|
||||
Identity model (architecture §2.4/§2.7):
|
||||
|
||||
effect_id = sha256(repo + "\\0" + commit_sha + "\\0" + context + "\\0" + evidence_revision)
|
||||
|
||||
The evidence_revision is a canonical sha256 over the evidence record *except*
|
||||
the volatile fields (observed_at / effect_id / evidence_ref), so a re-run that
|
||||
produces the identical result/inputs/version hashes to the same revision and is
|
||||
idempotently suppressed, while any change in result/input/producer version
|
||||
yields a new revision that legitimately re-publishes.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Mapping
|
||||
|
||||
STATUS_EVIDENCE_SCHEMA = "aios.status-evidence.v1"
|
||||
|
||||
# The authoritative set of status contexts governed by this pipeline.
|
||||
REQUIRED_CONTEXTS: tuple[str, ...] = (
|
||||
"py-compile",
|
||||
"git-diff-check",
|
||||
"secret-scan",
|
||||
"autonomy-unit-tests",
|
||||
"autonomy-smoke-tests",
|
||||
"reconciliation-contract",
|
||||
"release-state-machine",
|
||||
"campaign-approval-gate",
|
||||
"signer-validator-contract",
|
||||
)
|
||||
|
||||
ALLOWED_CONTEXTS: frozenset[str] = frozenset(REQUIRED_CONTEXTS)
|
||||
|
||||
# Results that may map to a Gitea success status. Everything else is never
|
||||
# published as success (architecture §2.6 red line).
|
||||
# APPLICABLE_PASS -> success
|
||||
# FORMALLY_NOT_APPLICABLE -> success ONLY when a na_contract_ref is present.
|
||||
# FAIL / UNKNOWN -> never success.
|
||||
SUCCESS_RESULTS: frozenset[str] = frozenset({"APPLICABLE_PASS", "FORMALLY_NOT_APPLICABLE"})
|
||||
|
||||
|
||||
def _canonical_json_bytes(value: Any) -> bytes:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
||||
|
||||
|
||||
def sha256_hex(value: Any) -> str:
|
||||
return hashlib.sha256(_canonical_json_bytes(value)).hexdigest()
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def validate_context(context: str) -> str:
|
||||
context = str(context or "").strip()
|
||||
if context not in ALLOWED_CONTEXTS:
|
||||
raise ValueError(f"UNKNOWN_STATUS_CONTEXT:{context}")
|
||||
return context
|
||||
|
||||
|
||||
def validate_repository(repository: str, allowed_repository: str = "aios/aios") -> str:
|
||||
repo = str(repository or "").strip()
|
||||
if repo != allowed_repository:
|
||||
raise ValueError(f"WRONG_REPOSITORY:{repo}")
|
||||
return repo
|
||||
|
||||
|
||||
def validate_commit_sha(commit_sha: str) -> str:
|
||||
sha = str(commit_sha or "").strip().lower()
|
||||
if len(sha) != 40 or any(ch not in "0123456789abcdef" for ch in sha):
|
||||
raise ValueError("WRONG_COMMIT_SHA")
|
||||
return sha
|
||||
|
||||
|
||||
def compute_evidence_revision(record: Mapping[str, Any]) -> str:
|
||||
"""Canonical sha256 over the deterministic evidence fields.
|
||||
|
||||
Volatile fields (observed_at / effect_id / evidence_ref / na_contract_ref
|
||||
legality is fixed at record time) are excluded so that equal results and
|
||||
inputs hash equal across runs.
|
||||
"""
|
||||
volatile = {"observed_at", "effect_id", "evidence_ref"}
|
||||
core = {k: v for k, v in dict(record).items() if k not in volatile}
|
||||
return "sha256:" + sha256_hex(core)
|
||||
|
||||
|
||||
def compute_effect_id(repository: str, commit_sha: str, context: str, evidence_revision: str) -> str:
|
||||
return "sha256:" + sha256_hex(f"{repository}\0{commit_sha}\0{context}\0{evidence_revision}")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StatusEvidence:
|
||||
schema: str = STATUS_EVIDENCE_SCHEMA
|
||||
repository: str = "aios/aios"
|
||||
commit_sha: str = ""
|
||||
context: str = ""
|
||||
result: str = ""
|
||||
producer_identity: str = ""
|
||||
producer_revision: str = ""
|
||||
evidence_revision: str = ""
|
||||
observed_at: str = ""
|
||||
evidence_ref: str = ""
|
||||
na_contract_ref: str | None = None
|
||||
effect_id: str = ""
|
||||
detail: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
def validated(self, *, observed_at: str | None = None) -> "StatusEvidence":
|
||||
repo = validate_repository(self.repository)
|
||||
sha = validate_commit_sha(self.commit_sha)
|
||||
ctx = validate_context(self.context)
|
||||
if self.result not in {"APPLICABLE_PASS", "FAIL", "UNKNOWN", "FORMALLY_NOT_APPLICABLE"}:
|
||||
raise ValueError(f"INVALID_EVIDENCE_RESULT:{self.result}")
|
||||
if self.result == "FORMALLY_NOT_APPLICABLE" and not self.na_contract_ref:
|
||||
raise ValueError("NA_WITHOUT_CONTRACT_REF")
|
||||
if not self.producer_identity or not self.producer_revision:
|
||||
raise ValueError("PRODUCER_IDENTITY_INCOMPLETE")
|
||||
ts = observed_at or utc_now()
|
||||
return StatusEvidence(
|
||||
schema=STATUS_EVIDENCE_SCHEMA,
|
||||
repository=repo,
|
||||
commit_sha=sha,
|
||||
context=ctx,
|
||||
result=self.result,
|
||||
producer_identity=self.producer_identity,
|
||||
producer_revision=self.producer_revision,
|
||||
evidence_revision=self.evidence_revision,
|
||||
observed_at=ts,
|
||||
evidence_ref=self.evidence_ref,
|
||||
na_contract_ref=self.na_contract_ref,
|
||||
effect_id=self.effect_id,
|
||||
detail=dict(self.detail),
|
||||
)
|
||||
|
||||
def finalize(self, *, observed_at: str | None = None, evidence_ref: str = "") -> "StatusEvidence":
|
||||
"""Compute evidence_revision + effect_id and return a validated record."""
|
||||
if self.result == "FORMALLY_NOT_APPLICABLE" and not self.na_contract_ref:
|
||||
raise ValueError("NA_WITHOUT_CONTRACT_REF")
|
||||
base = dict(
|
||||
repository=self.repository,
|
||||
commit_sha=self.commit_sha,
|
||||
context=self.context,
|
||||
result=self.result,
|
||||
producer_identity=self.producer_identity,
|
||||
producer_revision=self.producer_revision,
|
||||
na_contract_ref=self.na_contract_ref,
|
||||
detail=dict(self.detail),
|
||||
evidence_ref=evidence_ref or self.evidence_ref,
|
||||
)
|
||||
rev = compute_evidence_revision(base)
|
||||
effect = compute_effect_id(self.repository, self.commit_sha, self.context, rev)
|
||||
return StatusEvidence(
|
||||
schema=STATUS_EVIDENCE_SCHEMA,
|
||||
repository=self.repository,
|
||||
commit_sha=validate_commit_sha(self.commit_sha),
|
||||
context=validate_context(self.context),
|
||||
result=self.result,
|
||||
producer_identity=self.producer_identity,
|
||||
producer_revision=self.producer_revision,
|
||||
evidence_revision=rev,
|
||||
observed_at=observed_at or utc_now(),
|
||||
evidence_ref=evidence_ref or self.evidence_ref,
|
||||
na_contract_ref=self.na_contract_ref,
|
||||
effect_id=effect,
|
||||
detail=dict(self.detail),
|
||||
)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"schema": self.schema,
|
||||
"repository": self.repository,
|
||||
"commit_sha": self.commit_sha,
|
||||
"context": self.context,
|
||||
"result": self.result,
|
||||
"producer_identity": self.producer_identity,
|
||||
"producer_revision": self.producer_revision,
|
||||
"evidence_revision": self.evidence_revision,
|
||||
"observed_at": self.observed_at,
|
||||
"evidence_ref": self.evidence_ref,
|
||||
"na_contract_ref": self.na_contract_ref,
|
||||
"effect_id": self.effect_id,
|
||||
"detail": dict(self.detail),
|
||||
}
|
||||
|
||||
|
||||
def maps_to_success(result: str) -> bool:
|
||||
"""Whether a result may be published as a Gitea success status.
|
||||
|
||||
FAIL / UNKNOWN never map to success. FORMALLY_NOT_APPLICABLE only maps when
|
||||
a contract ref is present, which is enforced at record validation time.
|
||||
"""
|
||||
return result in SUCCESS_RESULTS
|
||||
@@ -0,0 +1,186 @@
|
||||
"""Protected-main status orchestrator.
|
||||
|
||||
Responsibilities (architecture §2.2 / §2.5):
|
||||
1. Resolve the PR's exact head SHA (never guessed locally).
|
||||
2. Schedule all 9 required producers, each owning exactly one context.
|
||||
3. Convert each ProducerOutput into a StatusEvidence and hand it to the
|
||||
Status Publisher (the only writer of Gitea statuses).
|
||||
4. Enforce stale-SHA safety: every run keys effects on the *current* head SHA;
|
||||
evidence is always regenerated for that head economy — old SHA statuses are
|
||||
never copied onto a new head. A head change requires a full re-run.
|
||||
5. Report merge eligibility: protected main may merge only when every required
|
||||
context resolved to a genuine ``success`` status on the current head.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Mapping
|
||||
|
||||
from status_evidence_contract import (
|
||||
REQUIRED_CONTEXTS,
|
||||
StatusEvidence,
|
||||
validate_context,
|
||||
validate_commit_sha,
|
||||
validate_repository,
|
||||
)
|
||||
from status_publisher import AppendOnlyEvidenceStore, StatusPublisher
|
||||
|
||||
|
||||
class OrchestratorError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ContextOutcome:
|
||||
context: str
|
||||
producer_output: Any
|
||||
evidence: StatusEvidence
|
||||
publish: Any # PublishOutcome
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RunResult:
|
||||
head_sha: str
|
||||
context_outcomes: Mapping[str, ContextOutcome]
|
||||
merge_eligible: bool
|
||||
|
||||
|
||||
def resolve_pr_head_sha(pr: Mapping[str, Any], *, expected_repository: str = "aios/aios") -> str:
|
||||
"""Resolve the exact head SHA from a Gitea PR object (authoritative)."""
|
||||
repo = (pr.get("base") or {}).get("repo") or {}
|
||||
full_name = repo.get("full_name") or repo.get("fullname") or ""
|
||||
if full_name and full_name != expected_repository:
|
||||
raise OrchestratorError(f"WRONG_REPOSITORY:{full_name}")
|
||||
head = (pr.get("head") or {}).get("sha")
|
||||
if not head:
|
||||
raise OrchestratorError("PR_HEAD_SHA_MISSING")
|
||||
return validate_commit_sha(head)
|
||||
|
||||
|
||||
# The 9 producers, keyed by their governed context name.
|
||||
def _default_producers() -> Mapping[str, Callable[[Any], Any]]:
|
||||
from status_producers import (
|
||||
py_compile,
|
||||
git_diff_check,
|
||||
secret_scan,
|
||||
autonomy_unit_tests,
|
||||
autonomy_smoke_tests,
|
||||
reconciliation_contract,
|
||||
release_state_machine,
|
||||
campaign_approval_gate,
|
||||
signer_validator_contract,
|
||||
)
|
||||
|
||||
return {
|
||||
"py-compile": py_compile.run,
|
||||
"git-diff-check": git_diff_check.run,
|
||||
"secret-scan": secret_scan.run,
|
||||
"autonomy-unit-tests": autonomy_unit_tests.run,
|
||||
"autonomy-smoke-tests": autonomy_smoke_tests.run,
|
||||
"reconciliation-contract": reconciliation_contract.run,
|
||||
"release-state-machine": release_state_machine.run,
|
||||
"campaign-approval-gate": campaign_approval_gate.run,
|
||||
"signer-validator-contract": signer_validator_contract.run,
|
||||
}
|
||||
|
||||
|
||||
class StatusOrchestrator:
|
||||
def __init__(
|
||||
self,
|
||||
publisher: StatusPublisher,
|
||||
*,
|
||||
producers: Mapping[str, Callable[[Any], Any]] | None = None,
|
||||
checkout: Path | None = None,
|
||||
base_sha: str | None = None,
|
||||
extra_facts: Mapping[str, Any] | None = None,
|
||||
) -> None:
|
||||
self.publisher = publisher
|
||||
self.producers = _default_producers() if producers is None else producers
|
||||
if set(self.producers) != set(REQUIRED_CONTEXTS):
|
||||
raise OrchestratorError("PRODUCER_CONTEXT_SET_MISALIGNED")
|
||||
self.checkout = checkout
|
||||
self.base_sha = base_sha
|
||||
self.extra_facts = dict(extra_facts or {})
|
||||
|
||||
def run_for_pr(self, pr: Mapping[str, Any]) -> RunResult:
|
||||
head_sha = resolve_pr_head_sha(pr)
|
||||
return self._run_head(head_sha)
|
||||
|
||||
def run_for_head(self, head_sha: str) -> RunResult:
|
||||
return self._run_head(validate_commit_sha(head_sha))
|
||||
|
||||
def _run_head(self, head_sha: str) -> RunResult:
|
||||
outcomes: dict[str, ContextOutcome] = {}
|
||||
for context in REQUIRED_CONTEXTS:
|
||||
validate_context(context)
|
||||
producer = self.producers[context]
|
||||
try:
|
||||
producer_output = producer(
|
||||
self._make_ctx(context, head_sha)
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
from status_producers import unknown_output
|
||||
producer_output = unknown_output(
|
||||
context, f"aios.{context}", f"producer_error:{exc}"
|
||||
)
|
||||
evidence = StatusEvidence(
|
||||
repository=validate_repository("aios/aios"),
|
||||
commit_sha=head_sha,
|
||||
context=context,
|
||||
result=producer_output.result,
|
||||
producer_identity=producer_output.producer_identity,
|
||||
producer_revision=producer_output.producer_revision,
|
||||
na_contract_ref=producer_output.na_contract_ref,
|
||||
detail=dict(producer_output.detail),
|
||||
)
|
||||
publish = self.publisher.publish(evidence)
|
||||
outcomes[context] = ContextOutcome(context, producer_output, evidence, publish)
|
||||
|
||||
merge_eligible = all(
|
||||
self._context_green(outcomes[ctx]) for ctx in REQUIRED_CONTEXTS
|
||||
)
|
||||
return RunResult(head_sha, outcomes, merge_eligible)
|
||||
|
||||
@staticmethod
|
||||
def _context_green(outcome: ContextOutcome) -> bool:
|
||||
"""A context is merge-green IFF its *verdict* genuinely passed AND its
|
||||
status publish was *confirmed*.
|
||||
|
||||
The verdict (evidence.result) is authoritative (architecture §5): a
|
||||
FAIL/UNKNOWN context can never be green no matter what a conflicting
|
||||
leftover Gitea status says. A green verdict additionally requires a
|
||||
confirmed publish (PUBLISHED / RECONCILED matched_expected / SUPPRESS_NOOP
|
||||
with a verified success readback) — never an unconfirmed
|
||||
existing_state_differs result.
|
||||
"""
|
||||
evidence = outcome.evidence
|
||||
verdict_green = (
|
||||
evidence.result == "APPLICABLE_PASS"
|
||||
or (
|
||||
evidence.result == "FORMALLY_NOT_APPLICABLE"
|
||||
and bool(evidence.na_contract_ref)
|
||||
)
|
||||
)
|
||||
if not verdict_green:
|
||||
return False
|
||||
publish = outcome.publish
|
||||
if not publish.confirmed:
|
||||
return False
|
||||
# Confirmation must carry the eligible action forms only.
|
||||
if publish.action not in {"PUBLISHED", "RECONCILED", "SUPPRESS_NOOP"}:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _make_ctx(self, context: str, head_sha: str) -> Any:
|
||||
from status_producers import ProducerContext
|
||||
facts = dict(self.extra_facts)
|
||||
facts.setdefault("commit_sha", head_sha)
|
||||
return ProducerContext(
|
||||
context=context,
|
||||
repository="aios/aios",
|
||||
commit_sha=head_sha,
|
||||
checkout=self.checkout,
|
||||
base_sha=self.base_sha,
|
||||
scope_facts=facts,
|
||||
)
|
||||
@@ -0,0 +1,197 @@
|
||||
"""Protected-main status pipeline entry point (S6 trigger integration skeleton).
|
||||
|
||||
This module wires the real, locally-validated pipeline to a production Gitea
|
||||
broker using environment-provided configuration. It is intentionally a *thin*
|
||||
adapter: it constructs a capability-minimal broker (only ``status_publish`` plus
|
||||
the readback needed for RECONCILE), the idempotent publisher, the append-only
|
||||
evidence store, and an orchestrator, then runs one PR.
|
||||
|
||||
It does NOT fake or shortcut production connectivity: any missing environment
|
||||
(exact PR head, checkout, gitleaks, signer, campaign authorization) is surfaced
|
||||
as an explicit error/UNKNOWN through the pipeline's fail-closed semantics. This
|
||||
file is the integration point a production trigger (webhook / durable job /
|
||||
poll loop) calls; the trigger mechanism itself is owned by the orchestrating
|
||||
service and is out of scope here.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from aios_gitea_machine_broker import (
|
||||
CampaignCapabilityReceipt,
|
||||
GiteaMachineBroker,
|
||||
MachineCredentialProjection,
|
||||
)
|
||||
from status_evidence_contract import REQUIRED_CONTEXTS
|
||||
from status_orchestrator import StatusOrchestrator
|
||||
from status_publisher import (
|
||||
AppendOnlyEvidenceStore,
|
||||
DANGEROUS_CAPABILITIES,
|
||||
StatusPublisher,
|
||||
)
|
||||
|
||||
REPOSITORY = os.environ.get("AIOS_GITEA_REPOSITORY", "aios/aios")
|
||||
|
||||
|
||||
def _json_env(name: str) -> dict | None:
|
||||
"""Parse a JSON object from an environment variable; None if absent/unparsable.
|
||||
|
||||
Unparsable input is ignored (fail-closed) so a malformed injection never
|
||||
silently degrades a check to green — it simply behaves as if not provided,
|
||||
which keeps the pipeline's UNKNOWN fail-closed semantics intact.
|
||||
"""
|
||||
raw = os.environ.get(name, "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
value = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return value if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def build_orchestrator(
|
||||
*,
|
||||
scope_facts: dict | None = None,
|
||||
smoke_results: dict | None = None,
|
||||
) -> StatusOrchestrator:
|
||||
"""Build the orchestrator from environment (capability-minimal broker).
|
||||
|
||||
``scope_facts`` (campaign_authorization_present / campaign_active /
|
||||
in_campaign_lineage / touches_signer_bound_change) and ``smoke_results`` are
|
||||
*optional injection channels* that let a production trigger supply real
|
||||
observed applicability / smoke findings so the 9 contexts can genuinely
|
||||
reach 9/9. They are never defaulted to fake pass values: when absent or
|
||||
malformed the pipeline stays UNKNOWN fail-closed. Env vars are merged
|
||||
first; explicit caller args take precedence.
|
||||
"""
|
||||
from aios_gitea_machine_transport import GiteaMachineTransport
|
||||
|
||||
account = os.environ.get("AIOS_GITEA_MACHINE_ACCOUNT", "aios-autonomy-agent")
|
||||
api = os.environ.get(
|
||||
"AIOS_GITEA_API", "https://g.cnnetmkt.top/git/api/v1"
|
||||
)
|
||||
remote = os.environ.get(
|
||||
"AIOS_GITEA_REMOTE", "https://g.cnnetmkt.top/git/aios/aios.git"
|
||||
)
|
||||
authority_dir = Path(
|
||||
os.environ.get("AIOS_GITOPS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git")
|
||||
)
|
||||
checkout = Path(os.environ.get("AIOS_STATUS_CHECKOUT", "")) or None
|
||||
evidence_root = Path(
|
||||
os.environ.get("AIOS_STATUS_EVIDENCE_ROOT", "/var/lib/aios-gitops/status-evidence")
|
||||
)
|
||||
|
||||
transport = GiteaMachineTransport(api, REPOSITORY, remote, authority_dir)
|
||||
credential = MachineCredentialProjection.from_mapping(
|
||||
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": account}
|
||||
)
|
||||
# Capability-minimal: only status_publish (+ readback for RECONCILE), and an
|
||||
# explicit denied set covering dangerous capabilities (aligned with the
|
||||
# gitops denied set). The publisher additionally rejects any dangerous cap
|
||||
# that leaks through here.
|
||||
receipt = CampaignCapabilityReceipt(
|
||||
os.environ.get("AIOS_CAMPAIGN_ID", "status-pipeline"),
|
||||
os.environ.get("AIOS_TASK_ID", "status-pipeline-task"),
|
||||
REPOSITORY,
|
||||
os.environ.get("AIOS_STATUS_BRANCH", "agent/status-pipeline"),
|
||||
os.environ.get("AIOS_EXPECTED_BRANCH_OID", ""),
|
||||
frozenset({"status_publish"}),
|
||||
frozenset(DANGEROUS_CAPABILITIES),
|
||||
)
|
||||
broker = GiteaMachineBroker(transport, credential, receipt)
|
||||
store = AppendOnlyEvidenceStore(evidence_root)
|
||||
publisher = StatusPublisher(
|
||||
broker,
|
||||
store,
|
||||
receipt_capabilities={
|
||||
"capabilities": set(receipt.capabilities),
|
||||
"denied": set(receipt.denied),
|
||||
},
|
||||
)
|
||||
|
||||
# Assemble optional injection channels (env first, caller args override).
|
||||
extra_facts: dict = {
|
||||
"pr_number": int(os.environ.get("AIOS_STATUS_PR_NUMBER", "0") or 0),
|
||||
}
|
||||
env_scope = _json_env("AIOS_STATUS_SCOPE_FACTS")
|
||||
if env_scope:
|
||||
extra_facts.update(env_scope)
|
||||
if scope_facts:
|
||||
extra_facts.update(scope_facts)
|
||||
|
||||
env_smoke = _json_env("AIOS_STATUS_SMOKE_RESULTS")
|
||||
smoke: dict = {}
|
||||
if env_smoke:
|
||||
smoke.update(env_smoke)
|
||||
if smoke_results:
|
||||
smoke.update(smoke_results)
|
||||
if smoke:
|
||||
extra_facts["smoke_results"] = smoke
|
||||
|
||||
return StatusOrchestrator(
|
||||
publisher,
|
||||
checkout=checkout,
|
||||
base_sha=os.environ.get("AIOS_STATUS_BASE_SHA"),
|
||||
extra_facts=extra_facts,
|
||||
)
|
||||
|
||||
|
||||
def _json_arg(value: str | None, name: str) -> dict | None:
|
||||
if value is None or not value.strip():
|
||||
return None
|
||||
try:
|
||||
parsed = json.loads(value)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise SystemExit(f"{name} must be a JSON object") from exc
|
||||
if not isinstance(parsed, dict):
|
||||
raise SystemExit(f"{name} must be a JSON object")
|
||||
return parsed
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Run the protected-main status pipeline for a PR")
|
||||
parser.add_argument("--pr-json", help="Gitea PR object JSON (must include base.repo and head.sha)")
|
||||
parser.add_argument("--head-sha", help="exact head SHA (alternative to --pr-json)")
|
||||
parser.add_argument("--scope-facts", help="JSON object of applicability scope facts (campaign/signer)")
|
||||
parser.add_argument("--smoke-results", help="JSON object of observed smoke check results")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
orchestrator = build_orchestrator(
|
||||
scope_facts=_json_arg(args.scope_facts, "--scope-facts"),
|
||||
smoke_results=_json_arg(args.smoke_results, "--smoke-results"),
|
||||
)
|
||||
|
||||
try:
|
||||
if args.head_sha:
|
||||
result = orchestrator.run_for_head(args.head_sha)
|
||||
elif args.pr_json:
|
||||
result = orchestrator.run_for_pr(json.loads(args.pr_json))
|
||||
else:
|
||||
raise SystemExit("require --pr-json or --head-sha")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(json.dumps({"status": "ERROR", "reason": str(exc)}, sort_keys=True))
|
||||
return 2
|
||||
|
||||
summary = {
|
||||
"head_sha": result.head_sha,
|
||||
"required_contexts": len(REQUIRED_CONTEXTS),
|
||||
"merge_eligible": result.merge_eligible,
|
||||
"contexts": {
|
||||
ctx: {
|
||||
"action": o.publish.action,
|
||||
"state": o.publish.gitea_state,
|
||||
"result": o.producer_output.result,
|
||||
}
|
||||
for ctx, o in result.context_outcomes.items()
|
||||
},
|
||||
}
|
||||
print(json.dumps(summary, sort_keys=True))
|
||||
return 0 if result.merge_eligible else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Protected-main status producers (thin, real-execution wrappers).
|
||||
|
||||
Each producer owns exactly one required context and derives a deterministic
|
||||
verdict from *real* tool output or a real validator — never a manufactured
|
||||
``success``. Producers never write Gitea statuses; they emit a
|
||||
:class:`ProducerOutput` that the Status Orchestrator converts into a
|
||||
:class:`~status_evidence_contract.StatusEvidence` for the Status Publisher.
|
||||
|
||||
Testability
|
||||
-----------
|
||||
The real-execution subprocess helper ``run_command`` is injectable via
|
||||
``set_command_runner`` so unit tests can resolve failures without invoking the
|
||||
real toolchain (there is no Gitea / gitleaks / signer binary on non-Linux dev
|
||||
hosts). Production loans the real runner.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Mapping, Sequence
|
||||
|
||||
# Every producer has a fixed producer_identity (audit attribution).
|
||||
PRODUCER_REVISION = "producer.2026-09-15.1"
|
||||
|
||||
CommandRunner = Callable[[Sequence[str], Path | None], subprocess.CompletedProcess[str]]
|
||||
|
||||
|
||||
def _default_command_runner(
|
||||
command: Sequence[str], cwd: Path | None = None, *, timeout: int = 300
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
list(command),
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
timeout=timeout,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
_ACTIVE_RUNNER: CommandRunner = _default_command_runner
|
||||
|
||||
|
||||
def set_command_runner(runner: CommandRunner) -> None:
|
||||
"""Override the subprocess runner (used by tests to avoid real toolchains)."""
|
||||
global _ACTIVE_RUNNER
|
||||
_ACTIVE_RUNNER = runner
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProducerContext:
|
||||
context: str = "" # governed context name this run serves
|
||||
repository: str = "aios/aios"
|
||||
commit_sha: str = ""
|
||||
checkout: Path | None = None # exact-head checkout (real exec target)
|
||||
base_sha: str | None = None # for base...head checks
|
||||
scope_facts: Mapping[str, Any] = field(default_factory=dict)
|
||||
runner_timeout: int = 300
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProducerOutput:
|
||||
context: str
|
||||
result: str # APPLICABLE_PASS | FAIL | UNKNOWN | FORMALLY_NOT_APPLICABLE
|
||||
producer_identity: str
|
||||
producer_revision: str
|
||||
detail: Mapping[str, Any] = field(default_factory=dict)
|
||||
na_contract_ref: str | None = None
|
||||
|
||||
|
||||
def fail_output(context: str, identity: str, reason: str, detail: Mapping[str, Any] | None = None) -> ProducerOutput:
|
||||
merged = dict(detail or {})
|
||||
merged["reason"] = reason
|
||||
return ProducerOutput(context, "FAIL", identity, PRODUCER_REVISION, merged)
|
||||
|
||||
|
||||
def unknown_output(context: str, identity: str, reason: str) -> ProducerOutput:
|
||||
return ProducerOutput(context, "UNKNOWN", identity, PRODUCER_REVISION, {"reason": reason})
|
||||
@@ -0,0 +1,57 @@
|
||||
"""autonomy-smoke-tests producer: real staging/smoke checklist.
|
||||
|
||||
This must never be fulfilled by rerunning the unit suite. Smoke checks are a
|
||||
separate, environment-dependent checklist; where a real staging environment is
|
||||
unavailable the producer reports UNKNOWN rather than an invented pass.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
# Smoke checklist entries as (label). Each must map to a real check at runtime;
|
||||
# the checklist itself is the contract, not the unit suite.
|
||||
SMOKE_CHECKS: tuple[str, ...] = (
|
||||
"imports_load",
|
||||
"core_services_respond",
|
||||
)
|
||||
|
||||
|
||||
def run(
|
||||
ctx: ProducerContext,
|
||||
*,
|
||||
results: dict[str, bool] | None = None,
|
||||
) -> ProducerOutput:
|
||||
identity = "aios.autonomy-smoke-tests"
|
||||
checkout = ctx.checkout
|
||||
if checkout is None or not checkout.is_dir():
|
||||
return unknown_output(ctx.context, identity, "checkout_unavailable")
|
||||
|
||||
# Production executes real staging probes and reports a results map via the
|
||||
# injected smoke_results (contract layer); fall back to scope_facts so an
|
||||
# orchestrator/entry can pass concrete observed results. If no real results
|
||||
# were observed, this is UNKNOWN — never an auto-pass.
|
||||
if results is None:
|
||||
injected = ctx.scope_facts.get("smoke_results")
|
||||
results = dict(injected) if isinstance(injected, dict) else None
|
||||
if results is None:
|
||||
return unknown_output(ctx.context, identity, "smoke_environment_unavailable")
|
||||
|
||||
missing = [label for label in SMOKE_CHECKS if label not in results]
|
||||
if missing:
|
||||
return fail_output(ctx.context, identity, "smoke_results_incomplete", {"missing": missing})
|
||||
all_passed = all(bool(results[label]) for label in SMOKE_CHECKS)
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if all_passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
{"smoke_checks": {label: bool(results[label]) for label in SMOKE_CHECKS}},
|
||||
)
|
||||
@@ -0,0 +1,60 @@
|
||||
"""autonomy-unit-tests producer: real pytest suite on a stable subset."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
# Stable pure-logic subset that does not drag in fcntl/systemd/signer binaries,
|
||||
# so it can run on non-Linux hosts and in CI. Production may extend this list,
|
||||
# but a producer pass must always result from a real suite that ran green.
|
||||
STABLE_UNIT_TESTS: tuple[str, ...] = (
|
||||
"tests/test_convergence_contracts.py",
|
||||
"tests/test_gitea_transport_consistency_provider.py",
|
||||
)
|
||||
|
||||
|
||||
def run(ctx: ProducerContext) -> ProducerOutput:
|
||||
identity = "aios.autonomy-unit-tests"
|
||||
checkout = ctx.checkout
|
||||
if checkout is None or not checkout.is_dir():
|
||||
return unknown_output(ctx.context, identity, "checkout_unavailable")
|
||||
missing = [t for t in STABLE_UNIT_TESTS if not (checkout / t).is_file()]
|
||||
if missing:
|
||||
return fail_output(ctx.context, identity, "test_targets_missing", {"missing": missing})
|
||||
|
||||
from . import _ACTIVE_RUNNER
|
||||
app = checkout / "active-runtime" / "app"
|
||||
env_paths = [str(checkout), str(app)]
|
||||
env = dict(os.environ)
|
||||
env["PYTHONPATH"] = os.pathsep.join(env_paths)
|
||||
result = _ACTIVE_RUNNER(
|
||||
["python", "-m", "pytest", "-q", "-p", "no:cacheprovider", *STABLE_UNIT_TESTS],
|
||||
checkout,
|
||||
)
|
||||
return _from_pytest(ctx, result, identity)
|
||||
|
||||
|
||||
def _from_pytest(ctx: ProducerContext, result, identity: str) -> ProducerOutput:
|
||||
# A pass requires a real, green run: exit 0 and no failures. The runner
|
||||
# returns a string-yield CompletedProcess; tests may attach a mock.
|
||||
passed = result.returncode == 0
|
||||
detail = {
|
||||
"exit_code": result.returncode,
|
||||
"tests_run": [t for t in STABLE_UNIT_TESTS],
|
||||
"summary": (result.stdout or "")[:2000],
|
||||
}
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
@@ -0,0 +1,92 @@
|
||||
"""campaign-approval-gate producer.
|
||||
|
||||
Scope is classified via ``producer_scope_classifier.classify_na``:
|
||||
- APPLICABLE -> run the real ``campaign_approval_gate.evaluate``; a missing
|
||||
CampaignAuthorization or unavailable toolchain is UNKNOWN, never success.
|
||||
- FORMALLY_NOT_APPLICABLE -> emit an NA verdict bound to the applicability
|
||||
contract ref. An N/A verdict with no contract ref is forbidden.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from producer_scope_classifier import (
|
||||
ScopeClassificationError,
|
||||
classify_na,
|
||||
)
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
|
||||
def run(
|
||||
ctx: ProducerContext,
|
||||
*,
|
||||
campaign_facts: dict | None = None,
|
||||
campaign_path: Path | None = None,
|
||||
) -> ProducerOutput:
|
||||
identity = "aios.campaign-approval-gate"
|
||||
facts = campaign_facts if campaign_facts is not None else dict(ctx.scope_facts)
|
||||
try:
|
||||
verdict = classify_na(ctx.context, facts)
|
||||
except ScopeClassificationError as exc:
|
||||
return unknown_output(ctx.context, identity, f"scope_unknown:{exc}")
|
||||
|
||||
if verdict.formally_not_applicable:
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"FORMALLY_NOT_APPLICABLE",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
{"reason": verdict.reason, "scope_facts": dict(facts), "applicable": False},
|
||||
na_contract_ref=verdict.na_contract_ref,
|
||||
)
|
||||
|
||||
# APPLICABLE: real gate requires a CampaignAuthorization artifact.
|
||||
if campaign_path is None or not campaign_path.is_file():
|
||||
return unknown_output(ctx.context, identity, "campaign_authorization_unavailable")
|
||||
try:
|
||||
from release_campaign_authorization import CampaignAuthorization
|
||||
from campaign_approval_gate import evaluate
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return unknown_output(ctx.context, identity, f"gate_module_unavailable:{exc}")
|
||||
try:
|
||||
campaign = CampaignAuthorization.load(campaign_path, require_active=True)
|
||||
gate_facts = {
|
||||
"repository": ctx.repository,
|
||||
"pr_number": int(facts.get("pr_number", 0)),
|
||||
"base_ref": "refs/heads/main",
|
||||
"base_oid": facts.get("base_oid", ctx.base_sha or ""),
|
||||
"head_oid": ctx.commit_sha,
|
||||
"required_checks_all_success": True,
|
||||
"required_checks_same_head": True,
|
||||
"unresolved_conversations": 0,
|
||||
"mergeable": True,
|
||||
"protected_target": "main",
|
||||
}
|
||||
if ctx.checkout is not None:
|
||||
from campaign_approval_gate import git_is_ancestor
|
||||
lineage_ok = (
|
||||
git_is_ancestor(ctx.checkout, campaign.anchor_oid, gate_facts["base_oid"])
|
||||
and git_is_ancestor(ctx.checkout, gate_facts["base_oid"], ctx.commit_sha)
|
||||
)
|
||||
else:
|
||||
lineage_ok = True
|
||||
if not lineage_ok:
|
||||
return fail_output(ctx.context, identity, "campaign_lineage_mismatch")
|
||||
result = evaluate(campaign, gate_facts, ctx.checkout)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return fail_output(ctx.context, identity, f"campaign_evaluate_failed:{exc}")
|
||||
|
||||
passed = result.get("status") == "SUCCESS"
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
{"gate_status": result.get("status"), "campaign_id": result.get("campaign_id"), "applicable": True},
|
||||
)
|
||||
@@ -0,0 +1,42 @@
|
||||
"""git-diff-check producer: genuine ``git diff --check`` on base...head."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
|
||||
def run(ctx: ProducerContext) -> ProducerOutput:
|
||||
identity = "aios.git-diff-check"
|
||||
checkout = ctx.checkout
|
||||
if checkout is None or not checkout.is_dir():
|
||||
return unknown_output(ctx.context, identity, "checkout_unavailable")
|
||||
base = ctx.base_sha
|
||||
if not base:
|
||||
return fail_output(ctx.context, identity, "base_sha_missing")
|
||||
|
||||
from . import _ACTIVE_RUNNER
|
||||
# Real whitespace/conflict-marker check over the PR range.
|
||||
result = _ACTIVE_RUNNER(
|
||||
["git", "diff", "--check", f"{base}...{ctx.commit_sha}"], checkout
|
||||
)
|
||||
passed = result.returncode == 0
|
||||
detail = {
|
||||
"base": base,
|
||||
"head": ctx.commit_sha,
|
||||
"exit_code": result.returncode,
|
||||
"whitespace_errors": (result.stdout + result.stderr).strip(),
|
||||
}
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
@@ -0,0 +1,40 @@
|
||||
"""py-compile producer: real byte-compilation of the PR head checkout."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
|
||||
def run(ctx: ProducerContext) -> ProducerOutput:
|
||||
identity = "aios.py-compile"
|
||||
checkout = ctx.checkout
|
||||
if checkout is None or not checkout.is_dir():
|
||||
return unknown_output(ctx.context, identity, "checkout_unavailable")
|
||||
app = checkout / "active-runtime" / "app"
|
||||
targets = sorted(str(p) for p in app.glob("*.py")) if app.is_dir() else []
|
||||
if not targets:
|
||||
return fail_output(ctx.context, identity, "no_python_targets")
|
||||
|
||||
from . import _ACTIVE_RUNNER
|
||||
result = _ACTIVE_RUNNER(["python", "-m", "py_compile", *targets], checkout)
|
||||
passed = result.returncode == 0
|
||||
detail = {
|
||||
"compiled_count": len(targets),
|
||||
"exit_code": result.returncode,
|
||||
"checked_abs": False,
|
||||
}
|
||||
producer = ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
return producer
|
||||
@@ -0,0 +1,99 @@
|
||||
"""reconciliation-contract producer: pure convergence-contract self-consistency.
|
||||
|
||||
Orchestrates the existing pure validators
|
||||
(``aios_gitops_reconciliation_core.decide_reconciliation``,
|
||||
``reconciliation_controller.ReconciliationController.reconcile`` and
|
||||
``aios_convergence_contracts.plan_reconciliation``) and asserts the contract is
|
||||
self-consistent (no drift / converged / no unknown observation). Pure and fully
|
||||
local-runnable.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
|
||||
def run(
|
||||
ctx: ProducerContext,
|
||||
*,
|
||||
observation: dict | None = None,
|
||||
controller_state: dict | None = None,
|
||||
) -> ProducerOutput:
|
||||
identity = "aios.reconciliation-contract"
|
||||
from aios_gitops_reconciliation_core import RemoteObservation, decide_reconciliation
|
||||
from reconciliation_controller import ReconciliationController
|
||||
from aios_convergence_contracts import (
|
||||
DesiredState,
|
||||
ObservedState,
|
||||
ObservedHealth,
|
||||
plan_reconciliation,
|
||||
)
|
||||
|
||||
# apply decide_reconciliation on a healthy observation
|
||||
obs = observation or {
|
||||
"repository_identity": ctx.repository,
|
||||
"branch": "agent/status-pipeline-check",
|
||||
"desired_oid": ctx.commit_sha,
|
||||
"authority_oid": ctx.commit_sha,
|
||||
"remote_git_oid": ctx.commit_sha,
|
||||
"remote_api_oid": ctx.commit_sha,
|
||||
"commit_visible": True,
|
||||
"observed_at": "1970-01-01T00:00:00Z",
|
||||
"expected_oid": ctx.commit_sha,
|
||||
}
|
||||
decision = decide_reconciliation(RemoteObservation(**obs))
|
||||
|
||||
# apply plan_reconciliation on healthy observed state
|
||||
desired = DesiredState("/status-pipeline", "git_status", 1, {"head": ctx.commit_sha})
|
||||
observed = ObservedState(
|
||||
"/status-pipeline",
|
||||
1,
|
||||
{"head": ctx.commit_sha},
|
||||
ObservedHealth.HEALTHY,
|
||||
"reconciliation-contract",
|
||||
"1970-01-01T00:00:00Z",
|
||||
)
|
||||
plan = plan_reconciliation(desired, observed)
|
||||
|
||||
# apply controller reconcile on a no-drift state set
|
||||
controller = ReconciliationController()
|
||||
state = controller_state or {
|
||||
"authority_main": ctx.commit_sha,
|
||||
"gitea_main": ctx.commit_sha,
|
||||
"local_tag_oid": "t",
|
||||
"remote_tag_oid": "t",
|
||||
"remote_peeled_target": ctx.commit_sha,
|
||||
"approved_target_oid": ctx.commit_sha,
|
||||
"registry_version": "1",
|
||||
"baseline_version": "1",
|
||||
"approved_artifact_digest": "d",
|
||||
"deployed_artifact_digest": "d",
|
||||
"declared_production_state": True,
|
||||
"service_healthy": True,
|
||||
}
|
||||
report = controller.reconcile(state)
|
||||
|
||||
consistent = (
|
||||
decision.state == "REMOTE_VERIFIED"
|
||||
and plan.state.value == "CONVERGED"
|
||||
and report.healthy is True
|
||||
)
|
||||
detail = {
|
||||
"decide_reconciliation": decision.state,
|
||||
"plan_reconciliation": plan.state.value,
|
||||
"controller_healthy": report.healthy,
|
||||
}
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if consistent else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
@@ -0,0 +1,46 @@
|
||||
"""release-state-machine producer: fail-closed reachability of a legal path.
|
||||
|
||||
Walks the explicit ``ReleaseStateMachine`` transitions and asserts a legal,
|
||||
non-fail-closed path exists from SOURCE_READY to PR_APPROVED. A gate that
|
||||
demands missing evidence fails closed and yields FAIL, never success.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
)
|
||||
|
||||
|
||||
def run(ctx: ProducerContext) -> ProducerOutput:
|
||||
identity = "aios.release-state-machine"
|
||||
from release_state_machine import (
|
||||
ReleaseState,
|
||||
ReleaseStateMachine,
|
||||
InvalidTransition,
|
||||
)
|
||||
|
||||
legal_path = (
|
||||
ReleaseState.SOURCE_READY,
|
||||
ReleaseState.TESTED,
|
||||
ReleaseState.PR_OPEN,
|
||||
ReleaseState.PR_APPROVED,
|
||||
)
|
||||
sm = ReleaseStateMachine()
|
||||
try:
|
||||
for target in legal_path[1:]:
|
||||
sm.transition(target)
|
||||
except InvalidTransition as exc:
|
||||
return fail_output(ctx.context, identity, f"illegal_transition:{exc}")
|
||||
|
||||
reached = sm.state == ReleaseState.PR_APPROVED
|
||||
detail = {"reached": sm.state.value, "legal_path": [s.value for s in legal_path]}
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if reached else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
@@ -0,0 +1,98 @@
|
||||
"""secret-scan producer: real gitleaks scanner on the head checkout.
|
||||
|
||||
Reuses the governed ``run_secret_scan_provider`` (gitleaks ``--no-git``) and the
|
||||
repository root ``.gitleaks.toml`` allowlist. A scanner that is unavailable or
|
||||
crashes is UNKNOWN/FAIL — never an automatic pass.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
_DEFAULT_GITLEAKS_BIN = "/usr/bin/gitleaks"
|
||||
|
||||
|
||||
def _resolve_gitleaks_bin(gitleaks_bin: str | None) -> str:
|
||||
"""Resolve the scanner binary from (in order): caller arg, env, config, default.
|
||||
|
||||
The explicit ``.gitleaks.toml``-style config in
|
||||
``config/gitops/status-pipeline.json`` is consumed as a fallback so the
|
||||
operator's declared path is honored. Returns the path string; presence on
|
||||
disk is validated separately so an unusable binary fails closed to UNKNOWN.
|
||||
"""
|
||||
if gitleaks_bin:
|
||||
return gitleaks_bin
|
||||
env_path = os.environ.get("AIOS_GITLEAKS_BIN", "").strip()
|
||||
if env_path:
|
||||
return env_path
|
||||
cfg = Path(__file__).resolve().parents[3] / "config" / "gitops" / "status-pipeline.json"
|
||||
if cfg.is_file():
|
||||
try:
|
||||
decoded = json.loads(cfg.read_text(encoding="utf-8"))
|
||||
configured = (decoded.get("secret_scan") or {}).get("gitleaks_bin")
|
||||
if isinstance(configured, str) and configured.strip():
|
||||
return configured.strip()
|
||||
except (OSError, json.JSONDecodeError):
|
||||
pass
|
||||
return _DEFAULT_GITLEAKS_BIN
|
||||
|
||||
|
||||
def run(
|
||||
ctx: ProducerContext,
|
||||
*,
|
||||
gitleaks_bin: str | None = None,
|
||||
allowlist_toml: str = ".gitleaks.toml",
|
||||
) -> ProducerOutput:
|
||||
identity = "aios.secret-scanner"
|
||||
checkout = ctx.checkout
|
||||
if checkout is None or not checkout.is_dir():
|
||||
return unknown_output(ctx.context, identity, "checkout_unavailable")
|
||||
|
||||
# Resolve the scanner binary and validate it really exists on disk.
|
||||
# Absent -> UNKNOWN (fail-closed, never success).
|
||||
resolved_bin = _resolve_gitleaks_bin(gitleaks_bin)
|
||||
if not Path(resolved_bin).is_file():
|
||||
return unknown_output(ctx.context, identity, "secret_scanner_unavailable")
|
||||
|
||||
allowlist = checkout / allowlist_toml
|
||||
if not allowlist.is_file():
|
||||
# The governed repo is expected to carry an allowlist; a manifest-proof
|
||||
# absence is fail-closed, not silently accepted.
|
||||
return fail_output(ctx.context, identity, "gitleaks_allowlist_missing")
|
||||
|
||||
try:
|
||||
from trusted_evidence_providers import run_secret_scan_provider
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return unknown_output(ctx.context, identity, f"provider_unavailable:{exc}")
|
||||
|
||||
try:
|
||||
result, ref = run_secret_scan_provider(
|
||||
gitleaks_bin=resolved_bin,
|
||||
target=checkout,
|
||||
source_oid=ctx.commit_sha,
|
||||
tree_oid="",
|
||||
provider_identity=identity,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return unknown_output(ctx.context, identity, f"secret_scan_error:{exc}")
|
||||
|
||||
passed = bool(result.get("passed"))
|
||||
detail = dict(result)
|
||||
detail["allowlist_toml"] = allowlist_toml
|
||||
detail["receipt_ref"] = ref
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
detail,
|
||||
)
|
||||
@@ -0,0 +1,75 @@
|
||||
"""signer-validator-contract producer.
|
||||
|
||||
Scope is classified via ``producer_scope_classifier.classify_na``:
|
||||
- APPLICABLE -> require a real signer/validator run
|
||||
(``signer_contract_compatibility.run_compatible_validator``); an unavailable
|
||||
toolchain is UNKNOWN, never an invented PASS.
|
||||
- FORMALLY_NOT_APPLICABLE -> emit an NA verdict bound to the applicability
|
||||
contract ref.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from producer_scope_classifier import (
|
||||
ScopeClassificationError,
|
||||
classify_na,
|
||||
)
|
||||
from . import (
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
fail_output,
|
||||
unknown_output,
|
||||
)
|
||||
|
||||
|
||||
def run(
|
||||
ctx: ProducerContext,
|
||||
*,
|
||||
signer_facts: dict | None = None,
|
||||
grant_dir: Path | None = None,
|
||||
) -> ProducerOutput:
|
||||
identity = "aios.signer-validator-contract"
|
||||
facts = signer_facts if signer_facts is not None else dict(ctx.scope_facts)
|
||||
try:
|
||||
verdict = classify_na(ctx.context, facts)
|
||||
except ScopeClassificationError as exc:
|
||||
return unknown_output(ctx.context, identity, f"scope_unknown:{exc}")
|
||||
|
||||
if verdict.formally_not_applicable:
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"FORMALLY_NOT_APPLICABLE",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
{"reason": verdict.reason, "scope_facts": dict(facts), "applicable": False},
|
||||
na_contract_ref=verdict.na_contract_ref,
|
||||
)
|
||||
|
||||
# APPLICABLE: requires a real signed grant to validate.
|
||||
if grant_dir is None or not grant_dir.is_dir():
|
||||
return unknown_output(ctx.context, identity, "signed_grant_unavailable")
|
||||
try:
|
||||
from signer_contract_compatibility import run_compatible_validator
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return unknown_output(ctx.context, identity, f"signer_module_unavailable:{exc}")
|
||||
try:
|
||||
result_path = grant_dir / "compat-validator-result.json"
|
||||
run_compatible_validator(
|
||||
["aios-compatible-validator"],
|
||||
grant_dir,
|
||||
result_path,
|
||||
)
|
||||
payload = (grant_dir / "grant-payload.json").is_file()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return fail_output(ctx.context, identity, f"validator_failed:{exc}")
|
||||
|
||||
passed = payload
|
||||
return ProducerOutput(
|
||||
ctx.context,
|
||||
"APPLICABLE_PASS" if passed else "FAIL",
|
||||
identity,
|
||||
PRODUCER_REVISION,
|
||||
{"grant_validated": bool(payload), "applicable": True},
|
||||
)
|
||||
@@ -0,0 +1,311 @@
|
||||
"""Idempotent, capability-minimal status publisher for protected main.
|
||||
|
||||
This is the *only* component licensed to write Gitea commit statuses for the
|
||||
protected-main pipeline (architecture §2.4). It:
|
||||
|
||||
- binds every status to an explicit exact commit SHA via
|
||||
``GiteaMachineBroker.status_for_sha`` (never a moving branch tip);
|
||||
- derives the Gitea state purely from the evidence ``result`` (never accepts a
|
||||
caller-supplied ``state=success``);
|
||||
- is idempotent: the same effect_id (repo\0sha\0context\0evidence_revision) is a
|
||||
NOOP and is not re-posted;
|
||||
- on an unknown POST outcome performs a bounded RECONCILE (GET statuses) and
|
||||
never blind-posts success after a timeout;
|
||||
- persists every evidence record to an append-only store (O_EXCL + fsync) and
|
||||
never deletes history.
|
||||
|
||||
Only the ``status_publish`` capability is required (enforced inside the broker's
|
||||
``status_for_sha``); the publisher additionally refuses a receipt that grants
|
||||
any dangerous capability that the gitops denied-set should have excluded.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Mapping
|
||||
|
||||
from status_evidence_contract import (
|
||||
StatusEvidence,
|
||||
maps_to_success,
|
||||
sha256_hex,
|
||||
validate_commit_sha,
|
||||
validate_context,
|
||||
validate_repository,
|
||||
)
|
||||
|
||||
# Dangerous capabilities that must never be co-granted with status_publish on
|
||||
# the publisher identity (aligned with aios_gitops_job_executor.py:502 denied set).
|
||||
DANGEROUS_CAPABILITIES: frozenset[str] = frozenset(
|
||||
{
|
||||
"direct_main_push",
|
||||
"force_push",
|
||||
"tag_update",
|
||||
"tag_delete",
|
||||
"baseline_write",
|
||||
"production_deploy",
|
||||
"protected_merge",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class StatusPublishError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def gitea_state_for_result(result: str, *, has_na_contract_ref: bool) -> str:
|
||||
"""Map a deterministic evidence result to a Gitea status state.
|
||||
|
||||
success is only ever produced for APPLICABLE_PASS or FORMALLY_NOT_APPLICABLE
|
||||
(the latter only when an applicability contract ref is present - enforced by
|
||||
StatusEvidence validation). FAIL/UNKNOWN yield failure/error and are never
|
||||
success. This is the single authorized projection point.
|
||||
"""
|
||||
if result == "APPLICABLE_PASS":
|
||||
return "success"
|
||||
if result == "FORMALLY_NOT_APPLICABLE":
|
||||
if not has_na_contract_ref:
|
||||
raise StatusPublishError("NA_WITHOUT_CONTRACT_REF")
|
||||
return "success"
|
||||
if result == "FAIL":
|
||||
return "failure"
|
||||
if result == "UNKNOWN":
|
||||
return "error"
|
||||
raise StatusPublishError(f"INVALID_EVIDENCE_RESULT:{result}")
|
||||
|
||||
|
||||
class StatusPublisherError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class AppendOnlyEvidenceStore:
|
||||
"""Append-only evidence store (O_EXCL + fsync), never deletes history.
|
||||
|
||||
In production this directory is root-owned and non-group-writable (mode
|
||||
0700). The vendor style mirrors ``trusted_evidence_providers._write_receipt_exclusive`` /
|
||||
``gitea_transport_consistency_provider._write_receipt_exclusive``; no root
|
||||
elevate is asserted here so the store remains testable on any platform,
|
||||
but the deployment must place it under a root-owned path.
|
||||
"""
|
||||
|
||||
def __init__(self, root: Path) -> None:
|
||||
self.root = Path(root)
|
||||
|
||||
def _path_for(self, effect_id: str) -> Path:
|
||||
digest = str(effect_id or "").strip()
|
||||
if not digest.startswith("sha256:") or len(digest) != 7 + 64:
|
||||
raise StatusPublisherError("EFFECT_ID_INVALID")
|
||||
# derive a safe relative ref with no traversal
|
||||
leaf = digest.split(":", 1)[1]
|
||||
return self.root / f"{leaf[:2]}" / f"{leaf}.json"
|
||||
|
||||
def has(self, effect_id: str) -> bool:
|
||||
return self._path_for(effect_id).is_file()
|
||||
|
||||
def append(self, effect_id: str, record: Mapping[str, Any]) -> str:
|
||||
target = self._path_for(effect_id)
|
||||
data = json.dumps(
|
||||
record, sort_keys=True, separators=(",", ":"), ensure_ascii=False
|
||||
).encode("utf-8")
|
||||
if target.exists() or target.is_symlink():
|
||||
raise StatusPublisherError("EFFECT_ALREADY_PERSISTED")
|
||||
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
||||
flags |= getattr(os, "O_NOFOLLOW", 0)
|
||||
flags |= getattr(os, "O_BINARY", 0)
|
||||
try:
|
||||
target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
fd = os.open(target, flags, 0o600)
|
||||
except OSError as exc:
|
||||
raise StatusPublisherError("EVIDENCE_WRITE_DENIED") from exc
|
||||
try:
|
||||
view = memoryview(data)
|
||||
while view:
|
||||
written = os.write(fd, view)
|
||||
if written <= 0:
|
||||
raise StatusPublisherError("EVIDENCE_WRITE_SHORT")
|
||||
view = view[written:]
|
||||
os.fsync(fd)
|
||||
except OSError as exc:
|
||||
raise StatusPublisherError("EVIDENCE_WRITE_DENIED") from exc
|
||||
finally:
|
||||
os.close(fd)
|
||||
os.chmod(target, 0o600)
|
||||
return str(target)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PublishOutcome:
|
||||
action: str # PUBLISHED | SUPPRESS_NOOP | RECONCILED | UNKNOWN
|
||||
effect_id: str
|
||||
gitea_state: str | None
|
||||
reason: str
|
||||
# Whether this context's status is *confirmed* as the derived success state.
|
||||
# False for UNKNOWN and for a RECONCILE that hit a conflicting leftover state
|
||||
# (existing_state_differs): such results must NEVER be credited as green.
|
||||
confirmed: bool = False
|
||||
reconciled_from_status: Mapping[str, Any] | None = None
|
||||
|
||||
|
||||
class StatusPublisher:
|
||||
def __init__(
|
||||
self,
|
||||
broker: Any,
|
||||
store: AppendOnlyEvidenceStore,
|
||||
*,
|
||||
receipt_capabilities: Mapping[str, Any] | None = None,
|
||||
retry_reconcile: bool = True,
|
||||
) -> None:
|
||||
"""broker must expose ``status_for_sha(sha, state, description, context)`` and
|
||||
``read_statuses(sha, context)`` (duck-typed for testability)."""
|
||||
self.broker = broker
|
||||
self.store = store
|
||||
self._receipt_caps = receipt_capabilities
|
||||
self.retry_reconcile = retry_reconcile
|
||||
|
||||
def verify_minimal_capability(self) -> None:
|
||||
caps = self._receipt_caps
|
||||
if caps is None:
|
||||
# Nothing to introspect (tests pass a fake); skip.
|
||||
return
|
||||
capabilities = caps.get("capabilities") or ()
|
||||
granted = set(capabilities)
|
||||
overlap = granted & DANGEROUS_CAPABILITIES
|
||||
if overlap:
|
||||
raise StatusPublisherError(
|
||||
"CAPABILITY_MINIMIZATION_VIOLATED:" + ",".join(sorted(overlap))
|
||||
)
|
||||
if "status_publish" not in granted:
|
||||
raise StatusPublisherError("CAPABILITY_STATUS_PUBLISH_MISSING")
|
||||
|
||||
def publish(self, evidence: StatusEvidence) -> PublishOutcome:
|
||||
self.verify_minimal_capability()
|
||||
record = evidence.finalize()
|
||||
effect_id = record.effect_id
|
||||
# Idempotency: identical effect already persisted -> SUPPRESS_NOOP.
|
||||
# A replay is only *confirmed* as green when a readback verifies the
|
||||
# derived state already exists; otherwise it stays unconfirmed (a prior
|
||||
# run may have persisted evidence but never confirmed the POST).
|
||||
if self.store.has(effect_id):
|
||||
confirmed = self._readback_confirmed(record)
|
||||
return PublishOutcome(
|
||||
"SUPPRESS_NOOP", effect_id, None, "duplicate_effect", confirmed=confirmed
|
||||
)
|
||||
|
||||
gitea_state = gitea_state_for_result(
|
||||
record.result, has_na_contract_ref=bool(record.na_contract_ref)
|
||||
)
|
||||
|
||||
# Persist the evidence record *before* any POST so a lost response never
|
||||
# loses the evidence itself. Ordering is explicit and not overwritten.
|
||||
try:
|
||||
evidence_ref = self.store.append(effect_id, record.as_dict())
|
||||
except StatusPublisherError as exc:
|
||||
# A concurrent writer persisted the same effect first (EEXIST). This
|
||||
# is an idempotent replay, not an ERROR: never abort the round and
|
||||
# never count green unless a readback confirms the success state.
|
||||
if "EFFECT_ALREADY_PERSISTED" in str(exc):
|
||||
confirmed = self._readback_confirmed(record)
|
||||
return PublishOutcome(
|
||||
"SUPPRESS_NOOP",
|
||||
effect_id,
|
||||
None,
|
||||
"concurrent_duplicate",
|
||||
confirmed=confirmed,
|
||||
)
|
||||
raise
|
||||
|
||||
description = f"{record.context} -> {record.result}"
|
||||
try:
|
||||
self.broker.status_for_sha(
|
||||
record.commit_sha, gitea_state, description, record.context
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 - any POST uncertainty -> RECONCILE
|
||||
resolved = self._reconcile(record, gitea_state, effect_id)
|
||||
if resolved is not None:
|
||||
return resolved
|
||||
# Fail closed: an unconfirmed POST that could not be reconciled is
|
||||
# reported as error (never success) and never confirmed.
|
||||
return PublishOutcome(
|
||||
"UNKNOWN",
|
||||
effect_id,
|
||||
"error",
|
||||
f"post_unknown:{exc}",
|
||||
confirmed=False,
|
||||
)
|
||||
return PublishOutcome(
|
||||
"PUBLISHED", effect_id, gitea_state, "posted", confirmed=True, reconciled_from_status=None
|
||||
)
|
||||
|
||||
def _readback_confirmed(self, record: StatusEvidence) -> bool:
|
||||
"""Confirm a persisted effect by reading Gitea's real status.
|
||||
|
||||
A replay is only a confirmed success when the readback shows the derived
|
||||
state (e.g. success) already present for this SHA+context. Absent,
|
||||
errored, or conflicting (e.g. leftover) reads are fail-closed -> False.
|
||||
"""
|
||||
expected = gitea_state_for_result(
|
||||
record.result, has_na_contract_ref=bool(record.na_contract_ref)
|
||||
)
|
||||
if expected != "success":
|
||||
return False
|
||||
try:
|
||||
statuses = self.broker.read_statuses(record.commit_sha, record.context)
|
||||
except Exception: # noqa: BLE001
|
||||
return False
|
||||
for row in statuses:
|
||||
if not isinstance(row, Mapping):
|
||||
continue
|
||||
if row.get("context") == record.context and row.get("state") == "success":
|
||||
return True
|
||||
return False
|
||||
|
||||
def _reconcile(
|
||||
self, record: StatusEvidence, gitea_state: str, effect_id: str
|
||||
) -> PublishOutcome | None:
|
||||
"""Bounded RECONCILE: query Gitea statuses for the exact SHA+context.
|
||||
|
||||
Returns a resolved outcome or None when still unknown. Never posts
|
||||
success blindly; never overwrites a conflicting existing status.
|
||||
"""
|
||||
try:
|
||||
statuses = self.broker.read_statuses(record.commit_sha, record.context)
|
||||
except Exception: # noqa: BLE001
|
||||
if self.retry_reconcile:
|
||||
try:
|
||||
statuses = self.broker.read_statuses(record.commit_sha, record.context)
|
||||
except Exception: # noqa: BLE001
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
for row in statuses:
|
||||
if not isinstance(row, Mapping):
|
||||
continue
|
||||
if row.get("context") != record.context:
|
||||
continue
|
||||
actual_state = row.get("state")
|
||||
if actual_state == gitea_state:
|
||||
# The Gitea state matches exactly what the verdict projected:
|
||||
# this is a genuine confirmation of the derived state.
|
||||
return PublishOutcome(
|
||||
"RECONCILED",
|
||||
effect_id,
|
||||
actual_state,
|
||||
"matched_expected",
|
||||
confirmed=True,
|
||||
reconciled_from_status=dict(row),
|
||||
)
|
||||
# Conflicting pre-existing / leftover state (e.g. a stray success while
|
||||
# the verdict projects failure). Report the real state but mark it
|
||||
# NOT confirmed so the orchestrator never credits it as green.
|
||||
return PublishOutcome(
|
||||
"RECONCILED",
|
||||
effect_id,
|
||||
actual_state,
|
||||
"existing_state_differs",
|
||||
confirmed=False,
|
||||
reconciled_from_status=dict(row),
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,42 @@
|
||||
{
|
||||
"schema": "aios.status-pipeline-config.v1",
|
||||
"repository": "aios/aios",
|
||||
"checkout": "/data/aios-gitops/status-checkout",
|
||||
"evidence_store_root": "/var/lib/aios-gitops/status-evidence",
|
||||
"required_contexts": [
|
||||
"py-compile",
|
||||
"git-diff-check",
|
||||
"secret-scan",
|
||||
"autonomy-unit-tests",
|
||||
"autonomy-smoke-tests",
|
||||
"reconciliation-contract",
|
||||
"release-state-machine",
|
||||
"campaign-approval-gate",
|
||||
"signer-validator-contract"
|
||||
],
|
||||
"capabilities": {
|
||||
"granted": ["status_publish"],
|
||||
"denied": [
|
||||
"direct_main_push",
|
||||
"force_push",
|
||||
"tag_update",
|
||||
"tag_delete",
|
||||
"baseline_write",
|
||||
"production_deploy",
|
||||
"protected_merge"
|
||||
]
|
||||
},
|
||||
"na_contracts": {
|
||||
"campaign-approval-gate": "aios.campaign-approval-gate.v1#formally-not-applicable",
|
||||
"signer-validator-contract": "aios.signer-validator-contract.v1#formally-not-applicable"
|
||||
},
|
||||
"secret_scan": {
|
||||
"gitleaks_bin": "/usr/bin/gitleaks",
|
||||
"allowlist": ".gitleaks.toml"
|
||||
},
|
||||
"integration": {
|
||||
"trigger": "external-hosted (webhook/durable-job/poll) — not part of this pipeline; see status_pipeline_entry.py",
|
||||
"producer_commit": "pending-team-commit",
|
||||
"reviewed_tree_oid": "pending-open-code-review"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,587 @@
|
||||
"""Negative / safety tests for the protected-main status pipeline.
|
||||
|
||||
Runs fully locally with a fake broker and temp evidence store — no real Gitea,
|
||||
gitleaks, fcntl, or systemd. Covers architecture §3 (authority req §13) safety
|
||||
semantics: no forgery, no green-reporter, UNKNOWN != success, N/A contract
|
||||
binding, stale SHA, idempotency, RECONCILE, capability minimization, non-plaintext
|
||||
credential, and merge-only-on-genuine-pass.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import sys
|
||||
from pathlib import Path as _P
|
||||
|
||||
ROOT = _P(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
||||
|
||||
from aios_gitea_machine_broker import ( # noqa: E402
|
||||
CampaignCapabilityReceipt,
|
||||
GiteaMachineBroker,
|
||||
MachineCredentialProjection,
|
||||
MachineBrokerError,
|
||||
)
|
||||
from status_evidence_contract import ( # noqa: E402
|
||||
REQUIRED_CONTEXTS,
|
||||
StatusEvidence,
|
||||
maps_to_success,
|
||||
)
|
||||
from status_publisher import ( # noqa: E402
|
||||
AppendOnlyEvidenceStore,
|
||||
StatusPublisher,
|
||||
StatusPublisherError,
|
||||
gitea_state_for_result,
|
||||
)
|
||||
from status_orchestrator import StatusOrchestrator # noqa: E402
|
||||
from status_producers import ( # noqa: E402
|
||||
PRODUCER_REVISION,
|
||||
ProducerContext,
|
||||
ProducerOutput,
|
||||
set_command_runner,
|
||||
)
|
||||
|
||||
|
||||
def _sha(value: int) -> str:
|
||||
return f"{value:040d}"
|
||||
|
||||
|
||||
HEAD_A = _sha(1)
|
||||
HEAD_B = _sha(2)
|
||||
|
||||
|
||||
class FakeBroker:
|
||||
def __init__(self, *, fail_post=False, statuses=None, raise_read=False):
|
||||
self.posts = []
|
||||
self.fail_post = fail_post
|
||||
self._statuses = statuses or []
|
||||
self.raise_read = raise_read
|
||||
self.read_calls = 0
|
||||
|
||||
def status_for_sha(self, sha, state, description, context):
|
||||
self.posts.append({"sha": sha, "state": state, "context": context})
|
||||
if self.fail_post:
|
||||
raise RuntimeError("post_unknown")
|
||||
return {"ok": True}
|
||||
|
||||
def read_statuses(self, sha, context=None):
|
||||
self.read_calls += 1
|
||||
if self.raise_read:
|
||||
raise RuntimeError("read_unknown")
|
||||
return [s for s in self._statuses if context is None or s.get("context") == context]
|
||||
|
||||
|
||||
def _store():
|
||||
return AppendOnlyEvidenceStore(Path(tempfile.mkdtemp()))
|
||||
|
||||
|
||||
def _publisher(broker, store, caps=None):
|
||||
return StatusPublisher(broker, store, receipt_capabilities=caps)
|
||||
|
||||
|
||||
def _evidence(context, result, commit_sha=HEAD_A, na_ref=None):
|
||||
return StatusEvidence(
|
||||
repository="aios/aios",
|
||||
commit_sha=commit_sha,
|
||||
context=context,
|
||||
result=result,
|
||||
producer_identity=f"aios.{context}",
|
||||
producer_revision=PRODUCER_REVISION,
|
||||
na_contract_ref=na_ref,
|
||||
)
|
||||
|
||||
|
||||
class TestStateProjection(unittest.TestCase):
|
||||
def test_applicable_pass_is_success(self):
|
||||
self.assertEqual(gitea_state_for_result("APPLICABLE_PASS", has_na_contract_ref=False), "success")
|
||||
|
||||
def test_formally_na_is_success_only_with_contract(self):
|
||||
self.assertEqual(gitea_state_for_result("FORMALLY_NOT_APPLICABLE", has_na_contract_ref=True), "success")
|
||||
with self.assertRaises(Exception):
|
||||
gitea_state_for_result("FORMALLY_NOT_APPLICABLE", has_na_contract_ref=False)
|
||||
|
||||
def test_unknown_never_success(self):
|
||||
self.assertEqual(gitea_state_for_result("UNKNOWN", has_na_contract_ref=False), "error")
|
||||
self.assertNotEqual(gitea_state_for_result("UNKNOWN", has_na_contract_ref=False), "success")
|
||||
|
||||
def test_fail_never_success(self):
|
||||
self.assertEqual(gitea_state_for_result("FAIL", has_na_contract_ref=False), "failure")
|
||||
self.assertNotEqual(gitea_state_for_result("FAIL", has_na_contract_ref=False), "success")
|
||||
|
||||
def test_maps_to_success_redline(self):
|
||||
self.assertTrue(maps_to_success("APPLICABLE_PASS"))
|
||||
self.assertFalse(maps_to_success("UNKNOWN"))
|
||||
self.assertFalse(maps_to_success("FAIL"))
|
||||
|
||||
def test_no_universal_green_reporter(self):
|
||||
# gitea_state_for_result never yields success for non-pass results; a
|
||||
# "publish success for all" loop cannot be expressed via this projection.
|
||||
for result in ("UNKNOWN", "FAIL"):
|
||||
self.assertNotEqual(gitea_state_for_result(result, has_na_contract_ref=False), "success")
|
||||
|
||||
|
||||
class TestPublisher(unittest.TestCase):
|
||||
def test_publisher_never_posts_success_for_unknown(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store())
|
||||
ev = _evidence("py-compile", "UNKNOWN")
|
||||
outc = pub.publish(ev)
|
||||
self.assertEqual(outc.action, "PUBLISHED")
|
||||
self.assertEqual(outc.gitea_state, "error")
|
||||
self.assertTrue(all(p["state"] == "error" for p in broker.posts))
|
||||
|
||||
def test_fail_posts_failure(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store())
|
||||
outc = pub.publish(_evidence("py-compile", "FAIL"))
|
||||
self.assertEqual(outc.gitea_state, "failure")
|
||||
self.assertEqual(broker.posts[0]["state"], "failure")
|
||||
|
||||
def test_na_without_contract_forbidden(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store())
|
||||
# build evidence directly: NA without contract raises at finalize
|
||||
ev = _evidence("campaign-approval-gate", "FORMALLY_NOT_APPLICABLE", na_ref=None)
|
||||
with self.assertRaises(Exception):
|
||||
ev.finalize()
|
||||
|
||||
def test_applicable_pass_posts_success(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.gitea_state, "success")
|
||||
|
||||
def test_idempotent_suppress_noop(self):
|
||||
broker = FakeBroker()
|
||||
store = _store()
|
||||
pub = _publisher(broker, store)
|
||||
ev = _evidence("py-compile", "APPLICABLE_PASS")
|
||||
first = pub.publish(ev)
|
||||
second = pub.publish(ev)
|
||||
self.assertEqual(first.action, "PUBLISHED")
|
||||
self.assertEqual(second.action, "SUPPRESS_NOOP")
|
||||
self.assertEqual(len(broker.posts), 1)
|
||||
|
||||
def test_unknown_post_reconciles_to_expected(self):
|
||||
broker = FakeBroker(
|
||||
fail_post=True,
|
||||
statuses=[{"context": "py-compile", "state": "success"}],
|
||||
)
|
||||
pub = _publisher(broker, _store())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.action, "RECONCILED")
|
||||
self.assertEqual(outc.gitea_state, "success")
|
||||
|
||||
def test_unknown_post_and_unknown_read_is_unknown(self):
|
||||
broker = FakeBroker(fail_post=True, raise_read=True)
|
||||
pub = _publisher(broker, _store())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.action, "UNKNOWN")
|
||||
self.assertNotEqual(outc.gitea_state, "success")
|
||||
|
||||
def test_stale_sha_not_copied(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store())
|
||||
pub.publish(_evidence("py-compile", "APPLICABLE_PASS", commit_sha=HEAD_A))
|
||||
pub.publish(_evidence("py-compile", "APPLICABLE_PASS", commit_sha=HEAD_B))
|
||||
self.assertEqual([p["sha"] for p in broker.posts], [HEAD_A, HEAD_B])
|
||||
|
||||
def test_capability_minimization_forbidden(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(
|
||||
broker,
|
||||
_store(),
|
||||
caps={"capabilities": {"status_publish", "direct_main_push"}, "denied": set()},
|
||||
)
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertIn("CAPABILITY_MINIMIZATION_VIOLATED", str(ctx.exception))
|
||||
|
||||
def test_capability_status_publish_required(self):
|
||||
broker = FakeBroker()
|
||||
pub = _publisher(broker, _store(), caps={"capabilities": {"pr_create_update"}, "denied": set()})
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertIn("CAPABILITY_STATUS_PUBLISH_MISSING", str(ctx.exception))
|
||||
|
||||
|
||||
class TestCredentialNonPlaintext(unittest.TestCase):
|
||||
def test_plaintext_scheme_rejected(self):
|
||||
with self.assertRaises(MachineBrokerError):
|
||||
MachineCredentialProjection.from_mapping(
|
||||
{"scheme": "plain", "name": "gitea-machine-token", "account": "aios-autonomy-agent"}
|
||||
)
|
||||
|
||||
def test_owner_identity_rejected(self):
|
||||
with self.assertRaises(MachineBrokerError):
|
||||
MachineCredentialProjection.from_mapping(
|
||||
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": "aiman"}
|
||||
)
|
||||
|
||||
|
||||
class TestBrokerPermissionDenied(unittest.TestCase):
|
||||
def _broker_without_status(self):
|
||||
transport = FakeBroker()
|
||||
credential = MachineCredentialProjection.from_mapping(
|
||||
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": "aios-autonomy-agent"}
|
||||
)
|
||||
receipt = CampaignCapabilityReceipt(
|
||||
"c1", "t1", "aios/aios", "agent/dev", _sha(3),
|
||||
frozenset({"pr_create_update"}), frozenset(),
|
||||
)
|
||||
return GiteaMachineBroker(transport, credential, receipt)
|
||||
|
||||
def test_status_for_sha_denied_without_capability(self):
|
||||
broker = self._broker_without_status()
|
||||
with self.assertRaises(MachineBrokerError) as ctx:
|
||||
broker.status_for_sha(HEAD_A, "success", "d", "py-compile")
|
||||
self.assertEqual(str(ctx.exception), "CAPABILITY_DENIED")
|
||||
|
||||
|
||||
def _pass_producer(context):
|
||||
def run(ctx):
|
||||
return ProducerOutput(context, "APPLICABLE_PASS", f"aios.{context}", PRODUCER_REVISION, {})
|
||||
return run
|
||||
|
||||
|
||||
def _unknown_producer(context):
|
||||
def run(ctx):
|
||||
return ProducerOutput(context, "UNKNOWN", f"aios.{context}", PRODUCER_REVISION, {})
|
||||
return run
|
||||
|
||||
|
||||
def _fail_producer(context):
|
||||
def run(ctx):
|
||||
return ProducerOutput(context, "FAIL", f"aios.{context}", PRODUCER_REVISION, {})
|
||||
return run
|
||||
|
||||
|
||||
class TestOrchestratorMerge(unittest.TestCase):
|
||||
def _make(self, producer_fn):
|
||||
producers = {c: producer_fn(c) for c in REQUIRED_CONTEXTS}
|
||||
broker = FakeBroker()
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
return orch, broker
|
||||
|
||||
def test_merge_only_when_all_genuine_pass(self):
|
||||
orch, _ = self._make(_pass_producer)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertTrue(result.merge_eligible)
|
||||
self.assertEqual(len(result.context_outcomes), 9)
|
||||
|
||||
def test_unknown_blocks_merge(self):
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
producers["py-compile"] = _unknown_producer("py-compile")
|
||||
broker = FakeBroker()
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
def test_head_change_invalidates_old(self):
|
||||
orch, broker = self._make(_pass_producer)
|
||||
orch.run_for_head(HEAD_A)
|
||||
broker.posts.clear()
|
||||
orch.run_for_head(HEAD_B)
|
||||
# second run must only touch the new head
|
||||
self.assertTrue(broker.posts)
|
||||
self.assertTrue(all(p["sha"] == HEAD_B for p in broker.posts))
|
||||
|
||||
def test_intended_success_but_post_unknown_blocks_merge(self):
|
||||
# Producer verdicts are all PASS, but the publisher cannot confirm any
|
||||
# POST and reconciliation is unknown -> merge must be blocked (never
|
||||
# treat an unconfirmed success as green).
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
broker = FakeBroker(fail_post=True, raise_read=True)
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
# every context ended UNKNOWN (not confirmed published), so no green is credited
|
||||
self.assertTrue(
|
||||
all(
|
||||
o.publish.action == "UNKNOWN"
|
||||
for o in result.context_outcomes.values()
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestExistingStateDiffersNotGreen(unittest.TestCase):
|
||||
"""Red-line: a conflicting leftover status must never credit a FAIL/UNKNOWN
|
||||
verdict as green, and an unconfirmed reconcile must never count."""
|
||||
|
||||
def test_reconcile_conflict_marks_unconfirmed(self):
|
||||
broker = FakeBroker(
|
||||
fail_post=True, statuses=[{"context": "py-compile", "state": "success"}]
|
||||
)
|
||||
pub = _publisher(broker, _store())
|
||||
ev = _evidence("py-compile", "FAIL")
|
||||
outc = pub.publish(ev)
|
||||
self.assertEqual(outc.action, "RECONCILED")
|
||||
self.assertEqual(outc.gitea_state, "success") # reports the leftover state
|
||||
self.assertFalse(outc.confirmed) # ...but NOT confirmed as green
|
||||
|
||||
def test_fail_verdict_with_leftover_success_blocks_merge(self):
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
producers["py-compile"] = _fail_producer("py-compile")
|
||||
broker = FakeBroker(
|
||||
fail_post=True,
|
||||
statuses=[{"context": c, "state": "success"} for c in REQUIRED_CONTEXTS],
|
||||
)
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
def test_unknown_verdict_with_leftover_success_blocks_merge(self):
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
producers["py-compile"] = _unknown_producer("py-compile")
|
||||
broker = FakeBroker(
|
||||
fail_post=True,
|
||||
statuses=[{"context": c, "state": "success"} for c in REQUIRED_CONTEXTS],
|
||||
)
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
def test_replay_of_confirmed_success_is_green(self):
|
||||
# Vulnerability 2: a replayed (idempotent) effect that was previously
|
||||
# confirmed as success must remain merge-eligible.
|
||||
broker = FakeBroker()
|
||||
store = _store()
|
||||
pub = _publisher(broker, store)
|
||||
ev = _evidence("py-compile", "APPLICABLE_PASS")
|
||||
first = pub.publish(ev) # PUBLISHED
|
||||
self.assertTrue(first.confirmed)
|
||||
# readback now reflects the posted success
|
||||
broker._statuses = [{"context": "py-compile", "state": "success"}]
|
||||
second = pub.publish(ev) # SUPPRESS_NOOP
|
||||
self.assertEqual(second.action, "SUPPRESS_NOOP")
|
||||
self.assertTrue(second.confirmed)
|
||||
|
||||
def test_replay_of_unconfirmed_is_not_green(self):
|
||||
broker = FakeBroker() # no readback status -> unconfirmed
|
||||
store = _store()
|
||||
pub = _publisher(broker, store)
|
||||
ev = _evidence("py-compile", "APPLICABLE_PASS")
|
||||
first = pub.publish(ev)
|
||||
second = pub.publish(ev) # SUPPRESS_NOOP but readback returns nothing
|
||||
self.assertEqual(second.action, "SUPPRESS_NOOP")
|
||||
self.assertFalse(second.confirmed)
|
||||
|
||||
|
||||
class TestConcurrentDuplicate(unittest.TestCase):
|
||||
"""EF#2: a concurrent EEXIST is an idempotent replay, never a round ERROR."""
|
||||
|
||||
class _ConcurrentStore:
|
||||
def has(self, effect_id):
|
||||
return False
|
||||
|
||||
def append(self, effect_id, record):
|
||||
raise StatusPublisherError("EFFECT_ALREADY_PERSISTED")
|
||||
|
||||
def test_concurrent_eexist_is_suppress_noop_not_error(self):
|
||||
broker = FakeBroker() # no readback -> unconfirmed
|
||||
pub = StatusPublisher(broker, self._ConcurrentStore())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.action, "SUPPRESS_NOOP")
|
||||
self.assertEqual(outc.reason, "concurrent_duplicate")
|
||||
self.assertFalse(outc.confirmed)
|
||||
|
||||
def test_concurrent_eexist_confirmed_when_readback_success(self):
|
||||
broker = FakeBroker(statuses=[{"context": "py-compile", "state": "success"}])
|
||||
pub = StatusPublisher(broker, self._ConcurrentStore())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.action, "SUPPRESS_NOOP")
|
||||
self.assertTrue(outc.confirmed)
|
||||
|
||||
|
||||
class TestInjectionContract(unittest.TestCase):
|
||||
"""EF#1: optional injection channels let real scope/smoke data flow in,
|
||||
without ever fabricating a pass (absent data stays UNKNOWN fail-closed)."""
|
||||
|
||||
def test_smoke_no_results_is_unknown(self):
|
||||
from status_producers import autonomy_smoke_tests, ProducerContext
|
||||
out = autonomy_smoke_tests.run(
|
||||
ProducerContext(context="autonomy-smoke-tests", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={})
|
||||
)
|
||||
self.assertEqual(out.result, "UNKNOWN")
|
||||
|
||||
def test_smoke_injected_pass_goes_green(self):
|
||||
from status_producers import autonomy_smoke_tests, ProducerContext
|
||||
out = autonomy_smoke_tests.run(
|
||||
ProducerContext(
|
||||
context="autonomy-smoke-tests",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={"smoke_results": {"imports_load": True, "core_services_respond": True}},
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "APPLICABLE_PASS")
|
||||
|
||||
def test_smoke_injected_partial_is_fail(self):
|
||||
from status_producers import autonomy_smoke_tests, ProducerContext
|
||||
out = autonomy_smoke_tests.run(
|
||||
ProducerContext(
|
||||
context="autonomy-smoke-tests",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={"smoke_results": {"imports_load": True}},
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "FAIL")
|
||||
|
||||
def test_signer_not_injected_is_unknown(self):
|
||||
from status_producers import signer_validator_contract, ProducerContext
|
||||
out = signer_validator_contract.run(
|
||||
ProducerContext(context="signer-validator-contract", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={})
|
||||
)
|
||||
self.assertEqual(out.result, "UNKNOWN")
|
||||
|
||||
def test_signer_injected_not_applicable_has_contract(self):
|
||||
from status_producers import signer_validator_contract, ProducerContext
|
||||
out = signer_validator_contract.run(
|
||||
ProducerContext(
|
||||
context="signer-validator-contract",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={"touches_signer_bound_change": False},
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "FORMALLY_NOT_APPLICABLE")
|
||||
self.assertTrue(out.na_contract_ref)
|
||||
|
||||
def test_campaign_injected_na_has_contract(self):
|
||||
from status_producers import campaign_approval_gate, ProducerContext
|
||||
out = campaign_approval_gate.run(
|
||||
ProducerContext(
|
||||
context="campaign-approval-gate",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={
|
||||
"campaign_authorization_present": False,
|
||||
"campaign_active": False,
|
||||
"in_campaign_lineage": False,
|
||||
},
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "FORMALLY_NOT_APPLICABLE")
|
||||
self.assertTrue(out.na_contract_ref)
|
||||
|
||||
def test_injected_na_contexts_and_smoke_reach_full_green(self):
|
||||
# Full 9/9 traceability: PASS producers everywhere, plus injected N/A and
|
||||
# smoke facts, yields merge-eligible — proving the contract layer can
|
||||
# genuinely drive 9/9 when real facts are supplied (nothing fabricated).
|
||||
from status_producers import ProducerOutput, PRODUCER_REVISION
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
|
||||
def na_campaign(ctx):
|
||||
return ProducerOutput(ctx.context, "FORMALLY_NOT_APPLICABLE", "aios.campaign-approval-gate", PRODUCER_REVISION, {}, na_contract_ref="aios.campaign-approval-gate.v1#formally-not-applicable")
|
||||
producers["campaign-approval-gate"] = na_campaign
|
||||
|
||||
def na_signer(ctx):
|
||||
return ProducerOutput(ctx.context, "FORMALLY_NOT_APPLICABLE", "aios.signer-validator-contract", PRODUCER_REVISION, {}, na_contract_ref="aios.signer-validator-contract.v1#formally-not-applicable")
|
||||
producers["signer-validator-contract"] = na_signer
|
||||
|
||||
def smoke_via_scope(ctx):
|
||||
from status_producers import autonomy_smoke_tests
|
||||
return autonomy_smoke_tests.run(ctx)
|
||||
producers["autonomy-smoke-tests"] = smoke_via_scope
|
||||
|
||||
broker = FakeBroker()
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
extra_facts={"smoke_results": {"imports_load": True, "core_services_respond": True}},
|
||||
)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertTrue(result.merge_eligible)
|
||||
self.assertEqual(
|
||||
result.context_outcomes["autonomy-smoke-tests"].producer_output.result,
|
||||
"APPLICABLE_PASS",
|
||||
)
|
||||
|
||||
|
||||
class TestSecretScanNoAutoPass(unittest.TestCase):
|
||||
def test_scanner_unavailable_is_unknown(self):
|
||||
from status_producers import secret_scan
|
||||
out = secret_scan.run(
|
||||
ProducerContext(context="secret-scan", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={}),
|
||||
gitleaks_bin="/nonexistent/gitleaks",
|
||||
)
|
||||
self.assertEqual(out.result, "UNKNOWN")
|
||||
self.assertNotEqual(out.result, "APPLICABLE_PASS")
|
||||
|
||||
|
||||
class TestCampaignSignerNa(unittest.TestCase):
|
||||
def _orchestrator_with(self, producer_out):
|
||||
def run(ctx):
|
||||
return producer_out
|
||||
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
|
||||
producers["campaign-approval-gate"] = run
|
||||
producers["signer-validator-contract"] = run
|
||||
broker = FakeBroker()
|
||||
orch = StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
return orch
|
||||
|
||||
def test_na_with_contract_is_success(self):
|
||||
from status_producers import ProducerOutput, PRODUCER_REVISION
|
||||
na = ProducerOutput(
|
||||
"campaign-approval-gate",
|
||||
"FORMALLY_NOT_APPLICABLE",
|
||||
"aios.campaign-approval-gate",
|
||||
PRODUCER_REVISION,
|
||||
{"applicable": False},
|
||||
na_contract_ref="aios.campaign-approval-gate.v1#formally-not-applicable",
|
||||
)
|
||||
orch = self._orchestrator_with(na)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertTrue(result.merge_eligible)
|
||||
self.assertEqual(result.context_outcomes["campaign-approval-gate"].publish.gitea_state, "success")
|
||||
|
||||
def test_campaign_applicable_but_failed_blocks(self):
|
||||
from status_producers import ProducerOutput, PRODUCER_REVISION
|
||||
fail = ProducerOutput(
|
||||
"campaign-approval-gate",
|
||||
"FAIL",
|
||||
"aios.campaign-approval-gate",
|
||||
PRODUCER_REVISION,
|
||||
{"reason": "campaign_evaluate_failed"},
|
||||
)
|
||||
orch = self._orchestrator_with(fail)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
self.assertEqual(result.context_outcomes["campaign-approval-gate"].publish.gitea_state, "failure")
|
||||
|
||||
def test_producer_scope_ambiguous_fail_closed(self):
|
||||
from producer_scope_classifier import ScopeClassificationError, classify_na
|
||||
with self.assertRaises(ScopeClassificationError):
|
||||
classify_na("campaign-approval-gate", {"campaign_authorization_present": True})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,288 @@
|
||||
"""QA independent verification for the protected-main status pipeline.
|
||||
|
||||
Written by QA (yan-guoguan) with a fresh-eyes mandate: hunt for red-line
|
||||
violations / merge-eligibility spoofing that Engineer's 27-item suite did not
|
||||
cover. Fully local: fake broker + temp evidence store, no real Gitea/gitleaks.
|
||||
|
||||
Targeted adversarial cases:
|
||||
C1. FAIL/UNKNOWN verdict but a conflicting existing "success" status on the
|
||||
same SHA+context plus an unknown POST -> RECONCILE reports the leftover
|
||||
status's state. merge_eligible MUST be False (a FAIL context must never
|
||||
be credited as green because a stray success status exists).
|
||||
C2. RECONCILE exhausted (both read attempts fail) -> action UNKNOWN, state
|
||||
!= success, and merge_eligible False at orchestrator level.
|
||||
C3. One producer raises a runtime exception -> orchestrator converts it to
|
||||
UNKNOWN (never success) and the other 8 producers still run.
|
||||
C4. Idempotency: same effect replayed twice -> 2nd is SUPPRESS_NOOP and the
|
||||
evidence store is not duplicated.
|
||||
C5. N/A but scope facts incomplete/ambiguous -> classify_na raises -> producer
|
||||
returns UNKNOWN -> not success and merge blocked.
|
||||
C6. Transport binds publish to the exact SHA (never a moving branch tip).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import sys
|
||||
from pathlib import Path as _P
|
||||
|
||||
ROOT = _P(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
||||
|
||||
from status_evidence_contract import ( # noqa: E402
|
||||
REQUIRED_CONTEXTS,
|
||||
StatusEvidence,
|
||||
)
|
||||
from status_publisher import ( # noqa: E402
|
||||
AppendOnlyEvidenceStore,
|
||||
StatusPublisher,
|
||||
)
|
||||
from status_orchestrator import StatusOrchestrator # noqa: E402
|
||||
from status_producers import ( # noqa: E402
|
||||
PRODUCER_REVISION,
|
||||
ProducerOutput,
|
||||
)
|
||||
|
||||
|
||||
def _sha(value: int) -> str:
|
||||
return f"{value:040d}"
|
||||
|
||||
|
||||
HEAD_A = _sha(101)
|
||||
HEAD_B = _sha(202)
|
||||
|
||||
|
||||
class FakeBroker:
|
||||
"""Duck-typed broker mirroring the publisher's required surface."""
|
||||
|
||||
def __init__(self, *, fail_post=False, statuses=None, raise_read=False):
|
||||
self.posts = []
|
||||
self.fail_post = fail_post
|
||||
self._statuses = statuses or []
|
||||
self.raise_read = raise_read
|
||||
self.read_calls = 0
|
||||
|
||||
def status_for_sha(self, sha, state, description, context):
|
||||
self.posts.append({"sha": sha, "state": state, "context": context})
|
||||
if self.fail_post:
|
||||
raise RuntimeError("post_unknown")
|
||||
return {"ok": True}
|
||||
|
||||
def read_statuses(self, sha, context=None):
|
||||
self.read_calls += 1
|
||||
if self.raise_read:
|
||||
raise RuntimeError("read_unknown")
|
||||
return [s for s in self._statuses if context is None or s.get("context") == context]
|
||||
|
||||
|
||||
def _store():
|
||||
return AppendOnlyEvidenceStore(Path(tempfile.mkdtemp()))
|
||||
|
||||
|
||||
def _evidence(context, result, commit_sha=HEAD_A, na_ref=None):
|
||||
return StatusEvidence(
|
||||
repository="aios/aios",
|
||||
commit_sha=commit_sha,
|
||||
context=context,
|
||||
result=result,
|
||||
producer_identity=f"aios.{context}",
|
||||
producer_revision=PRODUCER_REVISION,
|
||||
na_contract_ref=na_ref,
|
||||
)
|
||||
|
||||
|
||||
def _verdict_producer(context, result, na_ref=None):
|
||||
def run(ctx):
|
||||
return ProducerOutput(
|
||||
context,
|
||||
result,
|
||||
f"aios.{context}",
|
||||
PRODUCER_REVISION,
|
||||
{},
|
||||
na_contract_ref=na_ref,
|
||||
)
|
||||
return run
|
||||
|
||||
|
||||
def _orchestrator(broker, producers):
|
||||
return StatusOrchestrator(
|
||||
StatusPublisher(broker, _store()),
|
||||
producers=producers,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
|
||||
|
||||
class TestQaC1ReconciledConflictSpoof(unittest.TestCase):
|
||||
"""CRITICAL red-line: a FAIL/UNKNOWN verdict must never be credited as green
|
||||
just because a conflicting existing 'success' status sits on the same SHA+ctx
|
||||
and the POST outcome is unknown (RECONCILE reports the leftover state)."""
|
||||
|
||||
def _run_with(self, bad_result, existing_state):
|
||||
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
|
||||
producers["py-compile"] = _verdict_producer("py-compile", bad_result)
|
||||
# Every context happens to have a pre-existing status on this SHA+ctx.
|
||||
statuses = [{"context": c, "state": existing_state} for c in REQUIRED_CONTEXTS]
|
||||
broker = FakeBroker(fail_post=True, statuses=statuses)
|
||||
orch = _orchestrator(broker, producers)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
return result
|
||||
|
||||
def test_fail_verdict_with_leftover_success_still_blocks_merge(self):
|
||||
result = self._run_with("FAIL", "success")
|
||||
pc = result.context_outcomes["py-compile"].publish
|
||||
# The FAIL context reconciled to the conflicting leftover success:
|
||||
self.assertEqual(pc.action, "RECONCILED")
|
||||
self.assertEqual(pc.gitea_state, "success")
|
||||
# ...so merge must NOT be eligible (a FAIL verdict was never confirmed).
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
def test_unknown_verdict_with_leftover_success_still_blocks_merge(self):
|
||||
result = self._run_with("UNKNOWN", "success")
|
||||
pc = result.context_outcomes["py-compile"].publish
|
||||
self.assertEqual(pc.gitea_state, "success")
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
|
||||
class TestQaC2ReconcileExhausted(unittest.TestCase):
|
||||
def test_publisher_reconcile_exhausted_is_unknown_not_success(self):
|
||||
broker = FakeBroker(fail_post=True, raise_read=True)
|
||||
pub = StatusPublisher(broker, _store())
|
||||
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
|
||||
self.assertEqual(outc.action, "UNKNOWN")
|
||||
self.assertNotEqual(outc.gitea_state, "success")
|
||||
|
||||
def test_orchestrator_reconcile_exhausted_blocks_merge(self):
|
||||
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
|
||||
broker = FakeBroker(fail_post=True, raise_read=True)
|
||||
orch = _orchestrator(broker, producers)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
self.assertFalse(result.merge_eligible)
|
||||
self.assertTrue(
|
||||
all(
|
||||
o.publish.action == "UNKNOWN"
|
||||
for o in result.context_outcomes.values()
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestQaC3ProducerRaisesKeepsOthersRunning(unittest.TestCase):
|
||||
def test_exception_becomes_unknown_and_others_still_publish(self):
|
||||
def boom(ctx):
|
||||
raise RuntimeError("producer_bombed")
|
||||
|
||||
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
|
||||
producers["secret-scan"] = boom
|
||||
broker = FakeBroker()
|
||||
orch = _orchestrator(broker, producers)
|
||||
result = orch.run_for_head(HEAD_A)
|
||||
|
||||
self.assertEqual(len(result.context_outcomes), 9)
|
||||
# The crashed one became UNKNOWN and published "error", never success.
|
||||
ss = result.context_outcomes["secret-scan"]
|
||||
self.assertEqual(ss.producer_output.result, "UNKNOWN")
|
||||
self.assertEqual(ss.publish.gitea_state, "error")
|
||||
# The other 8 still ran and published success.
|
||||
good = [c for c in REQUIRED_CONTEXTS if c != "secret-scan"]
|
||||
self.assertEqual(
|
||||
{result.context_outcomes[c].publish.gitea_state for c in good},
|
||||
{"success"},
|
||||
)
|
||||
# 8 success posts (the crashed one posts error).
|
||||
posted = [p for p in broker.posts if p["state"] == "success"]
|
||||
self.assertEqual(len(posted), 8)
|
||||
# Merge blocked.
|
||||
self.assertFalse(result.merge_eligible)
|
||||
|
||||
|
||||
class TestQaC4IdempotencyStoreNotDuplicated(unittest.TestCase):
|
||||
def test_replay_is_suppress_noop_and_store_not_duplicated(self):
|
||||
broker = FakeBroker()
|
||||
store = _store()
|
||||
pub = StatusPublisher(broker, store)
|
||||
ev = _evidence("py-compile", "APPLICABLE_PASS")
|
||||
first = pub.publish(ev)
|
||||
second = pub.publish(ev)
|
||||
self.assertEqual(first.action, "PUBLISHED")
|
||||
self.assertEqual(second.action, "SUPPRESS_NOOP")
|
||||
self.assertEqual(len(broker.posts), 1)
|
||||
self.assertIsNone(second.gitea_state) # NOOP carries no projected state
|
||||
|
||||
def test_store_single_artifact_for_replayed_effect(self):
|
||||
broker = FakeBroker()
|
||||
store = _store()
|
||||
pub = StatusPublisher(broker, store)
|
||||
ev = _evidence("py-compile", "APPLICABLE_PASS")
|
||||
pub.publish(ev)
|
||||
pub.publish(ev)
|
||||
files = list(Path(store.root).rglob("*.json"))
|
||||
self.assertEqual(len(files), 1)
|
||||
|
||||
|
||||
class TestQaC5NaScopeFactsIncomplete(unittest.TestCase):
|
||||
def test_ambiguous_scope_facts_fail_closed(self):
|
||||
from status_producers import campaign_approval_gate, ProducerContext
|
||||
out = campaign_approval_gate.run(
|
||||
ProducerContext(
|
||||
context="campaign-approval-gate",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={"campaign_authorization_present": True, "campaign_active": None, "in_campaign_lineage": None},
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "UNKNOWN")
|
||||
self.assertNotEqual(out.result, "APPLICABLE_PASS")
|
||||
self.assertIsNone(out.na_contract_ref)
|
||||
|
||||
def test_incomplete_signer_facts_fail_closed(self):
|
||||
from status_producers import signer_validator_contract, ProducerContext
|
||||
out = signer_validator_contract.run(
|
||||
ProducerContext(
|
||||
context="signer-validator-contract",
|
||||
commit_sha=HEAD_A,
|
||||
checkout=Path(tempfile.mkdtemp()),
|
||||
scope_facts={}, # touches_signer missing
|
||||
)
|
||||
)
|
||||
self.assertEqual(out.result, "UNKNOWN")
|
||||
|
||||
|
||||
class TestQaC6TransportExactShaBinding(unittest.TestCase):
|
||||
def test_publish_status_for_sha_binds_to_exact_sha_not_branch_tip(self):
|
||||
from aios_gitea_machine_transport import GiteaMachineTransport
|
||||
|
||||
captured = {}
|
||||
|
||||
class StubTransport(GiteaMachineTransport):
|
||||
def _request(self, method, path, body=None):
|
||||
captured["method"] = method
|
||||
captured["path"] = path
|
||||
captured["body"] = body
|
||||
return 201, {}
|
||||
|
||||
t = StubTransport(
|
||||
api_base="https://127.0.0.1:1/",
|
||||
repository_id="aios/aios",
|
||||
remote_url="https://127.0.0.1:1/aios/aios.git",
|
||||
authority_git_dir=Path(tempfile.mkdtemp()),
|
||||
)
|
||||
t.credential_name = "gitea-machine-token"
|
||||
# Avoid touching the real credential path check in __post_init__; construct
|
||||
# the dataclass directly is not possible due to __post_init__, so reuse the
|
||||
# instance and call the method (which does not read the credential token here).
|
||||
# _safe_oid guards the SHA shape.
|
||||
t.publish_status_for_sha(HEAD_A, "success", "d", "py-compile")
|
||||
self.assertEqual(captured["method"], "POST")
|
||||
# The URL MUST pin the exact 40-hex SHA, never a branch name / branch tip.
|
||||
self.assertTrue(
|
||||
captured["path"].endswith(f"/statuses/{HEAD_A}"),
|
||||
captured["path"],
|
||||
)
|
||||
self.assertNotIn("/branches/", captured["path"])
|
||||
self.assertEqual(captured["body"]["state"], "success")
|
||||
self.assertEqual(captured["body"]["context"], "py-compile")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user