orchestrata_ultimate.promotion.identity_checks

Reference › orchestrata_ultimate › promotion › identity_checks

The static checks on a unit's configuration for the identity verifier. They ask whether anything a unit runs or reads points outside the deploy root and the system prefixes, which is how a production unit running from a developer checkout is caught.

check_c0 collects path-like tokens from ExecStart, ExecStartPre, ExecStartPost, WorkingDirectory, the environment files, Environment and the Standard* outputs, and records a mismatch for each one that resolves_safely rejects: it must be under the root, under /lib*, or under a system prefix. For a oneshot service the program path and path= field are held to a stricter rule (strictly under the root, no system-prefix exemption), because oneshots skip the process checks and this is their only identity proof. Template instances may also reference their own instance parameter as data. check_path_directives reads the unit's fragment file and every drop-in file and records a mismatch when a path directive references a /home/<user>/workspace/ checkout; continuation lines inherit the directive that started them, and unreadable files are errors.

The parsing helpers are extract_argv, extract_argv0 and extract_path (from systemd's rendered ExecStart value), normalize_directive_value, and environment_file_paths. tool_tree_root decides whether a template instance is judged against the tools tree instead of the product root: only when its program lives under that tree (ADR-0067). URL and Environment handling is in identity_urls.

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

"""ExecStart/path extraction, judging and the c0/path-directive checks 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 ``ExecStart*`` argv/`path=` extraction helpers, the
directive-value normalizer, the ROOT/system-prefix judge, and the two check
functions that are their only callers (``check_c0``, ``check_path_directives``)
live here. ``identity_core.py`` imports every public name straight back, so
the split is invisible to every importer (see that module's own docstring).
"""

from __future__ import annotations

import os
import re
from collections.abc import Mapping, Sequence
from pathlib import Path

from orchestrata_ultimate.promotion.identity_scope import (
    SYSTEM_PREFIXES,
    TOOLS_ROOT,
    IdentityReport,
    UnitFacts,
    instance_of,
)
from orchestrata_ultimate.promotion.identity_urls import (
    _path_like_segments,
    environment_path_values,
    standard_io_path,
)

#: `argv[]=` is terminated by the NEXT known field in the fixed record
#: layout systemd emits (`{ path=... ; argv[]=... ; ignore_errors=... ;
#: start_time=... ; ... }`) -- or, for an abbreviated `{ ... }` block that
#: omits every field after argv[] (used throughout this file's own tests),
#: by the block's own closing `}`. Never a bare `;`: a real argv commonly
#: contains one of its own (`-c "import time; time.sleep(20)"`), and a
#: naive `;`-split silently truncates every argument after it (amendment
#: finding 2, confirmed on a real systemd 257 `systemctl show`).
_ARGV_BLOCK = re.compile(r"argv\[\]=(.*?)\s*;\s*(?:ignore_errors=|\})", re.DOTALL)

#: Same span as `_ARGV_BLOCK`, but captured in three groups (the `argv[]=`
#: marker, the argv content, the terminator) so the content group alone can
#: be blanked out -- used by `extract_path` to keep an argv's own embedded
#: `; path=` text from ever reaching `_PATH_FIELD` (see `extract_path`).
_ARGV_BLOCK_SPAN = re.compile(
    r"(argv\[\]=)(.*?)(\s*;\s*(?:ignore_errors=|\}))", re.DOTALL
)


def extract_argv(exec_value: str) -> list[str]:
    """Every argv[] token across ALL `{ ... }` blocks in a resolved
    ExecStart*/ExecStartPre*/ExecStartPost* value -- one block per directive
    instance, so every argument of every invocation is checked."""
    argvs: list[str] = []
    for block in _ARGV_BLOCK.finditer(exec_value):
        argvs.extend(block.group(1).split())
    if argvs:
        return argvs
    match = re.search(r"path=(\S+)", exec_value)
    if match:
        return [match.group(1)]
    stripped = exec_value.strip()
    return stripped.split() if stripped else []


def extract_argv0(exec_value: str) -> str:
    argv = extract_argv(exec_value)
    return argv[0] if argv else ""


#: Anchored to the record's OWN field boundary -- a `path=` field is always
#: the first thing after a block's opening `{` or a `; ` field separator in
#: systemd's fixed layout (`{ path=... ; argv[]=... ; ... }`). Unanchored
#: (`path=(\S+)`, attempt-5), this matched `path=` as a SUBSTRING of an
#: ordinary argv option -- `--data-path=/var/lib/x` or `classpath=/usr/...`
#: -- and judged THAT as if it were the record's real `path=` field,
#: failing a correct prod oneshot forever (attempt-6 finding 2).
_PATH_FIELD = re.compile(r"[{;]\s*path=(\S+)")


def extract_path(exec_value: str) -> list[str]:
    """Every `path=` field across ALL `{ ... }` blocks in a resolved
    ExecStart*/ExecStartPre*/ExecStartPost* value -- the executable systemd
    ACTUALLY runs, which an `@`-prefixed ExecStart (``@/real/path argv0 ...``)
    deliberately makes differ from argv[0], the fake argv0 the process merely
    SEES (amendment attempt-5 finding 3).

    Searched ONLY over the part of each block BEFORE `argv[]=` -- never
    inside the argv value itself. An argv commonly carries its own literal
    `; path=` substring (`-c "import os; path=/usr/lib/x"`), which the
    anchor `[{;]\\s*path=` -- built to recognize the record's OWN field
    separator -- cannot distinguish from a genuine field boundary once it is
    embedded verbatim in an argv string; that produced a false `path=`
    mismatch on an otherwise-correct prod oneshot (attempt-7 finding). Each
    block's argv[] content is blanked (its own field boundaries kept) before
    `_PATH_FIELD` ever sees it."""
    blanked = _ARGV_BLOCK_SPAN.sub(r"\1\3", exec_value)
    return _PATH_FIELD.findall(blanked)


def normalize_directive_value(value: str) -> str:
    """Strip systemd's optional-directive markers before judging a path.

    `show`'s RESOLVED value renders an optional directive's marker as a
    leading `!` (verified live: ``WorkingDirectory=-/home/x`` in a unit file
    -> ``WorkingDirectory=!/home/x`` in ``systemctl show`` -- a bare
    ``.startswith("/")`` check is false and the value was silently never
    judged, amendment attempt-5 finding 1). A leading `-` is kept too, in
    case a caller ever feeds an UN-resolved unit-file value instead. A bare
    `~`/`~user` HOME shorthand (valid systemd WorkingDirectory syntax) is
    expanded so it can be judged as the real path it resolves to."""
    stripped = value.lstrip("!-")
    if stripped == "~" or stripped.startswith("~/"):
        stripped = os.path.expanduser(stripped)
    return stripped


def environment_file_paths(value: str) -> list[str]:
    """Paths out of `EnvironmentFiles=`: one+ `<path> (ignore_errors=no)` groups."""
    return re.findall(r"(\S+)\s*\(ignore_errors=\w+\)", value)


def resolves_safely(token: str, root: Path) -> bool:
    #: Belt-and-suspenders (attempt-7): normalize ANY doubled/tripled leading
    #: slash spelling to a single slash before comparing prefixes, so a
    #: doubled-slash remainder from an empty-authority URL -- or from any
    #: other source -- can never dodge a legitimate ROOT/system-prefix match
    #: by literal-string accident.
    token = re.sub(r"^/{2,}", "/", token)
    root_str = str(root)
    if token == root_str or token.startswith(f"{root_str}/"):
        return True
    if token.startswith("/lib"):  # /lib, /lib64, /lib32, /libexec
        return True
    return any(
        token == prefix or token.startswith(f"{prefix}/") for prefix in SYSTEM_PREFIXES
    )


def _strictly_under_root(value: str, root: Path) -> bool:
    """No system-prefix exemption -- ROOT itself or nothing."""
    root_str = str(root)
    return value == root_str or value.startswith(f"{root_str}/")


def tool_tree_root(
    name: str, properties: Mapping[str, str], tools_root: Path = TOOLS_ROOT
) -> Path | None:
    """The root a template INSTANCE is judged against (ADR-0067 R1/P3), else None.

    An instance (``@`` before ``.service``, non-empty ``%i``) whose ExecStart
    argv0 lives under ``tools_root`` executes deployed tool-tree code, so its
    code root is that tree. Any other unit -- including an instance whose argv0
    points elsewhere, which stays the defect -- keeps the product root. Pure
    string work: no filesystem access."""
    if not instance_of(name):
        return None
    argv0 = re.sub(r"^/{2,}", "/", extract_argv0(properties.get("ExecStart", "")))
    root_str = str(tools_root)
    if argv0 == root_str or argv0.startswith(f"{root_str}/"):
        return tools_root
    return None


def check_c0(
    unit: UnitFacts,
    root: Path,
    report: IdentityReport,
    *,
    data_prefixes: Sequence[str] = (),
) -> None:
    """Every path-like token of ExecStart*/WorkingDirectory/EnvironmentFiles/
    Environment/StandardOutput/StandardError/StandardInput must resolve
    under ROOT or a system prefix (D2; attempt-7b residual finding 3 closes
    the Standard* gap -- see identity_urls.py's `standard_io_path`).

    ExecStart's own argv0 AND its `path=` field are the one exception, and
    they are STRICTER, not looser: for a ``Type=oneshot`` unit, D4 skips
    every /proc check entirely, so c0 judging them here is the ONLY proof of
    identity it has -- the /usr system-prefix whitelist that governs
    everything else in this function does not apply to either (amendment
    finding 3, extended attempt-5 finding 3: an `@`-prefixed ExecStart makes
    systemd run `path=`, not argv[0] -- both must independently pass)."""
    props = unit.properties
    tokens: list[tuple[str, str]] = []
    is_oneshot = props.get("Type", "") == "oneshot"

    exec_start_argv = extract_argv(props.get("ExecStart", ""))
    if exec_start_argv:
        argv0, *rest = exec_start_argv
        if is_oneshot:
            if not _strictly_under_root(argv0, root):
                report.mismatches.append(
                    f"{unit.name}: ExecStart argv0: {argv0} does not resolve "
                    f"under ROOT {root} (no system-prefix exemption for a "
                    "oneshot's argv0 -- it is the only identity proof it has)"
                )
        else:
            tokens += [("ExecStart", seg) for seg in _path_like_segments(argv0)]
        tokens += [
            ("ExecStart", seg) for tok in rest for seg in _path_like_segments(tok)
        ]

    for path_value in extract_path(props.get("ExecStart", "")):
        if is_oneshot:
            if not _strictly_under_root(path_value, root):
                report.mismatches.append(
                    f"{unit.name}: ExecStart path=: {path_value} does not "
                    f"resolve under ROOT {root} (no system-prefix exemption "
                    "for a oneshot's path= -- it is the executable systemd "
                    "actually runs, which an @-prefixed ExecStart can make "
                    "differ from argv0)"
                )
        else:
            tokens += [
                ("ExecStart path=", seg) for seg in _path_like_segments(path_value)
            ]

    for prop in ("ExecStartPre", "ExecStartPost"):
        tokens += [
            (prop, seg)
            for tok in extract_argv(props.get(prop, ""))
            for seg in _path_like_segments(tok)
        ]

    wd = normalize_directive_value(props.get("WorkingDirectory", ""))
    if wd.startswith("/"):
        tokens.append(("WorkingDirectory", wd))
    tokens += [
        ("EnvironmentFiles", tok)
        for tok in environment_file_paths(props.get("EnvironmentFiles", ""))
    ]
    tokens += [
        ("Environment", tok)
        for tok in environment_path_values(props.get("Environment", ""))
    ]
    # Attempt-7b residual finding 3: StandardOutput=/StandardError=/
    # StandardInput= can redirect through a path into the dev tree without
    # touching WorkingDirectory or any Exec* directive at all -- only
    # `file:`/`append:`/`truncate:` carry one (see identity_urls.py's
    # `standard_io_path` docstring for the opaque-vocabulary rule).
    for prop in ("StandardOutput", "StandardError", "StandardInput"):
        path = standard_io_path(props.get(prop, ""))
        if path:
            tokens.append((prop, path))
    for source, token in tokens:
        if any(resolves_safely(token, Path(data)) for data in data_prefixes):
            continue  # the instance parameter is data (ADR-0067 R1)
        if not resolves_safely(token, root):
            report.mismatches.append(
                f"{unit.name}: {source}: {token} resolves outside ROOT and system prefixes"
            )


def check_path_directives(
    unit: UnitFacts, path_directives: Sequence[str], report: IdentityReport
) -> None:
    """Scan the fragment file AND every DropInPaths file for a PATH_DIRECTIVES
    line referencing a dev-checkout (/home/*/workspace/).

    A `\\`-continued directive's continuation lines carry no `key=` of their
    own (only the FIRST line of a multi-line value does), so they must
    inherit the directive that started them rather than being independently
    re-classified line by line -- amendment finding 3: the previous version
    reclassified every line on its own and silently skipped every
    continuation line whose bare text (e.g. `    --repo=/home/.../workspace`)
    did not itself look like a directive assignment."""
    paths = [p for p in (unit.properties.get("FragmentPath", ""),) if p]
    paths += unit.properties.get("DropInPaths", "").split()
    for path_str in paths:
        try:
            text = Path(path_str).read_text(encoding="utf-8")
        except OSError as exc:
            report.errors.append(f"cannot read {path_str} for {unit.name}: {exc}")
            continue
        current_directive: str | None = None
        continuing = False
        for raw in text.splitlines():
            stripped = raw.strip()
            if not continuing:
                if not stripped or stripped.startswith("#") or "=" not in stripped:
                    current_directive = None
                    continue
                key = stripped.split("=", 1)[0].strip().lstrip("-+!:@")
                current_directive = key if key in path_directives else None
            continuing = stripped.endswith("\\")
            line = stripped[:-1] if continuing else stripped
            if current_directive is not None and re.search(
                r"/home/[^/]+/workspace/", line
            ):
                report.mismatches.append(
                    f"{unit.name}: path-directive {current_directive} ({path_str}): "
                    f"references a workspace checkout: {line}"
                )