diff --git a/changelog.d/160-release-approve.md b/changelog.d/160-release-approve.md new file mode 100644 index 0000000..69509e3 --- /dev/null +++ b/changelog.d/160-release-approve.md @@ -0,0 +1,3 @@ +### Added + +- `cloudfall release approve --release ` records a person's approval of one built release as a `ReleaseApproval` (new `release-approval.schema.json`) under `decisions/releases/`: the application, component, release, archive SHA-256, git commit, servers, approver, `approvedAt`, `expiresAt` (`--valid-for` hours, 1 to 168, default 24) and `via: terminal`. It verifies the artifact first, refuses a component of an `automatic` application with `release_approval_not_required`, and writes the record only once a person typed the release id at a terminal; `--yes` never answers it, and it is no MCP tool. It deploys nothing, and `deploy` does not read the record yet (#160) diff --git a/config/schemas/v1/release-approval.schema.json b/config/schemas/v1/release-approval.schema.json new file mode 100644 index 0000000..3100956 --- /dev/null +++ b/config/schemas/v1/release-approval.schema.json @@ -0,0 +1,90 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://cloudfall.internal/schemas/v1/release-approval.schema.json", + "title": "Cloudfall release approval record", + "description": "A person's approval of one release of one component, written by cloudfall release approve. It names the archive digest and servers it approves and holds no secret.", + "type": "object", + "properties": { + "apiVersion": { + "$ref": "common.schema.json#/$defs/apiVersion" + }, + "kind": { + "const": "ReleaseApproval" + }, + "metadata": { + "$ref": "common.schema.json#/$defs/metadata" + }, + "spec": { + "type": "object", + "properties": { + "application": { + "$ref": "common.schema.json#/$defs/resourceId" + }, + "component": { + "$ref": "common.schema.json#/$defs/resourceId" + }, + "release": { + "type": "string", + "pattern": "^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{7,40}$" + }, + "artifactSha256": { + "description": "The archive's SHA-256 as verified against its artifact metadata when approved", + "$ref": "common.schema.json#/$defs/sha256" + }, + "gitCommit": { + "type": "string", + "pattern": "^[0-9a-f]{40}$" + }, + "servers": { + "description": "The component's declared servers when approved", + "type": "array", + "items": { + "$ref": "common.schema.json#/$defs/resourceId" + }, + "minItems": 1, + "uniqueItems": true + }, + "approver": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "pattern": "^[^\\u0000-\\u001f\\u007f-\\u009f]+$" + }, + "approvedAt": { + "type": "string", + "format": "date-time" + }, + "expiresAt": { + "type": "string", + "format": "date-time" + }, + "via": { + "description": "How the approval arrived: terminal when a person typed the release id at a terminal", + "enum": [ + "terminal" + ] + } + }, + "required": [ + "application", + "component", + "release", + "artifactSha256", + "gitCommit", + "servers", + "approver", + "approvedAt", + "expiresAt", + "via" + ], + "additionalProperties": false + } + }, + "required": [ + "apiVersion", + "kind", + "metadata", + "spec" + ], + "additionalProperties": false +} diff --git a/sdk/src/cloudfall/agent_server.py b/sdk/src/cloudfall/agent_server.py index 1db95d5..876faa6 100644 --- a/sdk/src/cloudfall/agent_server.py +++ b/sdk/src/cloudfall/agent_server.py @@ -303,7 +303,8 @@ def operation_tool_description(operation: Operation) -> str: ) """The commands an agent runs on a project. -``init``, the loops and ``operator approve`` are not tools: a person approves. +``init``, the loops, ``operator approve`` and ``release approve`` are not +tools: a person approves. """ FLEET_COMMANDS = frozenset( diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index da63d41..3d3ed1f 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -25,6 +25,7 @@ # handler's Iterator and a provided MCP tool's Mapping must exist at runtime. from collections.abc import Iterator, Mapping from dataclasses import dataclass, replace +from datetime import UTC, datetime from functools import partial from http import HTTPStatus from importlib.metadata import version @@ -245,6 +246,18 @@ resolve_project_directory, ) from cloudfall.recording import Recording, RecordingError +from cloudfall.release_approval import ( + APPROVALS_DIRECTORY, + VALID_FOR_DEFAULT, + VALID_FOR_MAXIMUM, + VALID_FOR_MINIMUM, + Approver, + ReleaseApprovalError, + ValidityHours, + approve_release, + release_candidate, + write_approval, +) from cloudfall.render_api import ( ERROR_API_UNREACHABLE, RENDER_API_URL, @@ -4031,6 +4044,151 @@ def restart_command(args: ComponentArgs, ctx: Ctx, fleet: Fleet) -> RestartPaylo return RestartPayload({**result.as_dict(), "effect": "updated"}) +# release approve: a person's record that one release may be deployed (S-008) + + +release_group = app.group( + "release", description="Record a person's approval of one built release" +) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class ReleaseApproveArgs(ProjectArgs): + """Arguments of ``release approve``.""" + + component: ResourceId = Arg(description="Component whose release is approved") + release: ReleaseId = Flag( + description="Release id produced by cloudfall-engine artifact build" + ) + artifacts: Path = Flag( + default=Path("tmp/artifacts"), + description="Artifact directory (default: tmp/artifacts)", + ) + approvals: Path = Flag( + default=Path(APPROVALS_DIRECTORY), + description=( + f"Directory the approval record is written to (default: " + f"{APPROVALS_DIRECTORY})" + ), + ) + valid_for: int = Flag( + default=VALID_FOR_DEFAULT, + description=( + f"Hours the approval counts for, {VALID_FOR_MINIMUM} to " + f"{VALID_FOR_MAXIMUM} (default: {VALID_FOR_DEFAULT})" + ), + ) + approver: str | None = Flag( + default=None, + description="Who is approving (default: the USER environment variable)", + ) + + def __post_init__(self) -> None: + """Keep relative paths inside the project; check the validity.""" + _inside_project(self.artifacts, "artifacts") + _inside_project(self.approvals, "approvals") + self.validity() + + def validity(self) -> ValidityHours: + """Return how long the approval counts, refusing an out-of-range count.""" + try: + return ValidityHours(self.valid_for) + except ReleaseApprovalError as error: + message = f"--valid-for: {error.detail}" + raise ParseError(message, context={"flag": "valid-for"}) from error + + +@dataclass(frozen=True, slots=True) +class ReleaseApproved: + """The approval record a person wrote, and where it is kept.""" + + effect: str + status: str + path: str + approval: dict[str, object] = Out(ordered=True) + + +@release_group.command( + "approve", + # A person approves; no agent tool stands in for it. As with decisions + # approve, they type the release id at a terminal, which no flag answers, + # --yes included. requires_person keeps it off MCP too; mcp=False says so + # here as well. + mcp=False, + requires_person=True, + description=( + "Approve one built release of one component for deploy; a person types " + "the release id at a terminal, and nothing is deployed" + ), + danger_level="mutating", + exit_codes=[ + "PROJECT_INVALID", + "CONFIG_INVALID", + "PRECONDITION", + "PERMISSION_DENIED", + ], + supports_raw_payload=True, + examples=[ + ( + "Approve a release, typing its id at a terminal", + "cloudfall release approve crm-backend --release 20260101T000000Z-abcdef0", + ), + ], +) +def release_approve( + args: ReleaseApproveArgs, ctx: Ctx, fleet: Fleet +) -> ReleaseApproved: + """Write the approval record once a person typed the release id.""" + valid_for = args.validity() + try: + approver = Approver( + args.approver if args.approver is not None else ctx.env.get("USER", "") + ) + candidate = release_candidate( + fleet.inventory, + args.component, + args.release, + fleet.path(args.artifacts), + args.schemas, + ) + except ReleaseApprovalError as error: + raise Exit.PRECONDITION(error.detail, context={"code": error.code}) from error + except LifecycleError as error: + raise _lifecycle_failed(error) from error + until = (datetime.now(UTC) + valid_for.duration).strftime("%Y-%m-%dT%H:%M:%SZ") + # Off a terminal this ends with PERSON_REQUIRED, and a mistyped id with + # ATTESTATION_MISMATCH, before anything is written: --yes never approves. + attestation = ctx.attest( + f"{candidate.describe()}. The approval counts for {valid_for.value} " + f"hours, until about {until}. Type the release id to approve " + f"{candidate.release.value}", + expected=candidate.release.value, + ) + approval = approve_release( + candidate, + approver, + ApprovalChannel(attestation.channel), + valid_for, + attestation.at, + ) + try: + path = write_approval( + approval, fleet.path(args.approvals), SchemaCatalog(args.schemas) + ) + except ReleaseApprovalError as error: + raise Exit.PRECONDITION(error.detail, context={"code": error.code}) from error + except OSError as error: + if error.errno not in _NOT_WRITABLE: + raise + raise _not_writable(error) from error + return ReleaseApproved( + effect="created", + status="ok", + path=str(path), + approval=approval.as_document(), + ) + + data = app.group("data", description="Migrate data into declared services") diff --git a/sdk/src/cloudfall/commands.py b/sdk/src/cloudfall/commands.py index e43702d..0602852 100644 --- a/sdk/src/cloudfall/commands.py +++ b/sdk/src/cloudfall/commands.py @@ -511,6 +511,15 @@ def _plan(subject: str, *extra: str) -> OutputShape: gate=_YES, output=(_plan("component"), _shape(*_EXECUTED, when="with `--yes`")), ), + CommandContract( + "cloudfall", + "release approve", + CommandEffect.PROJECT, + "record a person's approval of one built release under " + "`decisions/releases/`; a person types the release id at a terminal, " + "which no flag answers (`--yes` included), and nothing is deployed", + output=(_shape("status", "path", "approval:object"),), + ), CommandContract( "cloudfall", "data copy", diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index 126a5a3..9788c5f 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -178,6 +178,8 @@ class VerifiedArtifact: archive_path: Path metadata_path: Path sha256: str + git_commit: str + """The commit the metadata says the archive was built from.""" @dataclass(frozen=True, slots=True) @@ -421,6 +423,8 @@ def verify_artifact( archive_path=archive_path, metadata_path=metadata_path, sha256=actual_digest, + # The schema above requires it as 40 lowercase hex characters. + git_commit=cast("str", spec["gitCommit"]), ) @@ -1481,7 +1485,13 @@ def _component( context: EngineContext, component_id: ResourceId ) -> ComponentInventory: state = validate_config(context.project_directory, context.schema_directory) - inventory = PlatformInventory.from_state(state) + return find_component(PlatformInventory.from_state(state), component_id) + + +def find_component( + inventory: PlatformInventory, component_id: ResourceId +) -> ComponentInventory: + """Return the declared component, or fail as every lifecycle command does.""" component = next( ( candidate diff --git a/sdk/src/cloudfall/release_approval.py b/sdk/src/cloudfall/release_approval.py new file mode 100644 index 0000000..5ec4dbf --- /dev/null +++ b/sdk/src/cloudfall/release_approval.py @@ -0,0 +1,286 @@ +"""A person's approval of one release of one component (S-008). + +``cloudfall release approve`` writes one ``ReleaseApproval`` record once a +person typed the release id at a terminal. The record names the archive +digest, the git commit and the servers it approves, and when it expires; it +holds no secret. This module builds and writes the record; nothing here +deploys, and nothing here reads the records back. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from datetime import UTC, timedelta +from typing import TYPE_CHECKING + +from cloudfall.decision import ApprovalChannel +from cloudfall.domain import DeploymentApproval, ResourceId, Sha256Digest +from cloudfall.lifecycle import find_component, verify_artifact + +if TYPE_CHECKING: + from datetime import datetime + from pathlib import Path + + from cloudfall.domain import ReleaseId + from cloudfall.inventory import PlatformInventory + from cloudfall.validation import SchemaCatalog + +RELEASE_APPROVAL_SCHEMA = "release-approval.schema.json" +APPROVALS_DIRECTORY = "decisions/releases" +"""Where approvals are kept by default: beside the decision records.""" + +ERROR_APPROVAL_NOT_REQUIRED = "release_approval_not_required" +"""The component's application deploys ``automatic``: a record would gate nothing.""" +ERROR_APPROVER_MISSING = "release_approver_missing" +ERROR_APPROVER_INVALID = "release_approver_invalid" +ERROR_APPROVAL_EXISTS = "release_approval_exists" +"""Two approvals of one release in the same second share an id.""" +ERROR_CHANNEL_UNKNOWN = "release_approval_channel_unknown" +ERROR_VALIDITY_INVALID = "release_approval_validity_invalid" + +VALID_FOR_MINIMUM = 1 +VALID_FOR_MAXIMUM = 168 +"""An approval counts for one hour at least and a week at most.""" +VALID_FOR_DEFAULT = 24 + +_KIND = "ReleaseApproval" +_API_VERSION = "cloudfall/v1" +_ID_PREFIX = "approval" +_FINGERPRINT_LENGTH = 16 +_APPROVER_MAX_LENGTH = 200 +_GIT_COMMIT = re.compile(r"[0-9a-f]{40}") +_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ" + + +class ReleaseApprovalError(RuntimeError): + """Fail-fast release approval error with a stable machine-readable code.""" + + def __init__(self, code: str, message: str) -> None: + """Record the failure code and human-readable detail.""" + self.code = code + self.detail = message + super().__init__(f"{code}: {message}") + + +@dataclass(frozen=True, slots=True) +class ValidityHours: + """How long an approval counts: whole hours, from one to a week.""" + + value: int + + def __post_init__(self) -> None: + """Refuse a non-integer, or a count outside one to 168 hours.""" + if ( + isinstance(self.value, bool) + or not isinstance(self.value, int) + or not VALID_FOR_MINIMUM <= self.value <= VALID_FOR_MAXIMUM + ): + message = ( + f"an approval is valid for {VALID_FOR_MINIMUM} to " + f"{VALID_FOR_MAXIMUM} whole hours, got {self.value!r}" + ) + raise ReleaseApprovalError(ERROR_VALIDITY_INVALID, message) + + @property + def duration(self) -> timedelta: + """Return the validity as a time span.""" + return timedelta(hours=self.value) + + +@dataclass(frozen=True, slots=True) +class GitCommit: + """A full lowercase git commit id, as artifact metadata records it.""" + + value: str + + def __post_init__(self) -> None: + """Enforce 40 lowercase hexadecimal characters.""" + if not _GIT_COMMIT.fullmatch(self.value): + message = f"invalid git commit: {self.value!r}" + raise ValueError(message) + + +@dataclass(frozen=True, slots=True) +class Approver: + """Who approved: a non-blank name, as ``decisions approve`` records it.""" + + value: str + + def __post_init__(self) -> None: + """Refuse a blank, overlong or unprintable approver.""" + if not self.value.strip(): + message = "no approver: pass --approver or set USER" + raise ReleaseApprovalError(ERROR_APPROVER_MISSING, message) + if len(self.value) > _APPROVER_MAX_LENGTH: + message = f"the approver is longer than {_APPROVER_MAX_LENGTH} characters" + raise ReleaseApprovalError(ERROR_APPROVER_INVALID, message) + # A plan shows the approver later: a newline, an escape sequence or a + # bidi override could forge or hide a line there. + if not self.value.isprintable(): + message = f"the approver has a control character: {self.value!r}" + raise ReleaseApprovalError(ERROR_APPROVER_INVALID, message) + + +@dataclass(frozen=True, slots=True) +class ReleaseCandidate: + """One verified release of one component of a ``manual`` application.""" + + application_id: ResourceId + component_id: ResourceId + release: ReleaseId + artifact_sha256: Sha256Digest + git_commit: GitCommit + server_ids: tuple[ResourceId, ...] + + def describe(self) -> str: + """Name what a person approves, for the prompt.""" + return ( + f"Release {self.release.value} of component {self.component_id} " + f"(application {self.application_id}): git commit " + f"{self.git_commit.value}, archive SHA-256 " + f"{self.artifact_sha256.value}, servers " + f"{', '.join(server.value for server in self.server_ids)}" + ) + + +def release_candidate( + inventory: PlatformInventory, + component_id: ResourceId, + release: ReleaseId, + artifacts: Path, + schemas: Path, +) -> ReleaseCandidate: + """Return what would be approved, checked as ``deploy`` checks it. + + Raises ``LifecycleError`` for an undeclared component or an artifact + that does not verify, and ``ReleaseApprovalError`` for a component of + an ``automatic`` application. + """ + component = find_component(inventory, component_id) + applications = { + application.resource_id: application + for application in inventory.applications + } + application = applications[component.application_id] + if application.approval is DeploymentApproval.AUTOMATIC: + message = ( + f"application {application.resource_id} deploys automatic; an " + f"approval of {component_id} would gate nothing" + ) + raise ReleaseApprovalError(ERROR_APPROVAL_NOT_REQUIRED, message) + artifact = verify_artifact(artifacts, component_id, release, schemas) + return ReleaseCandidate( + application_id=application.resource_id, + component_id=component_id, + release=release, + artifact_sha256=Sha256Digest(artifact.sha256), + git_commit=GitCommit(artifact.git_commit), + server_ids=component.server_ids, + ) + + +@dataclass(frozen=True, slots=True) +class ReleaseApproval: + """One person's approval of one release, valid until it expires.""" + + approval_id: ResourceId + candidate: ReleaseCandidate + approver: Approver + approved_at: datetime + expires_at: datetime + via: ApprovalChannel + + def as_document(self) -> dict[str, object]: + """Serialize the record as ``release-approval.schema.json`` defines it.""" + return { + "apiVersion": _API_VERSION, + "kind": _KIND, + "metadata": {"id": self.approval_id.value}, + "spec": _spec( + self.candidate, + self.approver, + self.approved_at, + self.expires_at, + self.via, + ), + } + + def path(self, directory: Path) -> Path: + """Return where the record is kept under ``directory``.""" + return directory / f"{self.approval_id.value}.json" + + +def approve_release( + candidate: ReleaseCandidate, + approver: Approver, + via: ApprovalChannel, + valid_for: ValidityHours, + approved_at: datetime, +) -> ReleaseApproval: + """Return the approval a person gave at ``approved_at``, an aware time.""" + if via is ApprovalChannel.UNKNOWN: + message = "an approval must say how it arrived" + raise ReleaseApprovalError(ERROR_CHANNEL_UNKNOWN, message) + if approved_at.utcoffset() is None: + message = f"the approval time has no time zone: {approved_at!r}" + raise ValueError(message) + # The record keeps whole seconds, so its two times differ by exactly the + # validity. + moment = approved_at.astimezone(UTC).replace(microsecond=0) + expires_at = moment + valid_for.duration + spec = _spec(candidate, approver, moment, expires_at, via) + canonical = json.dumps(spec, sort_keys=True, separators=(",", ":")) + fingerprint = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + stamp = moment.strftime("%Y%m%dt%H%M%S") + return ReleaseApproval( + approval_id=ResourceId( + f"{_ID_PREFIX}-{stamp}-{fingerprint[:_FINGERPRINT_LENGTH]}" + ), + candidate=candidate, + approver=approver, + approved_at=moment, + expires_at=expires_at, + via=via, + ) + + +def write_approval( + approval: ReleaseApproval, directory: Path, catalog: SchemaCatalog +) -> Path: + """Validate the record and write it as a new file; never overwrite one.""" + document = approval.as_document() + catalog.validate_named(RELEASE_APPROVAL_SCHEMA, document) + directory.mkdir(parents=True, exist_ok=True) + path = approval.path(directory) + try: + handle = path.open("x", encoding="utf-8") + except FileExistsError as error: + message = f"an identical approval is already recorded at {path}" + raise ReleaseApprovalError(ERROR_APPROVAL_EXISTS, message) from error + with handle: + handle.write(json.dumps(document, indent=2) + "\n") + return path + + +def _spec( + candidate: ReleaseCandidate, + approver: Approver, + approved_at: datetime, + expires_at: datetime, + via: ApprovalChannel, +) -> dict[str, object]: + return { + "application": candidate.application_id.value, + "component": candidate.component_id.value, + "release": candidate.release.value, + "artifactSha256": candidate.artifact_sha256.value, + "gitCommit": candidate.git_commit.value, + "servers": [server.value for server in candidate.server_ids], + "approver": approver.value, + "approvedAt": approved_at.astimezone(UTC).strftime(_TIME_FORMAT), + "expiresAt": expires_at.astimezone(UTC).strftime(_TIME_FORMAT), + "via": via.value, + } diff --git a/sdk/tests/test_agent_server.py b/sdk/tests/test_agent_server.py index a3d5afa..98193fb 100644 --- a/sdk/tests/test_agent_server.py +++ b/sdk/tests/test_agent_server.py @@ -288,6 +288,18 @@ def test_a_project_call_cannot_approve_its_own_proposal(tmp_path: Path) -> None: assert approved["error"]["code"] == "UNKNOWN_TOOL" +def test_a_project_call_cannot_approve_a_release(tmp_path: Path) -> None: + # A person approves a release at a terminal; an agent that could call + # release approve would license the deploy it then runs itself. + tools, (approved,) = _talk( + _serve("--project", str(EXAMPLES), cwd=tmp_path), + ("release_approve", {"component": "crm-backend"}), + ) + + assert not any(name.startswith("release") for name in tools) + assert approved["error"]["code"] == "UNKNOWN_TOOL" + + def test_a_project_call_cannot_send_a_file_elsewhere_or_write_secrets_out( tmp_path: Path, ) -> None: diff --git a/sdk/tests/test_release_approval.py b/sdk/tests/test_release_approval.py new file mode 100644 index 0000000..95c8820 --- /dev/null +++ b/sdk/tests/test_release_approval.py @@ -0,0 +1,426 @@ +"""``release approve``: a person's record that one release may be deployed (S-008).""" + +from __future__ import annotations + +import hashlib +import io +import json +import shutil +import tarfile +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from cloudfall.agent_server import PROJECT_COMMANDS +from cloudfall.app import app +from cloudfall.cli import main +from cloudfall.decision import ApprovalChannel +from cloudfall.domain import ReleaseId, ResourceId, Sha256Digest +from cloudfall.release_approval import ( + Approver, + GitCommit, + ReleaseApproval, + ReleaseApprovalError, + ReleaseCandidate, + ValidityHours, + approve_release, + write_approval, +) +from cloudfall.validation import SchemaCatalog +from jsonschema import ValidationError +from treaty import CommandPath + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" +RELEASE = "20260101T000000Z-abcdef0" +COMMIT = "a" * 40 +MOMENT = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC) + + +class _Terminal(io.StringIO): + """A stream that says it is a terminal, as a person's shell is.""" + + def isatty(self) -> bool: + return True + + +def _project(tmp_path: Path, approval: str = "manual") -> Path: + project = tmp_path / "project" + shutil.copytree(EXAMPLES, project) + application = project / "applications" / "crm.yaml" + text = application.read_text(encoding="utf-8") + application.write_text( + text.replace("approval: manual", f"approval: {approval}"), encoding="utf-8" + ) + return project + + +def _write_artifact(project: Path) -> str: + """Build a consistent artifact under the project's default folder.""" + directory = project / "tmp" / "artifacts" / "crm-backend" + directory.mkdir(parents=True) + source = project / "tmp" / "source" + source.mkdir() + (source / "app.py").write_text("print('crm')\n", encoding="utf-8") + archive = directory / f"{RELEASE}.tar.gz" + with tarfile.open(archive, "w:gz") as tar: + tar.add(source, arcname=".") + digest = hashlib.sha256(archive.read_bytes()).hexdigest() + metadata = { + "apiVersion": "cloudfall/v1", + "kind": "Artifact", + "metadata": {"id": "crm-backend"}, + "spec": { + "component": "crm-backend", + "release": RELEASE, + "gitRef": "main", + "gitCommit": COMMIT, + "builtAt": "2026-01-01T00:00:00Z", + "archive": archive.name, + "archiveSha256": digest, + "sizeBytes": archive.stat().st_size, + }, + } + (directory / f"{RELEASE}.json").write_text(json.dumps(metadata), encoding="utf-8") + return digest + + +def _argv(project: Path, *flags: str) -> list[str]: + return [ + *("release", "approve", "crm-backend", "--release", RELEASE), + *("--project", str(project), "--schemas", str(SCHEMAS)), + *("--approver", "roman"), + *flags, + ] + + +def _at_terminal( + project: Path, typed: str, *flags: str +) -> tuple[int, dict[str, object], str]: + """Run ``release approve`` as a person at a terminal who types ``typed``.""" + stdout = _Terminal() + stderr = io.StringIO() + exit_code = app.run( + [*_argv(project, *flags), "--format", "json"], + stdin=_Terminal(f"{typed}\n"), + stdout=stdout, + stderr=stderr, + ) + return exit_code, json.loads(stdout.getvalue()), stderr.getvalue() + + +def _records(project: Path) -> list[Path]: + approvals = project / "decisions" / "releases" + return sorted(approvals.glob("*.json")) if approvals.is_dir() else [] + + +def test_s008_2_a_person_approves_a_release_by_typing_its_id(tmp_path: Path) -> None: + project = _project(tmp_path) + digest = _write_artifact(project) + + exit_code, payload, prompt = _at_terminal(project, RELEASE) + + assert exit_code == 0, payload + (path,) = _records(project) + record = json.loads(path.read_text(encoding="utf-8")) + SchemaCatalog(SCHEMAS).validate_named("release-approval.schema.json", record) + assert record["kind"] == "ReleaseApproval" + assert path.name == f"{record['metadata']['id']}.json" + assert record["metadata"]["id"].startswith("approval-") + spec = record["spec"] + assert spec["application"] == "crm" + assert spec["component"] == "crm-backend" + assert spec["release"] == RELEASE + assert spec["artifactSha256"] == digest + assert spec["gitCommit"] == COMMIT + assert spec["servers"] == ["h1", "h2"] + assert spec["approver"] == "roman" + assert spec["via"] == "terminal" + approved = datetime.fromisoformat(spec["approvedAt"]) + expires = datetime.fromisoformat(spec["expiresAt"]) + assert expires - approved == timedelta(hours=24) + data = payload["data"] + assert isinstance(data, dict) + assert data["approval"] == record + assert data["path"] == str(path) + for named in ("crm", "crm-backend", RELEASE, COMMIT, digest, "h1, h2"): + assert named in prompt + + +def test_s008_2_the_record_id_is_an_approval_time_and_16_hex() -> None: + candidate = ReleaseCandidate( + application_id=ResourceId("crm"), + component_id=ResourceId("crm-backend"), + release=ReleaseId(RELEASE), + artifact_sha256=Sha256Digest("b" * 64), + git_commit=GitCommit(COMMIT), + server_ids=(ResourceId("h1"),), + ) + + approval = approve_release( + candidate, + Approver("roman"), + ApprovalChannel.TERMINAL, + ValidityHours(24), + MOMENT, + ) + + document = approval.as_document() + assert approval.approval_id.value.startswith("approval-20260101t120000-") + assert len(approval.approval_id.value.rsplit("-", 1)[1]) == 16 + assert document["spec"] == { + "application": "crm", + "component": "crm-backend", + "release": RELEASE, + "artifactSha256": "b" * 64, + "gitCommit": COMMIT, + "servers": ["h1"], + "approver": "roman", + "approvedAt": "2026-01-01T12:00:00Z", + "expiresAt": "2026-01-02T12:00:00Z", + "via": "terminal", + } + + +@pytest.mark.parametrize("flags", [["--yes"], []], ids=["yes", "bare"]) +def test_s008_4_off_a_terminal_nothing_is_approved( + tmp_path: Path, capsys: pytest.CaptureFixture[str], flags: list[str] +) -> None: + project = _project(tmp_path) + _write_artifact(project) + + exit_code = main(_argv(project, *flags)) + + payload = json.loads(capsys.readouterr().out) + assert exit_code == 4 + assert payload["error"]["code"] == "PERSON_REQUIRED" + assert _records(project) == [] + + +@pytest.mark.parametrize("flags", [["--yes"], []], ids=["yes", "bare"]) +def test_s008_4_a_mistyped_release_id_approves_nothing( + tmp_path: Path, flags: list[str] +) -> None: + project = _project(tmp_path) + _write_artifact(project) + + exit_code, payload, _prompt = _at_terminal(project, "yes", *flags) + + error = payload["error"] + assert isinstance(error, dict) + assert exit_code == 4 + assert error["code"] == "ATTESTATION_MISMATCH" + assert _records(project) == [] + + +def test_s008_11_an_automatic_application_needs_no_approval(tmp_path: Path) -> None: + project = _project(tmp_path, approval="automatic") + _write_artifact(project) + + exit_code, payload, prompt = _at_terminal(project, RELEASE) + + error = payload["error"] + assert isinstance(error, dict) + assert exit_code == 4 + assert error["code"] == "PRECONDITION" + assert error["context"]["code"] == "release_approval_not_required" + assert prompt == "" + assert _records(project) == [] + + +def test_s008_11_a_missing_artifact_is_refused_before_asking(tmp_path: Path) -> None: + project = _project(tmp_path) + + exit_code, payload, prompt = _at_terminal(project, RELEASE) + + error = payload["error"] + assert isinstance(error, dict) + assert exit_code == 4 + assert error["code"] == "PRECONDITION" + assert error["context"]["code"] == "lifecycle_artifact_missing" + assert prompt == "" + assert _records(project) == [] + + +def test_s008_11_a_tampered_artifact_is_refused_before_asking(tmp_path: Path) -> None: + project = _project(tmp_path) + _write_artifact(project) + archive = project / "tmp" / "artifacts" / "crm-backend" / f"{RELEASE}.tar.gz" + archive.write_bytes(archive.read_bytes() + b"tampered") + + exit_code, payload, prompt = _at_terminal(project, RELEASE) + + error = payload["error"] + assert isinstance(error, dict) + assert exit_code == 4 + assert error["code"] == "PRECONDITION" + assert error["context"]["code"] == "lifecycle_artifact_digest_mismatch" + assert prompt == "" + assert _records(project) == [] + + +def test_an_undeclared_component_is_refused_before_asking(tmp_path: Path) -> None: + project = _project(tmp_path) + stdout = _Terminal() + stderr = io.StringIO() + + exit_code = app.run( + [ + *("release", "approve", "ghost", "--release", RELEASE), + *("--project", str(project), "--schemas", str(SCHEMAS)), + *("--approver", "roman", "--format", "json"), + ], + stdin=_Terminal(f"{RELEASE}\n"), + stdout=stdout, + stderr=stderr, + ) + + payload = json.loads(stdout.getvalue()) + assert exit_code == 4 + assert payload["error"]["context"]["code"] == "lifecycle_component_missing" + assert stderr.getvalue() == "" + + +@pytest.mark.parametrize("hours", ["0", "169"]) +def test_s008_12_a_validity_out_of_range_is_a_usage_error( + tmp_path: Path, capsys: pytest.CaptureFixture[str], hours: str +) -> None: + project = _project(tmp_path) + _write_artifact(project) + + exit_code = main(_argv(project, "--valid-for", hours)) + + payload = json.loads(capsys.readouterr().out) + assert exit_code == 2 + assert payload["error"]["code"] == "ARG_ERROR" + assert "--valid-for" in payload["error"]["message"] + assert _records(project) == [] + + +def test_s008_12_a_validity_of_two_hours_expires_two_hours_later( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + _write_artifact(project) + + exit_code, payload, _prompt = _at_terminal(project, RELEASE, "--valid-for", "2") + + assert exit_code == 0, payload + (path,) = _records(project) + spec = json.loads(path.read_text(encoding="utf-8"))["spec"] + approved = datetime.fromisoformat(spec["approvedAt"]) + expires = datetime.fromisoformat(spec["expiresAt"]) + assert expires - approved == timedelta(hours=2) + + +@pytest.mark.parametrize("hours", [0, 169, True]) +def test_the_validity_is_whole_hours_from_one_to_a_week(hours: int) -> None: + with pytest.raises(ReleaseApprovalError): + ValidityHours(hours) + + +def test_an_approval_needs_an_approver() -> None: + with pytest.raises(ReleaseApprovalError) as error: + Approver(" ") + + assert error.value.code == "release_approver_missing" + + +@pytest.mark.parametrize( + "name", ["roman\nadmin", "roman\x1b[31m", "roman\tx", "roman" + chr(0x202E)] +) +def test_an_approver_with_control_characters_is_refused(name: str) -> None: + # A later plan shows the approver; a newline or escape must not forge a line. + with pytest.raises(ReleaseApprovalError) as error: + Approver(name) + + assert error.value.code == "release_approver_invalid" + + +def test_the_schema_refuses_an_approver_with_control_characters() -> None: + approval = approve_release( + ReleaseCandidate( + application_id=ResourceId("crm"), + component_id=ResourceId("crm-backend"), + release=ReleaseId(RELEASE), + artifact_sha256=Sha256Digest("b" * 64), + git_commit=GitCommit(COMMIT), + server_ids=(ResourceId("h1"),), + ), + Approver("roman"), + ApprovalChannel.TERMINAL, + ValidityHours(24), + MOMENT, + ) + document = approval.as_document() + spec = document["spec"] + assert isinstance(spec, dict) + spec["approver"] = "roman\nadmin" + + with pytest.raises(ValidationError): + SchemaCatalog(SCHEMAS).validate_named("release-approval.schema.json", document) + + +def test_a_second_identical_approval_is_refused_not_overwritten( + tmp_path: Path, +) -> None: + candidate = ReleaseCandidate( + application_id=ResourceId("crm"), + component_id=ResourceId("crm-backend"), + release=ReleaseId(RELEASE), + artifact_sha256=Sha256Digest("b" * 64), + git_commit=GitCommit(COMMIT), + server_ids=(ResourceId("h1"),), + ) + catalog = SchemaCatalog(SCHEMAS) + + def at(moment: datetime) -> ReleaseApproval: + return approve_release( + candidate, + Approver("roman"), + ApprovalChannel.TERMINAL, + ValidityHours(24), + moment, + ) + + # Two approvals in the same second have the same id. + path = write_approval(at(MOMENT), tmp_path, catalog) + with pytest.raises(ReleaseApprovalError) as error: + write_approval(at(MOMENT.replace(microsecond=500)), tmp_path, catalog) + + assert error.value.code == "release_approval_exists" + assert str(path) in error.value.detail + assert list(tmp_path.iterdir()) == [path] + + +def test_a_blank_approver_is_refused_before_asking(tmp_path: Path) -> None: + project = _project(tmp_path) + _write_artifact(project) + stdout = _Terminal() + stderr = io.StringIO() + + exit_code = app.run( + [ + *("release", "approve", "crm-backend", "--release", RELEASE), + *("--project", str(project), "--schemas", str(SCHEMAS)), + *("--approver", " ", "--format", "json"), + ], + stdin=_Terminal(f"{RELEASE}\n"), + stdout=stdout, + stderr=stderr, + ) + + payload = json.loads(stdout.getvalue()) + assert exit_code == 4 + assert payload["error"]["context"]["code"] == "release_approver_missing" + assert stderr.getvalue() == "" + assert _records(project) == [] + + +def test_the_release_approval_command_is_a_persons_and_never_a_tool() -> None: + command = app.commands[CommandPath("release.approve")] + + assert command.requires_person is True + assert command.mcp is False + assert "release.approve" not in PROJECT_COMMANDS