Files
aios/active-runtime/app/authorized_task_writer_provisioning.py
T

954 lines
35 KiB
Python

"""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)