orchestrata_ultimate.promotion.identity_core

Reference › orchestrata_ultimate › promotion › identity_core

The two per-unit runtime checks of the identity verifier, both writing into an IdentityReport from identity_scope.

check_process(pid, proc_root, root, venv_python, name, report) inspects /proc/<pid>. The working directory must equal the deploy root, the first element of the command line must lie under the root, and the executable must be the root's virtual-environment Python. A difference is recorded as a mismatch, and a file that cannot be read is recorded as an error, which the report ranks above a mismatch.

check_liveness(unit, run, scope, name, report) applies the state rule per unit kind. A timer must be active, since it has no main process of its own. A oneshot service is judged by its sibling timer, which must be active, and the oneshot itself must not be in the failed state. Every other service type is long-running and must be active; there is no list of exempt types.

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_core.py

"""Identity-check engine for ``promotion/identity.py`` (U-T6).

Split out to keep ``identity.py`` -- the CLI-facing orchestration module --
under its 300-hand-line budget. Everything the checks (a, c0, c, d, e, f, g)
need is reachable from here: the dataclasses, the ``VERSION``/timestamp/
`systemctl show` parsers, the subprocess-shaped ``show``/``discover`` calls
(still driven by the caller's injected ``run``, never a module-level
subprocess of their own), and each individual check function. ``identity.py``
sequences these; it does not reimplement any of them.

U-T6c (2026-09-26): this module itself grew past its own 300-line budget
(RUL-021). The scoping constants, identity dataclasses, VERSION/timestamp/
`systemctl show` parsers and the ``show``/``discover`` calls now live in
``identity_scope.py``; the ExecStart/path extraction+judging helpers and
``check_c0``/``check_path_directives`` now live in ``identity_checks.py``.
``check_process`` and ``check_liveness`` -- this module's own check-function
entry points -- stay here. Every public name this module used to define is
imported straight back below (explicit re-exports), so every importer
(``identity.py``, ``scripts/check_prod_tree_separation.py``,
``scripts/verify_deployment_identity.py``, every test) is unchanged.
"""

from __future__ import annotations

import os
import re
from pathlib import Path

from orchestrata_ultimate.promotion.identity_checks import (
    check_c0 as check_c0,
)
from orchestrata_ultimate.promotion.identity_checks import (
    check_path_directives as check_path_directives,
)
from orchestrata_ultimate.promotion.identity_checks import (
    environment_file_paths as environment_file_paths,
)
from orchestrata_ultimate.promotion.identity_checks import (
    extract_argv as extract_argv,
)
from orchestrata_ultimate.promotion.identity_checks import (
    extract_argv0 as extract_argv0,
)
from orchestrata_ultimate.promotion.identity_checks import (
    extract_path as extract_path,
)
from orchestrata_ultimate.promotion.identity_checks import (
    normalize_directive_value as normalize_directive_value,
)
from orchestrata_ultimate.promotion.identity_checks import (
    resolves_safely as resolves_safely,
)
from orchestrata_ultimate.promotion.identity_scope import (
    DEFAULT_GLOBS as DEFAULT_GLOBS,
)
from orchestrata_ultimate.promotion.identity_scope import (
    PATH_DIRECTIVES as PATH_DIRECTIVES,
)
from orchestrata_ultimate.promotion.identity_scope import (
    SERVICE_GLOBS as SERVICE_GLOBS,
)
from orchestrata_ultimate.promotion.identity_scope import (
    SYSTEM_PREFIXES as SYSTEM_PREFIXES,
)
from orchestrata_ultimate.promotion.identity_scope import (
    UNIT_PROPERTIES as UNIT_PROPERTIES,
)
from orchestrata_ultimate.promotion.identity_scope import (
    IdentityReport,
    RunFn,
    UnitFacts,
    show,
)
from orchestrata_ultimate.promotion.identity_scope import (
    VersionRecord as VersionRecord,
)
from orchestrata_ultimate.promotion.identity_scope import (
    discover as discover,
)
from orchestrata_ultimate.promotion.identity_scope import (
    parse_show as parse_show,
)
from orchestrata_ultimate.promotion.identity_scope import (
    parse_timestamp as parse_timestamp,
)
from orchestrata_ultimate.promotion.identity_scope import (
    parse_version as parse_version,
)
from orchestrata_ultimate.promotion.identity_urls import (
    environment_path_values as environment_path_values,
)
from orchestrata_ultimate.promotion.identity_urls import (
    standard_io_path as standard_io_path,
)


