orchestrata_ultimate.promotion.identity_scope

Reference › orchestrata_ultimate › promotion › identity_scope

The data types and parsing helpers of the identity verifier, plus the systemd discovery it starts from. It also defines the constants that set the scope of the check, including the unit-name globs and property list.

The types are VersionRecord (commit, ref, repo and deployed time parsed from VERSION), UnitFacts (a discovered unit with its scope, systemctl show properties and optional tools-tree root) and IdentityReport. The report accumulates units, mismatches (something differs), errors (something could not be inspected) and skipped entries. proven counts the judged units, exit_code is 2 when there are errors, else 1 when there are mismatches, else 0, render() gives the text form (IDENTITY OK with the short sha, or one line per problem), and as_dict() the JSON form.

parse_version reads the space-separated key value lines that the installer writes and returns nothing when the commit or deployed stamp is missing. parse_timestamp accepts ISO-8601 and systemd's default timestamp format. read_tools_version parses a tools tree's own VERSION and appends the cause to the error list when it fails. systemd_unescape and instance_of decode a template instance name without a subprocess, so the instance parameter stays data. show runs systemctl show for a unit and parse_show parses the output. discover lists units with systemctl list-units --all --plain and keeps those matching the globs and an optional regular expression.

At a glance

Class diagram

Dependency diagram

API

Classes

VersionRecord (dataclass)

Decorators: dataclass(frozen=True) · Line 115

UnitFacts (dataclass)

Decorators: dataclass · Line 123

IdentityReport (dataclass)

Decorators: dataclass · Line 132

Functions

Source

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

"""Scope, identity records and discovery for ``identity_core.py`` (U-T6c).

Split out 2026-09-26 so ``identity_core.py`` stays under its own 300-line
budget (RUL-021): the module-level scoping constants (unit globs, resolved
``systemctl show`` property names, path-directive names), the identity
dataclasses (``VersionRecord``/``UnitFacts``/``IdentityReport``), the
``VERSION``/timestamp/``systemctl show`` parsers, and the subprocess-shaped
``show``/``discover`` calls (still driven by the caller's injected ``run``,
never a module-level subprocess of their own) all live here.
``identity_core.py`` imports every public name straight back, so the split
is invisible to every importer (``identity.py``,
``scripts/check_prod_tree_separation.py``,
``scripts/verify_deployment_identity.py``, every test) -- see that module's
own docstring for the full picture.
"""

from __future__ import annotations

import re
import subprocess
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path

#: The deployed tool tree (ADR-0067 P3): template instances whose ExecStart
#: argv0 lives here are judged against THIS root, not the product ROOT.
TOOLS_ROOT = Path("/opt/orchestrata-tools")

# Prefixes a path-like directive token may legitimately live under besides
# ROOT itself. Anything under /home is NEVER on this list (amendment D2).
SYSTEM_PREFIXES = (
    "/usr",
    "/bin",
    "/sbin",
    "/etc",
    "/run",
    "/var",
    "/tmp",
    "/dev",
    "/proc",
    "/sys",
    str(TOOLS_ROOT),
)

# Resolved values only -- drop-ins and `\`-continued directives are already
# merged into these by systemd itself (confirmed, systemd 257).
UNIT_PROPERTIES = (
    "FragmentPath",
    "DropInPaths",
    "ExecStart",
    "ExecStartPre",
    "ExecStartPost",
    "WorkingDirectory",
    "EnvironmentFiles",
    "Environment",
    "Type",
    "ActiveState",
    "SubState",
    "Result",
    "MainPID",
    "ExecMainStartTimestamp",
    "NRestarts",
    "StandardOutput",
    "StandardError",
    "StandardInput",
)

