126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from aios_autonomous_release_candidate.approval import (
|
|
ApprovalVerifier,
|
|
StaticRevocationSource,
|
|
UnavailableSignatureVerifier,
|
|
)
|
|
from aios_autonomous_release_candidate.errors import ApprovalRequired
|
|
from aios_autonomous_release_candidate.gitea_canary import (
|
|
EphemeralKeyProjection,
|
|
GiteaDeployKeyCanary,
|
|
)
|
|
from aios_autonomous_release_candidate.models import (
|
|
ApprovalEnvelope,
|
|
ResourceBinding,
|
|
RiskClass,
|
|
TaskPlan,
|
|
WorkflowStep,
|
|
)
|
|
from aios_autonomous_release_candidate.store import StateStore
|
|
|
|
|
|
NOW = datetime(2026, 7, 26, tzinfo=timezone.utc)
|
|
OID = "a" * 40
|
|
|
|
|
|
def task_plan() -> TaskPlan:
|
|
return TaskPlan(
|
|
"external-gate-task",
|
|
"external gate verification",
|
|
ResourceBinding("aios/aios", "patch/external-gate", OID, ("services/test",)),
|
|
RiskClass.HIGH,
|
|
(WorkflowStep("canary", "tester", ("native_session.spawn",)),),
|
|
("external-gates",),
|
|
)
|
|
|
|
|
|
def test_unknown_owner_verifier_fails_with_stable_blocker(tmp_path) -> None:
|
|
plan = task_plan()
|
|
envelope = ApprovalEnvelope(
|
|
approval_plan_id="owner-external-gate",
|
|
task_id=plan.task_id,
|
|
task_plan_sha256=plan.digest(),
|
|
binding=plan.binding,
|
|
capabilities=("native_session.spawn",),
|
|
risk=RiskClass.HIGH,
|
|
not_before=NOW - timedelta(minutes=1),
|
|
expires_at=NOW + timedelta(minutes=5),
|
|
nonce="nonce-owner-external",
|
|
signing_key_id="owner-kms-1",
|
|
signature="opaque-signature",
|
|
)
|
|
verifier = ApprovalVerifier(
|
|
StateStore(tmp_path / "state.sqlite3"),
|
|
UnavailableSignatureVerifier(),
|
|
StaticRevocationSource(),
|
|
)
|
|
with pytest.raises(
|
|
ApprovalRequired, match="BLOCKED_OWNER_APPROVAL_VERIFIER_UNAVAILABLE"
|
|
):
|
|
verifier.verify(plan, envelope, now=NOW)
|
|
|
|
|
|
class KeyProjection:
|
|
def __init__(self) -> None:
|
|
self.present = False
|
|
|
|
def create(self, task_id, branch):
|
|
self.present = True
|
|
return EphemeralKeyProjection(
|
|
public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICandidateCanaryOnly",
|
|
private_key_reference="systemd-credential:gitea-deploy-key-canary",
|
|
)
|
|
|
|
def delete(self, reference):
|
|
self.present = False
|
|
|
|
def is_present(self, reference):
|
|
return self.present
|
|
|
|
|
|
class Gitea:
|
|
simulation = True
|
|
|
|
def __init__(self) -> None:
|
|
self.keys = {}
|
|
|
|
def list_deploy_keys(self, repository, credential_reference):
|
|
return list(self.keys.values())
|
|
|
|
def create_deploy_key(self, repository, title, public_key, credential_reference):
|
|
item = {"id": 1, "title": title, "key": public_key, "read_only": False}
|
|
self.keys[1] = item
|
|
return item
|
|
|
|
def delete_deploy_key(self, repository, key_id, credential_reference):
|
|
self.keys.pop(key_id, None)
|
|
|
|
|
|
class SSHProbe:
|
|
def __init__(self, gitea, projection) -> None:
|
|
self.gitea = gitea
|
|
self.projection = projection
|
|
|
|
def authenticate(self, repository, branch, private_key_reference):
|
|
return bool(self.gitea.keys) and self.projection.present
|
|
|
|
|
|
def test_gitea_canary_always_revokes_key_and_projection() -> None:
|
|
projection = KeyProjection()
|
|
gitea = Gitea()
|
|
result = GiteaDeployKeyCanary(gitea, projection, SSHProbe(gitea, projection)).run(
|
|
task_id="external-gate-task",
|
|
branch="patch/external-gate",
|
|
credential_reference="systemd-credential:gitea-admin-api-candidate",
|
|
)
|
|
assert result.key_created is True
|
|
assert result.key_deleted is True
|
|
assert result.post_revoke_auth_rejected is True
|
|
assert result.credential_projection_deleted is True
|
|
assert result.gitea_admin_api_lifecycle_canary is False # simulation is not real evidence
|
|
assert gitea.keys == {}
|
|
assert projection.present is False
|