scripts.check_prod_tree_separation

Reference › scripts › check_prod_tree_separation

The prod/dev tree separation gate, the mechanism for ruling RUL-027: production systemd user units must not point at a development checkout. Four test modules import it: tests/scripts/test_check_prod_tree_separation.py, tests/scripts/test_check_prod_tree_separation_paths.py, tests/promotion/test_identity_check_path_directives.py and tests/promotion/test_identity_parsing.py. It relies on identity_core for unit parsing, which ties it to the promotion flow.

Invocation: main takes --unit-dir (default the user's systemd unit directory), --dev-tree (default the ORCHESTRATA_DEV_TREE variable, else this repository) and --glob (repeatable; the default globs cover the orchestrata units). A machine with no unit directory passes, with nothing to check.

find_violations and find_violations_with_reasons read each unit file (_examined_units), join continued lines (_logical_lines), pull candidate paths out of each directive value (_candidate_paths, _workspace_candidates) and test whether a candidate lies under the dev tree (_is_under_root) or under any workspace checkout (_is_workspace_path). Template units are skipped (_is_template_unit). The result is an OK line with the number of units examined, or each violation with its reason and exit 1.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_prod_tree_separation.py

#!/usr/bin/env python3
"""Prod/dev tree separation gate — the mechanism for ruling **RUL-027**.

"Separate dev code from prod" (user command, 2026-08-08).

Production must execute from its own checkout. Before RUL-027 all six
``orchestrata-*`` user units carried
``WorkingDirectory=/home/mohamed/workspace/orchestrata-ultimate-architecture``
together with ``Environment=ORCHESTRATA_ENV_PROFILE=prod``, so production ran
whatever the development tree happened to contain: every edit was a pending
deploy, and ``git checkout`` moved production silently.

This gate fails when any prod unit still points a path-bearing directive at the
development tree. It reads the INSTALLED units (``~/.config/systemd/user``),
not the repo's templates — a template can be correct while the installed copy
has drifted, and the installed copy is what actually runs.

It examines both ``orchestrata-*`` AND ``orchgraph-*`` services, plus every
``*.timer`` unit (a timer rarely carries a path directive itself, but it is
still counted as examined so its absence from the report is never mistaken
for "not checked").

Two rules, each with its own reason string in the report:

* ``--dev-tree``: any path-bearing directive resolving under the one dev tree
  this run was told about (RUL-027, the original rule).
* ``workspace``: any NON-template unit whose ``ExecStart*`` / ``ExecStop*`` /
  ``ExecReload`` / ``ExecCondition`` or ``WorkingDirectory`` names a path under
  ``/home/*/workspace/`` (ADR-0067 R1:
  a checkout is data, never the thing a plain unit executes from). Template
  (``*@.service``) and instance (``*@*.service``) units are exempt -- their
  checkout path is the ``%I`` runtime parameter, i.e. data.

Exit codes: 0 clean, 1 violations found, 2 could not inspect.
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from pathlib import Path

# PATH_DIRECTIVES and DEFAULT_GLOBS moved into orchestrata_ultimate.promotion
# .identity_core on 2026-09-23 (U-T6 amendment finding 4): identity.py needed
# them without requiring `scripts` to be importable at all, so the package is
# now their canonical home and this gate imports them back -- values and
# behaviour here are unchanged (see identity_core.py for the full rationale
# on each entry, e.g. why `Environment=`/`StandardOutput=` etc. are path
# directives too).
from orchestrata_ultimate.promotion.identity_core import DEFAULT_GLOBS, PATH_DIRECTIVES

# The workspace rule covers the directives that decide WHAT a unit executes
# and WHERE: ExecStart and its Pre/Post siblings, the stop/reload/condition
# hooks (ExecStop, ExecStopPost, ExecReload, ExecCondition), and
# WorkingDirectory. It does
# not widen to every PATH_DIRECTIVE (Environment=, StandardOutput=, ...): those
# stay governed by the --dev-tree rule only.
WORKSPACE_DIRECTIVES = frozenset(
    {
        "ExecStart",
        "ExecStartPre",
        "ExecStartPost",
        "ExecStop",
        "ExecStopPost",
        "ExecReload",
        "ExecCondition",
        "WorkingDirectory",
    }
)
# systemd.service(5) command prefixes: a path may be spelled `@/path name`,
# `-/path`, `:/path`, `+/path`, `!/path`, `|/path` (and stacked, `-@/path`).
_EXEC_PREFIXES = "@-:+!|"
REASON_DEV_TREE = "under the --dev-tree"
REASON_WORKSPACE = "under a /home/*/workspace/ checkout (non-template unit)"

DEFAULT_DEV_TREE = str(Path(__file__).resolve().parents[1])
DEFAULT_UNIT_DIR = Path.home() / ".config" / "systemd" / "user"


def _as_globs(glob: str | tuple[str, ...]) -> tuple[str, ...]:
    """Normalize the old single-pattern form to a tuple, unchanged otherwise."""
    return (glob,) if isinstance(glob, str) else tuple(glob)


def _examined_units(unit_dir: Path, glob: str | tuple[str, ...]) -> list[Path]:
    """Every unit matched by any of `glob`'s patterns, deduplicated and sorted."""
    seen: dict[Path, None] = {}
    for pattern in _as_globs(glob):
        for unit in sorted(unit_dir.glob(pattern)):
            seen.setdefault(unit, None)
    return sorted(seen)


def _candidate_paths(value: str) -> list[str]:
    """Path-like tokens embedded in a directive's raw value.

    A directive's value may be a bare path (``WorkingDirectory=``), a
    ``KEY=path`` pair (``Environment=PYTHONPATH=...``), a ``verb:path`` form
    (``StandardOutput=append:...``), an optional-marker-prefixed path
    (``EnvironmentFile=-...``), or a full command line with arguments
    (``ExecStart=<path> -m x``). Every colon/equals/whitespace-separated
    segment that itself looks like an absolute path is a candidate -- this
    is intentionally permissive (a non-path segment like ``-m`` or
    ``PYTHONPATH`` never starts with ``/`` and is simply not a candidate).
    """
    candidates: list[str] = []
    for word in value.split():
        for part in re.split(r"[:=]", word):
            part = part.lstrip("-+!")
            if part.startswith("/"):
                candidates.append(part)
    return candidates


def _workspace_candidates(value: str, home: Path | None = None) -> list[str]:
    """Path-like tokens for the workspace rule: `_candidate_paths` after
    undoing the ordinary ways a unit spells a checkout path.

    Surrounding single/double quotes are stripped from every token, the exec
    prefixes (``@ - : + ! |``) from the FIRST token (the executable), and the
    ``%h`` specifier (the manager's home directory) is expanded to `home`
    (default ``Path.home()``; the seam lets a caller state the home it means
    instead of depending on the host's). ``%u``/``%U`` are not expanded: they
    never stand in for a ``/home/<user>/workspace`` prefix on their own.
    """
    home_str = str(Path.home() if home is None else home)
    words: list[str] = []
    for index, word in enumerate(value.split()):
        word = word.strip("\"'")
        if index == 0:
            word = word.lstrip(_EXEC_PREFIXES).strip("\"'")
        words.append(word.replace("%h", home_str))
    return _candidate_paths(" ".join(words))


def _logical_lines(lines: list[str]) -> list[tuple[int, str]]:
    """(first physical line number, text) per logical line: physical lines
    ending in an UNESCAPED ``\\`` are joined with the next one (systemd's line
    continuation), so ``ExecStart=/usr/bin/env \\`` followed by the real path
    is one directive. A trailing ``\\\\`` (an escaped backslash) is not a
    continuation: only an odd count of trailing backslashes continues. A comment
    line outside a continuation is skipped; one INSIDE a continuation is skipped
    without ending it (systemd joins across it)."""
    logical: list[tuple[int, str]] = []
    start = 0
    buffer = ""
    for number, raw in enumerate(lines, start=1):
        line = raw.strip()
        if not buffer:
            if not line or line.startswith("#"):
                continue
            start = number
        elif line.startswith(("#", ";")):
            continue
        if (len(line) - len(line.rstrip("\\"))) % 2 == 1:
            buffer += line[:-1] + " "
            continue
        logical.append((start, (buffer + line).strip()))
        buffer = ""
    if buffer:
        logical.append((start, buffer.strip()))
    return logical


def _is_under_root(candidate: str, root: Path) -> bool:
    """True when `candidate`'s RESOLVED path components fall under `root`
    (itself already resolved) -- never a string-prefix check, so a sibling
    that merely shares a name prefix (``/opt/orchestrata-tools`` under
    ``/opt/orchestrata``), a ``..`` escape, or a symlinked route into root
    are each judged by what the path actually resolves to."""
    try:
        resolved = Path(candidate).resolve()
    except OSError:
        return False
    return resolved.is_relative_to(root)


def _is_template_unit(unit: Path) -> bool:
    """Template (``foo@.service``) and instance (``foo@bar.service``) units."""
    return "@" in unit.name


def _is_workspace_path(candidate: str) -> bool:
    """True when `candidate`, made absolute WITHOUT resolving symlinks, has the
    shape ``/home/<user>/workspace/...`` -- the unit file's text is the
    evidence, not whatever the path happens to point at today."""
    parts = Path(os.path.abspath(candidate)).parts
    return len(parts) >= 4 and parts[1] == "home" and parts[3] == "workspace"


def find_violations(
    unit_dir: Path,
    dev_tree: str,
    glob: str | tuple[str, ...] = DEFAULT_GLOBS,
    *,
    home: Path | None = None,
) -> list[tuple[Path, int, str]]:
    """Return (unit, line number, line) for every offending line, once each."""
    seen: dict[tuple[Path, int, str], None] = {}
    for unit, number, line, _reason in find_violations_with_reasons(
        unit_dir, dev_tree, glob, home=home
    ):
        seen.setdefault((unit, number, line), None)
    return list(seen)


def find_violations_with_reasons(
    unit_dir: Path,
    dev_tree: str,
    glob: str | tuple[str, ...] = DEFAULT_GLOBS,
    *,
    home: Path | None = None,
) -> list[tuple[Path, int, str, str]]:
    """Return (unit, line number, line, reason); a line that trips both rules
    appears once per reason. `home` is what ``%h`` expands to for the workspace
    rule (default ``Path.home()``)."""
    root = Path(dev_tree).resolve()
    violations: list[tuple[Path, int, str, str]] = []
    for unit in _examined_units(unit_dir, glob):
        try:
            lines = unit.read_text(encoding="utf-8").splitlines()
        except OSError:
            continue
        for number, line in _logical_lines(lines):
            directive, _, value = line.partition("=")
            # A systemd directive may repeat with `-` / `+` / `!` prefixes.
            directive = directive.strip().lstrip("-+!:@")
            if (
                directive not in PATH_DIRECTIVES
                and directive not in WORKSPACE_DIRECTIVES
            ):
                continue
            # The --dev-tree rule stays scoped to PATH_DIRECTIVES (its original
            # surface); the workspace rule alone widens to the other Exec* hooks.
            candidates = _candidate_paths(value)
            if directive in PATH_DIRECTIVES and any(
                _is_under_root(candidate, root) for candidate in candidates
            ):
                violations.append((unit, number, line, REASON_DEV_TREE))
            if (
                directive in WORKSPACE_DIRECTIVES
                and not _is_template_unit(unit)
                and any(
                    _is_workspace_path(candidate)
                    for candidate in _workspace_candidates(value, home)
                )
            ):
                violations.append((unit, number, line, REASON_WORKSPACE))
    return violations


def main(argv: list[str] | None = None) -> int:
    """Run the gate."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--unit-dir",
        default=str(DEFAULT_UNIT_DIR),
        help="systemd user unit directory (default: ~/.config/systemd/user)",
    )
    parser.add_argument(
        "--dev-tree",
        default=os.environ.get("ORCHESTRATA_DEV_TREE", DEFAULT_DEV_TREE),
        help="path that prod units must NOT reference (default: this repo)",
    )
    parser.add_argument(
        "--glob",
        action="append",
        default=None,
        help=("unit glob; repeatable (default: " + ", ".join(DEFAULT_GLOBS) + ")"),
    )
    args = parser.parse_args(argv)

    unit_dir = Path(args.unit_dir)
    globs: tuple[str, ...] = tuple(args.glob) if args.glob else DEFAULT_GLOBS
    if not unit_dir.is_dir():
        # No user units installed at all is not a violation — a machine that
        # does not run prod cannot have prod pointed at the wrong tree.
        print(
            f"[prod-separation] no unit dir at {unit_dir} — "
            "0 unit(s) examined, nothing to check"
        )
        return 0

    dev_tree = str(Path(args.dev_tree).resolve())
    units = _examined_units(unit_dir, globs)
    reasoned = find_violations_with_reasons(unit_dir, dev_tree, globs)
    violations = find_violations(unit_dir, dev_tree, globs)

    if not violations:
        print(
            f"[prod-separation] OK — RUL-027 satisfied: {len(units)} unit(s) "
            f"examined, none reference {dev_tree} or a /home/*/workspace/ checkout"
        )
        return 0

    has_dev_tree = any(reason == REASON_DEV_TREE for *_rest, reason in reasoned)
    has_workspace = any(reason == REASON_WORKSPACE for *_rest, reason in reasoned)
    # Name only the tree(s) actually hit: a workspace-only run never touched
    # --dev-tree, so the header must not claim it did.
    targets = []
    if has_dev_tree:
        targets.append(f"the DEVELOPMENT checkout (--dev-tree {dev_tree})")
    if has_workspace:
        targets.append("a /home/*/workspace/ checkout")
    print(
        f"[prod-separation] FAIL — RUL-027: {len(units)} unit(s) examined, "
        f"{len(violations)} directive(s) point production at " + " and ".join(targets),
        file=sys.stderr,
    )
    for unit, number, line, reason in reasoned:
        print(f"  {unit.name}:{number}: {line}  [{reason}]", file=sys.stderr)
    trailer = ["\nProduction must run from its own checkout (/opt/orchestrata)."]
    if has_dev_tree:
        trailer.append(
            "While a prod unit points here, every edit to this tree is a pending\n"
            "deploy and `git checkout` moves production silently."
        )
    if has_workspace:
        trailer.append(
            "A plain unit points at a /home/*/workspace/ checkout: `git checkout`\n"
            "there moves production silently. Serve a checkout as DATA through a\n"
            "template unit (%I), never as the thing a plain unit executes from."
        )
    print("\n".join(trailer), file=sys.stderr)
    return 1


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