def check_process(
    pid: int,
    proc_root: Path,
    root: Path,
    venv_python: Path,
    name: str,
    report: IdentityReport,
) -> None:

    proc_dir = proc_root / str(pid)
    try:
        cwd = os.readlink(proc_dir / "cwd")
    except OSError as exc:
        report.errors.append(f"cannot read {proc_dir / 'cwd'} for {name}: {exc}")
    else:
        if cwd != str(root):
            report.mismatches.append(
                f"{name}: /proc/{pid}/cwd: expected {root}, got {cwd}"
            )
    try:
        cmdline_bytes = (proc_dir / "cmdline").read_bytes()
    except OSError as exc:
        report.errors.append(f"cannot read {proc_dir / 'cmdline'} for {name}: {exc}")
    else:
        cmdline0 = cmdline_bytes.split(b"\0", 1)[0].decode(errors="replace")
        if not cmdline0.startswith(f"{root}/"):
            report.mismatches.append(
                f"{name}: /proc/{pid}/cmdline[0]: expected under {root}, got {cmdline0}"
            )
    try:
        exe = os.readlink(proc_dir / "exe")
    except OSError as exc:
        report.errors.append(f"cannot read {proc_dir / 'exe'} for {name}: {exc}")
    else:
        if exe != str(venv_python):
            report.mismatches.append(
                f"{name}: /proc/{pid}/exe: expected {venv_python}, got {exe}"
            )


def check_liveness(
    unit: UnitFacts, run: RunFn, scope: str, name: str, report: IdentityReport
) -> None:
    """Amendment finding 5: ANY Type that is not oneshot is long-running and
    judged the same way -- there is no allowlist of "checked" types (idle,
    dbus, notify-reload and an unset/empty Type on a real .service unit used
    to get no liveness check at all and counted as trivially proven). A
    ``.timer`` unit is the one structural exception: verified on this host,
    `systemctl show -p Type,MainPID <t>.timer` prints NEITHER line -- a
    timer never has a Type or a MainPID of its own, so its own ActiveState
    (active while armed/waiting) IS its whole liveness contract."""
    if name.endswith(".timer"):
        active_state = unit.properties.get("ActiveState", "")
        if active_state != "active":
            report.mismatches.append(
                f"{name}: ActiveState: expected active, got {active_state}"
            )
        return
    unit_type = unit.properties.get("Type", "")
    if unit_type == "oneshot":
        timer_name = re.sub(r"\.service$", ".timer", name)
        timer_properties, error = show(run, scope, timer_name, ("ActiveState",))
        if error is not None:
            report.errors.append(f"{error} (sibling timer for oneshot {name})")
            return
        if timer_properties.get("ActiveState") != "active":
            report.mismatches.append(
                f"{name}: sibling timer {timer_name} ActiveState: expected active, "
                f"got {timer_properties.get('ActiveState', '')}"
            )
        # `Result=` is a REASON code (success, exit-code, signal, ...) -- it
        # is never literally "failed" (systemd 257). ActiveState is.
        if unit.properties.get("ActiveState") == "failed":
            result = unit.properties.get("Result", "")
            report.mismatches.append(
                f"{name}: ActiveState: expected != failed, got failed (Result={result})"
            )
        return
    active_state = unit.properties.get("ActiveState", "")
    if active_state != "active":
        report.mismatches.append(
            f"{name}: ActiveState: expected active, got {active_state}"
        )