aios/services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/gitea_canary.py

"""API-only Gitea deploy-key lifecycle canary."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Mapping, Protocol, Sequence

from .errors import CredentialUnavailable, StateConflict


GITEA_REFERENCE = "systemd-credential:gitea-admin-api-candidate"


@dataclass(frozen=True, slots=True)
class EphemeralKeyProjection:
    public_key: str
    private_key_reference: str


class KeyProjectionProvider(Protocol):
    def create(self, task_id: str, branch: str) -> EphemeralKeyProjection: ...

    def delete(self, reference: str) -> None: ...

    def is_present(self, reference: str) -> bool: ...


class GiteaDeployKeyProvider(Protocol):
    simulation: bool

    def list_deploy_keys(
        self, repository: str, credential_reference: str
    ) -> Sequence[Mapping[str, Any]]: ...

    def create_deploy_key(
        self,
        repository: str,
        title: str,
        public_key: str,
        credential_reference: str,
    ) -> Mapping[str, Any]: ...

    def delete_deploy_key(
        self, repository: str, key_id: int, credential_reference: str
    ) -> None: ...


class SSHAuthenticationProbe(Protocol):
    def authenticate(
        self, repository: str, branch: str, private_key_reference: str
    ) -> bool: ...


@dataclass(frozen=True, slots=True)
class GiteaCanaryResult:
    key_created: bool
    key_deleted: bool
    api_key_absent_after_delete: bool
    post_revoke_auth_rejected: bool
    credential_projection_deleted: bool
    gitea_admin_api_lifecycle_canary: bool
    simulation: bool
    secrets_included: bool = False


class GiteaDeployKeyCanary:
    """Exercise only repository deploy-key list/create/delete provider operations."""

    repository = "aios/aios"

    def __init__(
        self,
        gitea: GiteaDeployKeyProvider,
        projections: KeyProjectionProvider,
        ssh_probe: SSHAuthenticationProbe,
    ) -> None:
        self.gitea = gitea
        self.projections = projections
        self.ssh_probe = ssh_probe

    def run(
        self, *, task_id: str, branch: str, credential_reference: str
    ) -> GiteaCanaryResult:
        if credential_reference != GITEA_REFERENCE:
            raise CredentialUnavailable(
                "BLOCKED_GITEA_ADMIN_CREDENTIAL_REQUIRED",
                "exact Candidate Gitea Admin API reference is required",
            )
        if not task_id or not branch.startswith("patch/"):
            raise StateConflict("GITEA_CANARY_BINDING_INVALID", "task and patch branch are required")
        projection: EphemeralKeyProjection | None = None
        key_id: int | None = None
        key_created = False
        key_deleted = False
        api_absent = False
        auth_rejected = False
        projection_deleted = False
        title = f"aios-candidate-{task_id}"
        try:
            projection = self.projections.create(task_id, branch)
            if (
                not projection.public_key.startswith("ssh-ed25519 ")
                or not projection.private_key_reference.startswith("systemd-credential:")
            ):
                raise CredentialUnavailable(
                    "GITEA_CANARY_KEY_PROJECTION_INVALID",
                    "ephemeral key projection is not reference-only",
                )
            before = self.gitea.list_deploy_keys(self.repository, credential_reference)
            if any(item.get("title") == title for item in before):
                raise StateConflict("GITEA_CANARY_KEY_COLLISION", "canary key title already exists")
            created = self.gitea.create_deploy_key(
                self.repository,
                title,
                projection.public_key,
                credential_reference,
            )
            raw_id = created.get("id")
            if not isinstance(raw_id, int) or created.get("key") != projection.public_key:
                raise StateConflict("GITEA_CANARY_CREATE_INVALID", "created key response is invalid")
            key_id = raw_id
            key_created = True
            visible = self.gitea.list_deploy_keys(self.repository, credential_reference)
            if not any(item.get("id") == key_id for item in visible):
                raise StateConflict("GITEA_CANARY_KEY_NOT_VISIBLE", "created key is not listed")
            if not self.ssh_probe.authenticate(
                self.repository, branch, projection.private_key_reference
            ):
                raise StateConflict("GITEA_CANARY_SSH_AUTH_FAILED", "deploy key could not authenticate")
            self.gitea.delete_deploy_key(self.repository, key_id, credential_reference)
            key_deleted = True
            key_id = None
            after = self.gitea.list_deploy_keys(self.repository, credential_reference)
            api_absent = not any(item.get("title") == title for item in after)
            if not api_absent:
                raise StateConflict("GITEA_CANARY_DELETE_UNVERIFIED", "deleted key remains visible")
            self.projections.delete(projection.private_key_reference)
            projection_deleted = not self.projections.is_present(projection.private_key_reference)
            auth_rejected = not self.ssh_probe.authenticate(
                self.repository, branch, projection.private_key_reference
            )
            if not projection_deleted or not auth_rejected:
                raise StateConflict(
                    "GITEA_CANARY_REVOCATION_UNVERIFIED",
                    "projection deletion or post-revocation SSH rejection failed",
                )
            simulated = bool(getattr(self.gitea, "simulation", True))
            return GiteaCanaryResult(
                key_created=key_created,
                key_deleted=key_deleted,
                api_key_absent_after_delete=api_absent,
                post_revoke_auth_rejected=auth_rejected,
                credential_projection_deleted=projection_deleted,
                gitea_admin_api_lifecycle_canary=not simulated,
                simulation=simulated,
                secrets_included=False,
            )
        finally:
            if key_id is not None:
                try:
                    self.gitea.delete_deploy_key(
                        self.repository, key_id, credential_reference
                    )
                except Exception:
                    pass
            if projection is not None and self.projections.is_present(
                projection.private_key_reference
            ):
                try:
                    self.projections.delete(projection.private_key_reference)
                except Exception:
                    pass