orchestrata_ultimate.promotion.identity

Reference › orchestrata_ultimate › promotion › identity

The deployment identity verifier. It proves, or disproves, that the deploy root, its VERSION record and every production-named systemd unit are the same deployed thing. The verify stage calls it as its last check, and it is also a command-line tool through main and build_parser (--root, a required --expect-sha, a test-only --pattern, and --json). The helper modules are identity_scope (data types, parsing, unit discovery), identity_core (process and liveness checks), identity_checks (unit-file path checks), identity_skips and identity_urls.

check_identity(root, expect_sha, pattern, run, proc_root, tools_root) returns an IdentityReport. The shell command runner and the /proc location are injected so the function is pure apart from those seams. It returns early with an error when the expected sha is empty, the root is not a directory, VERSION is unreadable or not in the space-separated key value format, or git rev-parse HEAD fails. It records a mismatch when HEAD differs from VERSION's commit or from the expected sha. It then discovers units in user and system scope whose names match the service globs, adds each service's own sibling timer, and judges every unit: the skip rules, the path directives, the unit-file path check, the liveness check and the process check. A template instance whose program lives under the tools tree is judged against that tree's own VERSION and virtual environment (ADR-0067).

_check_process_identity handles long-running units: a unit with no main process is an error, the process must pass the checks in identity_core, and its start time must not precede the deployed stamp in VERSION. Oneshots and timers are exempt from the process check. Exit codes follow the report: 0 proven, 1 a mismatch was found, 2 the verifier could not inspect something, and an inspection failure outranks a mismatch.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/promotion/identity.py

