activation_transaction.py
Rama main · réplica de lectura del repositorio original.
"""Crash-safe provenance for one AgentsServer release activation.
The installer deliberately keeps this small ledger outside any release tree.
It exists only from the last pre-takeover snapshot of links/configuration until
either candidate health or verified rollback. A later installer must recover
that exact transaction before it may choose a new rollback baseline.
"""
from __future__ import annotations
import argparse
import hashlib
import ipaddress
import json
import os
from pathlib import Path
import re
import secrets
import shutil
import stat
import sys
from typing import Any
FORMAT = 1
MAX_CONTROL_BYTES = 256 * 1024
MAX_CONFIG_BYTES = 16 * 1024 * 1024
PHASES = {
"prepared",
"guarded",
"linking",
"linked",
"stopping",
"stopped",
"fencing",
"fenced",
"authorizing",
"authority",
"candidate-starting",
"candidate-healthy",
"committing",
"committed",
"rolling-back",
"rolled-back",
"rollback-healthy",
}
TRANSITIONS = {
"prepared": {"guarded", "linking", "rolling-back"},
"guarded": {"linking", "rolling-back"},
"linking": {"linked", "rolling-back"},
"linked": {"stopping", "candidate-starting", "rolling-back"},
"stopping": {"stopped", "rolling-back"},
"stopped": {"fencing", "fenced", "authorizing", "rolling-back"},
"fencing": {"fenced", "rolling-back"},
"fenced": {"authorizing", "rolling-back"},
"authorizing": {"authority", "rolling-back"},
"authority": {"candidate-starting", "rolling-back"},
"candidate-starting": {"candidate-healthy", "rolling-back"},
"candidate-healthy": {"committing", "rolling-back"},
"committing": {"committed"},
"committed": set(),
"rolling-back": {"rolled-back"},
"rolled-back": {"rollback-healthy"},
"rollback-healthy": set(),
}
MANIFEST_TEMP_RE = re.compile(r"^\.manifest\.json\.[0-9a-f]{24}\.tmp$")
GC_DIRECTORY_RE = re.compile(r"^\.activation-transaction-gc-activation-[0-9a-f]{24}$")
BEGIN_DIRECTORY_RE = re.compile(r"^\.activation-transaction\.[0-9a-f]{24}\.tmp$")
def _absolute(path: str | Path) -> Path:
return Path(os.path.abspath(os.path.expanduser(os.fspath(path))))
def _fsync_directory(path: Path) -> None:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_DIRECTORY", 0)
descriptor = os.open(path, flags)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def _safe_text(value: Any, name: str, *, maximum: int = 4096) -> str:
if not isinstance(value, str) or not value or len(value) > maximum:
raise RuntimeError(f"activation transaction {name} is invalid")
if "\x00" in value or "\n" in value or "\r" in value:
raise RuntimeError(f"activation transaction {name} is invalid")
return value
def _read_private(path: Path, *, maximum: int = MAX_CONTROL_BYTES) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_uid != os.getuid()
or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) != 0o600
or not 0 <= before.st_size <= maximum
):
raise PermissionError("activation transaction file is unsafe")
chunks: list[bytes] = []
remaining = maximum + 1
while remaining:
chunk = os.read(descriptor, min(remaining, 1024 * 1024))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
value = b"".join(chunks)
after = os.fstat(descriptor)
linked = path.lstat()
if (
len(value) > maximum
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
or (before.st_dev, before.st_ino) != (linked.st_dev, linked.st_ino)
):
raise RuntimeError("activation transaction file changed while reading")
return value
finally:
os.close(descriptor)
def _read_owned_regular(path: Path, *, maximum: int = MAX_CONFIG_BYTES) -> bytes:
"""Read an installer-owned config, allowing 0600 or ordinary 0644 mode."""
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_uid != os.getuid()
or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) & 0o022
or not 0 <= before.st_size <= maximum
):
raise PermissionError("activation configuration file is unsafe")
chunks: list[bytes] = []
remaining = maximum + 1
while remaining:
chunk = os.read(descriptor, min(remaining, 1024 * 1024))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
value = b"".join(chunks)
after = os.fstat(descriptor)
linked = path.lstat()
if (
len(value) > maximum
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
or (before.st_dev, before.st_ino) != (linked.st_dev, linked.st_ino)
):
raise RuntimeError("activation configuration changed while reading")
return value
finally:
os.close(descriptor)
def _write_new_private(path: Path, value: bytes) -> None:
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
flags |= getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags, 0o600)
try:
os.fchmod(descriptor, 0o600)
written = 0
while written < len(value):
written += os.write(descriptor, value[written:])
os.fsync(descriptor)
finally:
os.close(descriptor)
def _replace_private(path: Path, value: bytes) -> None:
temporary = path.parent / f".{path.name}.{secrets.token_hex(12)}.tmp"
try:
_write_new_private(temporary, value)
os.replace(temporary, path)
_fsync_directory(path.parent)
finally:
try:
temporary.unlink()
except FileNotFoundError:
pass
def _sha256(path: Path) -> str:
return hashlib.sha256(_read_private(path, maximum=MAX_CONFIG_BYTES)).hexdigest()
def _owned_regular_sha256(path: Path, *, maximum: int = 4096) -> str:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_uid != os.getuid()
or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) & 0o022
or not 1 <= before.st_size <= maximum
):
raise PermissionError("activation release identity is unsafe")
chunks: list[bytes] = []
remaining = maximum + 1
while remaining:
chunk = os.read(descriptor, min(remaining, 1024 * 1024))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
value = b"".join(chunks)
after = os.fstat(descriptor)
linked = path.lstat()
if (
not 1 <= len(value) <= maximum
or len(value) != before.st_size
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
or (before.st_dev, before.st_ino) != (linked.st_dev, linked.st_ino)
):
raise RuntimeError("activation release identity changed while reading")
return hashlib.sha256(value).hexdigest()
finally:
os.close(descriptor)
def _release_identity(source: Path | None, target: Path | None) -> dict[str, Any]:
if source is None or target is None:
return {}
info = source.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) & 0o022
):
raise PermissionError("activation rollback release is unsafe")
return {
"source": str(source),
"target": str(target),
"device": info.st_dev,
"inode": info.st_ino,
"version_sha256": _owned_regular_sha256(source / "VERSION"),
}
def _release_matches(path: Path, release: dict[str, Any]) -> bool:
try:
info = path.lstat()
except FileNotFoundError:
return False
return (
stat.S_ISDIR(info.st_mode)
and info.st_uid == os.getuid()
and not stat.S_IMODE(info.st_mode) & 0o022
and info.st_dev == release["device"]
and info.st_ino == release["inode"]
and _owned_regular_sha256(path / "VERSION")
== release["version_sha256"]
)
def _locate_release(
release: dict[str, Any],
*,
extras: tuple[Path, ...] = (),
) -> Path | None:
candidates = [
_absolute(release["source"]),
_absolute(release["target"]),
*extras,
]
seen: set[Path] = set()
for candidate in candidates:
if candidate in seen:
continue
seen.add(candidate)
if _release_matches(candidate, release):
return candidate
return None
def _copy_config(source: Path, destination: Path) -> tuple[str, int]:
source_info = source.lstat()
value = _read_owned_regular(source, maximum=MAX_CONFIG_BYTES)
linked_info = source.lstat()
if (source_info.st_dev, source_info.st_ino) != (
linked_info.st_dev,
linked_info.st_ino,
):
raise RuntimeError("activation configuration changed while capturing it")
_write_new_private(destination, value)
return hashlib.sha256(value).hexdigest(), stat.S_IMODE(source_info.st_mode)
def _configuration(
path: Path,
backup: Path,
*,
allowed_modes: set[int],
) -> dict[str, Any]:
try:
info = path.lstat()
except FileNotFoundError:
return {
"existed": False,
"backup": backup.name,
"sha256": None,
"mode": None,
}
if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid():
raise PermissionError(f"configuration path is unsafe: {path}")
digest, mode = _copy_config(path, backup)
if mode not in allowed_modes:
raise PermissionError(f"configuration mode cannot be restored safely: {path}")
return {
"existed": True,
"backup": backup.name,
"sha256": digest,
"mode": mode,
}
def _link_state(path: Path, *, allow_directory: bool) -> dict[str, str]:
try:
before = path.lstat()
except FileNotFoundError:
return {"kind": "missing", "target": ""}
if stat.S_ISLNK(before.st_mode):
target = _safe_text(os.readlink(path), "link target")
after = path.lstat()
if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
raise RuntimeError("activation link changed while reading")
return {"kind": "symlink", "target": target}
if allow_directory and stat.S_ISDIR(before.st_mode) and before.st_uid == os.getuid():
return {"kind": "directory", "target": ""}
raise RuntimeError(f"activation path is not supported: {path}")
def _linked_release_identity(path: Path) -> dict[str, Any]:
state = _link_state(path, allow_directory=False)
if state["kind"] != "symlink":
return {}
target = Path(state["target"])
if not target.is_absolute():
target = path.parent / target
target = _absolute(target)
try:
return _release_identity(target, target)
except FileNotFoundError:
# A pre-existing dangling previous link remains restorable as an exact
# link value, but it carries no directory-identity claim.
return {}
def _validate_root(root: Path) -> None:
info = root.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) & 0o022
):
raise PermissionError("activation transaction root is unsafe")
def _validate_releases_root(root: Path) -> Path:
releases = root / "releases"
info = releases.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) & 0o022
):
raise PermissionError("activation releases root is unsafe")
return releases
def _validate_guard_metadata(guard: Any) -> None:
if guard == {}:
return
if (
not isinstance(guard, dict)
or set(guard) != {"id", "device", "inode"}
or not isinstance(guard.get("id"), str)
or re.fullmatch(r"cold-handoff-[0-9a-f]{24}", guard["id"]) is None
or isinstance(guard.get("device"), bool)
or not isinstance(guard.get("device"), int)
or guard["device"] <= 0
or isinstance(guard.get("inode"), bool)
or not isinstance(guard.get("inode"), int)
or guard["inode"] <= 0
):
raise RuntimeError("activation transaction guard state is invalid")
def _validate_hub_metadata(hub: Any, *, intent: str) -> None:
if hub == {}:
return
if not isinstance(hub, dict) or set(hub) != {
"kind",
"data_dir",
"hub_id",
"host_identity",
"operation_id",
"snapshot",
"fence_device",
"fence_inode",
}:
raise RuntimeError("activation transaction Hub state is invalid")
data_dir_raw = hub.get("data_dir")
snapshot_raw = hub.get("snapshot")
if not isinstance(data_dir_raw, str) or not isinstance(snapshot_raw, str):
raise RuntimeError("activation transaction Hub state is invalid")
data_dir = _absolute(data_dir_raw)
snapshot = _absolute(snapshot_raw)
if (
data_dir_raw != str(data_dir)
or snapshot_raw != str(snapshot)
or snapshot.parent != data_dir / "maintenance-backups"
or re.fullmatch(r"snapshot_[A-Za-z0-9_]+", snapshot.name) is None
or hub.get("kind") not in {"server-update", "host-reactivation"}
or not isinstance(hub.get("hub_id"), str)
or re.fullmatch(r"[A-Za-z0-9_.:-]{8,240}", hub["hub_id"]) is None
or not isinstance(hub.get("host_identity"), str)
or re.fullmatch(
r"[A-Za-z0-9][A-Za-z0-9._:@/-]{7,239}",
hub["host_identity"],
)
is None
or not isinstance(hub.get("operation_id"), str)
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", hub["operation_id"])
is None
or isinstance(hub.get("fence_device"), bool)
or not isinstance(hub.get("fence_device"), int)
or hub["fence_device"] <= 0
or isinstance(hub.get("fence_inode"), bool)
or not isinstance(hub.get("fence_inode"), int)
or hub["fence_inode"] <= 0
):
raise RuntimeError("activation transaction Hub state is invalid")
if (
(hub["kind"] == "server-update" and intent != "server-update")
or (
hub["kind"] == "host-reactivation"
and intent not in {"host-reactivation", "failed-host-repair"}
)
):
raise RuntimeError("activation transaction Hub intent changed")
def _manifest_path(root: Path) -> Path:
return root / ".activation-transaction" / "manifest.json"
def _cleanup_manifest_temps(directory: Path) -> None:
removed = False
for entry in list(os.scandir(directory)):
if MANIFEST_TEMP_RE.fullmatch(entry.name) is None:
continue
path = directory / entry.name
info = path.lstat()
if (
not stat.S_ISREG(info.st_mode)
or info.st_uid != os.getuid()
or info.st_nlink != 1
or stat.S_IMODE(info.st_mode) != 0o600
):
raise PermissionError("activation transaction staging file is unsafe")
path.unlink()
removed = True
if removed:
_fsync_directory(directory)
def _remove_private_transaction_directory(directory: Path) -> None:
info = directory.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) != 0o700
):
raise PermissionError("retired activation transaction is unsafe")
for child in list(os.scandir(directory)):
child_path = directory / child.name
child_info = child_path.lstat()
if child.name == "candidate.retired":
if (
not stat.S_ISDIR(child_info.st_mode)
or child_info.st_uid != os.getuid()
or not shutil.rmtree.avoids_symlink_attacks
):
raise PermissionError("retired activation candidate is unsafe")
shutil.rmtree(child_path)
continue
if (
child.name
not in {"manifest.json", "env.backup", "service.backup"}
and MANIFEST_TEMP_RE.fullmatch(child.name) is None
) or (
not stat.S_ISREG(child_info.st_mode)
or child_info.st_uid != os.getuid()
or child_info.st_nlink != 1
or stat.S_IMODE(child_info.st_mode) != 0o600
):
raise PermissionError("retired activation transaction is unsafe")
child_path.unlink()
directory.rmdir()
def _cleanup_retired_transactions(root: Path) -> None:
removed = False
for entry in list(os.scandir(root)):
if (
GC_DIRECTORY_RE.fullmatch(entry.name) is None
and BEGIN_DIRECTORY_RE.fullmatch(entry.name) is None
):
continue
directory = root / entry.name
_remove_private_transaction_directory(directory)
removed = True
if removed:
_fsync_directory(root)
def _canonical(value: dict[str, Any]) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n"
def _read_manifest(
root: Path,
*,
directory: Path | None = None,
) -> tuple[Path, dict[str, Any]]:
_validate_root(root)
releases = _validate_releases_root(root)
directory = directory or (root / ".activation-transaction")
info = directory.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) != 0o700
):
raise PermissionError("activation transaction directory is unsafe")
_cleanup_manifest_temps(directory)
try:
value = json.loads(_read_private(directory / "manifest.json"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimeError("activation transaction manifest is invalid") from exc
expected = {
"format",
"transaction_id",
"release_version",
"release_dir",
"candidate_release",
"old_target",
"old_release",
"current",
"previous",
"previous_release",
"env_path",
"env",
"service_path",
"service",
"service_state",
"service_enabled",
"legacy_service_state",
"legacy_service_enabled",
"prior_port",
"prior_bind_address",
"intent",
"client_binding",
"desired_env",
"desired_service",
"phase",
"rollback_from",
"hub",
"guard",
"authority_pending",
"observed_env_sha256",
"observed_service_sha256",
}
if not isinstance(value, dict) or set(value) != expected or value.get("format") != FORMAT:
raise RuntimeError("activation transaction manifest is invalid")
if value.get("phase") not in PHASES:
raise RuntimeError("activation transaction phase is invalid")
rollback_from = value.get("rollback_from")
if rollback_from is not None and (
rollback_from not in PHASES
or rollback_from
in {
"rolling-back",
"rolled-back",
"rollback-healthy",
"committing",
"committed",
}
or value["phase"]
not in {"rolling-back", "rolled-back", "rollback-healthy"}
):
raise RuntimeError("activation transaction rollback origin is invalid")
if value["phase"] in {"rolling-back", "rolled-back", "rollback-healthy"} and rollback_from is None:
raise RuntimeError("activation transaction rollback origin is missing")
transaction_id = _safe_text(value.get("transaction_id"), "id", maximum=64)
if re.fullmatch(r"activation-[0-9a-f]{24}", transaction_id) is None:
raise RuntimeError("activation transaction id is invalid")
_safe_text(value.get("release_version"), "version", maximum=128)
for field in ("release_dir", "env_path", "service_path"):
_safe_text(value.get(field), field)
if not isinstance(value.get("old_target"), str) or any(
character in value["old_target"] for character in "\x00\r\n"
):
raise RuntimeError("activation transaction old target is invalid")
def valid_release(release: Any) -> bool:
return (
isinstance(release, dict)
and set(release)
== {"source", "target", "device", "inode", "version_sha256"}
and isinstance(release.get("source"), str)
and isinstance(release.get("target"), str)
and release["source"].startswith("/")
and release["target"].startswith("/")
and isinstance(release.get("device"), int)
and release["device"] >= 0
and isinstance(release.get("inode"), int)
and release["inode"] > 0
and isinstance(release.get("version_sha256"), str)
and re.fullmatch(r"[0-9a-f]{64}", release["version_sha256"])
is not None
)
old_release = value.get("old_release")
if old_release != {} and not valid_release(old_release):
raise RuntimeError("activation transaction rollback release is invalid")
previous_release = value.get("previous_release")
if previous_release != {} and not valid_release(previous_release):
raise RuntimeError("activation transaction previous release is invalid")
for field in ("current", "previous"):
item = value.get(field)
if (
not isinstance(item, dict)
or set(item) != {"kind", "target"}
or item.get("kind") not in {"missing", "symlink", "directory"}
or not isinstance(item.get("target"), str)
or "\n" in item["target"]
or "\r" in item["target"]
or "\x00" in item["target"]
or (field == "previous" and item["kind"] == "directory")
):
raise RuntimeError("activation transaction link state is invalid")
for field in ("env", "service"):
item = value.get(field)
if (
not isinstance(item, dict)
or set(item) != {"existed", "backup", "sha256", "mode"}
or not isinstance(item.get("existed"), bool)
or item.get("backup") not in {"env.backup", "service.backup"}
or (
item["existed"]
and (
not isinstance(item.get("sha256"), str)
or re.fullmatch(r"[0-9a-f]{64}", item["sha256"])
is None
)
)
or (
item["existed"]
and (
not isinstance(item.get("mode"), int)
or item["mode"] not in {0o600, 0o640, 0o644}
)
)
or (
not item["existed"]
and (item.get("sha256") is not None or item.get("mode") is not None)
)
):
raise RuntimeError("activation transaction backup metadata is invalid")
if item["existed"] and _sha256(directory / item["backup"]) != item["sha256"]:
raise RuntimeError("activation transaction backup changed")
if value.get("service_state") not in {"absent", "stopped", "running"}:
raise RuntimeError("activation transaction service state is invalid")
if not isinstance(value.get("service_enabled"), bool):
raise RuntimeError("activation transaction service enablement is invalid")
if value.get("legacy_service_state") not in {"absent", "stopped", "running"}:
raise RuntimeError("activation transaction legacy service state is invalid")
if not isinstance(value.get("legacy_service_enabled"), bool):
raise RuntimeError("activation transaction legacy enablement is invalid")
prior_port = value.get("prior_port")
if not isinstance(prior_port, int) or not 1 <= prior_port <= 65535:
raise RuntimeError("activation transaction prior port is invalid")
prior_bind = _safe_text(
value.get("prior_bind_address"), "prior bind", maximum=64
)
if prior_bind != "localhost":
try:
parsed_bind = ipaddress.ip_address(prior_bind)
except ValueError as exc:
raise RuntimeError(
"activation transaction prior bind is invalid"
) from exc
if str(parsed_bind) != prior_bind:
raise RuntimeError("activation transaction prior bind is noncanonical")
if value.get("intent") not in {
"ordinary",
"server-update",
"host-reactivation",
"failed-host-repair",
}:
raise RuntimeError("activation transaction intent is invalid")
client_binding = value.get("client_binding")
if (
not isinstance(client_binding, str)
or len(client_binding) > 64 * 1024
or any(character in client_binding for character in "\x00\r\n")
):
raise RuntimeError("activation transaction client binding is invalid")
for field in ("desired_env", "desired_service"):
item = value.get(field)
kind = field.removeprefix("desired_")
destination = _absolute(
value["env_path" if kind == "env" else "service_path"]
)
expected_source, _expected_secured = _config_staging_paths(
destination,
value["transaction_id"],
kind,
)
if item is not None and (
not isinstance(item, dict)
or set(item) != {"sha256", "mode", "source", "device", "inode"}
or not isinstance(item.get("sha256"), str)
or re.fullmatch(r"[0-9a-f]{64}", item["sha256"]) is None
or not isinstance(item.get("mode"), int)
or item["mode"] not in (
{0o600} if field == "desired_env" else {0o600, 0o644}
)
or item.get("source") != str(expected_source)
or isinstance(item.get("device"), bool)
or not isinstance(item.get("device"), int)
or item["device"] <= 0
or isinstance(item.get("inode"), bool)
or not isinstance(item.get("inode"), int)
or item["inode"] <= 0
):
raise RuntimeError("activation desired configuration is invalid")
if not isinstance(value.get("authority_pending"), bool):
raise RuntimeError("activation transaction authority state is invalid")
for field in ("observed_env_sha256", "observed_service_sha256"):
items = value.get(field)
if not isinstance(items, list) or any(
item is not None
and (
not isinstance(item, str)
or re.fullmatch(r"[0-9a-f]{64}", item) is None
)
for item in items
):
raise RuntimeError("activation transaction configuration history is invalid")
candidate_release = value.get("candidate_release")
if not valid_release(candidate_release):
raise RuntimeError("activation transaction candidate release is invalid")
release_version = value["release_version"]
release_dir = _absolute(value["release_dir"])
candidate_source = _absolute(candidate_release["source"])
candidate_target = _absolute(candidate_release["target"])
if (
value["release_dir"] != str(release_dir)
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,127}", release_version)
is None
or release_dir != releases / release_version
or candidate_release["source"] != str(candidate_source)
or candidate_release["target"] != str(candidate_target)
or candidate_source.parent != releases
or candidate_source == candidate_target
or candidate_target != release_dir
):
raise RuntimeError("activation transaction release path is invalid")
if old_release:
old_source = _absolute(old_release["source"])
old_target = _absolute(old_release["target"])
if (
old_release["source"] != str(old_source)
or old_release["target"] != str(old_target)
or old_target.parent != releases
or (old_source.parent != releases and old_source != root / "current")
or value["old_target"] != str(old_target)
):
raise RuntimeError("activation transaction rollback release path is invalid")
elif value["old_target"]:
raise RuntimeError("activation transaction rollback release path is invalid")
if previous_release:
previous_source = _absolute(previous_release["source"])
previous_target = _absolute(previous_release["target"])
if (
previous_release["source"] != str(previous_source)
or previous_release["target"] != str(previous_target)
or previous_source.parent != releases
or previous_target.parent != releases
):
raise RuntimeError("activation transaction previous release path is invalid")
hub = value.get("hub")
_validate_hub_metadata(hub, intent=value["intent"])
guard = value.get("guard")
_validate_guard_metadata(guard)
return directory, value
def _current_config_metadata(path: Path) -> dict[str, Any] | None:
try:
info = path.lstat()
except FileNotFoundError:
return None
payload = _read_owned_regular(path, maximum=MAX_CONFIG_BYTES)
linked_info = path.lstat()
if (info.st_dev, info.st_ino) != (linked_info.st_dev, linked_info.st_ino):
raise RuntimeError("activation configuration changed while verifying it")
return {
"sha256": hashlib.sha256(payload).hexdigest(),
"mode": stat.S_IMODE(info.st_mode),
}
def _current_config_digest(path: Path) -> str | None:
metadata = _current_config_metadata(path)
return None if metadata is None else str(metadata["sha256"])
def _config_staging_paths(
destination: Path,
transaction_id: str,
kind: str,
) -> tuple[Path, Path]:
prefix = f".{destination.name}.activation-{transaction_id}-{kind}"
return (
destination.parent / f"{prefix}.source",
destination.parent / f"{prefix}.publication",
)
def _desired_content_metadata(desired: dict[str, Any]) -> dict[str, Any]:
return {"sha256": desired["sha256"], "mode": desired["mode"]}
def _config_identity_matches(info: os.stat_result, desired: dict[str, Any]) -> bool:
return (info.st_dev, info.st_ino) == (desired["device"], desired["inode"])
def _config_path_matches_desired(path: Path, desired: dict[str, Any]) -> bool:
try:
before = path.lstat()
except FileNotFoundError:
return False
if not _config_identity_matches(before, desired):
return False
if _current_config_metadata(path) != _desired_content_metadata(desired):
return False
after = path.lstat()
return (before.st_dev, before.st_ino) == (after.st_dev, after.st_ino)
def _open_config_staging_source(path: Path) -> tuple[int, os.stat_result, bytes]:
flags = os.O_RDWR | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_uid != os.getuid()
or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) & 0o022
or not 0 <= before.st_size <= MAX_CONFIG_BYTES
):
raise PermissionError("activation configuration staging file is unsafe")
chunks: list[bytes] = []
remaining = MAX_CONFIG_BYTES + 1
while remaining:
chunk = os.read(descriptor, min(remaining, 1024 * 1024))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
payload = b"".join(chunks)
after = os.fstat(descriptor)
linked = path.lstat()
if (
len(payload) > MAX_CONFIG_BYTES
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
or (before.st_dev, before.st_ino) != (linked.st_dev, linked.st_ino)
):
raise RuntimeError("activation configuration staging file changed")
return descriptor, before, payload
except BaseException:
os.close(descriptor)
raise
def _staged_config_metadata(path: Path) -> tuple[dict[str, Any], os.stat_result]:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_uid != os.getuid()
or before.st_nlink not in {1, 2}
or stat.S_IMODE(before.st_mode) & 0o022
or not 0 <= before.st_size <= MAX_CONFIG_BYTES
):
raise PermissionError("activation configuration publication is unsafe")
chunks: list[bytes] = []
remaining = MAX_CONFIG_BYTES + 1
while remaining:
chunk = os.read(descriptor, min(remaining, 1024 * 1024))
if not chunk:
break
chunks.append(chunk)
remaining -= len(chunk)
payload = b"".join(chunks)
after = os.fstat(descriptor)
linked = path.lstat()
if (
len(payload) > MAX_CONFIG_BYTES
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
or (before.st_dev, before.st_ino) != (linked.st_dev, linked.st_ino)
):
raise RuntimeError("activation configuration publication changed")
return {
"sha256": hashlib.sha256(payload).hexdigest(),
"mode": stat.S_IMODE(before.st_mode),
}, before
finally:
os.close(descriptor)
def _unlink_if_same_file(path: Path, identity: os.stat_result) -> None:
try:
linked = path.lstat()
except FileNotFoundError:
return
if (linked.st_dev, linked.st_ino) == (identity.st_dev, identity.st_ino):
path.unlink()
def _fsync_staging_directory(path: Path) -> None:
"""Persist the private staging link before the final publish rename."""
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_DIRECTORY", 0)
descriptor = os.open(path, flags)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def _publish_staged_config(
*,
staged: Path,
source: Path,
destination: Path,
desired: dict[str, Any],
) -> None:
metadata, identity = _staged_config_metadata(staged)
if (
metadata != _desired_content_metadata(desired)
or not _config_identity_matches(identity, desired)
):
raise RuntimeError("activation staged configuration does not match its journal")
_unlink_if_same_file(source, identity)
_fsync_staging_directory(source.parent)
if staged.parent != source.parent:
_fsync_staging_directory(staged.parent)
try:
os.replace(staged, destination)
except BaseException:
# A normal rename failure leaves the exact staged inode available for
# retry. Restore the caller's source name when possible so an in-process
# retry retains the original interface; a process crash instead leaves
# the fsynced deterministic staging name for recovery.
try:
staged.lstat()
source.lstat()
except FileNotFoundError:
try:
staged.lstat()
except FileNotFoundError:
pass
else:
os.rename(staged, source)
_fsync_staging_directory(source.parent)
raise
_fsync_directory(destination.parent)
if not _config_path_matches_desired(destination, desired):
raise RuntimeError("activation configuration changed after publication")
def _cleanup_config_staging(
directory: Path,
value: dict[str, Any],
) -> None:
"""Remove only schema-closed staging inodes owned by this transaction."""
for kind in ("env", "service"):
desired = value[f"desired_{kind}"]
destination = _absolute(
value["env_path" if kind == "env" else "service_path"]
)
source, secured = _config_staging_paths(
destination,
value["transaction_id"],
kind,
)
try:
secured_metadata, secured_info = _staged_config_metadata(secured)
except FileNotFoundError:
secured_metadata = None
secured_info = None
try:
source_metadata, source_info = _staged_config_metadata(source)
except FileNotFoundError:
pass
else:
if desired is None:
if (
source_info.st_nlink != 1
or stat.S_IMODE(source_info.st_mode) != 0o600
):
raise PermissionError(
"activation unjournaled configuration staging is unsafe"
)
elif (
source_metadata != _desired_content_metadata(desired)
or not _config_identity_matches(source_info, desired)
):
# A later rewrite may have populated this private reserved
# source and crashed before its new intent was journaled. No
# matching publication inode can exist before that
# write-ahead update, so an independent nlink-1 draft is safe
# to discard even while the prior exact publication is being
# resumed.
if (
source_info.st_nlink != 1
or stat.S_IMODE(source_info.st_mode) != 0o600
or (
secured_info is not None
and (
desired is None
or secured_metadata
!= _desired_content_metadata(desired)
or not _config_identity_matches(
secured_info,
desired,
)
)
)
):
raise RuntimeError(
"activation configuration staging ownership changed"
)
source.unlink()
_fsync_directory(source.parent)
if secured_info is None:
continue
if desired is None or (
secured_metadata != _desired_content_metadata(desired)
or not _config_identity_matches(secured_info, desired)
):
raise RuntimeError(
"activation configuration publication ownership changed"
)
secured.unlink()
_fsync_directory(destination.parent)
def begin(args: argparse.Namespace) -> None:
root = _absolute(args.root)
_validate_root(root)
releases = _validate_releases_root(root)
_cleanup_retired_transactions(root)
current = _absolute(args.current)
previous = _absolute(args.previous)
env_path = _absolute(args.env)
service_path = _absolute(args.service)
release_dir = _absolute(args.release_dir)
candidate_source = _absolute(args.candidate_source)
old_source = _absolute(args.old_source) if args.old_source else None
old_target = _absolute(args.old_target) if args.old_target else None
release_version = _safe_text(args.release_version, "version", maximum=128)
if (
current != root / "current"
or previous != root / "previous"
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,127}", release_version)
is None
or release_dir != releases / release_version
or candidate_source.parent != releases
or candidate_source == release_dir
or (
old_target is not None and old_target.parent != releases
)
or (
old_source is not None
and old_source.parent != releases
and old_source != current
)
or ((old_source is None) != (old_target is None))
):
raise RuntimeError("activation transaction paths are invalid")
final = root / ".activation-transaction"
try:
final.lstat()
except FileNotFoundError:
pass
else:
raise RuntimeError("an activation transaction is already pending")
temporary = root / f".activation-transaction.{secrets.token_hex(12)}.tmp"
os.mkdir(temporary, 0o700)
os.chmod(temporary, 0o700)
try:
env = _configuration(
env_path,
temporary / "env.backup",
allowed_modes={0o600},
)
service = _configuration(
service_path,
temporary / "service.backup",
allowed_modes={0o600, 0o640, 0o644},
)
current_state = _link_state(current, allow_directory=True)
old_release = _release_identity(old_source, old_target)
if current_state["kind"] == "missing":
if old_release:
raise RuntimeError(
"activation rollback identity exists without a current release"
)
elif not old_release:
raise RuntimeError("activation current release has no rollback identity")
elif current_state["kind"] == "directory":
if old_source != current or not _release_matches(current, old_release):
raise RuntimeError("activation legacy current rollback identity changed")
else:
current_target = Path(current_state["target"])
if not current_target.is_absolute():
current_target = current.parent / current_target
if not _release_matches(_absolute(current_target), old_release):
raise RuntimeError("activation current rollback identity changed")
candidate_release = _release_identity(candidate_source, release_dir)
version_payload = _read_owned_regular(
candidate_source / "VERSION",
maximum=4096,
)
try:
candidate_version = version_payload.decode("utf-8").strip()
except UnicodeDecodeError as exc:
raise RuntimeError("activation candidate version is invalid") from exc
if candidate_version != release_version:
raise RuntimeError("activation candidate release version changed")
value: dict[str, Any] = {
"format": FORMAT,
"transaction_id": f"activation-{secrets.token_hex(12)}",
"release_version": release_version,
"release_dir": str(release_dir),
"candidate_release": candidate_release,
"old_target": str(old_target) if old_target is not None else "",
"old_release": old_release,
"current": current_state,
"previous": _link_state(previous, allow_directory=False),
"previous_release": _linked_release_identity(previous),
"env_path": str(env_path),
"env": env,
"service_path": str(service_path),
"service": service,
"service_state": args.service_state,
"service_enabled": args.service_enabled == "true",
"legacy_service_state": args.legacy_service_state,
"legacy_service_enabled": args.legacy_service_enabled == "true",
"prior_port": args.prior_port,
"prior_bind_address": args.prior_bind_address,
"intent": args.intent,
"client_binding": args.client_binding,
"desired_env": None,
"desired_service": None,
"phase": "prepared",
"rollback_from": None,
"hub": {},
"guard": {},
"authority_pending": False,
"observed_env_sha256": [env["sha256"]],
"observed_service_sha256": [service["sha256"]],
}
_write_new_private(temporary / "manifest.json", _canonical(value))
_fsync_directory(temporary)
_verified_directory, verified_value = _read_manifest(
root,
directory=temporary,
)
_validate_invocation(
args,
verified_value,
transaction_directory=temporary,
)
os.rename(temporary, final)
_fsync_directory(root)
sys.stdout.write(str(value["transaction_id"]) + "\n")
except BaseException:
shutil.rmtree(temporary, ignore_errors=True)
raise
def _validate_invocation(
args: argparse.Namespace,
value: dict[str, Any],
*,
allow_missing_candidate: bool = False,
transaction_directory: Path | None = None,
) -> None:
root = _absolute(args.root)
if (
_absolute(value["env_path"]) != _absolute(args.env)
or _absolute(value["service_path"]) != _absolute(args.service)
or _absolute(args.current) != root / "current"
or _absolute(args.previous) != root / "previous"
):
raise RuntimeError("pending activation belongs to a different layout")
if getattr(args, "transaction_id", None) not in {None, ""} and (
args.transaction_id != value["transaction_id"]
):
raise RuntimeError("activation transaction ownership changed")
if getattr(args, "release_version", None) not in {None, ""} and (
value["release_version"] != args.release_version
):
raise RuntimeError("activation transaction release version changed")
if getattr(args, "release_dir", None) not in {None, ""} and (
_absolute(value["release_dir"]) != _absolute(args.release_dir)
):
raise RuntimeError("activation transaction release path changed")
candidate_retired = root / "releases" / (
".activation-candidate-retired-" + str(value["transaction_id"])
)
candidate_parked = (
transaction_directory or (root / ".activation-transaction")
) / "candidate.retired"
old_release = value["old_release"]
if old_release and _locate_release(old_release) is None:
raise RuntimeError("activation rollback release identity changed")
candidate_location = _locate_release(
value["candidate_release"],
extras=(candidate_retired, candidate_parked),
)
if candidate_location is None and not allow_missing_candidate:
raise RuntimeError("activation candidate release identity changed")
def link_points_to(path: Path, release: dict[str, Any]) -> bool:
try:
info = path.lstat()
except FileNotFoundError:
return False
if not stat.S_ISLNK(info.st_mode):
return False
raw_target = os.readlink(path)
target = Path(raw_target)
if not target.is_absolute():
target = path.parent / target
return _release_matches(_absolute(target), release)
current_path = _absolute(args.current)
previous_path = _absolute(args.previous)
current_state = _link_state(current_path, allow_directory=True)
original_current = value["current"]
candidate_current = {"kind": "symlink", "target": value["release_dir"]}
phase = value["phase"]
candidate_link_target_missing = False
if current_state == candidate_current:
raw_target = Path(current_state["target"])
if not raw_target.is_absolute():
raw_target = current_path.parent / raw_target
try:
raw_target.lstat()
except FileNotFoundError:
candidate_link_target_missing = True
points_to_candidate = (
current_state == candidate_current
and link_points_to(current_path, value["candidate_release"])
)
points_to_old = (
bool(old_release)
and current_state == original_current
and (
(
original_current["kind"] == "symlink"
and link_points_to(current_path, old_release)
)
or (
original_current["kind"] == "directory"
and _release_matches(current_path, old_release)
)
)
)
original_current_valid = current_state == original_current and (
original_current["kind"] == "missing" or points_to_old
)
candidate_current_valid = points_to_candidate or (
allow_missing_candidate
and current_state == candidate_current
and candidate_link_target_missing
)
current_missing = current_state == {"kind": "missing", "target": ""}
if phase in {"prepared", "guarded"}:
if not original_current_valid:
raise RuntimeError("current release changed before link takeover")
elif phase in {"linking", "rolling-back"}:
if not (
original_current_valid
or candidate_current_valid
or current_missing
or (
current_state == candidate_current
and candidate_link_target_missing
)
):
raise RuntimeError("current release target changed during activation")
elif phase in {"rolled-back", "rollback-healthy"}:
if not original_current_valid:
raise RuntimeError("current release was not restored")
elif not candidate_current_valid:
raise RuntimeError("current release does not resolve to the candidate")
previous_state = _link_state(previous_path, allow_directory=False)
candidate_previous = (
{"kind": "symlink", "target": value["old_target"]}
if value["old_target"]
else None
)
previous_release = value["previous_release"]
original_previous_valid = previous_state == value["previous"] and (
not previous_release or link_points_to(previous_path, previous_release)
)
candidate_previous_valid = (
candidate_previous is None
and original_previous_valid
) or (
candidate_previous is not None
and previous_state == candidate_previous
and bool(old_release)
and link_points_to(previous_path, old_release)
)
dangling_candidate_previous = (
phase == "rolling-back"
and candidate_previous is not None
and previous_state == candidate_previous
and bool(old_release)
and _release_matches(_absolute(old_release["source"]), old_release)
)
previous_missing = previous_state == {"kind": "missing", "target": ""}
if phase in {"prepared", "guarded"}:
if not original_previous_valid:
raise RuntimeError("previous release changed before link takeover")
elif phase in {"linking", "rolling-back"}:
if not (
original_previous_valid
or candidate_previous_valid
or dangling_candidate_previous
or previous_missing
):
raise RuntimeError("previous release changed during activation")
elif phase in {"rolled-back", "rollback-healthy"}:
if not original_previous_valid:
raise RuntimeError("previous release link was not restored")
elif not candidate_previous_valid:
raise RuntimeError("previous release does not resolve to the rollback release")
env_metadata = _current_config_metadata(_absolute(args.env))
service_metadata = _current_config_metadata(_absolute(args.service))
def original_configuration(field: str) -> dict[str, Any] | None:
item = value[field]
if not item["existed"]:
return None
return {"sha256": item["sha256"], "mode": item["mode"]}
def authorized_configuration(
metadata: dict[str, Any] | None,
history_field: str,
*,
modes: set[int],
) -> bool:
if metadata is None:
return None in value[history_field]
return (
metadata["sha256"] in value[history_field]
and metadata["mode"] in modes
)
original_env = original_configuration("env")
original_service = original_configuration("service")
desired_env = value["desired_env"]
desired_service = value["desired_service"]
if phase in {"prepared", "guarded", "linking", "stopping", "stopped", "fencing"}:
configs_valid = (
env_metadata == original_env and service_metadata == original_service
)
elif phase in {"linked", "fenced", "authorizing", "authority", "rolling-back"}:
configs_valid = authorized_configuration(
env_metadata,
"observed_env_sha256",
modes={0o600},
) and authorized_configuration(
service_metadata,
"observed_service_sha256",
modes={0o600, 0o640, 0o644},
)
elif phase in {"rolled-back", "rollback-healthy"}:
configs_valid = (
env_metadata == original_env and service_metadata == original_service
)
else:
configs_valid = (
desired_env is not None
and desired_service is not None
and _config_path_matches_desired(_absolute(args.env), desired_env)
and _config_path_matches_desired(
_absolute(args.service), desired_service
)
)
if not configs_valid:
raise RuntimeError("activation configuration does not match its phase")
def load(args: argparse.Namespace) -> None:
root = _absolute(args.root)
_directory, value = _read_manifest(root)
_validate_invocation(args, value, allow_missing_candidate=True)
hub = value["hub"]
guard = value["guard"]
candidate_retired = root / "releases" / (
".activation-candidate-retired-" + str(value["transaction_id"])
)
candidate_parked = root / ".activation-transaction" / "candidate.retired"
candidate_path = _locate_release(
value["candidate_release"],
extras=(candidate_retired, candidate_parked),
)
fields = [
value["transaction_id"],
value["release_version"],
value["release_dir"],
value["phase"],
str(value.get("rollback_from") or ""),
value["old_target"],
str(value["old_release"].get("source", "")),
value["current"]["kind"],
value["current"]["target"],
value["previous"]["kind"],
value["previous"]["target"],
"true" if value["env"]["existed"] else "false",
str(root / ".activation-transaction" / value["env"]["backup"]),
"true" if value["service"]["existed"] else "false",
str(root / ".activation-transaction" / value["service"]["backup"]),
value["service_state"],
"true" if value["service_enabled"] else "false",
value["legacy_service_state"],
"true" if value["legacy_service_enabled"] else "false",
value["intent"],
value["client_binding"],
str(hub.get("kind", "")),
str(hub.get("data_dir", "")),
str(hub.get("hub_id", "")),
str(hub.get("host_identity", "")),
str(hub.get("operation_id", "")),
str(hub.get("snapshot", "")),
str(hub.get("fence_device", "")),
str(hub.get("fence_inode", "")),
str(guard.get("id", "")),
str(guard.get("device", "")),
str(guard.get("inode", "")),
"true" if value["authority_pending"] else "false",
(
"true"
if (
value["rollback_from"]
if value["phase"]
in {"rolling-back", "rolled-back", "rollback-healthy"}
else value["phase"]
)
in {
"linking",
"linked",
"stopping",
"stopped",
"fencing",
"fenced",
"authorizing",
"authority",
"candidate-starting",
"candidate-healthy",
"committing",
"committed",
"rolling-back",
}
else "false"
),
str(candidate_path) if candidate_path is not None else "",
str(value["prior_port"]),
value["prior_bind_address"],
"activation-end",
]
for index, field in enumerate(fields):
if "\n" in field or "\r" in field or "\x00" in field:
raise RuntimeError(f"activation transaction output field {index} is invalid")
sys.stdout.write("\n".join(fields) + "\n")
def record(args: argparse.Namespace) -> None:
root = _absolute(args.root)
directory, value = _read_manifest(root)
_validate_invocation(
args,
value,
allow_missing_candidate=args.phase
in {"rolling-back", "rolled-back", "rollback-healthy"},
)
if args.phase not in PHASES:
raise RuntimeError("activation transaction phase is invalid")
original_value = json.loads(json.dumps(value))
previous_phase = value["phase"]
if args.phase != value["phase"]:
if args.phase not in TRANSITIONS[value["phase"]]:
raise RuntimeError("activation transaction phase transition is invalid")
value["phase"] = args.phase
if args.phase == "rolling-back":
value["rollback_from"] = previous_phase
if args.hub_kind:
try:
fence_device = int(args.fence_device or 0)
fence_inode = int(args.fence_inode or 0)
except ValueError as exc:
raise RuntimeError("activation transaction Hub coordinates are invalid") from exc
data_dir = _absolute(args.hub_data_dir)
snapshot = _absolute(args.snapshot)
hub = {
"kind": args.hub_kind,
"data_dir": str(data_dir),
"hub_id": _safe_text(args.hub_id, "Hub id", maximum=240),
"host_identity": _safe_text(args.host_identity, "host identity", maximum=240),
"operation_id": _safe_text(args.operation_id, "operation id", maximum=128),
"snapshot": str(snapshot),
"fence_device": fence_device or 1,
"fence_inode": fence_inode or 1,
}
_validate_hub_metadata(hub, intent=value["intent"])
if args.hub_data_dir != str(data_dir) or args.snapshot != str(snapshot):
raise RuntimeError("activation transaction Hub path is invalid")
for hub_directory in (data_dir, data_dir / "maintenance-backups"):
hub_directory_info = hub_directory.lstat()
if (
not stat.S_ISDIR(hub_directory_info.st_mode)
or hub_directory_info.st_uid != os.getuid()
):
raise PermissionError("activation Team Hub directory is unsafe")
fence_path = data_dir / "maintenance-fence.json"
fence_info = fence_path.lstat()
if (
not stat.S_ISREG(fence_info.st_mode)
or fence_info.st_uid != os.getuid()
or fence_info.st_nlink != 1
or stat.S_IMODE(fence_info.st_mode) != 0o600
):
raise PermissionError("activation Team Hub fence is unsafe")
if fence_device == 0 and fence_inode == 0:
fence_device = fence_info.st_dev
fence_inode = fence_info.st_ino
elif (
fence_device != fence_info.st_dev
or fence_inode != fence_info.st_ino
):
raise RuntimeError("activation Team Hub fence ownership changed")
hub["fence_device"] = fence_device
hub["fence_inode"] = fence_inode
_validate_hub_metadata(hub, intent=value["intent"])
if value["hub"] and value["hub"] != hub:
existing_without_inode = {
key: item
for key, item in value["hub"].items()
if key not in {"fence_device", "fence_inode"}
}
new_without_inode = {
key: item
for key, item in hub.items()
if key not in {"fence_device", "fence_inode"}
}
if not (
previous_phase == "fencing"
and args.phase in {"fenced", "rolling-back"}
and existing_without_inode == new_without_inode
):
raise RuntimeError("activation Team Hub ownership changed")
value["hub"] = hub
elif any(
(
args.hub_data_dir,
args.hub_id,
args.host_identity,
args.operation_id,
args.snapshot,
args.fence_device,
args.fence_inode,
)
):
raise RuntimeError("activation transaction Hub metadata has no kind")
if args.guard_id:
try:
guard = {
"id": _safe_text(args.guard_id, "guard id", maximum=64),
"device": int(args.guard_device),
"inode": int(args.guard_inode),
}
except ValueError as exc:
raise RuntimeError("activation transaction guard coordinates are invalid") from exc
_validate_guard_metadata(guard)
if value["guard"] and value["guard"] != guard:
raise RuntimeError("activation Team Hub guard ownership changed")
value["guard"] = guard
elif args.guard_device or args.guard_inode:
raise RuntimeError("activation transaction guard coordinates have no id")
if args.authority_pending != "unchanged":
value["authority_pending"] = args.authority_pending == "true"
_validate_hub_metadata(value["hub"], intent=value["intent"])
_validate_guard_metadata(value["guard"])
if previous_phase in {"committed", "rollback-healthy"} and value != original_value:
raise RuntimeError("terminal activation transaction metadata is immutable")
_validate_invocation(
args,
value,
allow_missing_candidate=args.phase
in {"rolling-back", "rolled-back", "rollback-healthy"},
)
_replace_private(directory / "manifest.json", _canonical(value))
if args.hub_kind:
sys.stdout.write(
f"{value['hub']['fence_device']}\n{value['hub']['fence_inode']}\n"
)
def replace_config(args: argparse.Namespace) -> None:
"""Write-ahead and atomically publish one installer-generated config."""
root = _absolute(args.root)
directory, value = _read_manifest(root)
_validate_invocation(args, value)
if value["phase"] in {
"committing",
"committed",
"rolling-back",
"rolled-back",
"rollback-healthy",
}:
raise RuntimeError("activation configuration cannot change in this phase")
source = _absolute(args.source)
field = "env_path" if args.kind == "env" else "service_path"
history_field = (
"observed_env_sha256"
if args.kind == "env"
else "observed_service_sha256"
)
destination = _absolute(value[field])
expected_source, secured = _config_staging_paths(
destination,
value["transaction_id"],
args.kind,
)
if source != expected_source:
raise RuntimeError("activation configuration staging path is invalid")
mode = int(args.mode, 8)
expected_modes = {0o600} if args.kind == "env" else {0o600, 0o644}
if mode not in expected_modes:
raise RuntimeError("activation configuration mode is invalid")
desired_field = "desired_env" if args.kind == "env" else "desired_service"
journaled_desired = value[desired_field]
try:
source.lstat()
except FileNotFoundError:
if journaled_desired is None or journaled_desired.get("mode") != mode:
raise RuntimeError("activation configuration staging file is missing")
try:
secured.lstat()
except FileNotFoundError:
if not _config_path_matches_desired(destination, journaled_desired):
raise RuntimeError(
"activation configuration publication is incomplete"
)
_fsync_directory(destination.parent)
return
_publish_staged_config(
staged=secured,
source=source,
destination=destination,
desired=journaled_desired,
)
return
try:
secured.lstat()
except FileNotFoundError:
pass
else:
if journaled_desired is None or journaled_desired.get("mode") != mode:
raise RuntimeError("activation configuration publication path exists")
_publish_staged_config(
staged=secured,
source=source,
destination=destination,
desired=journaled_desired,
)
return
descriptor, source_identity, payload = _open_config_staging_source(source)
digest = hashlib.sha256(payload).hexdigest()
desired = {
"sha256": digest,
"mode": mode,
"source": str(source),
"device": source_identity.st_dev,
"inode": source_identity.st_ino,
}
value[desired_field] = desired
if digest not in value[history_field]:
value[history_field].append(digest)
try:
_replace_private(directory / "manifest.json", _canonical(value))
os.fchmod(descriptor, mode)
os.fsync(descriptor)
os.link(source, secured, follow_symlinks=False)
_fsync_staging_directory(source.parent)
secured_info = secured.lstat()
source_after = os.fstat(descriptor)
if (
(source_identity.st_dev, source_identity.st_ino)
!= (source_after.st_dev, source_after.st_ino)
or (source_identity.st_dev, source_identity.st_ino)
!= (secured_info.st_dev, secured_info.st_ino)
or hashlib.sha256(payload).hexdigest() != digest
):
raise RuntimeError("activation configuration staging identity changed")
finally:
os.close(descriptor)
_publish_staged_config(
staged=secured,
source=source,
destination=destination,
desired=desired,
)
def _replace_symlink(path: Path, target: str) -> None:
temporary = path.parent / f".{path.name}.{secrets.token_hex(12)}.tmp"
try:
os.symlink(target, temporary)
os.replace(temporary, path)
_fsync_directory(path.parent)
finally:
try:
temporary.unlink()
except FileNotFoundError:
pass
def _rename_and_fsync(source: Path, destination: Path) -> None:
"""Rename one entry and durably publish both affected directories."""
os.rename(source, destination)
_fsync_directory(source.parent)
if destination.parent != source.parent:
_fsync_directory(destination.parent)
def activate_files(args: argparse.Namespace) -> None:
"""Idempotently perform the release moves and exact link takeover."""
root = _absolute(args.root)
_directory, value = _read_manifest(root)
_validate_invocation(args, value)
if value["phase"] != "linking":
raise RuntimeError("activation transaction is not linking")
old_release = value["old_release"]
if old_release and old_release["source"] != old_release["target"]:
old_source = _absolute(old_release["source"])
old_target = _absolute(old_release["target"])
if _release_matches(old_source, old_release):
try:
old_target.lstat()
except FileNotFoundError:
_rename_and_fsync(old_source, old_target)
else:
raise RuntimeError("activation rollback quarantine already exists")
elif not _release_matches(old_target, old_release):
raise RuntimeError("activation rollback release is missing")
candidate = value["candidate_release"]
candidate_source = _absolute(candidate["source"])
candidate_target = _absolute(candidate["target"])
if not _release_matches(candidate_target, candidate):
if not _release_matches(candidate_source, candidate):
raise RuntimeError("activation candidate release is missing")
try:
candidate_target.lstat()
except FileNotFoundError:
_rename_and_fsync(candidate_source, candidate_target)
else:
raise RuntimeError("activation candidate destination changed")
old_target = value["old_target"]
if old_target:
_replace_symlink(_absolute(args.previous), old_target)
_replace_symlink(_absolute(args.current), value["release_dir"])
if not _release_matches(candidate_target, candidate):
raise RuntimeError("activation candidate release changed after link takeover")
def _restore_configuration(
destination: Path,
directory: Path,
metadata: dict[str, Any],
) -> None:
if not metadata["existed"]:
try:
info = destination.lstat()
except FileNotFoundError:
return
if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid():
raise PermissionError("activation configuration destination is unsafe")
destination.unlink()
_fsync_directory(destination.parent)
return
value = _read_private(
directory / metadata["backup"],
maximum=MAX_CONFIG_BYTES,
)
temporary = destination.parent / f".{destination.name}.{secrets.token_hex(12)}.tmp"
descriptor: int | None = None
try:
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
flags |= getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(temporary, flags, int(metadata["mode"]))
os.fchmod(descriptor, int(metadata["mode"]))
written = 0
while written < len(value):
written += os.write(descriptor, value[written:])
os.fsync(descriptor)
os.close(descriptor)
descriptor = None
os.replace(temporary, destination)
_fsync_directory(destination.parent)
finally:
if descriptor is not None:
os.close(descriptor)
try:
temporary.unlink()
except FileNotFoundError:
pass
def restore_files(args: argparse.Namespace) -> None:
"""Restore exact pre-activation links/configuration under the journal."""
root = _absolute(args.root)
directory, value = _read_manifest(root)
_validate_invocation(args, value, allow_missing_candidate=True)
if value["phase"] != "rolling-back":
raise RuntimeError("activation transaction is not rolling back")
current = _absolute(args.current)
previous = _absolute(args.previous)
old_release = value["old_release"]
if old_release and old_release["source"] != old_release["target"]:
old_source = _absolute(old_release["source"])
old_target = _absolute(old_release["target"])
if not _release_matches(old_source, old_release):
if not _release_matches(old_target, old_release):
raise RuntimeError("activation rollback release is missing")
try:
old_source_info = old_source.lstat()
except FileNotFoundError:
pass
else:
candidate = value["candidate_release"]
if stat.S_ISLNK(old_source_info.st_mode) and old_source == current:
old_source.unlink()
_fsync_directory(old_source.parent)
else:
if not _release_matches(old_source, candidate):
raise RuntimeError("activation candidate release changed")
retired_candidate = directory / "candidate.retired"
if not _release_matches(retired_candidate, candidate):
try:
retired_candidate.lstat()
except FileNotFoundError:
_rename_and_fsync(old_source, retired_candidate)
else:
raise RuntimeError(
"activation candidate retirement path exists"
)
_rename_and_fsync(old_target, old_source)
original_current = value["current"]
if original_current["kind"] == "symlink":
_replace_symlink(current, original_current["target"])
elif original_current["kind"] == "missing":
try:
current_info = current.lstat()
except FileNotFoundError:
pass
else:
if not stat.S_ISLNK(current_info.st_mode):
raise RuntimeError("current release path changed during rollback")
current.unlink()
_fsync_directory(current.parent)
else:
old_target = _absolute(value["old_target"])
try:
current_info = current.lstat()
except FileNotFoundError:
pass
else:
if stat.S_ISLNK(current_info.st_mode):
current.unlink()
_fsync_directory(current.parent)
elif stat.S_ISDIR(current_info.st_mode):
# The legacy directory was never moved.
old_target = current
else:
raise RuntimeError("legacy current path changed during rollback")
if old_target != current:
_rename_and_fsync(old_target, current)
original_previous = value["previous"]
if original_previous["kind"] == "symlink":
_replace_symlink(previous, original_previous["target"])
else:
try:
previous_info = previous.lstat()
except FileNotFoundError:
pass
else:
if not stat.S_ISLNK(previous_info.st_mode):
raise RuntimeError("previous release path changed during rollback")
previous.unlink()
_fsync_directory(previous.parent)
_cleanup_config_staging(directory, value)
_restore_configuration(
_absolute(value["env_path"]),
directory,
value["env"],
)
_restore_configuration(
_absolute(value["service_path"]),
directory,
value["service"],
)
def finish(args: argparse.Namespace) -> None:
root = _absolute(args.root)
_validate_root(root)
_validate_releases_root(root)
if re.fullmatch(r"activation-[0-9a-f]{24}", args.transaction_id) is None:
raise RuntimeError("activation transaction id is invalid")
active = root / ".activation-transaction"
retired = root / (
".activation-transaction-gc-" + str(args.transaction_id)
)
try:
active.lstat()
except FileNotFoundError:
try:
retired.lstat()
except FileNotFoundError:
# A prior exact finish may have retired and removed the directory
# before its caller observed success. The unguessable transaction
# id is the idempotency key; absence is already the terminal state.
_fsync_directory(root)
return
try:
directory, value = _read_manifest(root, directory=retired)
except FileNotFoundError:
# Retired-directory cleanup can itself be interrupted after the
# manifest unlink. Only this exact transaction-id directory is
# eligible for bounded, schema-closed cleanup.
_remove_private_transaction_directory(retired)
_fsync_directory(root)
return
else:
# Any missing or malformed control in the live transaction is an
# unsettled activation, never evidence that finish already succeeded.
directory, value = _read_manifest(root, directory=active)
_validate_invocation(
args,
value,
allow_missing_candidate=value["phase"] == "rollback-healthy",
transaction_directory=directory,
)
if value["phase"] not in {"committed", "rollback-healthy"}:
raise RuntimeError("activation transaction is not terminal")
if value["phase"] == "rollback-healthy":
candidate = value["candidate_release"]
retired_candidate = root / "releases" / (
".activation-candidate-retired-" + value["transaction_id"]
)
parked_candidate = directory / "candidate.retired"
candidate_path = _locate_release(
candidate,
extras=(retired_candidate, parked_candidate),
)
if candidate_path is not None and candidate_path != parked_candidate:
try:
parked_candidate.lstat()
except FileNotFoundError:
_rename_and_fsync(candidate_path, parked_candidate)
else:
raise RuntimeError("activation candidate retirement path exists")
_cleanup_config_staging(directory, value)
allowed = {"manifest.json"}
if value["env"]["existed"]:
allowed.add("env.backup")
if value["service"]["existed"]:
allowed.add("service.backup")
if value["phase"] == "rollback-healthy" and (directory / "candidate.retired").exists():
allowed.add("candidate.retired")
if {entry.name for entry in directory.iterdir()} != allowed:
raise RuntimeError("activation transaction directory contains unexpected files")
if directory != retired:
try:
retired.lstat()
except FileNotFoundError:
pass
else:
raise RuntimeError("activation transaction retirement path exists")
os.rename(directory, retired)
_fsync_directory(root)
_cleanup_retired_transactions(root)
def parser() -> argparse.ArgumentParser:
result = argparse.ArgumentParser()
subparsers = result.add_subparsers(dest="command", required=True)
def layout(command: argparse.ArgumentParser) -> None:
command.add_argument("--root", required=True)
command.add_argument("--current", required=True)
command.add_argument("--previous", required=True)
command.add_argument("--env", required=True)
command.add_argument("--service", required=True)
def owned(command: argparse.ArgumentParser) -> None:
layout(command)
command.add_argument("--release-dir", required=True)
command.add_argument("--release-version", required=True)
command.add_argument("--transaction-id", required=True)
begin_parser = subparsers.add_parser("begin")
layout(begin_parser)
begin_parser.add_argument("--release-dir", required=True)
begin_parser.add_argument("--release-version", required=True)
begin_parser.add_argument("--old-target", required=True)
begin_parser.add_argument("--old-source", required=True)
begin_parser.add_argument("--candidate-source", required=True)
begin_parser.add_argument(
"--service-state",
choices=("absent", "stopped", "running"),
required=True,
)
begin_parser.add_argument(
"--service-enabled",
choices=("true", "false"),
required=True,
)
begin_parser.add_argument(
"--legacy-service-state",
choices=("absent", "stopped", "running"),
required=True,
)
begin_parser.add_argument(
"--legacy-service-enabled",
choices=("true", "false"),
required=True,
)
begin_parser.add_argument("--prior-port", type=int, required=True)
begin_parser.add_argument("--prior-bind-address", required=True)
begin_parser.add_argument(
"--intent",
choices=(
"ordinary",
"server-update",
"host-reactivation",
"failed-host-repair",
),
required=True,
)
begin_parser.add_argument("--client-binding", default="")
load_parser = subparsers.add_parser("load")
layout(load_parser)
record_parser = subparsers.add_parser("record")
owned(record_parser)
record_parser.add_argument("--phase", required=True)
record_parser.add_argument("--hub-kind", default="")
record_parser.add_argument("--hub-data-dir", default="")
record_parser.add_argument("--hub-id", default="")
record_parser.add_argument("--host-identity", default="")
record_parser.add_argument("--operation-id", default="")
record_parser.add_argument("--snapshot", default="")
record_parser.add_argument("--fence-device", default="")
record_parser.add_argument("--fence-inode", default="")
record_parser.add_argument("--guard-id", default="")
record_parser.add_argument("--guard-device", default="")
record_parser.add_argument("--guard-inode", default="")
record_parser.add_argument(
"--authority-pending",
choices=("true", "false", "unchanged"),
default="unchanged",
)
replace_parser = subparsers.add_parser("replace-config")
owned(replace_parser)
replace_parser.add_argument("--kind", choices=("env", "service"), required=True)
replace_parser.add_argument("--source", required=True)
replace_parser.add_argument("--mode", required=True)
activate_parser = subparsers.add_parser("activate-files")
owned(activate_parser)
restore_parser = subparsers.add_parser("restore-files")
owned(restore_parser)
finish_parser = subparsers.add_parser("finish")
owned(finish_parser)
return result
def main(argv: list[str] | None = None) -> int:
args = parser().parse_args(argv)
{
"begin": begin,
"load": load,
"record": record,
"replace-config": replace_config,
"activate-files": activate_files,
"restore-files": restore_files,
"finish": finish,
}[args.command](args)
return 0
if __name__ == "__main__":
raise SystemExit(main())