scripts.check_env_subnet_scheme

Reference › scripts › check_env_subnet_scheme

The env-subnet-scheme gate, the mechanism for the backlog-413 ruling. It checks that each environment profile env/<name>.env obeys the scheme in which every environment has its own subnet octet. No module or test imports it; it takes no arguments and exits 2 if any is given.

main runs _scheme_self_check on the scheme table itself, then reads each profile named in the scheme and passes its text to check_profile, which returns the failures. A missing profile returns 2. Environments listed as pending migration are named in the summary. The output is a PASS line with the profile count, or an [error] line per violation followed by a FAIL line and exit 1. It is read-only.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Functions

Source

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

#!/usr/bin/env python3
"""Env-subnet-scheme gate — the mechanism for the backlog-413 ruling
(Mohamed, 2026-08-15): environment subnets are FORCED apart by a dedicated
environment octet so they can never be confused.

Ratified scheme (``plans/subnet-renumbering-design.md``): all environments
share the ``10.53.<env>.`` frame — first two octets identical, third octet
IS the environment:

    prod -> 10.53.1.0/24     test -> 10.53.2.0/24     dev -> 10.53.3.0/24

The old scheme (prod 10.92.53.x / test 10.93.53.x / dev 10.95.53.x) put the
discriminating digit mid-address with no dedicated meaning; one mistyped
digit aimed a test client at the prod kernel.

Scope: ``env/{dev,test,prod}.env`` — the authoritative endpoint layer
(RUL-003/no-infra-env-defaults forces code to read these, check_env_contract
forces the reading). Enforcing the authoritative layer is enforcing the
system; literal stragglers elsewhere are swept by the migration phases.

Ratchet: an environment whose fleet has not yet been cut over is listed in
``PENDING_MIGRATION`` and is pinned to its EXACT legacy subnet — any address
outside that legacy subnet fails, so the window cannot absorb drift. When a
phase lands, its env is removed from PENDING and the target scheme is
asserted unconditionally. An env on the old subnet that is NOT listed fails.

Exit codes: 0 pass, 1 violations, 2 invalid invocation/missing profile.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

# env name -> required address prefix under the ratified scheme.
SCHEME: dict[str, str] = {
    "prod": "10.53.1.",
    "test": "10.53.2.",
    "dev": "10.53.3.",
}

# Environments not yet cut over: env -> (exact legacy prefix, tracking note).
# Remove an entry in the SAME commit as that environment's cutover.
PENDING_MIGRATION: dict[str, tuple[str, str]] = {
    # EMPTY since 2026-08-16: all three environments migrated, the scheme is
    # asserted unconditionally. Migration history:
    # - dev  2026-08-15 (phase 2): orch-dev-br -> 10.53.3.1/24, containers
    #   .20/.21/.22, orch-no-dev ACL moved, probes green.
    # - test 2026-08-15 (phase 3): orch-test-br -> 10.53.2.1/24, containers
    #   .10-.13, orch-no-test ACL moved, app container re-provisioned,
    #   full authoritative suite green on the new fleet (33k+ tests).
    # - prod 2026-08-16 (phase 4): incusbr0 -> 10.53.1.1/24, 17 containers
    #   re-pinned (3 had in-container static networkd configs), UFW rules
    #   moved, orchgraph cluster coordination re-bootstrapped at the new
    #   addresses with replicas registered AS ASYNC, data verified intact
    #   (orchkg 254,560 / memgraph 768,462 nodes).
    # The mechanism stays: a future new environment lists itself here with
    # its legacy pin until its cutover phase lands.
}

_IPV4_RE = re.compile(r"\b(10\.\d{1,3}\.\d{1,3}\.\d{1,3})\b")


def _scheme_self_check() -> list[str]:
    """The scheme itself must obey the ruling: shared two-octet frame,
    unique env octet per environment."""
    failures: list[str] = []
    frames = {p.split(".")[0] + "." + p.split(".")[1] for p in SCHEME.values()}
    if len(frames) != 1:
        failures.append(
            f"scheme frame split: first two octets must be identical across "
            f"envs, got {sorted(frames)}"
        )
    env_octets = [p.split(".")[2] for p in SCHEME.values()]
    if len(set(env_octets)) != len(env_octets):
        failures.append(
            f"env octet collision: third octets must be unique per env, "
            f"got {env_octets}"
        )
    return failures


def check_profile(
    env: str,
    text: str,
    pending: dict[str, tuple[str, str]] | None = None,
) -> list[str]:
    """All 10.x literals in one profile obey the scheme (or the pinned
    legacy subnet while that env is listed as pending)."""
    if pending is None:
        pending = PENDING_MIGRATION
    failures: list[str] = []
    if env in pending:
        required, note = pending[env]
        label = f"legacy pin {required}* ({note})"
    else:
        required, label = SCHEME[env], f"scheme {SCHEME[env]}*"
    for lineno, line in enumerate(text.splitlines(), 1):
        for ip in _IPV4_RE.findall(line):
            if not ip.startswith(required):
                failures.append(f"env/{env}.env:{lineno}: {ip} outside {label}")
    return failures


def main(argv: list[str]) -> int:
    if argv:
        print(f"unknown arguments: {argv}", file=sys.stderr)
        return 2
    failures = _scheme_self_check()
    checked = 0
    for env in sorted(SCHEME):
        profile = REPO_ROOT / "env" / f"{env}.env"
        if not profile.is_file():
            print(f"FAIL: missing profile env/{env}.env")
            return 2
        failures.extend(check_profile(env, profile.read_text(encoding="utf-8")))
        checked += 1
    pending = ", ".join(sorted(PENDING_MIGRATION)) or "none"
    if failures:
        for f in failures:
            print(f"  [error] {f}")
        print(
            f"FAIL: {len(failures)} subnet-scheme violation(s) across "
            f"{checked} profiles (pending migration: {pending})"
        )
        return 1
    print(
        f"PASS: {checked} profiles obey the env-octet scheme "
        f"(pending migration: {pending})"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))