# Directives whose value is a path into a checkout. Canonical home of this
# constant -- scripts/check_prod_tree_separation.py imports it FROM here
# (moved 2026-09-23, amendment finding 4: identity.py must not need
# `scripts` importable at all to run).
#
# `Environment=` is here because of a real miss: orchestrata-history-mining
# carried `Environment=PYTHONPATH=<dev tree>/src`. An exported PYTHONPATH
# OUTRANKS a checkout's own .pth, so that one line makes a unit execute the
# development tree's source no matter how correct its WorkingDirectory and
# ExecStart are. `StandardOutput`/`StandardError`/`StandardInput`/
# `RootDirectory` are here for the same reason: each can carry (or redirect
# through) a path into the dev tree without touching WorkingDirectory or any
# Exec* directive at all.
PATH_DIRECTIVES = (
    "WorkingDirectory",
    "ExecStart",
    "ExecStartPre",
    "ExecStartPost",
    "ExecStop",
    "ExecReload",
    "EnvironmentFile",
    "Environment",
    "StandardOutput",
    "StandardError",
    "StandardInput",
    "RootDirectory",
)

# orchestrata-*/orchgraph-* services, plus every timer regardless of prefix.
# NOTE this is deliberately BROAD for check_prod_tree_separation.py's own
# purpose (examine every timer's own path directives, of which it has none
# in practice, so nothing is ever mistaken for "not checked"). identity.py's
# SCOPING rule is narrower and lives separately as SERVICE_GLOBS below --
# identity.py never treats "matches DEFAULT_GLOBS" alone as in-scope.
DEFAULT_GLOBS = ("orchestrata-*.service", "orchgraph-*.service", "*.timer")

# identity.py's actual scope (amendment finding 1): orchestrata-*/orchgraph-*
# SERVICES, and -- 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 it matches "*.timer".
SERVICE_GLOBS = tuple(g for g in DEFAULT_GLOBS if g.endswith(".service"))

RunFn = Callable[[Sequence[str]], "subprocess.CompletedProcess[str]"]


@dataclass(frozen=True)
class VersionRecord:
    commit: str
    ref: str
    repo: str
    deployed: datetime


@dataclass
class UnitFacts:
    name: str
    scope: str  # "user" | "system"
    properties: dict[str, str]
    #: Set only for a tool-tree template instance: the root its verdicts used.
    root: Path | None = None


@dataclass
class IdentityReport:
    root: Path
    sha: str = ""
    units: list[UnitFacts] = field(default_factory=list)
    mismatches: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)
    skipped: list[str] = field(default_factory=list)

    @property
    def proven(self) -> int:
        """Every discovered unit is judged (D2) -- at exit 0 all of them proven."""
        return len(self.units)

    @property
    def exit_code(self) -> int:
        """D5: cannot-inspect always outranks a found mismatch."""
        if self.errors:
            return 2
        if self.mismatches:
            return 1
        return 0

    def render(self) -> str:
        if not self.errors and not self.mismatches:
            lines = [f"IDENTITY OK {self.sha[:12]} · {self.proven} unit(s) proven"]
        else:
            lines = [f"cannot inspect: {problem}" for problem in self.errors]
            lines.extend(self.mismatches)
        lines.extend(self.skipped)
        lines.extend(
            f"judged {u.name} against {u.root} "
            "(ADR-0067 R1: instance parameter is data)"
            for u in self.units
            if u.root is not None
        )
        return "\n".join(lines)

    def as_dict(self) -> dict[str, object]:
        return {
            "root": str(self.root),
            "sha": self.sha,
            "exit_code": self.exit_code,
            "proven": self.proven,
            "units": [
                {
                    "name": u.name,
                    "scope": u.scope,
                    "properties": u.properties,
                    "judged_against": str(u.root) if u.root else None,
                }
                for u in self.units
            ],
            "mismatches": list(self.mismatches),
            "errors": list(self.errors),
            "skipped": list(self.skipped),
        }


def systemd_unescape(text: str) -> str:
    """Pure ``systemd-escape --unescape``: ``-`` is ``/`` and ``\\xHH`` is the
    byte HH (no subprocess -- the instance parameter is data, ADR-0067 R1)."""
    out = bytearray()
    i = 0
    while i < len(text):
        hex_pair = text[i + 2 : i + 4]
        if text[i : i + 2] == "\\x" and re.fullmatch(r"[0-9a-fA-F]{2}", hex_pair):
            out.append(int(hex_pair, 16))
            i += 4
            continue
        out += b"/" if text[i] == "-" else text[i].encode()
        i += 1
    return out.decode("utf-8", errors="replace")


