numos/distro.py
back to source
# SPDX-License-Identifier: MIT
"""Graph-resolved self-init distro profiles.
A DistroProfile is the composition of:
- tool floor (multi-call binary installed as numos-floor, not as "busybox")
- required applets
- BootPhase + OSUnit + HealthPredicate set for that profile
"self-init" is the offline bare-metal/QEMU profile that completes the phase
walk without eth0, ntpd, or an external busybox brand. The floor *implementation*
for published arches is still a multi-call static binary resolved from the graph
(URL + sha256); the image layout and unit graph no longer treat busybox as the
product identity.
Fleet profile keeps the original infra set (net/clock/join) for topology nodes.
"""
from __future__ import annotations
from numos import seed
# Applets the shell floor needs. Graph-exported so packers do not invent lists.
FLOOR_APPLETS = [
"sh", "ash", "grep", "cut", "head", "sort", "sed", "tr",
"mount", "umount", "sha256sum", "echo", "cat", "ls", "mkdir", "rm",
"sleep", "printf", "uname", "od", "mv", "date", "dirname", "basename",
"readlink", "wc", "test", "[", "true", "false", "kill", "wget",
"timeout", "ln", "chmod", "cp",
]
def _floor(arch, url, sha256=None, license_id="GPL-2.0-only",
source_offer="https://busybox.net/downloads/"):
"""Declare a floor provider for one arch.
The binary may be built from busybox or another multi-call source; in the
image it is always installed as numos-floor. source_offer documents
corresponding-source obligations when the implementation is GPL.
"""
return {
"arch": arch,
"install_name": "numos-floor",
"url": url,
"sha256": sha256,
"license": license_id,
"source_offer": source_offer,
"applets": list(FLOOR_APPLETS),
}
# Pinned floor artifacts (implementation binary). x86_64 sha256 measured 2026-08-09.
FLOOR_PROVIDERS = {
"x86_64": _floor(
"x86_64",
"https://busybox.net/downloads/binaries/1.35.0-x86_64-linux-musl/busybox",
sha256="6e123e7f3202a8c1e9b1f94d8941580a25135382b99e8d3e34fb858bba311348",
),
"x86": _floor(
"x86",
"https://busybox.net/downloads/binaries/1.35.0-i686-linux-musl/busybox",
sha256=None, # pin when measured for an official release
),
}
def _kernel(
arch,
url,
*,
kind="deb",
sha256=None,
install_name="vmlinuz",
extract_glob="boot/vmlinuz-*",
license_id="GPL-2.0-only",
source_offer=None,
notes=None,
):
"""Declare a third-party kernel provider for one arch (graph META).
NumericalOS does not ship a kernel build. KernelProvider records are the
M/G/S product surface for *which* kernel bytes ISO/QEMU paths may fetch,
with optional sha256 pin and extract rules for .deb packages.
"""
return {
"arch": arch,
"kind": kind, # deb | raw
"install_name": install_name,
"url": url,
"sha256": sha256,
"extract_glob": extract_glob,
"license": license_id,
"source_offer": source_offer or url,
"product": "third-party-kernel",
"notes": notes
or [
"Not built by NumericalOS; graph-pinned for reproducible ISO/QEMU.",
"ASEC residual: kernel integrity beyond sha256 of package is Tarski "
"(TPM/Secure Boot not Spec 1).",
],
}
# Kernel packages used for hybrid ISO (GENESIS media path). x86_64 pin matches
# observed QEMU boots (linux-image-6.1.0-50-amd64). Package .deb sha256 may be
# null until re-measured; packers still verify when supplied.
KERNEL_PROVIDERS = {
"x86_64": _kernel(
"x86_64",
"https://deb.debian.org/debian/pool/main/l/linux-signed-amd64/"
"linux-image-6.1.0-50-amd64_6.1.176-1_amd64.deb",
kind="deb",
sha256=None,
extract_glob="boot/vmlinuz-*",
source_offer="https://packages.debian.org/bookworm/linux-image-amd64",
),
}
def _unit(name, exec_, kind="oneshot", restart="never", requires=None, after=None,
health_probe=None):
return {
"name": name,
"kind": kind,
"restart": restart,
"backoff_ms": 0 if restart == "never" else 100,
"backoff_max_ms": 0 if restart == "never" else 8000,
"arch_mask": [],
"health_probe": health_probe,
"requires": requires or [],
"after": after or [],
"exec": exec_,
}
# Idempotent mounts: succeed if already present (initrd/kernel may pre-mount).
SELF_INIT_UNITS = [
_unit(
"mount-proc",
"sh -c 'test -e /proc/self || mount -t proc proc /proc'",
),
_unit(
"mount-sys",
"sh -c 'test -e /sys/kernel || mount -t sysfs sys /sys'",
requires=["mount-proc"],
),
_unit(
"mount-dev",
"sh -c 'test -e /dev/null || mount -t devtmpfs dev /dev'",
requires=["mount-proc"],
),
_unit(
"identity",
"numctl identity-init",
requires=["mount-proc"],
),
# Hold PID 1 steady state: a longrun that does not exit. Replaces join/net
# for the self-init profile so the phase walk can finish with degraded=0
# on a bare emulator with no eth0 and no ntpd.
_unit(
"hold",
"sh -c 'while true; do sleep 3600; done'",
kind="longrun",
restart="always",
requires=["identity"],
health_probe="hold-alive",
),
]
SELF_INIT_PHASES = [
{
"ordinal": 10,
"name": "mount",
"on_failure": "halt",
"required_units": ["mount-proc", "mount-sys", "mount-dev"],
},
{
"ordinal": 20,
"name": "identity",
"on_failure": "halt",
"required_units": ["identity"],
},
{
"ordinal": 30,
"name": "complete",
"on_failure": "halt",
"required_units": ["hold"],
},
]
SELF_INIT_HEALTH = [
{
"name": "hold-alive",
"interval_s": 15,
"threshold": 3,
"local_action": "degrade-node",
"on_fire": None,
"probe": (
"kill -0 $(cat ${NUMOS_RUNDIR:-/run/numos}/units/hold.pid "
"2>/dev/null) 2>/dev/null"
),
},
]
DISTRO_PROFILES = {
"self-init": {
"name": "self-init",
"title": "Graph-resolved self-init distro",
"description": (
"Offline bare init: mounts, identity, hold longrun. Completes the "
"phase walk without network or ntpd. Tool floor is graph-resolved "
"numos-floor (not busybox product layout)."
),
"include_op_units": False,
"units": SELF_INIT_UNITS,
"boot_phases": SELF_INIT_PHASES,
"health": SELF_INIT_HEALTH,
"floor_providers": FLOOR_PROVIDERS,
"kernel_providers": KERNEL_PROVIDERS,
},
"fleet": {
"name": "fleet",
"title": "Fleet / topology node",
"description": (
"Original infra set: mounts, net, clock, identity, join + optional "
"op catalogue. Expect degrade on bare QEMU without eth0/ntpd."
),
"include_op_units": True,
"units": seed.INFRA_UNITS,
"boot_phases": seed.BOOT_PHASES,
"health": seed.HEALTH_PREDICATES,
"floor_providers": FLOOR_PROVIDERS,
"kernel_providers": KERNEL_PROVIDERS,
},
}
def get_profile(name="self-init"):
if name not in DISTRO_PROFILES:
raise KeyError(
"unknown distro profile %r; known: %s"
% (name, ", ".join(sorted(DISTRO_PROFILES)))
)
return DISTRO_PROFILES[name]
def _normalize_arch(arch):
if arch in ("amd64",):
return "x86_64"
if arch in ("i686", "i386"):
return "x86"
return arch
def resolve_floor(arch, profile_name="self-init"):
"""Return floor provider dict for arch, or raise KeyError."""
profile = get_profile(profile_name)
providers = profile["floor_providers"]
arch = _normalize_arch(arch)
if arch not in providers:
raise KeyError(
"no floor provider for arch=%s profile=%s; declare one in "
"numos.distro.FLOOR_PROVIDERS or pass an explicit floor URL"
% (arch, profile_name)
)
return dict(providers[arch])
def resolve_kernel(arch, profile_name="self-init"):
"""Return kernel provider dict for arch, or raise KeyError."""
profile = get_profile(profile_name)
providers = profile.get("kernel_providers") or {}
arch = _normalize_arch(arch)
if arch not in providers:
raise KeyError(
"no kernel provider for arch=%s profile=%s; declare one in "
"numos.distro.KERNEL_PROVIDERS or pass KERNEL_URL"
% (arch, profile_name)
)
return dict(providers[arch])
def kernel_manifest(arch, profile_name="self-init"):
"""JSON-serializable kernel plan for ISO packers (graph META)."""
k = resolve_kernel(arch, profile_name)
profile = get_profile(profile_name)
return {
"profile": profile["name"],
"title": profile["title"],
"arch": k["arch"],
"kind": k["kind"],
"install_name": k["install_name"],
"url": k["url"],
"sha256": k["sha256"],
"extract_glob": k["extract_glob"],
"license": k["license"],
"source_offer": k["source_offer"],
"product": k["product"],
"notes": list(k["notes"]),
"asec": {
"layer": "META+GENESIS",
"role": "media kernel pin for ISO/QEMU, not PID-1 userspace",
"residuals": [
"kernel package sha256 may be unpinned until measured",
"Secure Boot / module signature chain not Spec 1",
"non-x86_64 providers absent until declared",
],
},
}
def floor_manifest(arch, profile_name="self-init"):
"""JSON-serializable floor + applet plan for packers (CF build, skills)."""
floor = resolve_floor(arch, profile_name)
profile = get_profile(profile_name)
return {
"profile": profile["name"],
"title": profile["title"],
"arch": floor["arch"],
"install_name": floor["install_name"],
"url": floor["url"],
"sha256": floor["sha256"],
"license": floor["license"],
"source_offer": floor["source_offer"],
"applets": floor["applets"],
"notes": [
"Install binary as /bin/%s; symlink applets to it."
% floor["install_name"],
"Do not brand the image as busybox; the graph product is numos-floor.",
"Corresponding source obligations follow the floor license field.",
],
}
def profile_summary(profile_name="self-init"):
"""Compact profile metadata for distro.profile.json and site tooling."""
p = get_profile(profile_name)
return {
"profile": p["name"],
"title": p["title"],
"description": p["description"],
"include_op_units": bool(p.get("include_op_units")),
"phase_names": [ph["name"] for ph in p["boot_phases"]],
"unit_names": sorted(u["name"] for u in p["units"]),
"health_names": sorted(h["name"] for h in p["health"]),
"floor_arches": sorted(p["floor_providers"].keys()),
"kernel_arches": sorted((p.get("kernel_providers") or {}).keys()),
}
def profile_graph(profile_name="self-init"):
"""Return a pure graph view: nodes + edges from the distro profile.
Layers (M/G/S/MGS analogy for the export product, not live Intrikata):
GENESIS — BootPhase chain (ordinal order)
MGS — OSUnit nodes + requires/after edges
SHADOW — HealthPredicate nodes + unit→probe edges
META — profile identity + floor providers
"""
p = get_profile(profile_name)
nodes = []
edges = []
nodes.append({
"id": "profile:%s" % p["name"],
"kind": "DistroProfile",
"layer": "META",
"label": p["title"],
})
for arch, floor in sorted(p["floor_providers"].items()):
nid = "floor:%s" % arch
nodes.append({
"id": nid,
"kind": "FloorProvider",
"layer": "META",
"label": "%s (%s)" % (floor["install_name"], arch),
"arch": arch,
"has_sha256": bool(floor.get("sha256")),
})
edges.append({
"from": "profile:%s" % p["name"],
"to": nid,
"rel": "floor_for",
})
for arch, kern in sorted((p.get("kernel_providers") or {}).items()):
nid = "kernel:%s" % arch
nodes.append({
"id": nid,
"kind": "KernelProvider",
"layer": "META",
"label": "kernel %s (%s)" % (arch, kern.get("kind", "?")),
"arch": arch,
"has_sha256": bool(kern.get("sha256")),
"product": kern.get("product"),
})
edges.append({
"from": "profile:%s" % p["name"],
"to": nid,
"rel": "kernel_for",
})
phase_ids = []
sorted_phases = sorted(p["boot_phases"], key=lambda x: x["ordinal"])
for ph in sorted_phases:
pid = "phase:%s" % ph["name"]
phase_ids.append(pid)
nodes.append({
"id": pid,
"kind": "BootPhase",
"layer": "GENESIS",
"label": "P%d %s" % (ph["ordinal"], ph["name"]),
"ordinal": ph["ordinal"],
"on_failure": ph["on_failure"],
})
for uname in ph.get("required_units") or []:
edges.append({
"from": pid,
"to": "unit:%s" % uname,
"rel": "requires_unit",
})
for a, b in zip(phase_ids, phase_ids[1:]):
edges.append({"from": a, "to": b, "rel": "next_phase"})
# GENESIS media (ISO) depends on META kernel pin when providers exist.
if sorted_phases and (p.get("kernel_providers") or {}):
first_phase = "phase:%s" % sorted_phases[0]["name"]
for arch in sorted(p["kernel_providers"].keys()):
edges.append({
"from": "kernel:%s" % arch,
"to": first_phase,
"rel": "enables_media",
})
for u in p["units"]:
uid = "unit:%s" % u["name"]
nodes.append({
"id": uid,
"kind": "OSUnit",
"layer": "MGS",
"label": u["name"],
"unit_kind": u["kind"],
"restart": u["restart"],
})
for dep in u.get("requires") or []:
edges.append({
"from": uid,
"to": "unit:%s" % dep,
"rel": "requires",
})
for dep in u.get("after") or []:
edges.append({
"from": uid,
"to": "unit:%s" % dep,
"rel": "after",
})
if u.get("health_probe"):
edges.append({
"from": uid,
"to": "health:%s" % u["health_probe"],
"rel": "health_probe",
})
for h in p["health"]:
hid = "health:%s" % h["name"]
nodes.append({
"id": hid,
"kind": "HealthPredicate",
"layer": "SHADOW",
"label": h["name"],
"local_action": h.get("local_action"),
"interval_s": h.get("interval_s"),
"threshold": h.get("threshold"),
})
return {
"profile": p["name"],
"title": p["title"],
"description": p["description"],
"nodes": nodes,
"edges": edges,
"counts": {
"nodes": len(nodes),
"edges": len(edges),
"phases": len(p["boot_phases"]),
"units": len(p["units"]),
"health": len(p["health"]),
},
}
def mermaid_for_profile(profile_name="self-init"):
"""Mermaid flowchart of phases + units (requires edges only for clarity)."""
p = get_profile(profile_name)
lines = [
"flowchart TD",
" %% NumericalOS distro profile: " + p["name"],
" %% Pure graph product - no unit files on the target",
]
for ph in sorted(p["boot_phases"], key=lambda x: x["ordinal"]):
pid = "PH_%s" % ph["name"].replace("-", "_")
lines.append(
' %s["P%d %s\\n(on_failure=%s)"]'
% (pid, ph["ordinal"], ph["name"], ph["on_failure"])
)
phases = sorted(p["boot_phases"], key=lambda x: x["ordinal"])
for a, b in zip(phases, phases[1:]):
lines.append(
" PH_%s --> PH_%s"
% (a["name"].replace("-", "_"), b["name"].replace("-", "_"))
)
for u in p["units"]:
uid = "U_%s" % u["name"].replace("-", "_")
kind = u["kind"]
lines.append(' %s("%s\\n%s")' % (uid, u["name"], kind))
for dep in u.get("requires") or []:
lines.append(
" U_%s --> U_%s"
% (u["name"].replace("-", "_"), dep.replace("-", "_"))
)
if u.get("health_probe"):
hid = "X_%s" % u["health_probe"].replace("-", "_")
lines.append(
' %s{{"%s"}}' % (hid, u["health_probe"])
)
lines.append(" %s -.-> %s" % (uid, hid))
for ph in p["boot_phases"]:
pid = "PH_%s" % ph["name"].replace("-", "_")
for uname in ph.get("required_units") or []:
lines.append(
" %s -.-> U_%s" % (pid, uname.replace("-", "_"))
)
return "\n".join(lines) + "\n"
def main(argv=None):
"""CLI: python -m numos.distro [--profile self-init] [summary|graph|mermaid|list]"""
import argparse
import json
import sys
parser = argparse.ArgumentParser(
prog="python -m numos.distro",
description="Inspect NumericalOS graph-resolved distro profiles",
)
parser.add_argument(
"--profile",
default="self-init",
choices=sorted(DISTRO_PROFILES),
)
parser.add_argument(
"command",
nargs="?",
default="summary",
choices=["summary", "graph", "mermaid", "list", "floor", "kernel"],
)
parser.add_argument("--arch", default="x86_64")
args = parser.parse_args(argv)
if args.command == "list":
for name, prof in sorted(DISTRO_PROFILES.items()):
sys.stdout.write("%s\t%s\n" % (name, prof["title"]))
return 0
if args.command == "summary":
sys.stdout.write(
json.dumps(profile_summary(args.profile), indent=2, sort_keys=True)
+ "\n"
)
return 0
if args.command == "graph":
sys.stdout.write(
json.dumps(profile_graph(args.profile), indent=2, sort_keys=True)
+ "\n"
)
return 0
if args.command == "mermaid":
sys.stdout.write(mermaid_for_profile(args.profile))
return 0
if args.command == "floor":
try:
sys.stdout.write(
json.dumps(
floor_manifest(args.arch, args.profile),
indent=2,
sort_keys=True,
)
+ "\n"
)
except KeyError as exc:
sys.stderr.write("error: %s\n" % exc)
return 2
return 0
if args.command == "kernel":
try:
sys.stdout.write(
json.dumps(
kernel_manifest(args.arch, args.profile),
indent=2,
sort_keys=True,
)
+ "\n"
)
except KeyError as exc:
sys.stderr.write("error: %s\n" % exc)
return 2
return 0
return 1
if __name__ == "__main__":
raise SystemExit(main())