"""Install a release, verify it, and undo it if the verification fails. THE SHAPE, AND WHY IT IS THIS SHAPE verify signature ─ before the archive is opened at all extract ─ into a staging directory, never over a live release migrate ─ using the NEW release's own migrator health gate ─ `python -m app check`, the same assertions the daemon makes at boot switch ─ one atomic pointer move restart ─ whatever the deployment uses verify again ─ readiness, over HTTP, against the running service roll back ─ on any failure from the health gate onwards Two verifications, not one. The gate before the switch catches a release that cannot possibly work — wrong Python, unreachable database, missing migration — while the old version is still live and nothing has been disturbed. The check after the restart catches what only shows up once the new code is actually running. A single check placed either side would miss one of those. WHAT ROLLBACK HONESTLY COVERS Code, and the pointer. Not the database. The CMS restores a SQL dump because it can afford to: it takes the site down, and a tax firm's CMS being read-only for two minutes is survivable. A rollback here is meant to take seconds, and restoring a dump of a workflow database would discard every run that started since the install — including documents a client sent while it was happening. So the rule is the expand/contract one: a migration must leave the previous release able to run. Add columns, do not rename them; add tables, do not drop them; remove things one release after nothing reads them. `state.json` records which migrations each release applied so an operator can see exactly what a rollback leaves in place. docs/distribution.md carries the policy in full. This is a deliberate trade, not an omission. """ from __future__ import annotations import json import logging import os import shutil import subprocess import sys import tarfile import tempfile import time import urllib.error import urllib.request from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from app.release import signing from app.release import version as versions from app.release.layout import Layout, LayoutError from app.release.manifest import ( MANIFEST_NAME, Manifest, ManifestError, Signature, parse_signing_string, ) from app.release.package import sha256_of from app.release.restart import NoRestart, RestartStrategy logger = logging.getLogger(__name__) #: How long a migration or a preflight check may take before it is abandoned. #: #: Generous: a preflight check loads the OCR models, which is tens of seconds on #: a cold VPS, and a migration on a large workflow table is not instant either. STEP_TIMEOUT_SECONDS = 600.0 #: How long to wait for the service to report ready after a restart. READINESS_TIMEOUT_SECONDS = 90.0 READINESS_INTERVAL_SECONDS = 3.0 #: A health probe. Given the version that should now be running, returns #: (ready, detail). #: #: The version is an argument rather than something the probe closes over #: because "is something listening?" is not the question. After a restart that #: silently did nothing, the OLD process is still listening and still answering #: 200 — so a probe without the version reports a release healthy while the code #: it replaced is what is actually running. Probe = Callable[[str], tuple[bool, str]] @dataclass(frozen=True, slots=True) class InstallResult: ok: bool version: str | None = None previous: str | None = None reason: str = "" detail: str = "" rolled_back: bool = False def summary(self) -> str: if self.ok: return f"{self.previous or 'nothing'} -> {self.version}" rolled = " (rolled back)" if self.rolled_back else "" return f"{self.reason}: {self.detail}{rolled}" class Installer: """Applies releases to one deployment directory.""" def __init__( self, layout: Layout, public_key: str | None, restart: RestartStrategy | None = None, probe: Probe | None = None, python: str | None = None, require_signature: bool = True, ) -> None: self._layout = layout self._public_key = public_key self._restart = restart or NoRestart() self._probe = probe self._python = python or sys.executable self._require_signature = require_signature # ── Installing ──────────────────────────────────────────────────────── def install(self, archive: Path, signature: Signature | None = None) -> InstallResult: archive = Path(archive) if not archive.is_file(): return InstallResult(False, reason="missing_archive", detail=str(archive)) try: claim = self._authenticate(archive, signature) except (ManifestError, signing.SignatureError) as exc: return InstallResult(False, reason="unverified", detail=str(exc)) version = claim.version state = self._layout.state() previous = self._layout.current_version() if version == previous: return InstallResult(False, version=version, reason="already_current", detail="") self._layout.prepare() target = self._layout.release(version) if target.exists(): # A previous attempt left it behind. Removing it is safe precisely # because it is not the live release — the pointer says so. shutil.rmtree(target, ignore_errors=True) staging = Path(tempfile.mkdtemp(prefix=f".{version}.", dir=self._layout.releases)) try: manifest = self._unpack(archive, staging, version) self._check_runtime(manifest) os.replace(staging, target) except (InstallError, OSError) as exc: shutil.rmtree(staging, ignore_errors=True) state.record("install", version, "failed", str(exc)) self._layout.save(state) return InstallResult(False, version=version, previous=previous, reason="rejected", detail=str(exc)) # From here the release is on disk, so every failure has to clean up. try: self._migrate(target) self._gate(target) except InstallError as exc: # Nothing has been switched yet: the old release is still live and # still serving. Removing the new directory is the entire undo. shutil.rmtree(target, ignore_errors=True) state.record("install", version, "failed", str(exc)) self._layout.save(state) return InstallResult(False, version=version, previous=previous, reason="failed_gate", detail=str(exc)) try: self._layout.point_to(version) except LayoutError as exc: shutil.rmtree(target, ignore_errors=True) state.record("install", version, "failed", str(exc)) self._layout.save(state) return InstallResult(False, version=version, previous=previous, reason="switch_failed", detail=str(exc)) restarted = self._restart.restart() ready, detail = self._verify_ready(version) if restarted.ok else (False, restarted.detail) if not ready: rolled = self._revert(previous) # When there is nothing to revert to, the new release stays live — # there is no better thing to point at, and unpointing would leave # the deployment with no application at all. The state file follows # the pointer so the two cannot disagree, and the outcome records # that it was never confirmed healthy. if not rolled: state.previous = previous state.version = version state.record("install", version, "rolled_back" if rolled else "unhealthy", detail) self._layout.save(state) return InstallResult( False, version=version, previous=previous, reason="unhealthy", detail=detail, rolled_back=rolled, ) # Both halves. Reporting only the restart would drop "not verified: no # readiness probe configured" — which is the difference between a # release that was confirmed running and one that merely got as far as # being switched to. outcome = "; ".join(part for part in (restarted.detail, detail) if part) state.previous = previous state.version = version state.record("install", version, "installed", outcome) self._layout.save(state) return InstallResult(True, version=version, previous=previous, detail=outcome) # ── Rolling back ────────────────────────────────────────────────────── def rollback(self, to: str | None = None) -> InstallResult: """Return to the previous release, or to a named one. Deliberately does NOT run migrations. There is nothing to run — the schema is already ahead, and the release being restored is required to work against it. See this module's docstring. """ state = self._layout.state() current = self._layout.current_version() target = to or state.previous if not target: return InstallResult(False, version=current, reason="nothing_to_roll_back_to", detail="") if target == current: return InstallResult(False, version=current, reason="already_current", detail="") if not self._layout.release(target).is_dir(): return InstallResult( False, version=current, reason="not_installed", detail=f"Release {target} is not on this deployment.", ) try: self._layout.point_to(target) except LayoutError as exc: return InstallResult(False, version=current, reason="switch_failed", detail=str(exc)) restarted = self._restart.restart() ready, detail = self._verify_ready(target) if restarted.ok else (False, restarted.detail) state.previous = current state.version = target state.record("rollback", target, "rolled_back" if ready else "unverified", detail) self._layout.save(state) # Reported even when the probe fails: the rollback itself happened, and # an operator rolling back is already dealing with a broken service — # telling them the rollback "failed" when the pointer did move would # send them looking in the wrong place. return InstallResult( ok=True, version=target, previous=current, reason="" if ready else "unverified", detail=detail, rolled_back=True, ) # ── Steps ───────────────────────────────────────────────────────────── def _authenticate(self, archive: Path, signature: Signature | None): """Prove who built this, before anything opens it.""" if signature is None: sidecar = archive.with_suffix(archive.suffix + ".sig") if sidecar.is_file(): signature = Signature.read(sidecar) if signature is None: if self._require_signature: raise ManifestError( f"No signature for {archive.name}. Expected {archive.name}.sig beside it." ) # Only reachable when an operator has explicitly turned the # requirement off — for a locally built package during development. # It is not a fallback for a missing key. logger.warning("Installing %s WITHOUT verifying its signature.", archive.name) return self._unsigned_claim(archive) if not self._public_key: # Fail closed. "Cannot verify" must never collapse into "assume # fine", which would hand back exactly the property the offline key # exists to provide. raise signing.SignatureError( "Signature verification is required but no public key is configured." ) if not signing.verify(signature.manifest, signature.signature, self._public_key): raise signing.SignatureError("The signature does not match the release manifest.") claim = parse_signing_string(signature.manifest) digest = sha256_of(archive) size = archive.stat().st_size # The signature attests to a digest and a size. This is where the file # on disk is held to that claim — without it, a valid signature for one # release would authenticate any bytes at all. if digest != claim.sha256: raise ManifestError(f"Digest mismatch: signed {claim.sha256}, found {digest}.") if size != claim.size: raise ManifestError(f"Size mismatch: signed {claim.size} bytes, found {size}.") return claim def _unsigned_claim(self, archive: Path): from app.release.manifest import SignedString with tarfile.open(archive, "r:gz") as tar: manifest = _read_manifest(tar, archive) return SignedString( version=manifest.version, sha256=sha256_of(archive), size=archive.stat().st_size, ) def _unpack(self, archive: Path, into: Path, expected_version: str) -> Manifest: try: with tarfile.open(archive, "r:gz") as tar: # filter="data" refuses absolute paths, `..` traversal, links # pointing outside the destination, and device files. Without it # a crafted archive writes anywhere the process can (CVE-2007-4559), # and this one runs where it can write the live application. tar.extractall(into, filter="data") except (tarfile.TarError, OSError) as exc: raise InstallError(f"The archive could not be extracted: {exc}") from exc manifest_path = into / MANIFEST_NAME if not manifest_path.is_file(): raise InstallError("The archive has no manifest.") try: manifest = Manifest.read(manifest_path) except ManifestError as exc: raise InstallError(str(exc)) from exc if manifest.version != expected_version: # The signature covers the digest and the version together, so this # can only mean the archive was labelled one thing and signed as # another. Refusing keeps `state.json` and the release directory # agreeing about what is installed. raise InstallError( f"The archive says it is {manifest.version}, but it was signed as {expected_version}." ) if not (into / "app").is_dir(): raise InstallError("The archive does not contain an application.") return manifest def _check_runtime(self, manifest: Manifest) -> None: running = ".".join(str(p) for p in sys.version_info[:3]) if not versions.at_least(running, manifest.requires_python): raise InstallError( f"This release needs Python {manifest.requires_python}; this is {running}." ) def _migrate(self, release: Path) -> None: """Run the new release's own migrator. A subprocess in the new release directory rather than an import, so the migrations that run are the ones being installed — importing them into the currently-running process would run the OLD code against the new SQL, which is the subtle version of this going wrong. """ result = self._run(release, "migrate") if result.returncode != 0: raise InstallError(f"Migrations failed: {_tail(result)}") def _gate(self, release: Path) -> None: """`python -m app check` — the same assertions the daemon makes at boot. Reused rather than reimplemented: a gate that checks something slightly different from what the service requires is a gate that passes releases which then fail to start. """ result = self._run(release, "check") if result.returncode != 0: raise InstallError(f"The release failed its preflight check: {_tail(result)}") def _run(self, release: Path, command: str) -> subprocess.CompletedProcess: environment = dict(os.environ) environment.update(read_env_file(self._layout.shared / ".env")) environment["PYTHONPATH"] = str(release) try: return subprocess.run( # noqa: S603 - argv is fixed; only cwd and env vary [self._python, "-m", "app", command], cwd=str(release), env=environment, capture_output=True, text=True, timeout=STEP_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired as exc: raise InstallError(f"`{command}` did not finish within {STEP_TIMEOUT_SECONDS:.0f}s.") from exc except OSError as exc: raise InstallError(f"Could not run `{command}`: {exc}") from exc def _verify_ready(self, version: str) -> tuple[bool, str]: if self._probe is None: # Nothing to probe against — a manual install, or a deployment whose # service this installer does not restart. Saying so is the honest # answer; probing for a restart that never happened would either # time out or, worse, pass against the old process. return True, "not verified: no readiness probe configured" return self._probe(version) def _revert(self, previous: str | None) -> bool: if not previous or not self._layout.release(previous).is_dir(): return False try: self._layout.point_to(previous) except LayoutError: return False self._restart.restart() return True class InstallError(RuntimeError): """A release could not be installed. Carries the reason an operator needs.""" # ── Helpers ─────────────────────────────────────────────────────────────── def http_readiness( url: str, timeout: float = READINESS_TIMEOUT_SECONDS, interval: float = READINESS_INTERVAL_SECONDS, sleep: Callable[[float], None] = time.sleep, ) -> Probe: """Poll `/health/ready` until the EXPECTED version answers 200. Two things are being checked, and the second is the one that matters. A 200 alone only proves something is listening on that port. If the restart silently did nothing — a unit that failed to reload, a container that was never replaced — the old process is still there, still healthy, and still running the code this install was meant to replace. An install verified that way reports success and changes nothing. So the health payload's own version has to match. A release that does not report one is accepted on the 200: that is an older build being rolled back to, and refusing it would make the rollback path fail exactly when it is needed. Polls rather than checks once, because a service that loads OCR models takes tens of seconds to become ready and a single immediate check would roll back every healthy install. """ def probe(expected: str) -> tuple[bool, str]: deadline = time.monotonic() + timeout last = "no response" while True: try: with urllib.request.urlopen(url, timeout=10) as response: # noqa: S310 - operator-configured if response.status == 200: running = _reported_version(response.read()) if running is None or running == expected: return True, f"ready on {running or 'an unreported version'}" # Listening, healthy, and the wrong code. Almost # certainly the previous process: the restart did not # take. last = f"still running {running}" else: last = f"HTTP {response.status}" except urllib.error.HTTPError as exc: last = f"HTTP {exc.code}" except (urllib.error.URLError, OSError) as exc: last = str(getattr(exc, "reason", exc)) if time.monotonic() >= deadline: return False, f"not ready after {timeout:.0f}s ({last})" sleep(interval) return probe def _reported_version(body: bytes) -> str | None: """The version out of a health response, or None if it does not say.""" try: payload = json.loads(body) except (json.JSONDecodeError, ValueError): return None version = payload.get("version") if isinstance(payload, dict) else None return str(version) if version else None def read_env_file(path: Path) -> dict[str, str]: """Read `KEY=value` lines from the deployment's shared configuration. The service itself is configured by whatever runs it — systemd's EnvironmentFile, Docker's --env-file. The installer reads the same file so that `migrate` and `check` see the configuration the service will see; without it the preflight check would fail on a missing CMS URL and reject a perfectly good release. Deliberately small: no interpolation, no `export`, no multi-line values. It reads a file this project writes, and a half-implemented shell parser that silently mis-reads a quoted password is worse than one that does less. """ if not path.is_file(): return {} values: dict[str, str] = {} try: lines = path.read_text(encoding="utf-8").splitlines() except OSError: return {} for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, _, value = stripped.partition("=") value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": value = value[1:-1] values[key.strip()] = value return values def _read_manifest(tar: tarfile.TarFile, archive: Path) -> Manifest: try: member = tar.extractfile(MANIFEST_NAME) except KeyError as exc: raise ManifestError(f"{archive.name} has no manifest.") from exc if member is None: raise ManifestError(f"{archive.name} has no readable manifest.") with member: return Manifest.from_json(member.read()) def _tail(result: subprocess.CompletedProcess) -> str: """The last line that said anything, which is almost always the reason.""" output = (result.stderr or "") + (result.stdout or "") lines = [line.strip() for line in output.splitlines() if line.strip()] return lines[-1] if lines else f"exit {result.returncode}"