numos/state.py

back to source

"""Render and parse the line-oriented numos.state format."""

from numos.canonical import content_hash


class StateError(Exception):
    """Raised when a state file is malformed or fails hash verification."""


def _csv(values):
    return ",".join(values) if values else "-"


def _uncsv(field):
    return [] if field == "-" else field.split(",")


def _opt(value):
    return value if value else "-"


def _unopt(field):
    return None if field == "-" else field


def render(state):
    """Serialize a state dict to canonical text with the C line stamped."""
    lines = ["V %d" % state["version"]]

    lines += sorted(
        "A %s %s %s %s %s" % (a["canonical"], _csv(a["aliases"]),
                              _opt(a["zig_triple"]), a["busybox_variant"],
                              _opt(a["artifact_sha256"]))
        for a in state["arch_targets"]
    )
    lines += sorted(
        "P %d %s %s %s" % (p["ordinal"], p["name"], p["on_failure"],
                           _csv(p["required_units"]))
        for p in state["boot_phases"]
    )
    lines += sorted(
        "U %s %s %s %d %d %s %s %s %s %s" % (
            u["name"], u["kind"], u["restart"], u["backoff_ms"], u["backoff_max_ms"],
            _csv(u["arch_mask"]), _opt(u["health_probe"]),
            _csv(u["requires"]), _csv(u["after"]), u["exec"])
        for u in state["units"]
    )
    lines += sorted(
        "X %s %d %d %s %s %s" % (h["name"], h["interval_s"], h["threshold"],
                                 h["local_action"], _opt(h["on_fire"]), h["probe"])
        for h in state["health"]
    )

    stamped = ["V %d" % state["version"], "C " + content_hash(lines)] + lines[1:]
    return "".join(ln + "\n" for ln in stamped)


def parse(text):
    """Parse canonical text back into a state dict. Ignores the C line."""
    state = {"version": None, "arch_targets": [], "boot_phases": [],
             "units": [], "health": []}
    for raw in text.split("\n"):
        if not raw:
            continue
        tag, _, rest = raw.partition(" ")
        if tag == "V":
            state["version"] = int(rest)
        elif tag == "C":
            continue
        elif tag == "A":
            f = rest.split(" ", 4)
            state["arch_targets"].append({
                "canonical": f[0], "aliases": _uncsv(f[1]),
                "zig_triple": _unopt(f[2]), "busybox_variant": f[3],
                "artifact_sha256": _unopt(f[4])})
        elif tag == "P":
            f = rest.split(" ", 3)
            state["boot_phases"].append({
                "ordinal": int(f[0]), "name": f[1], "on_failure": f[2],
                "required_units": _uncsv(f[3])})
        elif tag == "U":
            f = rest.split(" ", 9)
            state["units"].append({
                "name": f[0], "kind": f[1], "restart": f[2],
                "backoff_ms": int(f[3]), "backoff_max_ms": int(f[4]),
                "arch_mask": _uncsv(f[5]), "health_probe": _unopt(f[6]),
                "requires": _uncsv(f[7]), "after": _uncsv(f[8]), "exec": f[9]})
        elif tag == "X":
            f = rest.split(" ", 5)
            state["health"].append({
                "name": f[0], "interval_s": int(f[1]), "threshold": int(f[2]),
                "local_action": f[3], "on_fire": _unopt(f[4]), "probe": f[5]})
        else:
            raise StateError("unknown record tag: %r" % tag)
    if state["version"] is None:
        raise StateError("missing V record")
    return state


HEX = "0123456789abcdef"


def _file_lines(text):
    """Split text the way the shell verifier's `grep` sees the file.

    Exactly one trailing newline terminates the last line; anything past it
    is a real (blank) line that grep would emit and hash. Blank lines are
    NOT dropped: boot/bootstrap.sh keeps them in the hashed body, and a
    build machine must not certify a file the target would refuse.
    """
    lines = text.split("\n")
    if lines and lines[-1] == "":
        lines.pop()
    return lines


def verify(text):
    """Raise StateError unless the stamped C hash matches the body."""
    lines = _file_lines(text)
    stamped = [ln for ln in lines if ln.startswith("C ")]
    if len(stamped) != 1:
        raise StateError("expected exactly one C record, found %d" % len(stamped))
    want = stamped[0][2:]
    # The hash is the whole remainder of the line, and it must be exactly
    # 64 lowercase hex characters -- boot/bootstrap.sh applies the identical
    # rule, so `C <valid-hash> ANYTHING` is rejected by both verifiers.
    if len(want) != 64 or any(c not in HEX for c in want):
        raise StateError("C record is not a 64-character lowercase hex sha256: %r" % want)
    got = content_hash(lines)
    if want != got:
        raise StateError("content hash mismatch: stamped %s computed %s" % (want, got))