"""Deployment identity verifier (U-T6).

Proves that the tree at a deploy ROOT, its ``VERSION`` record, and every
prod-named systemd unit are provably the SAME deployed thing. Health
(``active``/``inactive``) is not identity: a unit can sit perfectly healthy
while serving from the wrong checkout.

Amendment 2026-09-23 (reviewer verdict on 32acfb5d7, "safe to merge: NO"):
scope is decided by unit NAME -- ``orchestrata-*.service``/``orchgraph-*.service``
plus, separately, ONLY each discovered service's own sibling ``.timer`` --
never by ``ExecStart``'s argv0 and never by the bare presence of ``*.timer``
(that would count every unrelated host timer as "proven" too). A prod-named
unit is judged no matter where its ``ExecStart`` actually points, and every
path-like token of ``ExecStart*``, ``WorkingDirectory``, ``EnvironmentFiles``
and ``Environment`` must resolve under ROOT or a known system prefix (check
c0); ``ExecStart``'s own argv0 is the exception -- for a ``Type=oneshot``
unit it must resolve strictly under ROOT with no system-prefix exemption,
since D4 skips every /proc check for a oneshot and c0 is the only identity
proof it has. ``VERSION`` is the repo's real ``key value``-per-line format
(``deploy/install-orchestrata.sh``), not ``key=value``.

Amendment 2026-09-23 attempt 4 (reviewer findings against 7998cc949):
``PATH_DIRECTIVES``/``DEFAULT_GLOBS``/``SERVICE_GLOBS`` now live in
:mod:`identity_core` and are imported at module level here -- the earlier
"lazy import inside check_identity" only moved the ``ModuleNotFoundError``
from import time to call time (``scripts`` still was not on ``sys.path`` at
the call site either), so ``check_identity`` now needs nothing from
``scripts`` at all. ``--expect-sha`` is a required, positional argument (the
contract's CLI shape always required it; the optional keyword form was a
bug). Every discovered unit whose Type is not ``oneshot`` is judged as
long-running (proc identity + start-time floor), not just an allowlisted
subset of Types -- and an unparsable/absent ``ExecMainStartTimestamp`` on a
unit with a live ``MainPID`` is cannot-inspect, never a silent pass.

All the actual parsing/checking logic lives in
:mod:`orchestrata_ultimate.promotion.identity_core`; this module sequences
it and is the CLI surface. Every I/O boundary is an injectable seam so tests
run against real transient systemd units, real git repos and real /proc,
never a double of this module's own code (RUL-002):

* ``run`` -- a ``Sequence[str] -> subprocess.CompletedProcess[str]`` callable
  for every ``systemctl``/``git`` invocation.
* ``proc_root`` -- the directory that stands in for ``/proc``.
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
from collections.abc import Sequence
from pathlib import Path

from orchestrata_ultimate.promotion.identity_checks import tool_tree_root
from orchestrata_ultimate.promotion.identity_core import (
    PATH_DIRECTIVES as _PATH_DIRECTIVES,
)
from orchestrata_ultimate.promotion.identity_core import (
    SERVICE_GLOBS,
    UNIT_PROPERTIES,
    IdentityReport,
    RunFn,
    UnitFacts,
    VersionRecord,
    check_c0,
    check_liveness,
    check_path_directives,
    check_process,
    discover,
    parse_timestamp,
    parse_version,
)
from orchestrata_ultimate.promotion.identity_scope import (
    TOOLS_ROOT,
    instance_of,
    read_tools_version,
    systemd_unescape,
)
from orchestrata_ultimate.promotion.identity_skips import resolve_skip

__all__ = ["default_run", "check_identity", "build_parser", "main"]


def default_run(argv: Sequence[str]) -> subprocess.CompletedProcess[str]:
    """The real seam: shell out to systemctl / git.

    `TZ=UTC`: `systemctl show` formats timestamps in the caller's local
    timezone (confirmed "... EDT" unset vs "... UTC" set, systemd 257).
    `LC_ALL=C`: output must not vary by locale. GIT_DIR/GIT_WORK_TREE/
    GIT_INDEX_FILE are scrubbed so an inherited git-hook environment can
    never redirect `git -C ROOT` at a different repo.
    """
    env = dict(os.environ)
    env["TZ"] = "UTC"
    env["LC_ALL"] = "C"
    for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
        env.pop(key, None)
    return subprocess.run(  # noqa: S603
        list(argv), capture_output=True, text=True, timeout=30, check=False, env=env
    )


def check_identity(
    root: Path | str,
    expect_sha: str,
    *,
    pattern: str | None = None,
    run: RunFn = default_run,
    proc_root: Path = Path("/proc"),
    tools_root: Path = TOOLS_ROOT,
) -> IdentityReport:
    """Prove (or disprove) that ``root``, its ``VERSION`` and every prod-named
    systemd unit are the same deployed thing. Pure aside from the injected
    ``run``/``proc_root`` seams. ``pattern`` narrows discovery FURTHER than
    the globs (test isolation only -- production passes None).

    A template instance whose argv0 lives under ``tools_root`` is judged
    against that tree's own VERSION/venv/cwd, and its ``%I`` parameter is
    data, not code (ADR-0067 R1/P3); ``tools_root`` is a test seam only.

    ``expect_sha`` is required (the contract's CLI shape always required it;
    :func:`build_parser` refuses to parse without it too) -- a verifier that
    can silently skip its OWN sha comparison is not a verifier."""
    root_path = Path(root)
    report = IdentityReport(root=root_path)

    if expect_sha == "":
        report.errors.append("--expect-sha must not be empty")
        return report
    if not root_path.is_dir():
        report.errors.append(f"ROOT {root_path} is not a directory")
        return report

    root_resolved = root_path.resolve()
    version_path = root_path / "VERSION"
    try:
        version_text = version_path.read_text(encoding="utf-8")
    except OSError as exc:
        report.errors.append(f"cannot read {version_path}: {exc}")
        return report
    version = parse_version(version_text)
    if version is None:
        report.errors.append(
            f"{version_path} is missing or not in the `key value` format"
        )
        return report

    git_result = run(["git", "-C", str(root_path), "rev-parse", "HEAD"])
    if git_result.returncode != 0:
        problem = (git_result.stderr or git_result.stdout or "").strip()
        report.errors.append(f"git -C {root_path} rev-parse HEAD failed: {problem}")
        return report
    sha = git_result.stdout.strip()
    report.sha = sha
    if sha != version.commit:
        report.mismatches.append(
            f"ROOT: VERSION.commit: expected {sha}, got {version.commit}"
        )
    if expect_sha and sha != expect_sha:
        report.mismatches.append(f"ROOT: git HEAD: expected {expect_sha}, got {sha}")

    compiled_pattern = re.compile(pattern) if pattern else None
    venv_python = (root_path / ".venv" / "bin" / "python").resolve()

    # Amendment finding 1: scope is orchestrata-*/orchgraph-* SERVICES, plus
    # -- separately, per discovered service -- ONLY that service's own
    # sibling `.timer` (same base name). A bare host timer (logrotate.timer,
    # man-db.timer, ...) is never in scope just because a unit named that
    # exists; the previous "*.timer" glob discovered EVERY timer on the
    # host and counted every one of them as "proven".
    discovered: list[tuple[str, str]] = []
    for scope in ("user", "system"):
        names, error = discover(run, scope, SERVICE_GLOBS, compiled_pattern)
        if error is not None:
            report.errors.append(error)
            continue
        for name in names:
            discovered.append((name, scope))
            timer_name = re.sub(r"\.service$", ".timer", name)
            timer_names, timer_error = discover(run, scope, (timer_name,), None)
            if timer_error is not None:
                report.errors.append(timer_error)
                continue
            discovered.extend((timer, scope) for timer in timer_names)

    tools_versions: dict[Path, VersionRecord | None] = {}  # read lazily, once
    for name, scope in discovered:
        properties, skip_reason, error = resolve_skip(name, scope, run, UNIT_PROPERTIES)
        if skip_reason is not None:
            report.skipped.append(f"skipped {name} ({skip_reason})")
            continue
        if error is not None:
            report.errors.append(error)
            continue
        unit_root, unit_python, unit_version = root_resolved, venv_python, version
        data: tuple[str, ...] = ()
        tools = tool_tree_root(name, properties, tools_root)
        if tools is not None:
            if tools not in tools_versions:
                tools_versions[tools] = read_tools_version(tools, report.errors)
            tools_version = tools_versions[tools]
            if tools_version is None:
                continue  # cause already in report.errors, once per tree
            unit_root = tools.resolve()
            unit_python = (unit_root / ".venv" / "bin" / "python").resolve()
            unit_version = tools_version
            data = ("/" + systemd_unescape(instance_of(name)),)
        unit = UnitFacts(name=name, scope=scope, properties=properties, root=tools)
        report.units.append(unit)

        check_path_directives(unit, _PATH_DIRECTIVES, report)  # check g
        check_c0(unit, unit_root, report, data_prefixes=data)  # check c0
        check_liveness(unit, run, scope, name, report)  # check f
        _check_process_identity(
            name,
            properties,
            unit_root,
            unit_python,
            unit_version,
            proc_root,
            report,
        )

    if not report.units and not report.errors and not report.skipped:
        report.errors.append("no unit discovered in scope")

    return report


def _check_process_identity(
    name: str,
    properties: dict[str, str],
    root_resolved: Path,
    venv_python: Path,
    version: VersionRecord,
    proc_root: Path,
    report: IdentityReport,
) -> None:
    """Checks c/d/e -- proc identity + start-time floor.

    Amendment finding 5: EVERY Type that is not ``oneshot`` is long-running
    and judged here -- there is no allowlist of "checked" Types (idle, dbus,
    notify-reload and an unset Type used to skip this entirely and count as
    trivially proven). A ``.timer`` unit is the one structural exception: it
    has no ``MainPID``/``Type`` of its own (verified on this host, see
    ``identity_core.check_liveness``'s docstring) so it is exempted by NAME,
    not by Type."""
    if name.endswith(".timer"):
        return
    if properties.get("Type", "") == "oneshot":
        return  # D4: an in-flight oneshot legitimately runs something else
    try:
        pid = int(properties.get("MainPID", "0") or "0")
    except ValueError:
        pid = 0
    if pid <= 0:
        # D2: never silently dropped -- a long-running unit with no process
        # cannot have its identity proven at all.
        report.errors.append(
            f"{name}: no process (MainPID=0) -- cannot verify cwd/cmdline/exe"
        )
        return
    check_process(pid, proc_root, root_resolved, venv_python, name, report)
    raw_start = properties.get("ExecMainStartTimestamp", "")
    start_time = parse_timestamp(raw_start)
    if start_time is None:
        # Amendment finding 9: systemd always sets this the moment a real
        # process starts, so an unparsable/absent value on a unit that DOES
        # have a live MainPID is a real inconsistency -- cannot-inspect,
        # never a silent pass.
        report.errors.append(
            f"{name}: ExecMainStartTimestamp: cannot parse {raw_start!r} "
            f"for a running process (MainPID={pid})"
        )
    elif start_time < version.deployed:
        report.mismatches.append(
            f"{name}: start-time: expected >= {version.deployed.isoformat()}, "
            f"got {start_time.isoformat()}"
        )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="verify_deployment_identity.py",
        description=(
            "Prove that ROOT, its VERSION record, and every prod-named "
            "systemd unit are the same deployed thing."
        ),
    )
    parser.add_argument(
        "--root",
        default="/opt/orchestrata",
        help="deploy root (default: /opt/orchestrata)",
    )
    parser.add_argument(
        "--expect-sha",
        required=True,
        help="git sha the deploy must match (required)",
    )
    parser.add_argument(
        "--pattern", default=None, help="extra regex narrowing discovery (tests only)"
    )
    parser.add_argument(
        "--json", action="store_true", help="emit the full report as JSON"
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    report = check_identity(
        args.root, args.expect_sha, pattern=args.pattern, run=default_run
    )
    if args.json:
        import json

        print(json.dumps(report.as_dict(), indent=2, default=str))
    else:
        print(report.render())
    return report.exit_code


if __name__ == "__main__":
    raise SystemExit(main())