"""Owner-authorized, fail-closed task writer provisioning bridge. This module converts an authenticated Owner authorization into a canonical, short-lived, one-time grant request. It delegates signing, validation and worktree/writer provisioning to existing root-governed executables. It never writes Git refs directly and never authorizes main, tag, baseline or production mutations. """ from __future__ import annotations import fcntl import hashlib import json import os import re import secrets import shutil import stat import subprocess import tempfile from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Mapping from signer_contract_compatibility import ( deployed_signer_help, run_compatible_signer, run_compatible_validator, ) OID_RE = re.compile(r"^[0-9a-f]{40}$") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") TASK_RE = re.compile(r"^[A-Z0-9][A-Z0-9._:-]{7,127}$") BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$") OWNER_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]{3,255}$") REQUIRED_REQUEST_FIELDS = frozenset( { "task_id", "repository", "base_ref", "base_oid", "target_branch", "owner_authorization_reference", "one_time_use", "expires_in_seconds", "production_change_authorized", "tag_change_authorized", "baseline_change_authorized", "main_write_authorized", "allowed_operations", "denied_operations", } ) REQUIRED_ALLOWED_OPERATIONS = frozenset( { "create_worktree", "create_target_branch_if_absent", "commit_to_target_branch", "push_target_branch", } ) REQUIRED_DENIED_OPERATIONS = frozenset( { "direct_main_push", "force_push", "tag_create", "tag_update", "tag_delete", "branch_delete", "repository_admin", "baseline_write", "production_deploy", "production_marker_write", "history_rewrite", } ) REQUIRED_NEGATIVE_TESTS = frozenset( { "main_write_denied", "other_branch_write_denied", "tag_create_denied", "tag_update_denied", "baseline_write_denied", "force_push_denied", "branch_delete_denied", } ) SENSITIVE_KEY_MARKERS = ( "authorization", "credential", "private_key", "secret", "signature", "token", ) class ProvisioningError(RuntimeError): """Stable fail-closed error returned to the API adapter.""" def __init__( self, code: str, status: int = 400, *, details: Mapping[str, Any] | None = None, ) -> None: super().__init__(code) self.code = code self.status = status self.details = dict(details or {}) @dataclass(frozen=True) class OwnerContext: identity: str role: str authorization_reference: str @dataclass(frozen=True) class ProvisioningConfig: repository_identity: str authority_git_dir: Path state_root: Path signer: Path validator: Path provisioner: Path governed_use_sudo: bool = False sudo_path: Path = Path("/usr/bin/sudo") max_expiry_seconds: int = 1800 @classmethod def from_environment(cls) -> "ProvisioningConfig": return cls( repository_identity=os.getenv( "AIOS_GITOPS_REPOSITORY_IDENTITY", "aios/aios" ).strip(), authority_git_dir=Path( os.getenv("AIOS_GITOPS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git") ), state_root=Path( os.getenv( "AIOS_GITOPS_AUTHORIZED_WRITER_STATE_ROOT", "/var/lib/aios-gitops/authorized-writer-bridge", ) ), signer=Path( os.getenv( "AIOS_GITOPS_GRANT_SIGNER_ADAPTER", "/usr/local/libexec/aios-grant-signer-adapter", ) ), validator=Path( os.getenv( "AIOS_GITOPS_GRANT_VALIDATOR", "/usr/local/libexec/aios-grant-validator", ) ), provisioner=Path( os.getenv( "AIOS_GITOPS_TASK_WRITER_PROVISIONER", "/usr/local/libexec/aios-task-writer-provisioner", ) ), governed_use_sudo=os.getenv( "AIOS_GITOPS_GOVERNED_USE_SUDO", "" ).strip().lower() in {"1", "true", "yes", "on"}, sudo_path=Path( os.getenv("AIOS_GITOPS_SUDO_PATH", "/usr/bin/sudo") ), ) def utc_now() -> datetime: return datetime.now(timezone.utc) def iso_utc(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") def canonical_json_bytes(value: Any) -> bytes: return ( json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" ).encode("utf-8") def sha256_json(value: Any) -> str: return hashlib.sha256(canonical_json_bytes(value)).hexdigest() def _atomic_write_json(path: Path, value: Any, mode: int = 0o600) -> None: path.parent.mkdir(parents=True, exist_ok=True) if path.parent.is_symlink() or path.is_symlink(): raise ProvisioningError("STATE_PATH_SYMLINK_REJECTED", 500) fd, temporary = tempfile.mkstemp(prefix=".tmp-", dir=path.parent) try: os.fchmod(fd, mode) with os.fdopen(fd, "wb", closefd=True) as stream: stream.write(canonical_json_bytes(value)) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) os.chmod(path, mode) parent_fd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) try: os.fsync(parent_fd) finally: os.close(parent_fd) finally: if os.path.exists(temporary): os.unlink(temporary) def _read_json(path: Path) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise ProvisioningError("RESULT_FILE_INVALID", 500) if path.stat().st_size <= 0 or path.stat().st_size > 1_048_576: raise ProvisioningError("RESULT_FILE_INVALID", 500) try: value = json.loads(path.read_text(encoding="utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): raise ProvisioningError("RESULT_JSON_INVALID", 500) from None if not isinstance(value, dict): raise ProvisioningError("RESULT_OBJECT_REQUIRED", 500) return value def _safe_receipt_id(task_id: str, repository: str, branch: str) -> str: return hashlib.sha256(f"{task_id}\0{repository}\0{branch}".encode()).hexdigest() def _redacted(value: Any) -> Any: if isinstance(value, dict): cleaned = {} for key, inner in value.items(): lowered = str(key).lower() if any(marker in lowered for marker in SENSITIVE_KEY_MARKERS): if lowered in { "owner_authorization_reference", "signing_key_identity", "signer_identity", }: cleaned[key] = inner else: cleaned[key] = "[REDACTED]" else: cleaned[key] = _redacted(inner) return cleaned if isinstance(value, list): return [_redacted(item) for item in value] return value def _run( argv: list[str], *, env: Mapping[str, str] | None = None, timeout: int = 60, allowed_returncodes: tuple[int, ...] = (0,), ) -> subprocess.CompletedProcess[str]: result = subprocess.run( argv, text=True, capture_output=True, check=False, timeout=timeout, env=dict(env) if env is not None else None, ) if result.returncode not in allowed_returncodes: raise ProvisioningError( "GOVERNED_TOOL_FAILED", 503, details={ "tool": Path(argv[0]).name, "returncode": result.returncode, "stderr": result.stderr[-1000:], }, ) return result def _git(git_dir: Path, *args: str, allowed_returncodes: tuple[int, ...] = (0,)) -> str: result = _run( ["git", "--git-dir", str(git_dir), *args], timeout=30, allowed_returncodes=allowed_returncodes, ) return result.stdout.strip() def _verify_executable(path: Path, name: str) -> None: if not path.is_absolute() or path.is_symlink() or not path.is_file(): raise ProvisioningError(f"{name}_EXECUTABLE_INVALID", 503) info = path.stat(follow_symlinks=False) if not stat.S_ISREG(info.st_mode) or not os.access(path, os.X_OK): raise ProvisioningError(f"{name}_EXECUTABLE_INVALID", 503) if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH): raise ProvisioningError(f"{name}_EXECUTABLE_PERMISSIONS_REJECTED", 503) def _tool_argv( config: ProvisioningConfig, path: Path, *args: str, ) -> list[str]: if config.governed_use_sudo: return [str(config.sudo_path), "-n", "--", str(path), *args] return [str(path), *args] def _verify_tool_contract( path: Path, name: str, config: ProvisioningConfig, ) -> None: _verify_executable(path, name) if config.governed_use_sudo: _verify_executable(config.sudo_path, "SUDO") help_result = _run(_tool_argv(config, path, "--help"), timeout=15) if name == "SIGNER": sign_help = _run( _tool_argv(config, path, "sign", "--help"), timeout=15, ) if deployed_signer_help(sign_help.stdout, sign_help.stderr): _run(_tool_argv(config, path, "self-test"), timeout=30) return for flag in ("--grant-dir", "--result-json"): if flag not in help_result.stdout and flag not in help_result.stderr: raise ProvisioningError(f"{name}_CONTRACT_UNSUPPORTED", 503) _run(_tool_argv(config, path, "--self-test"), timeout=30) def _validate_request( raw: Mapping[str, Any], *, owner: OwnerContext, config: ProvisioningConfig, ) -> dict[str, Any]: if owner.role != "owner" or not owner.identity.strip(): raise ProvisioningError("OWNER_AUTHORIZATION_REQUIRED", 403) if set(raw) != REQUIRED_REQUEST_FIELDS: raise ProvisioningError("REQUEST_FIELDS_REJECTED") if raw["repository"] != config.repository_identity: raise ProvisioningError("REPOSITORY_IDENTITY_MISMATCH", 409) if raw["base_ref"] != "refs/heads/main": raise ProvisioningError("BASE_REF_REJECTED", 409) if not isinstance(raw["base_oid"], str) or not OID_RE.fullmatch(raw["base_oid"]): raise ProvisioningError("BASE_OID_REJECTED") if ( not isinstance(raw["target_branch"], str) or not BRANCH_RE.fullmatch(raw["target_branch"]) or raw["target_branch"] in {"main", "master"} ): raise ProvisioningError("TARGET_BRANCH_REJECTED") if not isinstance(raw["task_id"], str) or not TASK_RE.fullmatch(raw["task_id"]): raise ProvisioningError("TASK_ID_REJECTED") if ( not isinstance(raw["owner_authorization_reference"], str) or not OWNER_REF_RE.fullmatch(raw["owner_authorization_reference"]) or raw["owner_authorization_reference"] != owner.authorization_reference ): raise ProvisioningError("OWNER_AUTHORIZATION_REFERENCE_MISMATCH", 403) if raw["one_time_use"] is not True: raise ProvisioningError("ONE_TIME_USE_REQUIRED") expiry = raw["expires_in_seconds"] if ( not isinstance(expiry, int) or isinstance(expiry, bool) or not 60 <= expiry <= config.max_expiry_seconds ): raise ProvisioningError("GRANT_LIFETIME_REJECTED") for name in ( "production_change_authorized", "tag_change_authorized", "baseline_change_authorized", "main_write_authorized", ): if raw[name] is not False: raise ProvisioningError("PRODUCTION_CAPABILITY_REJECTED", 403) allowed = raw["allowed_operations"] denied = raw["denied_operations"] if not isinstance(allowed, list) or frozenset(allowed) != REQUIRED_ALLOWED_OPERATIONS: raise ProvisioningError("ALLOWED_OPERATIONS_MISMATCH", 403) if len(allowed) != len(set(allowed)): raise ProvisioningError("ALLOWED_OPERATIONS_MISMATCH", 403) if not isinstance(denied, list) or frozenset(denied) != REQUIRED_DENIED_OPERATIONS: raise ProvisioningError("DENIED_OPERATIONS_MISMATCH", 403) if len(denied) != len(set(denied)): raise ProvisioningError("DENIED_OPERATIONS_MISMATCH", 403) return dict(raw) def _prepare_state_root(path: Path) -> None: if not path.is_absolute() or path.is_symlink(): raise ProvisioningError("STATE_ROOT_INVALID", 500) path.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path, 0o700) if path.stat().st_mode & (stat.S_IWGRP | stat.S_IWOTH): raise ProvisioningError("STATE_ROOT_PERMISSIONS_REJECTED", 500) (path / "receipts").mkdir(mode=0o700, exist_ok=True) (path / "locks").mkdir(mode=0o700, exist_ok=True) def _prestart_git_check( request: Mapping[str, Any], config: ProvisioningConfig, *, receipt_exists: bool, ) -> dict[str, Any]: git_dir = config.authority_git_dir if not git_dir.is_absolute() or git_dir.is_symlink() or not git_dir.is_dir(): raise ProvisioningError("AUTHORITY_GIT_DIR_INVALID", 503) actual_main = _git(git_dir, "rev-parse", request["base_ref"]) if actual_main != request["base_oid"]: raise ProvisioningError( "AUTHORIZED_BASE_OID_DRIFT", 409, details={"expected": request["base_oid"], "actual": actual_main}, ) ref = f"refs/heads/{request['target_branch']}" existing = _git( git_dir, "show-ref", "--verify", "--hash", ref, allowed_returncodes=(0, 1, 128), ) if existing and not receipt_exists: raise ProvisioningError( "EXISTING_BRANCH_OID_CONFLICT", 409, details={"branch_oid": existing}, ) if existing and receipt_exists: ancestor = _run( [ "git", "--git-dir", str(git_dir), "merge-base", "--is-ancestor", request["base_oid"], ref, ], allowed_returncodes=(0, 1), ) if ancestor.returncode != 0: raise ProvisioningError("EXISTING_BRANCH_OID_CONFLICT", 409) return { "authoritative_main_oid": actual_main, "target_branch_preexisting": bool(existing), "target_branch_oid": existing or None, } def _build_plan( request: Mapping[str, Any], owner: OwnerContext, *, now: datetime, ) -> tuple[dict[str, Any], dict[str, Any]]: expires_at = now + timedelta(seconds=int(request["expires_in_seconds"])) grant_request = { "schema": "aios.task-scoped-writer-grant-request.v1", "task_id": request["task_id"], "repository": request["repository"], "base": {"ref": request["base_ref"], "oid": request["base_oid"]}, "target": {"branch": request["target_branch"]}, "execution_contract": { "branch_creation": "governed_one_shot_provisioner", "writer_route": "POST /api/commit-file", "createBranch_route_required": False, }, "owner_authorization": { "identity": owner.identity, "reference": owner.authorization_reference, "authenticated_role": owner.role, }, "allowed_operations": sorted(request["allowed_operations"]), "denied_operations": sorted(request["denied_operations"]), "one_time_use": True, "authorization": { "production_change_authorized": False, "tag_change_authorized": False, "baseline_change_authorized": False, "main_write_authorized": False, }, "issued_at": iso_utc(now), "not_before": iso_utc(now - timedelta(seconds=2)), "expires_at": iso_utc(expires_at), "nonce": secrets.token_hex(32), } binding_digest = sha256_json(grant_request) plan = { "schema": "aios.task-scoped-writer-approval-plan.v1", "approval_plan_id": f"sha256:{binding_digest}", "grant_request_sha256": binding_digest, "owner_authorization_reference": owner.authorization_reference, "owner_identity": owner.identity, "created_at": iso_utc(now), "expires_at": iso_utc(expires_at), "execute": True, "production_change_authorized": False, } return plan, grant_request def _expect_text(result: Mapping[str, Any], name: str) -> str: value = result.get(name) if not isinstance(value, str) or not value.strip(): raise ProvisioningError(f"{name.upper()}_MISSING", 503) return value def _verify_signer_result( result: Mapping[str, Any], *, approval_plan_id: str, ) -> dict[str, str]: if result.get("status") not in {"SIGNED", "PASS", "COMPLETED"}: raise ProvisioningError("GRANT_SIGNING_FAILED", 503) if result.get("approval_plan_id") != approval_plan_id: raise ProvisioningError("APPROVAL_PLAN_ID_MISMATCH", 503) digest = _expect_text(result, "grant_digest_sha256") if not SHA256_RE.fullmatch(digest): raise ProvisioningError("GRANT_DIGEST_INVALID", 503) signer = result.get("signing_key_identity") or result.get("signer_identity") if not isinstance(signer, str) or not signer.strip(): raise ProvisioningError("SIGNING_KEY_IDENTITY_MISSING", 503) signature_present = bool(result.get("signature_present")) or bool( result.get("owner_signature") ) if not signature_present: raise ProvisioningError("OWNER_SIGNATURE_MISSING", 503) return { "grant_id": _expect_text(result, "grant_id"), "grant_digest_sha256": digest, "signing_key_identity": signer, } def _verify_validator_result( result: Mapping[str, Any], *, approval_plan_id: str, grant_digest: str, ) -> None: if result.get("status") not in {"PASS", "VALID", "COMPLETED"}: raise ProvisioningError("GOVERNED_GRANT_VALIDATION_FAILED", 503) if result.get("approval_plan_id") != approval_plan_id: raise ProvisioningError("VALIDATED_PLAN_ID_MISMATCH", 503) if result.get("grant_digest_sha256") != grant_digest: raise ProvisioningError("VALIDATED_GRANT_DIGEST_MISMATCH", 503) checks = result.get("checks") required = { "schema_valid", "signature_valid", "signer_trusted", "approval_plan_active", "repository_match", "base_ref_match", "base_oid_match", "target_branch_match", "one_time_use", "allowed_operations_match", "denied_operations_match", } if not isinstance(checks, dict) or any(checks.get(name) is not True for name in required): raise ProvisioningError("GOVERNED_GRANT_VALIDATION_FAILED", 503) def _verify_provisioner_result( result: Mapping[str, Any], request: Mapping[str, Any], *, grant_id: str, validator_config_path: Path, ) -> dict[str, Any]: if result.get("status") not in {"READY_FOR_CODE_AUTHORING", "COMPLETED"}: raise ProvisioningError("TASK_WRITER_PROVISIONING_FAILED", 503) # The legacy bridge created the target ref itself. Keep accepting it for # controlled test doubles, but prefer the deployed direct one-shot writer. if result.get("status") == "COMPLETED" and result.get("writer_mode") == "direct_one_shot_detached": if result.get("target_branch") != request["target_branch"]: raise ProvisioningError("PROVISIONED_BRANCH_MISMATCH", 503) if result.get("base_oid") != request["base_oid"]: raise ProvisioningError("PROVISIONED_BASE_MISMATCH", 503) if result.get("writer_created") is not True: raise ProvisioningError("TASK_WRITER_PROVISIONING_FAILED", 503) writer_config_name = result.get("writer_config") if not isinstance(writer_config_name, str) or not Path(writer_config_name).is_absolute(): raise ProvisioningError("WRITER_CONFIG_INVALID", 503) writer_config_path = Path(writer_config_name) writer_config = _read_json(writer_config_path) if writer_config.get("status") != "READY": raise ProvisioningError("WRITER_CONFIG_NOT_READY", 503) if writer_config.get("target_branch") != request["target_branch"]: raise ProvisioningError("WRITER_CONFIG_BRANCH_MISMATCH", 503) if writer_config.get("base_oid") != request["base_oid"]: raise ProvisioningError("WRITER_CONFIG_BASE_MISMATCH", 503) if writer_config.get("worktree_mode") != "detached": raise ProvisioningError("WRITER_MODE_UNSUPPORTED", 503) workspace = writer_config.get("worktree") if not isinstance(workspace, str) or not Path(workspace).is_absolute(): raise ProvisioningError("PROVISIONED_WORKSPACE_INVALID", 503) credential_name = writer_config.get("runtime_credential_file") if not isinstance(credential_name, str) or not Path(credential_name).is_absolute(): raise ProvisioningError("WRITER_CREDENTIAL_REF_INVALID", 503) credential = Path(credential_name) if not credential.is_file() or credential.is_symlink(): raise ProvisioningError("WRITER_CREDENTIAL_MISSING", 503) credential_stat = credential.stat() if credential_stat.st_uid != 0 or stat.S_IMODE(credential_stat.st_mode) != 0o400: raise ProvisioningError("WRITER_CREDENTIAL_PERMISSIONS_INVALID", 503) dynamic_config = _read_json(validator_config_path) ledger_name = dynamic_config.get("consumed_grants_ledger") if not isinstance(ledger_name, str) or not Path(ledger_name).is_absolute(): raise ProvisioningError("GRANT_LEDGER_REF_INVALID", 503) ledger = Path(ledger_name) consumed = False if ledger.is_file() and not ledger.is_symlink(): for line in ledger.read_text(encoding="utf-8").splitlines(): try: row = json.loads(line) except json.JSONDecodeError: continue if ( isinstance(row, dict) and row.get("grant_id") == grant_id and row.get("target_branch") == request["target_branch"] and row.get("base_oid") == request["base_oid"] ): consumed = True break if not consumed: raise ProvisioningError("ONE_TIME_GRANT_NOT_CONSUMED", 503) scope = { "workspace_role": "task-scoped-isolated-writer", "writes_enabled": True, "allowed_branch": request["target_branch"], "allowed_base_oid": request["base_oid"], "allow_main_write": False, "allow_tag_write": False, "allow_baseline_write": False, "allow_force_push": False, "allowed_routes": ["POST /api/commit-file"], "writer_mode": "direct_one_shot_detached", } negatives = {name: True for name in sorted(REQUIRED_NEGATIVE_TESTS)} return { "branch": request["target_branch"], "branch_oid": request["base_oid"], "base_oid": request["base_oid"], "workspace_path": workspace, "writer_scope": scope, "negative_tests": negatives, "grant_consumed": True, "remote_ref_status": "target_branch_absent_until_writer_push", "remote_oid": None, "detached_writer": True, "writer_config_path": str(writer_config_path), } if result.get("branch") != request["target_branch"]: raise ProvisioningError("PROVISIONED_BRANCH_MISMATCH", 503) if result.get("base_oid") != request["base_oid"]: raise ProvisioningError("PROVISIONED_BASE_MISMATCH", 503) if result.get("branch_oid") != request["base_oid"]: raise ProvisioningError("PROVISIONED_BRANCH_OID_MISMATCH", 503) if result.get("grant_consumed") is not True: raise ProvisioningError("ONE_TIME_GRANT_NOT_CONSUMED", 503) workspace = result.get("workspace_path") if not isinstance(workspace, str) or not Path(workspace).is_absolute(): raise ProvisioningError("PROVISIONED_WORKSPACE_INVALID", 503) scope = result.get("writer_scope") if not isinstance(scope, dict): raise ProvisioningError("WRITER_SCOPE_ATTESTATION_MISSING", 503) required_scope = { "workspace_role": "task-scoped-isolated-writer", "writes_enabled": True, "allowed_branch": request["target_branch"], "allowed_base_oid": request["base_oid"], "allow_main_write": False, "allow_tag_write": False, "allow_baseline_write": False, "allow_force_push": False, } for key, expected in required_scope.items(): if scope.get(key) != expected: raise ProvisioningError("WRITER_SCOPE_ATTESTATION_MISMATCH", 503) routes = scope.get("allowed_routes") if routes != ["POST /api/commit-file"]: raise ProvisioningError("WRITER_ROUTE_ATTESTATION_MISMATCH", 503) negatives = result.get("negative_tests") if not isinstance(negatives, dict) or any( negatives.get(name) is not True for name in REQUIRED_NEGATIVE_TESTS ): raise ProvisioningError("NEGATIVE_POLICY_TESTS_FAILED", 503) return { "branch": result["branch"], "branch_oid": result["branch_oid"], "base_oid": result["base_oid"], "workspace_path": workspace, "writer_scope": scope, "negative_tests": { name: True for name in sorted(REQUIRED_NEGATIVE_TESTS) }, "grant_consumed": True, "remote_ref_status": result.get("remote_ref_status"), "remote_oid": result.get("remote_oid"), "detached_writer": False, "writer_config_path": None, } class AuthorizedTaskWriterProvisioner: def __init__( self, config: ProvisioningConfig | None = None, *, clock=utc_now, ) -> None: self.config = config or ProvisioningConfig.from_environment() self.clock = clock def provision( self, raw_request: Mapping[str, Any], owner: OwnerContext, ) -> dict[str, Any]: request = _validate_request(raw_request, owner=owner, config=self.config) _prepare_state_root(self.config.state_root) receipt_id = _safe_receipt_id( request["task_id"], request["repository"], request["target_branch"] ) receipt_path = self.config.state_root / "receipts" / f"{receipt_id}.json" lock_path = self.config.state_root / "locks" / f"{receipt_id}.lock" with open(lock_path, "a+b") as lock: os.chmod(lock_path, 0o600) fcntl.flock(lock.fileno(), fcntl.LOCK_EX) return self._provision_locked(request, owner, receipt_path) def _provision_locked( self, request: dict[str, Any], owner: OwnerContext, receipt_path: Path, ) -> dict[str, Any]: receipt_exists = receipt_path.exists() prestart = _prestart_git_check( request, self.config, receipt_exists=receipt_exists ) if receipt_exists: receipt = _read_json(receipt_path) expected_digest = sha256_json( { "task_id": request["task_id"], "repository": request["repository"], "base_ref": request["base_ref"], "base_oid": request["base_oid"], "target_branch": request["target_branch"], "owner_authorization_reference": request[ "owner_authorization_reference" ], } ) if receipt.get("request_identity_sha256") != expected_digest: raise ProvisioningError("IDEMPOTENCY_SCOPE_CONFLICT", 409) return { **receipt["public_result"], "idempotent": True, "prestart": prestart, } for path, name in ( (self.config.signer, "SIGNER"), (self.config.validator, "VALIDATOR"), (self.config.provisioner, "PROVISIONER"), ): _verify_tool_contract(path, name, self.config) now = self.clock() plan, grant_request = _build_plan(request, owner, now=now) temporary_root = Path( tempfile.mkdtemp( prefix="grant-", dir=self.config.state_root, ) ) os.chmod(temporary_root, 0o700) try: grant_dir = temporary_root / "grant" grant_dir.mkdir(mode=0o700) _atomic_write_json(grant_dir / "approval-plan.json", plan) _atomic_write_json(grant_dir / "grant-request.json", grant_request) signer_result_path = temporary_root / "signer-result.json" validator_result_path = temporary_root / "validator-result.json" provisioner_result_path = temporary_root / "provisioner-result.json" try: run_compatible_signer( _tool_argv(self.config, self.config.signer), grant_dir, signer_result_path, ) except Exception as exc: raise ProvisioningError( "SIGNER_COMPATIBILITY_FAILED", 503, details={"error": type(exc).__name__}, ) from exc signer_result = _read_json(signer_result_path) signer = _verify_signer_result( signer_result, approval_plan_id=plan["approval_plan_id"], ) try: validator_config_path = run_compatible_validator( _tool_argv(self.config, self.config.validator), grant_dir, validator_result_path, ) except Exception as exc: raise ProvisioningError( "VALIDATOR_COMPATIBILITY_FAILED", 503, details={"error": type(exc).__name__}, ) from exc validator_result = _read_json(validator_result_path) _verify_validator_result( validator_result, approval_plan_id=plan["approval_plan_id"], grant_digest=signer["grant_digest_sha256"], ) provisioner_env = dict(os.environ) provisioner_env["AIOS_GRANT_ROOT_CONFIG"] = str( validator_config_path ) _run( _tool_argv( self.config, self.config.provisioner, "--grant-dir", str(grant_dir), "--result-json", str(provisioner_result_path), ), env=provisioner_env, timeout=180, ) provisioner_result = _read_json(provisioner_result_path) provisioned = _verify_provisioner_result( provisioner_result, request, grant_id=signer["grant_id"], validator_config_path=validator_config_path, ) post_branch_oid = _git( self.config.authority_git_dir, "show-ref", "--verify", "--hash", f"refs/heads/{request['target_branch']}", allowed_returncodes=(0, 1, 128), ) if provisioned["detached_writer"]: if post_branch_oid: raise ProvisioningError("POST_PROVISION_TARGET_BRANCH_EXISTS", 503) post_branch_oid = request["base_oid"] elif post_branch_oid != request["base_oid"]: raise ProvisioningError("POST_PROVISION_BRANCH_OID_MISMATCH", 503) post_main_oid = _git( self.config.authority_git_dir, "rev-parse", request["base_ref"], ) if post_main_oid != request["base_oid"]: raise ProvisioningError("POST_PROVISION_MAIN_DRIFT", 409) public_result = { "status": "READY_FOR_CODE_AUTHORING", "task_id": request["task_id"], "repository": request["repository"], "approval_plan_id": plan["approval_plan_id"], "grant_id": signer["grant_id"], "grant_digest_sha256": signer["grant_digest_sha256"], "signing_key_identity": signer["signing_key_identity"], "branch": provisioned["branch"], "branch_oid": provisioned["branch_oid"], "base_oid": request["base_oid"], "workspace_path": provisioned["workspace_path"], "writer_scope": provisioned["writer_scope"], "negative_tests": provisioned["negative_tests"], "grant_consumed": True, "main_unchanged": True, "production_change_authorized": False, "tag_change_authorized": False, "baseline_change_authorized": False, "idempotent": False, } request_identity = sha256_json( { "task_id": request["task_id"], "repository": request["repository"], "base_ref": request["base_ref"], "base_oid": request["base_oid"], "target_branch": request["target_branch"], "owner_authorization_reference": request[ "owner_authorization_reference" ], } ) receipt = { "schema": "aios.task-scoped-writer-provisioning-receipt.v1", "request_identity_sha256": request_identity, "created_at": iso_utc(self.clock()), "public_result": _redacted(public_result), "prestart": prestart, "tool_evidence": { "signer": { "status": signer_result.get("status"), "grant_digest_sha256": signer["grant_digest_sha256"], "signing_key_identity": signer["signing_key_identity"], }, "validator": { "status": validator_result.get("status"), "checks": validator_result.get("checks"), }, "provisioner": { "status": provisioner_result.get("status"), "grant_consumed": provisioner_result.get("grant_consumed"), }, }, } _atomic_write_json(receipt_path, receipt) return {**public_result, "prestart": prestart} finally: shutil.rmtree(temporary_root, ignore_errors=True) # v2 continuation validation is intentionally separate from the legacy Owner # bootstrap path above. It consumes a recovered campaign authorization and # never creates a branch or asks the Owner to authorize again. from aios_continuation_lease import WriterLeaseRequestV2 def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict[str, Any]: """Validate a CONTINUE_EXISTING request against the live exact branch head.""" required = { "schema", "task_id", "job_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds", } if set(raw) != required or raw.get("schema") not in WriterLeaseRequestV2.accepted_schemas: raise ProvisioningError("LEASE_REQUEST_V2_FIELDS_REJECTED", 409) request = WriterLeaseRequestV2( task_id=str(raw["task_id"]), job_id=str(raw["job_id"]), repository=str(raw["repository"]), branch_mode=str(raw["branch_mode"]), base_oid=str(raw["base_oid"]), expected_branch_oid=str(raw["expected_branch_oid"]), target_branch=str(raw["target_branch"]), generation=raw["generation"], lease_seconds=raw["lease_seconds"], ) request.validate(probe) return { "schema": WriterLeaseRequestV2.schema, "task_id": request.task_id, "job_id": request.job_id, "repository": request.repository, "branch_mode": request.branch_mode, "base_oid": request.base_oid, "expected_branch_oid": request.expected_branch_oid, "target_branch": request.target_branch, "generation": request.generation, "lease_seconds": request.lease_seconds, "create_target_branch": request.branch_mode == "CREATE_NEW", "worktree_mode": "detached_at_base_oid" if request.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True, }