def instance_of(name: str) -> str:
    """The raw ``%i`` of a template-instance unit name, ``""`` for any other."""
    return name.split(".service")[0].partition("@")[2]


def read_tools_version(tools_root: Path, errors: list[str]) -> VersionRecord | None:
    """``<tools_root>/VERSION`` parsed, or None with the cause appended to
    ``errors`` -- never a silent pass (the tool tree is its own deploy record)."""
    path = tools_root / "VERSION"
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        errors.append(f"cannot read {path}: {exc}")
        return None
    record = parse_version(text)
    if record is None:
        errors.append(f"{path} is missing or not in the `key value` format")
    return record


def parse_version(text: str) -> VersionRecord | None:
    """Space-separated `key value` lines, as install-orchestrata.sh's printf
    writes them (D3 -- NOT key=value)."""
    fields: dict[str, str] = {}
    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        key, _, value = line.partition(" ")
        fields[key.strip()] = value.strip()
    commit = fields.get("commit", "")
    deployed_raw = fields.get("deployed", "")
    if not commit or not deployed_raw:
        return None
    deployed = parse_timestamp(deployed_raw)
    if deployed is None:
        return None
    return VersionRecord(
        commit=commit,
        ref=fields.get("ref", ""),
        repo=fields.get("repo", ""),
        deployed=deployed,
    )


def parse_timestamp(value: str) -> datetime | None:
    """Either an ISO-8601 stamp (VERSION) or systemd's default show format."""
    text = value.strip()
    if not text or text in ("n/a", "0"):
        return None
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        pass
    else:
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
    match = re.match(
        r"^\w{3}\s+(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2}:\d{2})\s+(\S+)$", text
    )
    if not match:
        return None
    date_part, time_part, _tz = match.groups()
    try:
        parsed = datetime.strptime(f"{date_part} {time_part}", "%Y-%m-%d %H:%M:%S")
    except ValueError:
        return None
    return parsed.replace(tzinfo=UTC)


def parse_show(output: str) -> dict[str, str]:
    """A property that repeats (two ExecStartPre= lines, EnvironmentFiles=
    per file) is CONCATENATED, not overwritten (confirmed, systemd 257)."""
    properties: dict[str, str] = {}
    for line in output.splitlines():
        if "=" not in line:
            continue
        key, _, value = line.partition("=")
        key, value = key.strip(), value.strip()
        properties[key] = f"{properties[key]} {value}" if key in properties else value
    return properties


def show(
    run: RunFn, scope: str, unit: str, properties: Sequence[str]
) -> tuple[dict[str, str], str | None]:
    flags = ["--user"] if scope == "user" else []
    result = run(["systemctl", *flags, "show", "-p", ",".join(properties), unit])
    if result.returncode != 0:
        problem = (result.stderr or result.stdout or "").strip()
        return {}, f"systemctl show {scope} {unit} failed: {problem}"
    return parse_show(result.stdout), None


def discover(
    run: RunFn, scope: str, globs: Sequence[str], pattern: re.Pattern[str] | None
) -> tuple[list[str], str | None]:
    """Every unit (any type) whose NAME matches one of `globs`.

    `--plain` is load-bearing: without it, systemctl prefixes a "● " marker
    on units in a "bad" state, shifting `line.split()[0]` off the unit name
    (discovered against this host's real, already-failed orchestrata units).
    """
    import fnmatch

    flags = ["--user"] if scope == "user" else []
    result = run(["systemctl", *flags, "list-units", "--all", "--no-legend", "--plain"])
    if result.returncode != 0:
        problem = (result.stderr or result.stdout or "").strip()
        return [], f"systemctl list-units ({scope}) failed: {problem}"
    names: list[str] = []
    for raw in result.stdout.splitlines():
        line = raw.strip()
        if not line:
            continue
        unit = line.split()[0]
        if not any(fnmatch.fnmatch(unit, glob) for glob in globs):
            continue
        if pattern is not None and not pattern.match(unit):
            continue
        names.append(unit)
    return names, None