MIT License
-Copyright (c) 2019, 2024 Reasoning Technology
+Copyright (c) 2019, 2024, 2025 Reasoning Technology
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
--- /dev/null
+# Man_In_Grey acceptance filter (default template)
+# Return True to include a config file ,False to skip it.
+# You receive a PlanProvenance object named `prov`.
+#
+# Common fields:
+# prov.stage_root_dpath : Path
+# prov.config_abs_fpath : Path
+# prov.config_rel_fpath : Path
+# prov.read_dir_dpath : Path
+# prov.read_fname : str
+#
+# 1) Accept everything (default):
+# def accept(prov):
+# return True
+#
+# 2) Only a namespace:
+# def accept(prov):
+# return prov.config_rel_fpath.as_posix().startswith("dns/")
+#
+# 3) Exclude editor junk:
+# def accept(prov):
+# r = prov.config_rel_fpath.as_posix()
+# return not (r.endswith("~") or r.endswith(".swp"))
+#
+def accept(prov):
+ return True
--- /dev/null
+# write /etc/nftables.d/10-block-IPv6.nft — drop all IPv6
+def configure(prov, planner, WriteFileMeta):
+ wfm = WriteFileMeta(
+ dpath="/etc/nftables.d",
+ fname="10-block-IPv6.nft",
+ owner="root",
+ mode=0o644,
+ )
+ planner.displace(wfm)
+ planner.copy(wfm, content="""\
+table inet NO-IPV6 {
+ chain input {
+ type filter hook input priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 inbound" drop;
+ }
+
+ chain output {
+ type filter hook output priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 outbound" drop;
+ }
+
+ chain forward {
+ type filter hook forward priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 forward" drop;
+ }
+}
+""")
--- /dev/null
+table inet SUBU-DNS-REDIRECT {
+ chain output {
+ type nat hook output priority -100; policy accept;
+
+ # Redirect DNS for the subu UIDs to local Unbound listeners
+ meta skuid 2017 udp dport 53 redirect to :5301
+ meta skuid 2018 udp dport 53 redirect to :5302
+ meta skuid 2017 tcp dport 53 redirect to :5301
+ meta skuid 2018 tcp dport 53 redirect to :5302
+ }
+}
+
+table inet SUBU-PORT-EGRESS {
+ chain output {
+ type filter hook output priority 0; policy accept;
+
+ # Always allow loopback on egress
+ oifname "lo" accept
+
+ # No IPv6 for subu (until you reintroduce v6)
+ meta skuid {2017,2018} meta nfproto ipv6 counter comment "no IPv6 for subu" drop
+
+ ##### x6 (UID 2018)
+ # Block some exfil channels regardless of iface
+ meta skuid 2018 tcp dport {25,465,587} counter comment "block SMTP/Submission" drop
+ meta skuid 2018 udp dport {3478,5349,19302-19309} counter comment "block STUN/TURN" drop
+ meta skuid 2018 tcp dport 853 counter comment "block DoT (TCP/853)" drop
+
+ # (Optional) allow ICMP echo out via x6
+ meta skuid 2018 oifname "x6" ip protocol icmp icmp type echo-request accept
+
+ # Enforce interface binding
+ meta skuid 2018 oifname "x6" accept
+ meta skuid 2018 oifname != "x6" counter comment "x6 must use wg x6" drop
+
+ ##### US (UID 2017)
+ meta skuid 2017 tcp dport {25,465,587} counter drop comment "block SMTP/Submission"
+ meta skuid 2017 udp dport {3478,5349,19302-19309} counter drop comment "block STUN/TURN"
+ meta skuid 2017 tcp dport 853 counter drop comment "block DoT (TCP/853)"
+
+ # (Optional) ICMP via US
+ meta skuid 2017 oifname "US" ip protocol icmp icmp type echo-request accept
+
+ meta skuid 2017 oifname "US" accept
+ meta skuid 2017 oifname != "US" counter comment "US must use wg US" drop
+ }
+}
--- /dev/null
+# Man_In_Grey acceptance filter (default template)
+# Return True to include a config file ,False to skip it.
+# You receive a PlanProvenance object named `prov`.
+#
+# Common fields:
+# prov.stage_root_dpath : Path
+# prov.config_abs_fpath : Path
+# prov.config_rel_fpath : Path
+# prov.read_dir_dpath : Path
+# prov.read_fname : str
+#
+# 1) Accept everything (default):
+# def accept(prov):
+# return True
+#
+# 2) Only a namespace:
+# def accept(prov):
+# return prov.config_rel_fpath.as_posix().startswith("dns/")
+#
+# 3) Exclude editor junk:
+# def accept(prov):
+# r = prov.config_rel_fpath.as_posix()
+# return not (r.endswith("~") or r.endswith(".swp"))
+#
+def accept(prov):
+ return True
--- /dev/null
+# write /etc/nftables.d/10-block-IPv6.nft — drop all IPv6
+def configure(prov, planner, WriteFileMeta):
+ wfm = WriteFileMeta(
+ dpath="/etc/nftables.d",
+ fname="10-block-IPv6.nft",
+ owner="root",
+ mode=0o644,
+ )
+ planner.displace(wfm)
+ planner.copy(wfm, content="""\
+table inet NO-IPV6 {
+ chain input {
+ type filter hook input priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 inbound" drop;
+ }
+
+ chain output {
+ type filter hook output priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 outbound" drop;
+ }
+
+ chain forward {
+ type filter hook forward priority -300; policy accept;
+ meta nfproto ipv6 counter comment "drop all IPv6 forward" drop;
+ }
+}
+""")
--- /dev/null
+# write /etc/nftables.d/20-SUBU-ports.nft — DNS redirect + strict egress
+def configure(prov, planner, WriteFileMeta):
+ wfm = WriteFileMeta(
+ dpath="/etc/nftables.d",
+ fname="20-SUBU-ports.nft",
+ owner="root",
+ mode=0o644,
+ )
+ planner.displace(wfm)
+ planner.copy(wfm, content="""\
+# DNS per-UID redirect to local Unbound
+table inet SUBU-DNS-REDIRECT {
+ chain output {
+ type nat hook output priority -100; policy accept;
+
+ # US (uid 2017) -> 127.0.0.1:5301
+ meta skuid 2017 udp dport 53 redirect to :5301
+ meta skuid 2017 tcp dport 53 redirect to :5301
+ # x6 (uid 2018) -> 127.0.0.1:5302
+ meta skuid 2018 udp dport 53 redirect to :5302
+ meta skuid 2018 tcp dport 53 redirect to :5302
+ }
+}
+
+# Egress policy: subu UIDs must use their WireGuard iface; block exfil channels
+table inet SUBU-PORT-EGRESS {
+ chain output {
+ type filter hook output priority 0; policy accept;
+
+ # Always allow loopback
+ oifname "lo" accept;
+
+ # No IPv6 for subu (until you reintroduce v6)
+ meta skuid {2017,2018} meta nfproto ipv6 counter comment "no IPv6 for subu" drop;
+
+ ##### x6 (UID 2018)
+ meta skuid 2018 tcp dport {25,465,587} counter comment "block SMTP/Submission" drop;
+ meta skuid 2018 udp dport {3478,5349,19302-19309} counter comment "block STUN/TURN" drop;
+ meta skuid 2018 tcp dport 853 counter comment "block DoT (TCP/853)" drop;
+ meta skuid 2018 oifname "x6" accept;
+ meta skuid 2018 oifname != "x6" counter comment "x6 must use wg x6" drop;
+
+ ##### US (UID 2017)
+ meta skuid 2017 tcp dport {25,465,587} counter comment "block SMTP/Submission" drop;
+ meta skuid 2017 udp dport {3478,5349,19302-19309} counter comment "block STUN/TURN" drop;
+ meta skuid 2017 tcp dport 853 counter comment "block DoT (TCP/853)" drop;
+ meta skuid 2017 oifname "US" accept;
+ meta skuid 2017 oifname != "US" counter comment "US must use wg US" drop;
+ }
+}
+""")
--- /dev/null
+# write /etc/unbound/unbound-US.conf — local listener that egresses via US WG
+def configure(prov, planner, WriteFileMeta):
+ wfm = WriteFileMeta(
+ dpath="/etc/unbound",
+ fname="unbound-US.conf",
+ owner="root",
+ mode=0o644,
+ )
+ planner.displace(wfm)
+ planner.copy(wfm, content="""\
+server:
+ verbosity: 1
+ username: "unbound"
+ directory: "/etc/unbound"
+ chroot: ""
+
+ do-ip6: no
+ do-udp: yes
+ do-tcp: yes
+ prefer-ip6: no
+
+ # Listen only on loopback (US instance)
+ interface: 127.0.0.1@5301
+ access-control: 127.0.0.0/8 allow
+
+ # Egress via US tunnel address (policy rules ensure it leaves on wg US)
+ outgoing-interface: 10.0.0.1
+
+ # Hardening/cache
+ hide-identity: yes
+ hide-version: yes
+ harden-referral-path: yes
+ harden-dnssec-stripped: yes
+ qname-minimisation: yes
+ aggressive-nsec: yes
+ prefetch: yes
+ cache-min-ttl: 60
+ cache-max-ttl: 86400
+
+ # DNSSEC trust anchor
+ auto-trust-anchor-file: "/var/lib/unbound/root.key"
+""")
--- /dev/null
+# /etc/systemd/system/wg-policy-US.service — run after wg-quick@US to install policy rules
+def configure(prov, planner, WriteFileMeta):
+ content = """[Unit]
+Description=Policy routing for Unbound egress (US)
+After=wg-quick@US.service
+Wants=wg-quick@US.service
+
+[Service]
+Type=oneshot
+ExecStart=/usr/local/sbin/wg-policy-US.sh
+RemainAfterExit=yes
+
+[Install]
+WantedBy=multi-user.target
+"""
+ wfm = WriteFileMeta(dpath="/etc/systemd/system", fname="wg-policy-US.service", owner="root", mode="0644")
+ planner.displace(wfm)
+ planner.copy(wfm, content=content)
--- /dev/null
+# /usr/local/sbin/wg-policy-US.sh — source-policy routing for Unbound's egress
+def configure(prov, planner, WriteFileMeta):
+ # EDIT if your interface/IP differ:
+ WG_IFACE = "US"
+ WG_SRC_IP = "10.0.0.1"
+ TABLE = 100
+
+ content = f"""#!/usr/bin/env bash
+set -euo pipefail
+WG_IFACE="{WG_IFACE}"
+WG_SRC_IP="{WG_SRC_IP}"
+TABLE={TABLE}
+
+ip rule replace from "$WG_SRC_IP" lookup "$TABLE" priority 10010
+ip route replace default dev "$WG_IFACE" table "$TABLE"
+"""
+ wfm = WriteFileMeta(dpath="/usr/local/sbin", fname="wg-policy-US.sh", owner="root", mode="0755")
+ planner.displace(wfm)
+ planner.copy(wfm, content=content)
--- /dev/null
+# Redirect DNS traffic per-UID to local Unbound instances.
+# US (uid 2017) -> 127.0.0.1:5301
+# x6 (uid 2018) -> 127.0.0.1:5302
+table inet nat {
+ chain output {
+ type nat hook output priority -100;
+ # US
+ meta skuid 2017 udp dport 53 redirect to :5301
+ meta skuid 2017 tcp dport 53 redirect to :5301
+ # x6
+ meta skuid 2018 udp dport 53 redirect to :5302
+ meta skuid 2018 tcp dport 53 redirect to :5302
+ }
+}
--- /dev/null
+server:
+ verbosity: 1
+ username: "unbound"
+ directory: "/etc/unbound"
+ chroot: ""
+
+ do-ip6: no
+ do-udp: yes
+ do-tcp: yes
+ prefer-ip6: no
+
+ # Listen only on loopback (US instance)
+ interface: 127.0.0.1@5301
+ access-control: 127.0.0.0/8 allow
+
+ # Egress via US tunnel address (policy routing will carry it out the WG table)
+ outgoing-interface: 10.0.0.1
+
+ # Sensible hardening/cache
+ hide-identity: yes
+ hide-version: yes
+ harden-referral-path: yes
+ harden-dnssec-stripped: yes
+ qname-minimisation: yes
+ aggressive-nsec: yes
+ prefetch: yes
+ cache-min-ttl: 60
+ cache-max-ttl: 86400
+
+ # DNSSEC TA (create with unbound-anchor)
+ auto-trust-anchor-file: "/var/lib/unbound/root.key"
+ # Optional root hints (download separately)
+ # root-hints: "/var/lib/unbound/root.hints"
+
+# To use forwarding instead of full recursion, uncomment and edit:
+# forward-zone:
+# name: "."
+# forward-tls-upstream: no
+# forward-addr: 9.9.9.9
+# forward-addr: 1.1.1.1
--- /dev/null
+server:
+ verbosity: 1
+ username: "unbound"
+ directory: "/etc/unbound"
+ chroot: ""
+
+ do-ip6: no
+ do-udp: yes
+ do-tcp: yes
+ prefer-ip6: no
+
+ # Listen only on loopback (x6 instance)
+ interface: 127.0.0.1@5302
+ access-control: 127.0.0.0/8 allow
+
+ # Egress via x6 tunnel address (policy routing will carry it out the WG table)
+ outgoing-interface: 10.8.0.2
+
+ # Sensible hardening/cache
+ hide-identity: yes
+ hide-version: yes
+ harden-referral-path: yes
+ harden-dnssec-stripped: yes
+ qname-minimisation: yes
+ aggressive-nsec: yes
+ prefetch: yes
+ cache-min-ttl: 60
+ cache-max-ttl: 86400
+
+ # DNSSEC TA (create with unbound-anchor)
+ auto-trust-anchor-file: "/var/lib/unbound/root.key"
+ # Optional root hints (download separately)
+ # root-hints: "/var/lib/unbound/root.hints"
+
+# To use forwarding instead of full recursion, uncomment and edit:
+# forward-zone:
+# name: "."
+# forward-tls-upstream: no
+# forward-addr: 9.9.9.9
+# forward-addr: 1.1.1.1
--- /dev/null
+[Unit]
+Description=Unbound DNS (%i)
+Documentation=man:unbound(8)
+After=network-online.target wg-quick@%i.service
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=/usr/sbin/unbound -d -p -c /etc/unbound/unbound-%i.conf
+Restart=on-failure
+# Lock down a bit
+CapabilityBoundingSet=CAP_NET_BIND_SERVICE CAP_SETGID CAP_SETUID
+AmbientCapabilities=CAP_NET_BIND_SERVICE
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=full
+ProtectHome=true
+
+[Install]
+WantedBy=multi-user.target
--- /dev/null
+# unbound/unbound_US.py
+def configure(prov, planner, WriteFileMeta):
+ conf = """server:
+ verbosity: 1
+ username: "unbound"
+ directory: "/etc/unbound"
+ chroot: ""
+
+ do-ip6: no
+ do-udp: yes
+ do-tcp: yes
+ prefer-ip6: no
+
+ interface: 127.0.0.1@5301
+ access-control: 127.0.0.0/8 allow
+
+ outgoing-interface: 10.0.0.1
+
+ hide-identity: yes
+ hide-version: yes
+ harden-referral-path: yes
+ harden-dnssec-stripped: yes
+ qname-minimisation: yes
+ aggressive-nsec: yes
+ prefetch: yes
+ cache-min-ttl: 60
+ cache-max-ttl: 86400
+
+ auto-trust-anchor-file: "/var/lib/unbound/root.key"
+"""
+ wfm = WriteFileMeta(dpath="/etc/unbound", fname="unbound-US.conf",
+ owner="root", mode="0644")
+ planner.copy(wfm, content=conf)
--- /dev/null
+# systemd/unbound_at_template.py
+def configure(prov, planner, WriteFileMeta):
+ service = """[Unit]
+Description=Unbound DNS (%i)
+Documentation=man:unbound(8)
+After=network-online.target wg-quick@%i.service
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=/usr/sbin/unbound -d -p -c /etc/unbound/unbound-%i.conf
+Restart=on-failure
+CapabilityBoundingSet=CAP_NET_BIND_SERVICE CAP_SETGID CAP_SETUID
+AmbientCapabilities=CAP_NET_BIND_SERVICE
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=full
+ProtectHome=true
+
+[Install]
+WantedBy=multi-user.target
+"""
+ wfm = WriteFileMeta(dpath="/etc/systemd/system", fname="unbound@.service",
+ owner="root", mode="0644")
+ planner.copy(wfm, content=service)
--- /dev/null
+# unbound/unbound_x6.py
+def configure(prov, planner, WriteFileMeta):
+ conf = """server:
+ verbosity: 1
+ username: "unbound"
+ directory: "/etc/unbound"
+ chroot: ""
+
+ do-ip6: no
+ do-udp: yes
+ do-tcp: yes
+ prefer-ip6: no
+
+ interface: 127.0.0.1@5302
+ access-control: 127.0.0.0/8 allow
+
+ outgoing-interface: 10.8.0.2
+
+ hide-identity: yes
+ hide-version: yes
+ harden-referral-path: yes
+ harden-dnssec-stripped: yes
+ qname-minimisation: yes
+ aggressive-nsec: yes
+ prefetch: yes
+ cache-min-ttl: 60
+ cache-max-ttl: 86400
+
+ auto-trust-anchor-file: "/var/lib/unbound/root.key"
+"""
+ wfm = WriteFileMeta(dpath="/etc/unbound", fname="unbound-x6.conf",
+ owner="root", mode="0644")
+ planner.copy(wfm, content=conf)
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-Planner.py — plan builder for staged configuration (UNPRIVILEGED).
-
-Given: runner-side provenance (PlanProvenance) and optional defaults (WriteFileMeta).
-Does: expose Planner whose command methods (copy/displace/delete) build Command entries,
- resolving arguments with precedence: kwarg > per-call WriteFileMeta > planner default
- (and for filename, fallback to provenance-derived basename). On any argument error,
- the Command is returned with errors and NOT appended to the Journal.
-Returns: Journal (model only; dict in/out) via planner.journal().
-"""
-
-from __future__ import annotations
-
-# no bytecode anywhere (works under sudo/root shells too)
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from pathlib import Path
-import getpass
-
-# ===== Utilities =====
-
-def norm_perm(value: int|str)-> tuple[int ,str]|None:
- "Given int or 3/4-char octal string (optionally 0o-prefixed). Does validate/normalize. Returns (int ,'%04o') or None."
- if isinstance(value ,int):
- if 0 <= value <= 0o7777:
- return value ,f"{value:04o}"
- return None
- if isinstance(value ,str):
- s = value.strip().lower()
- if s.startswith("0o"):
- try:
- v = int(s ,8)
- return v ,f"{v:04o}"
- except Exception:
- return None
- if len(s) in (3 ,4) and all(ch in "01234567" for ch in s):
- try:
- v = int(s ,8)
- return v ,f"{v:04o}"
- except Exception:
- return None
- return None
-
-def is_abs_dpath(dpath_str: str|None)-> bool:
- "Given path string. Does quick abs dir check. Returns bool."
- return isinstance(dpath_str ,str) and dpath_str.startswith("/") and "\x00" not in dpath_str
-
-def norm_abs_dpath_str(value: str|Path|None)-> str|None:
- "Given str/Path/None. Does normalize absolute dir path string. Returns str or None."
- if value is None: return None
- s = value.as_posix() if isinstance(value ,Path) else str(value)
- return s if is_abs_dpath(s) else None
-
-def norm_dpath_str(value: str|Path|None)-> str|None:
- "Given str/Path/None. Does minimal sanitize; allows relative. Returns str or None."
- if value is None: return None
- s = value.as_posix() if isinstance(value ,Path) else str(value)
- if not s or "\x00" in s: return None
- return s
-
-def norm_fname_or_none(value: str|None)-> str|None:
- "Given candidate filename or None. Does validate bare filename. Returns str or None."
- if value is None: return None
- s = str(value)
- if not s: return None
- if "/" in s or s in ("." ,"..") or "\x00" in s: return None
- return s
-
-def norm_nonempty_owner(value: str|None)-> str|None:
- "Given owner string or None. Does minimally validate (non-empty). Returns str or None."
- if value is None: return None
- s = str(value).strip()
- return s if s else None
-
-def parse_mode(value: int|str|None)-> tuple[int|None ,str|None]:
- "Given int/str/None. Does normalize via norm_perm. Returns (int ,'%04o') or (None ,None)."
- if value is None: return None ,None
- r = norm_perm(value)
- return r if r is not None else (None ,None)
-
-def norm_content_bytes(value: bytes|str|None)-> bytes|None:
- "Given bytes/str/None. Does normalize to UTF-8 bytes or None. Returns bytes|None."
- if value is None: return None
- if isinstance(value ,bytes): return value
- return value.encode("utf-8")
-
-# ===== Wire-ready model types (no CBOR here) =====
-
-class Command:
- """
- Command — a single planned operation.
-
- Given name_str ('copy'|'displace'|'delete'), optional arg_dict, optional errors_list.
- Does hold op name, own a fresh arg_dict, collect per-entry errors.
- Returns dictionary via as_dictionary().
- """
- __slots__ = (
- "name_str"
- ,"arg_dict"
- ,"errors_list"
- )
-
- def __init__(self ,name_str: str ,arg_dict: dict|None=None ,errors_list: list[str]|None=None)-> None:
- self.name_str = name_str
- self.arg_dict = dict(arg_dict) if arg_dict is not None else {}
- self.errors_list = list(errors_list) if errors_list is not None else []
-
- def add_error(self ,msg_str: str)-> None:
- self.errors_list.append(msg_str)
-
- def as_dictionary(self)-> dict:
- return {
- "op": self.name_str
- ,"arg_dict": dict(self.arg_dict)
- ,"errors_list": list(self.errors_list)
- }
-
- def print(self ,* ,index: int|None=None ,file=None)-> None:
- """
- Given: optional index for numbering and optional file-like (defaults to stdout).
- Does: print a compact, human-readable one-line summary of this command; prints any errors indented below.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- op = self.name_str
- ad = self.arg_dict or {}
-
- # Compose destination path for display (normalize to collapse '..')
- d = ad.get("write_file_dpath_str") or ""
- f = ad.get("write_file_fname") or ""
- try:
- from pathlib import Path as _Path
- if d and f and "/" not in f:
- dst = (_Path(d)/f).resolve().as_posix()
- else:
- dst = "?"
- except Exception:
- dst = "?"
-
- prefix = f"{index:02d}. " if index is not None else ""
-
- if op == "copy":
- mode = ad.get("mode_int")
- owner = ad.get("owner_name")
- size = len(ad.get("content_bytes") or b"")
- line = f"{prefix}copy -> {dst} mode {mode:04o} owner {owner} bytes {size}"
- elif op == "displace":
- line = f"{prefix}displace -> {dst}"
- elif op == "delete":
- line = f"{prefix}delete -> {dst}"
- else:
- line = f"{prefix}?op? -> {dst}"
-
- print(line ,file=file)
-
- for err in self.errors_list:
- print(f" ! {err}" ,file=file)
-
-class Journal:
- """
- Journal — ordered list of Command plus provenance metadata (model only; no CBOR).
-
- Given optional plan_dict in wire shape (for reconstruction).
- Does manage meta, append commands, expose entries, and pack to dict.
- Returns dict via as_dictionary().
- """
- __slots__ = (
- "meta_dict"
- ,"command_list"
- )
-
- def __init__(self ,plan_dict: dict|None=None)-> None:
- self.meta_dict = {}
- self.command_list = []
- if plan_dict is not None:
- self._init_from_dict(plan_dict)
-
- def _init_from_dict(self ,plan_dict: dict)-> None:
- if not isinstance(plan_dict ,dict):
- raise ValueError("plan_dict must be a dict")
- meta = dict(plan_dict.get("meta_dict") or {})
- entries = plan_dict.get("entries_list") or []
- self.meta_dict.update(meta)
- for e in entries:
- if not isinstance(e ,dict):
- continue
- op = e.get("op") or "?"
- args = e.get("arg_dict") or {}
- errs = e.get("errors_list") or []
- self.command_list.append(Command(name_str=op ,arg_dict=dict(args) ,errors_list=list(errs)))
-
- def set_meta(self ,**kv)-> None:
- self.meta_dict.update(kv)
-
- def append(self ,cmd: Command)-> None:
- self.command_list.append(cmd)
-
- def entries_list(self)-> list[dict]:
- return [c.as_dictionary() for c in self.command_list]
-
- def as_dictionary(self)-> dict:
- return {
- "version_int": 1
- ,"meta_dict": dict(self.meta_dict)
- ,"entries_list": self.entries_list()
- }
-
- def print(self ,* ,index_start: int=1 ,file=None)-> None:
- """
- Given: optional starting index and optional file-like (defaults to stdout).
- Does: print each Command on a single line via Command.print(), numbered.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- if not self.command_list:
- print("(plan is empty)" ,file=file)
- return
-
- for i ,cmd in enumerate(self.command_list ,start=index_start):
- cmd.print(index=i ,file=file)
-
-# ===== Runner-provided provenance =====
-
-class PlanProvenance:
- """
- Runner-provided, read-only provenance for a single config script.
- """
- __slots__ = (
- "stage_root_dpath"
- ,"config_abs_fpath"
- ,"config_rel_fpath"
- ,"read_dir_dpath"
- ,"read_fname"
- ,"process_user"
- ,"cwd_dpath"
- )
-
- def __init__(self ,* ,stage_root: Path ,config_path: Path):
- self.stage_root_dpath = stage_root.resolve()
- self.config_abs_fpath = config_path.resolve()
- try:
- self.config_rel_fpath = self.config_abs_fpath.relative_to(self.stage_root_dpath)
- except Exception:
- self.config_rel_fpath = Path(self.config_abs_fpath.name)
-
- self.read_dir_dpath = self.config_abs_fpath.parent
-
- name = self.config_abs_fpath.name
- if name.endswith(".stage.py"):
- self.read_fname = name[:-len(".stage.py")]
- elif name.endswith(".py"):
- self.read_fname = name[:-3]
- else:
- self.read_fname = name
-
- self.process_user = getpass.getuser()
- self.cwd_dpath = Path.cwd().resolve()
-
- def print(self ,* ,file=None)-> None:
- if file is None:
- import sys as _sys
- file = _sys.stdout
- print(f"Stage root: {self.stage_root_dpath}" ,file=file)
- print(f"Config (rel): {self.config_rel_fpath.as_posix()}" ,file=file)
- print(f"Config (abs): {self.config_abs_fpath}" ,file=file)
- print(f"Read dir: {self.read_dir_dpath}" ,file=file)
- print(f"Read fname: {self.read_fname}" ,file=file)
- print(f"Process user: {self.process_user}" ,file=file)
-
-# ===== Admin-facing defaults carrier =====
-
-class WriteFileMeta:
- """
- WriteFileMeta — per-call or planner-default write-file attributes.
-
- Given dpath (str/Path, may be relative) ,fname (bare name or None) ,owner (str)
- ,mode (int|'0644') ,content (bytes|str|None).
- Does normalize into fields (may remain None if absent/invalid).
- Returns object suitable for providing defaults to Planner methods.
- """
- __slots__ = (
- "dpath_str"
- ,"fname"
- ,"owner_name_str"
- ,"mode_int"
- ,"mode_octal_str"
- ,"content_bytes"
- )
-
- def __init__(self
- ,*
- ,dpath="/"
- ,fname=None
- ,owner="root"
- ,mode=0o444
- ,content=None
- ):
- self.dpath_str = norm_dpath_str(dpath)
- self.fname = norm_fname_or_none(fname)
- self.owner_name_str = norm_nonempty_owner(owner)
- self.mode_int ,self.mode_octal_str = parse_mode(mode)
- self.content_bytes = norm_content_bytes(content)
-
- def print(self ,* ,label: str|None=None ,file=None)-> None:
- if file is None:
- import sys as _sys
- file = _sys.stdout
- dpath = self.dpath_str or "?"
- fname = self.fname or "?"
- owner = self.owner_name_str or "?"
- mode_str = f"{self.mode_int:04o}" if isinstance(self.mode_int ,int) else (self.mode_octal_str or "?")
- size = len(self.content_bytes) if isinstance(self.content_bytes ,(bytes ,bytearray)) else 0
- prefix = (label + ": ") if label else ""
- print(f"{prefix}dpath={dpath} fname={fname} owner={owner} mode={mode_str} bytes={size}" ,file=file)
-
-# ===== Planner =====
-
-class Planner:
- """
- Planner — constructs a Journal of Commands from config scripts.
-
- Given provenance (PlanProvenance) and optional default WriteFileMeta.
- Does resolve command parameters by precedence: kwarg > per-call WriteFileMeta > planner default,
- with a final filename fallback to provenance basename if still missing.
- On any argument error, returns the Command with errors and DOES NOT append it to Journal.
- Returns live Journal via journal().
- """
- __slots__ = (
- "_prov"
- ,"_defaults"
- ,"_journal"
- )
-
- def __init__(self ,provenance: PlanProvenance ,defaults: WriteFileMeta|None=None)-> None:
- self._prov = provenance
- self._defaults = defaults if defaults is not None else WriteFileMeta(
- dpath="/"
- ,fname=provenance.read_fname
- ,owner="root"
- ,mode=0o444
- ,content=None
- )
- self._journal = Journal()
- self._journal.set_meta(
- stage_root_dpath_str=str(self._prov.stage_root_dpath)
- ,config_rel_fpath_str=self._prov.config_rel_fpath.as_posix()
- )
-
- # --- provenance/defaults/journal access ---
-
- def set_provenance(self ,prov: PlanProvenance)-> None:
- self._prov = prov
-
- def set_defaults(self ,defaults: WriteFileMeta)-> None:
- self._defaults = defaults
-
- def defaults(self)-> WriteFileMeta:
- return self._defaults
-
- def journal(self)-> Journal:
- return self._journal
-
- # --- resolution helpers ---
-
- def _pick(self ,kw ,meta_attr ,default_attr):
- "Pick first non-None among kw ,meta_attr ,default_attr."
- return kw if kw is not None else (meta_attr if meta_attr is not None else default_attr)
-
- def _resolve_write_file(self ,wfm ,dpath ,fname)-> tuple[str|None ,str|None]:
- # normalize explicit kwargs
- dpath_str = norm_dpath_str(dpath) if dpath is not None else None
- fname_str = norm_fname_or_none(fname) if fname is not None else None
-
- # precedence: kwarg > per-call meta > planner default
- dpath_val = self._pick(dpath_str ,(wfm.dpath_str if wfm else None) ,self._defaults.dpath_str)
- fname_val = self._pick(fname_str ,(wfm.fname if wfm else None) ,self._defaults.fname)
-
- # final fallback for filename: derive from config name
- if fname_val is None:
- fname_val = self._prov.read_fname
-
- # anchor/normalize dpath
- if dpath_val is not None:
- p = Path(dpath_val)
- if not p.is_absolute():
- p = (self._prov.cwd_dpath/p)
- dpath_val = p.resolve().as_posix()
-
- return dpath_val ,fname_val
-
- def _resolve_owner_mode_content(self
- ,wfm: WriteFileMeta|None
- ,owner: str|None
- ,mode: int|str|None
- ,content: bytes|str|None
- )-> tuple[str|None ,tuple[int|None ,str|None] ,bytes|None]:
- owner_norm = norm_nonempty_owner(owner) if owner is not None else None
- mode_norm = parse_mode(mode) if mode is not None else (None ,None)
- content_b = norm_content_bytes(content) if content is not None else None
-
- owner_v = self._pick(owner_norm ,(wfm.owner_name_str if wfm else None) ,self._defaults.owner_name_str)
- mode_v = (mode_norm if mode_norm != (None ,None) else
- ((wfm.mode_int ,wfm.mode_octal_str) if wfm else (self._defaults.mode_int ,self._defaults.mode_octal_str)))
- content_v = self._pick(content_b ,(wfm.content_bytes if wfm else None) ,self._defaults.content_bytes)
- return owner_v ,mode_v ,content_v
-
- # --- printing ---
-
- def print(self ,* ,show_journal: bool=True ,file=None)-> None:
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- print("== Provenance ==" ,file=file)
- self._prov.print(file=file)
-
- print("\n== Defaults ==" ,file=file)
- self._defaults.print(label="defaults" ,file=file)
-
- if show_journal:
- entries = getattr(self._journal ,"command_list" ,[])
- n_total = len(entries)
- n_copy = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "copy")
- n_disp = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "displace")
- n_del = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "delete")
-
- print("\n== Journal ==" ,file=file)
- print(f"entries: {n_total} copy:{n_copy} displace:{n_disp} delete:{n_del}" ,file=file)
- if n_total:
- self._journal.print(index_start=1 ,file=file)
- else:
- print("(plan is empty)" ,file=file)
-
- # --- Command builders (first arg may be WriteFileMeta) ---
-
- def copy(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- ,owner: str|None=None
- ,mode: int|str|None=None
- ,content: bytes|str|None=None
- )-> Command:
- cmd = Command("copy")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- owner_v ,(mode_int ,mode_oct) ,content_b = self._resolve_owner_mode_content(wfm ,owner ,mode ,content)
-
- # well-formed checks
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- if not owner_v: cmd.add_error("owner must be non-empty")
- if (mode_int ,mode_oct) == (None ,None):
- cmd.add_error("mode must be int <= 0o7777 or 3/4-digit octal string")
- if content_b is None:
- cmd.add_error("content is required for copy() (bytes or str)")
-
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath
- ,"write_file_fname": fname
- ,"owner_name": owner_v
- ,"mode_int": mode_int
- ,"mode_octal_str": mode_oct
- ,"content_bytes": content_b
- ,"provenance_config_rel_fpath_str": self._prov.config_rel_fpath.as_posix()
- })
-
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
-
- def displace(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- )-> Command:
- cmd = Command("displace")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath
- ,"write_file_fname": fname
- })
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
-
- def delete(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- )-> Command:
- cmd = Command("delete")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath
- ,"write_file_fname": fname
- })
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-Planner.py — plan builder for staged configuration (UNPRIVILEGED).
-
-Given: runner-side provenance (PlanProvenance) and optional defaults (WriteFileMeta).
-Does: expose Planner whose command methods (copy/displace/delete) build Command entries,
- resolving arguments with precedence: kwarg > per-call WriteFileMeta > planner default
- (and for filename, fallback to provenance-derived basename). On any argument error,
- the Command is returned with errors and NOT appended to the Journal.
-Returns: Journal (model only; dict in/out) via planner.journal().
-"""
-
-from __future__ import annotations
-
-# no bytecode anywhere (works under sudo/root shells too)
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from pathlib import Path
-import getpass
-
-
-# ===== Utilities =====
-
-def norm_perm(value: int|str)-> tuple[int,str]|None:
- "Given int or 3/4-char octal string (optionally 0o-prefixed). Does validate/normalize. Returns (int,'%04o') or None."
- if isinstance(value ,int):
- if 0 <= value <= 0o7777:
- return value ,f"{value:04o}"
- return None
- if isinstance(value ,str):
- s = value.strip().lower()
- if s.startswith("0o"):
- try:
- v = int(s ,8)
- return v ,f"{v:04o}"
- except Exception:
- return None
- if len(s) in (3 ,4) and all(ch in "01234567" for ch in s):
- try:
- v = int(s ,8)
- return v ,f"{v:04o}"
- except Exception:
- return None
- return None
-
-def is_abs_dpath(dpath_str: str|None)-> bool:
- "Given path string. Does quick abs dir check. Returns bool."
- return isinstance(dpath_str ,str) and dpath_str.startswith("/") and "\x00" not in dpath_str
-
-def norm_abs_dpath_str(value: str|Path|None)-> str|None:
- "Given str/Path/None. Does normalize absolute dir path string. Returns str or None."
- if value is None: return None
- s = value.as_posix() if isinstance(value ,Path) else str(value)
- return s if is_abs_dpath(s) else None
-
-def norm_fname_or_none(value: str|None)-> str|None:
- "Given candidate filename or None. Does validate bare filename. Returns str or None."
- if value is None: return None
- s = str(value)
- if not s: return None
- if "/" in s or s in ("." ,"..") or "\x00" in s: return None
- return s
-
-def norm_nonempty_owner(value: str|None)-> str|None:
- "Given owner string or None. Does minimally validate (non-empty). Returns str or None."
- if value is None: return None
- s = str(value).strip()
- return s if s else None
-
-def parse_mode(value: int|str|None)-> tuple[int|None ,str|None]:
- "Given int/str/None. Does normalize via norm_perm. Returns (int,'%04o') or (None,None)."
- if value is None: return None ,None
- r = norm_perm(value)
- return r if r is not None else (None ,None)
-
-def norm_content_bytes(value: bytes|str|None)-> bytes|None:
- "Given bytes/str/None. Does normalize to UTF-8 bytes or None. Returns bytes|None."
- if value is None: return None
- if isinstance(value ,bytes): return value
- return value.encode("utf-8")
-
-def norm_dpath_str(value: str|Path|None)-> str|None:
- "Given str/Path/None. Does minimal sanitize; allows relative. Returns str or None."
- if value is None: return None
- s = value.as_posix() if isinstance(value ,Path) else str(value)
- if not s or "\x00" in s: return None
- return s
-
-
-# ===== Wire-ready model types (no CBOR here) =====
-
-class Command:
- """
- Command — a single planned operation.
-
- Given name_str ('copy'|'displace'|'delete'), optional arg_dict, optional errors_list.
- Does hold op name, own a fresh arg_dict, collect per-entry errors.
- Returns dictionary via as_dictionary().
- """
- __slots__ = ("name_str" ,"arg_dict" ,"errors_list")
-
- def __init__(self ,name_str: str ,arg_dict: dict|None=None ,errors_list: list[str]|None=None)-> None:
- self.name_str = name_str
- self.arg_dict = dict(arg_dict) if arg_dict is not None else {}
- self.errors_list = list(errors_list) if errors_list is not None else []
-
- def add_error(self ,msg_str: str)-> None:
- self.errors_list.append(msg_str)
-
- def as_dictionary(self)-> dict:
- return {
- "op": self.name_str
- ,"arg_dict": dict(self.arg_dict)
- ,"errors_list": list(self.errors_list)
- }
-
- def print(self, *, index: int|None=None, file=None)-> None:
- """
- Given: optional index for numbering and optional file-like (defaults to stdout).
- Does: print a compact, human-readable one-line summary of this command; prints any errors indented below.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- op = self.name_str
- ad = self.arg_dict or {}
-
- # Compose destination path for display
- d = ad.get("write_file_dpath_str") or ""
- f = ad.get("write_file_fname") or ""
- try:
- from pathlib import Path as _Path
- dst = (_Path(d)/f).as_posix() if d and f and "/" not in f else "?"
- except Exception:
- dst = "?"
-
- # Numbering prefix
- prefix = f"{index:02d}. " if index is not None else ""
-
- if op == "copy":
- mode = ad.get("mode_int")
- owner = ad.get("owner_name")
- size = len(ad.get("content_bytes") or b"")
- line = f"{prefix}copy -> {dst} mode {mode:04o} owner {owner} bytes {size}"
- elif op == "displace":
- line = f"{prefix}displace -> {dst}"
- elif op == "delete":
- line = f"{prefix}delete -> {dst}"
- else:
- line = f"{prefix}?op? -> {dst}"
-
- print(line, file=file)
-
- # Print any per-entry errors underneath
- for err in self.errors_list:
- print(f" ! {err}", file=file)
-
-
-class Journal:
- """
- Journal — ordered list of Command plus provenance metadata (model only; no CBOR).
-
- Given optional plan_dict in wire shape (for reconstruction).
- Does manage meta, append commands, expose entries, and pack to dict.
- Returns dict via as_dictionary().
- """
- __slots__ = ("meta_dict" ,"command_list")
-
- def __init__(self ,plan_dict: dict|None=None)-> None:
- self.meta_dict = {}
- self.command_list = []
- if plan_dict is not None:
- self._init_from_dict(plan_dict)
-
- def _init_from_dict(self ,plan_dict: dict)-> None:
- if not isinstance(plan_dict ,dict):
- raise ValueError("plan_dict must be a dict")
- meta = dict(plan_dict.get("meta_dict") or {})
- entries = plan_dict.get("entries_list") or []
- self.meta_dict.update(meta)
- for e in entries:
- if not isinstance(e ,dict):
- continue
- op = e.get("op") or "?"
- args = e.get("arg_dict") or {}
- errs = e.get("errors_list") or []
- self.command_list.append(Command(name_str=op ,arg_dict=dict(args) ,errors_list=list(errs)))
-
- def set_meta(self ,**kv)-> None:
- self.meta_dict.update(kv)
-
- def append(self ,cmd: Command)-> None:
- self.command_list.append(cmd)
-
- def entries_list(self)-> list[dict]:
- return [c.as_dictionary() for c in self.command_list]
-
- def as_dictionary(self)-> dict:
- return {
- "version_int": 1
- ,"meta_dict": dict(self.meta_dict)
- ,"entries_list": self.entries_list()
- }
-
- def print(self, *, index_start: int = 1, file=None) -> None:
- """
- Given: optional starting index and optional file-like (defaults to stdout).
- Does: print each Command on a single line via Command.print(), numbered.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- if not self.command_list:
- print("(plan is empty)", file=file)
- return
-
- for i, cmd in enumerate(self.command_list, start=index_start):
- cmd.print(index=i, file=file)
-
-# ===== Runner-provided provenance =====
-
-# Planner.py
-class PlanProvenance:
- """
- Runner-provided, read-only provenance for a single config script.
- """
- __slots__ = ("stage_root_dpath","config_abs_fpath","config_rel_fpath",
- "read_dir_dpath","read_fname","process_user")
-
- def __init__(self, *, stage_root: Path, config_path: Path):
- import getpass
- self.stage_root_dpath = stage_root.resolve()
- self.config_abs_fpath = config_path.resolve()
- try:
- self.config_rel_fpath = self.config_abs_fpath.relative_to(self.stage_root_dpath)
- except Exception:
- self.config_rel_fpath = Path(self.config_abs_fpath.name)
-
- self.read_dir_dpath = self.config_abs_fpath.parent
-
- name = self.config_abs_fpath.name
- if name.endswith(".stage.py"):
- self.read_fname = name[:-len(".stage.py")]
- elif name.endswith(".py"):
- self.read_fname = name[:-3]
- else:
- self.read_fname = name
-
- # NEW: owner of the StageHand process
- self.process_user = getpass.getuser()
-
- def print(self, *, file=None) -> None:
- if file is None:
- import sys as _sys
- file = _sys.stdout
- print(f"Stage root: {self.stage_root_dpath}", file=file)
- print(f"Config (rel): {self.config_rel_fpath.as_posix()}", file=file)
- print(f"Config (abs): {self.config_abs_fpath}", file=file)
- print(f"Read dir: {self.read_dir_dpath}", file=file)
- print(f"Read fname: {self.read_fname}", file=file)
- print(f"Process user: {self.process_user}", file=file) # NEW
-
-# ===== Admin-facing defaults carrier =====
-
-class WriteFileMeta:
- """
- WriteFileMeta — per-call or planner-default write-file attributes.
-
- Given dpath (abs str/Path) ,fname (bare name or None) ,owner (str)
- ,mode (int|'0644') ,content (bytes|str|None).
- Does normalize into fields (may remain None if absent/invalid).
- Returns object suitable for providing defaults to Planner methods.
- """
- __slots__ = ("dpath_str" ,"fname" ,"owner_name_str" ,"mode_int" ,"mode_octal_str" ,"content_bytes")
-
- def __init__(self
- ,*
- ,dpath="/"
- ,fname=None # None → let Planner/provenance choose
- ,owner="root"
- ,mode=0o444
- ,content=None
- ):
- self.dpath_str = norm_dpath_str(dpath)
- self.fname = norm_fname_or_none(fname) # '.' no longer special → None
- self.owner_name_str = norm_nonempty_owner(owner) # '.' rejected → None
- self.mode_int, self.mode_octal_str = parse_mode(mode)
- self.content_bytes = norm_content_bytes(content)
-
- def print(self, *, label: str | None = None, file=None) -> None:
- """
- Given: optional label and optional file-like (defaults to stdout).
- Does: print a single-line summary of defaults/overrides.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- dpath = self.dpath_str or "?"
- fname = self.fname or "?"
- owner = self.owner_name_str or "?"
- mode_str = f"{self.mode_int:04o}" if isinstance(self.mode_int, int) else (self.mode_octal_str or "?")
- size = len(self.content_bytes) if isinstance(self.content_bytes, (bytes, bytearray)) else 0
- prefix = (label + ": ") if label else ""
- print(f"{prefix}dpath={dpath} fname={fname} owner={owner} mode={mode_str} bytes={size}", file=file)
-
-
-# ===== Planner =====
-
-class Planner:
- """
- Planner — constructs a Journal of Commands from config scripts.
-
- Given provenance (PlanProvenance) and optional default WriteFileMeta.
- Does resolve command parameters by precedence: kwarg > per-call WriteFileMeta > planner default,
- with a final filename fallback to provenance basename if still missing.
- On any argument error, returns the Command with errors and DOES NOT append it to Journal.
- Returns live Journal via journal().
- """
- __slots__ = ("_prov" ,"_defaults" ,"_journal")
-
- def __init__(self ,provenance: PlanProvenance ,defaults: WriteFileMeta|None=None)-> None:
- self._prov = provenance
- self._defaults = defaults if defaults is not None else WriteFileMeta(
- dpath="/"
- ,fname=provenance.read_fname
- ,owner="root"
- ,mode=0o444
- ,content=None
- )
- self._journal = Journal()
- self._journal.set_meta(
- stage_root_dpath_str=str(self._prov.stage_root_dpath)
- ,config_rel_fpath_str=self._prov.config_rel_fpath.as_posix()
- )
-
- # --- defaults management / access ---
-
- # in Planner.py, inside class Planner
- def set_provenance(self, prov: PlanProvenance) -> None:
- """Switch the current provenance used for fallbacks & per-command provenance tagging."""
- self._prov = prov
-
- def set_defaults(self ,defaults: WriteFileMeta)-> None:
- "Given WriteFileMeta. Does replace planner defaults. Returns None."
- self._defaults = defaults
-
- def defaults(self)-> WriteFileMeta:
- "Given n/a. Does return current WriteFileMeta defaults. Returns WriteFileMeta."
- return self._defaults
-
- def journal(self)-> Journal:
- "Given n/a. Returns Journal reference (live, still being modified here)."
- return self._journal
-
- # --- resolution helpers ---
-
- def _pick(self ,kw ,meta_attr ,default_attr):
- "Given three sources. Does pick first non-None. Returns value or None."
- return kw if kw is not None else (meta_attr if meta_attr is not None else default_attr)
-
- def _resolve_write_file(self, wfm, dpath, fname) -> tuple[str|None, str|None]:
- dpath_str = norm_dpath_str(dpath) if dpath is not None else None
- fname = norm_fname_or_none(fname) if fname is not None else None
-
- dpath_val = self._pick(dpath_str, (wfm.dpath_str if wfm else None), self._defaults.dpath_str)
- fname_val = self._pick(fname, (wfm.fname if wfm else None), self._defaults.fname)
-
- # final fallback for filename: derive from config name
- if fname_val is None:
- fname_val = self._prov.read_fname
-
- # anchor relative dpaths against the config’s directory
- if dpath_val is not None and not is_abs_dpath(dpath_val):
- dpath_val = (self._prov.read_dir_dpath / dpath_val).as_posix()
-
- return dpath_val, fname_val
-
- def _resolve_owner_mode_content(self
- ,wfm: WriteFileMeta|None
- ,owner: str|None
- ,mode: int|str|None
- ,content: bytes|str|None
- )-> tuple[str|None ,tuple[int|None ,str|None] ,bytes|None]:
- owner_norm = norm_nonempty_owner(owner) if owner is not None else None
- mode_norm = parse_mode(mode) if mode is not None else (None ,None)
- content_b = norm_content_bytes(content) if content is not None else None
-
- owner_v = self._pick(owner_norm, (wfm.owner_name_str if wfm else None), self._defaults.owner_name_str)
- mode_v = (mode_norm if mode_norm != (None ,None) else
- ((wfm.mode_int ,wfm.mode_octal_str) if wfm else (self._defaults.mode_int ,self._defaults.mode_octal_str)))
- content_v = self._pick(content_b ,(wfm.content_bytes if wfm else None) ,self._defaults.content_bytes)
- return owner_v ,mode_v ,content_v
-
- def print(self, *, show_journal: bool = True, file=None) -> None:
- """
- Given: flags (show_journal) and optional file-like (defaults to stdout).
- Does: print provenance, defaults, and optionally the journal via delegation.
- Returns: None.
- """
- if file is None:
- import sys as _sys
- file = _sys.stdout
-
- print("== Provenance ==", file=file)
- self._prov.print(file=file)
-
- print("\n== Defaults ==", file=file)
- self._defaults.print(label="defaults", file=file)
-
- if show_journal:
- entries = getattr(self._journal, "command_list", [])
- n_total = len(entries)
- n_copy = sum(1 for c in entries if getattr(c, "name_str", None) == "copy")
- n_disp = sum(1 for c in entries if getattr(c, "name_str", None) == "displace")
- n_del = sum(1 for c in entries if getattr(c, "name_str", None) == "delete")
-
- print("\n== Journal ==", file=file)
- print(f"entries: {n_total} copy:{n_copy} displace:{n_disp} delete:{n_del}", file=file)
- if n_total:
- self._journal.print(index_start=1, file=file)
- else:
- print("(plan is empty)", file=file)
-
- # --- Command builders (first arg may be WriteFileMeta) ---
-
- def copy(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- ,owner: str|None=None
- ,mode: int|str|None=None
- ,content: bytes|str|None=None
- )-> Command:
- """
- Given optional WriteFileMeta plus keyword overrides.
- Does build a 'copy' command; on any argument error the command is returned with errors and NOT appended.
- Returns Command.
- """
- cmd = Command("copy")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- owner_v ,(mode_int ,mode_oct) ,content_b = self._resolve_owner_mode_content(wfm ,owner ,mode ,content)
-
- # well-formed checks
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- if not owner_v: cmd.add_error("owner must be non-empty")
- if (mode_int ,mode_oct) == (None ,None):
- cmd.add_error("mode must be int <= 0o7777 or 3/4-digit octal string")
- if content_b is None:
- cmd.add_error("content is required for copy() (bytes or str)")
-
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath,
- "write_file_fname": fname, # was write_file_fname
- "owner_name": owner_v, # was owner_name_str
- "mode_int": mode_int,
- "mode_octal_str": mode_oct,
- "content_bytes": content_b,
- "provenance_config_rel_fpath_str": self._prov.config_rel_fpath.as_posix(),
- })
-
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
-
- def displace(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- )-> Command:
- "Given optional WriteFileMeta plus overrides. Does build 'displace' entry or return errors. Returns Command."
- cmd = Command("displace")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath,
- "write_file_fname": fname,
- })
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
-
- def delete(self
- ,wfm: WriteFileMeta|None=None
- ,*
- ,write_file_dpath: str|Path|None=None
- ,write_file_fname: str|None=None
- )-> Command:
- "Given optional WriteFileMeta plus overrides. Does build 'delete' entry or return errors. Returns Command."
- cmd = Command("delete")
- dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname)
- if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute")
- if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename")
- cmd.arg_dict.update({
- "write_file_dpath_str": dpath,
- "write_file_fname": fname,
- })
- if not cmd.errors_list:
- self._journal.append(cmd)
- return cmd
-
-
-
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-Stage.py — planner runtime for staged config programs (UNPRIVILEGED).
-
-Config usage:
- import Stage
-
- Stage.init(
- write_file_name="."
- , write_dpath="/etc/unbound"
- , write_file_owner_name="root"
- , write_file_permissions=0o644 # or "0644"
- , read_file_contents=b"...bytes..."# bytes preferred; str is utf-8 encoded
- )
- Stage.displace()
- Stage.copy()
- # Stage.delete()
-
-Notes:
- - This module only RECORDS plan steps using native Python values (ints/bytes/str).
- - The outer tool CBOR-encodes the accumulated plan AFTER all configs run.
-"""
-
-from __future__ import annotations
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from dataclasses import dataclass ,field
-from pathlib import Path
-from typing import Any
-
-# ---------- helpers ----------
-
-def _norm_perm(value: int|str)-> tuple[int,str]|None:
- "Given: an int or a 4-char octal string. Does: validate/normalize to (int,'%04o'). Returns: tuple or None."
- if isinstance(value ,int):
- if 0 <= value <= 0o7777:
- return value ,f"{value:04o}"
- return None
- if isinstance(value ,str):
- s = value.strip()
- if len(s)==4 and all(ch in "01234567" for ch in s):
- try:
- v = int(s ,8)
- return v ,s
- except Exception:
- return None
- return None
-
-@dataclass
-class _Ctx:
- "Information used by many entries in the plan, plan specific command defaults, i.e. the plan context."
- read_rel_fpath: Path
- stage_root_dpath: Path
- defaults_map: dict[str,Any] = field(default_factory=dict) # this syntax gives each context instance a distinct dictionary.
-
-# ---------- planner singleton ----------
-
-class _Planner:
- "Given: staged config executions. Does: accumulate plan entries. Returns: plan map."
- def __init__(self)-> None:
- self._ctx: _Ctx|None = None
- self._entries_list: list[dict[str,Any]] = []
- self._meta_map: dict[str,Any] = {}
-
- # ---- framework (called by outer tools) ----
- def _begin(self ,read_rel_fpath: Path ,stage_root_dpath: Path)-> None:
- "Given: a config’s relative file path and stage root. Does: start context. Returns: None."
- self._ctx = _Ctx(read_rel_fpath=read_rel_fpath ,stage_root_dpath=stage_root_dpath)
-
- def _end(self)-> None:
- "Given: active context. Does: end it. Returns: None."
- self._ctx = None
-
- def _reset(self)-> None:
- "Given: n/a. Does: clear meta and entries. Returns: None."
- self._entries_list.clear()
- self._meta_map.clear()
- self._ctx = None
-
- # ---- exported for outer tools ----
- def plan_entries(self)-> list[dict[str,Any]]:
- "Given: n/a. Does: return a shallow copy of current entries. Returns: list[dict]."
- return list(self._entries_list)
-
- def set_meta(self ,**kv)-> None:
- "Given: keyword meta. Does: merge into meta_map. Returns: None."
- self._meta_map.update(kv)
-
- def plan_object(self)-> dict[str,Any]:
- "Packages a self-contained plan map ready for CBOR encoding.
- Given: accumulated meta/entries. Does: freeze a copy and stamp a version. Returns: dict.
- "
- return {
- "version_int": 1
- ,"meta_map": dict(self._meta_map)
- ,"entries_list": list(self._entries_list)
- }
-
- # ---- config API ----
- def init(
- self
- ,write_file_name: str
- ,write_dpath: str
- ,write_file_owner_name: str
- ,write_file_permissions: int|str
- ,read_file_contents: bytes|str|None=None
- )-> None:
- """
- Given: write filename ('.' → basename of config), destination dir path, owner name,
- permissions (int or '0644'), and optional read content (bytes or str).
- Does: store per-config defaults used by subsequent Stage.* calls.
- Returns: None.
- """
- if self._ctx is None:
- raise RuntimeError("Stage.init used without active context")
- fname = self._ctx.read_rel_fpath.name if write_file_name == "." else write_file_name
- if isinstance(read_file_contents ,str):
- content_bytes = read_file_contents.encode("utf-8")
- else:
- content_bytes = read_file_contents
- perm_norm = _norm_perm(write_file_permissions)
- if perm_norm is None:
- mode_int ,mode_octal_str = None ,None
- else:
- mode_int ,mode_octal_str = perm_norm
- self._ctx.defaults_map = {
- "dst_fname": fname
- ,"dst_dpath": write_dpath
- ,"owner_name": write_file_owner_name
- ,"mode_int": mode_int
- ,"mode_octal_str": mode_octal_str
- ,"content_bytes": content_bytes
- }
-
- def _require_defaults(self)-> dict[str,Any]:
- "Given: current ctx. Does: ensure Stage.init ran. Returns: defaults_map."
- if self._ctx is None or not self._ctx.defaults_map:
- raise RuntimeError("Stage.* called before Stage.init in this config")
- return self._ctx.defaults_map
-
- def displace(self)-> None:
- "Given: defaults. Does: append a displace op. Returns: None."
- d = self._require_defaults()
- self._entries_list.append({
- "op":"displace"
- ,"dst_dpath": d["dst_dpath"]
- ,"dst_fname": d["dst_fname"]
- })
-
- def copy(self)-> None:
- "Given: defaults. Does: append a copy op. Returns: None."
- d = self._require_defaults()
- self._entries_list.append({
- "op":"copy"
- ,"dst_dpath": d["dst_dpath"]
- ,"dst_fname": d["dst_fname"]
- ,"owner_name": d["owner_name"]
- ,"mode_int": d["mode_int"]
- ,"mode_octal_str": d["mode_octal_str"]
- ,"content_bytes": d["content_bytes"]
- })
-
- def delete(self)-> None:
- "Given: defaults. Does: append a delete op. Returns: None."
- d = self._require_defaults()
- self._entries_list.append({
- "op":"delete"
- ,"dst_dpath": d["dst_dpath"]
- ,"dst_fname": d["dst_fname"]
- })
-
-# exported singleton
-Stage = _Planner()
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-executor.py — StageHand outer/inner executor (MVP; UNPRIVILEGED for now)
-
-Phase 0 (bootstrap):
- - Ensure filter program exists (create default in CWD if --filter omitted)
- - Validate --stage exists
- - If --phase-0-then-stop: exit here (no scan ,no execution)
-
-Phase 1 (outer):
- - Discover every file under --stage; acceptance filter decides which to include
- - Execute each config’s configure(prov ,planner ,WriteFileMeta) into ONE Planner
- - Optionally print the planner; optionally stop
-
-Phase 2 (inner shim in same program for now; no privilege yet):
- - Encode plan to CBOR and hand to inner path
- - Inner decodes to a Journal and can print it
-"""
-
-from __future__ import annotations
-
-# no bytecode anywhere
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from pathlib import Path
-import argparse
-import getpass
-import tempfile
-import runpy
-import subprocess
-import datetime as _dt
-import stat
-
-# Local module: Planner.py (same directory)
-from Planner import (
- Planner ,PlanProvenance ,WriteFileMeta ,Journal ,Command,
-)
-
-# -------- default filter template (written to CWD when --filter not provided) --------
-
-DEFAULT_FILTER_FILENAME = "stagehand_filter.py"
-
-DEFAULT_FILTER_SOURCE = """# StageHand acceptance filter (default template)
-# Return True to include a config file ,False to skip it.
-# You receive a PlanProvenance object named `prov`.
-#
-# prov fields commonly used here:
-# prov.stage_root_dpath : Path → absolute path to the stage root
-# prov.config_abs_fpath : Path → absolute path to the candidate file
-# prov.config_rel_fpath : Path → path relative to the stage root
-# prov.read_dir_dpath : Path → directory of the candidate file
-# prov.read_fname : str → filename with trailing '.py' stripped (if present)
-#
-# Examples:
-#
-# 1) Accept everything (default behavior):
-# def accept(prov):
-# return True
-#
-# 2) Only accept configs in a 'dns/' namespace under the stage:
-# def accept(prov):
-# return prov.config_rel_fpath.as_posix().startswith("dns/")
-#
-# 3) Exclude editor backup files:
-# def accept(prov):
-# rel = prov.config_rel_fpath.as_posix()
-# return not (rel.endswith("~") or rel.endswith(".swp"))
-#
-# 4) Only accept Python files + a few non-Python names:
-# def accept(prov):
-# name = prov.config_abs_fpath.name
-# return name.endswith(".py") or name in {"hosts" ,"resolv.conf"}
-#
-# Choose ONE 'accept' definition. Below is the default:
-
-def accept(prov):
- return True
-"""
-
-# -------- utilities --------
-
-def iso_utc_now_str() -> str:
- return _dt.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
-
-def _ensure_filter_file(filter_arg: str|None) -> Path:
- """
- If --filter is provided ,return that path (must exist).
- Otherwise ,create ./stagehand_filter.py in the CWD if missing (writing a helpful template),
- and return its path.
- """
- if filter_arg:
- p = Path(filter_arg)
- if not p.is_file():
- raise RuntimeError(f"--filter file not found: {p}")
- return p
-
- p = Path.cwd() / DEFAULT_FILTER_FILENAME
- if not p.exists():
- try:
- p.write_text(DEFAULT_FILTER_SOURCE ,encoding="utf-8")
- print(f"(created default filter at {p})")
- except Exception as e:
- raise RuntimeError(f"failed to create default filter {p}: {e}")
- return p
-
-def _load_accept_func(filter_path: Path):
- env = runpy.run_path(str(filter_path))
- fn = env.get("accept")
- if not callable(fn):
- raise RuntimeError(f"{filter_path}: missing callable 'accept(prov)'")
- return fn
-
-def _walk_all_files(stage_root: Path):
- """
- Yield every file (regular or symlink) under stage_root recursively.
- We do not follow symlinked directories to avoid cycles.
- """
- root = stage_root.resolve()
- for dirpath ,dirnames ,filenames in os.walk(root ,followlinks=False):
- # prune symlinked dirs (files can still be symlinks)
- dirnames[:] = [d for d in dirnames if not os.path.islink(os.path.join(dirpath ,d))]
- for fname in filenames:
- p = Path(dirpath ,fname)
- try:
- st = p.lstat()
- if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
- yield p.resolve()
- except Exception:
- # unreadable/broken entries skipped
- continue
-
-def find_config_paths(stage_root: Path ,accept_func) -> list[Path]:
- out: list[tuple[int ,str ,Path]] = []
- root = stage_root.resolve()
- for p in _walk_all_files(stage_root):
- prov = PlanProvenance(stage_root=stage_root ,config_path=p)
- try:
- if accept_func(prov):
- rel = p.resolve().relative_to(root)
- out.append((len(rel.parts) ,rel.as_posix() ,p.resolve()))
- except Exception as e:
- raise RuntimeError(f"accept() failed on {prov.config_rel_fpath.as_posix()}: {e}")
- out.sort(key=lambda t: (t[0] ,t[1])) # (depth ,name)
- return [t[2] for t in out]
-
-# --- run all configs into ONE planner ---
-
-def _run_all_configs_into_single_planner(stage_root: Path ,cfgs: list[Path]) -> Planner:
- """
- Create a single Planner and execute each config's configure(prov ,planner ,WriteFileMeta)
- against it. Returns that single Planner containing the entire plan.
- """
- # seed with synthetic provenance; we overwrite per config before execution
- aggregate_prov = PlanProvenance(stage_root=stage_root ,config_path=stage_root / "(aggregate).py")
- planner = Planner(provenance=aggregate_prov)
-
- for cfg in cfgs:
- prov = PlanProvenance(stage_root=stage_root ,config_path=cfg)
- planner.set_provenance(prov)
-
- env = runpy.run_path(str(cfg))
- fn = env.get("configure")
- if not callable(fn):
- raise RuntimeError(f"{cfg}: missing callable configure(prov ,planner ,WriteFileMeta)")
-
- fn(prov ,planner ,WriteFileMeta)
-
- # annotate meta once ,on the single planner's journal
- j = planner.journal()
- j.set_meta(
- generator_prog_str="executor.py",
- generated_at_utc_str=iso_utc_now_str(),
- user_name_str=getpass.getuser(),
- host_name_str=os.uname().nodename if hasattr(os ,"uname") else "unknown",
- stage_root_dpath_str=str(stage_root.resolve()),
- configs_list=[str(p.resolve().relative_to(stage_root.resolve())) for p in cfgs],
- )
- return planner
-
-# ----- CBOR “matchbox” (simple wrapper kept local to executor) -----
-
-def _plan_to_cbor_bytes(planner: Planner) -> bytes:
- """Serialize a Planner's Journal to CBOR bytes."""
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- plan_dict = planner.journal().as_dictionary()
- return cbor2.dumps(plan_dict ,canonical=True)
-
-def _journal_from_cbor_bytes(data: bytes) -> Journal:
- """Rebuild a Journal from CBOR bytes."""
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- obj = cbor2.loads(data)
- if not isinstance(obj ,dict):
- raise ValueError("CBOR root must be a dict")
- return Journal(plan_dict=obj)
-
-# -------- inner executor (phase 2) --------
-
-def _inner_main(plan_path: Path ,phase2_print: bool ,phase2_then_stop: bool) -> int:
- """Inner executor path: decode CBOR → Journal; optionally print; (apply TBD)."""
- try:
- data = Path(plan_path).read_bytes()
- except Exception as e:
- print(f"error: failed to read plan file: {e}" ,file=sys.stderr)
- return 2
-
- try:
- journal = _journal_from_cbor_bytes(data)
- except Exception as e:
- print(f"error: failed to decode CBOR: {e}" ,file=sys.stderr)
- return 2
-
- if phase2_print:
- journal.print()
-
- if phase2_then_stop:
- return 0
-
- # (Stage 3 apply would go here; omitted in MVP)
- return 0
-
-# -------- outer executor (phase 1 & handoff) --------
-
-def _outer_main(stage_root: Path ,accept_func ,args) -> int:
- if not stage_root.is_dir():
- print(f"error: --stage not a directory: {stage_root}" ,file=sys.stderr)
- return 2
-
- cfgs = find_config_paths(stage_root ,accept_func)
- if not cfgs:
- print("No configuration files found.")
- return 0
-
- try:
- master = _run_all_configs_into_single_planner(stage_root ,cfgs)
- except SystemExit:
- raise
- except Exception as e:
- print(f"error: executing configs: {e}" ,file=sys.stderr)
- return 2
-
- if args.phase_1_print:
- master.print()
-
- if args.phase_1_then_stop:
- return 0
-
- # Phase 2: encode CBOR and invoke inner path (same script ,--inner)
- try:
- cbor_bytes = _plan_to_cbor_bytes(master)
- except Exception as e:
- print(f"error: CBOR encode failed: {e}" ,file=sys.stderr)
- return 2
-
- with tempfile.NamedTemporaryFile(prefix="stagehand_plan_" ,suffix=".cbor" ,delete=False) as tf:
- tf.write(cbor_bytes)
- plan_path = tf.name
-
- try:
- cmd = [
- sys.executable,
- str(Path(__file__).resolve()),
- "--inner",
- "--plan" ,plan_path,
- ]
- if args.phase_2_print:
- cmd.append("--phase-2-print")
- if args.phase_2_then_stop:
- cmd.append("--phase-2-then-stop")
-
- proc = subprocess.run(cmd)
- return proc.returncode
- finally:
- try:
- os.unlink(plan_path)
- except Exception:
- pass
-
-# -------- CLI --------
-
-def main(argv: list[str] | None = None) -> int:
- ap = argparse.ArgumentParser(
- prog="executor.py",
- description="StageHand outer/inner executor (plan → CBOR → decode).",
- )
- ap.add_argument("--stage" ,default="stage",
- help="stage root directory (default: ./stage)")
- ap.add_argument(
- "--filter",
- default="",
- help=f"path to acceptance filter program exporting accept(prov) "
- f"(default: ./{DEFAULT_FILTER_FILENAME}; created if missing)"
- )
- ap.add_argument(
- "--phase-0-then-stop",
- action="store_true",
- help="stop after arg checks & filter bootstrap (no stage scan)"
- )
-
- # Phase-1 (outer) controls
- ap.add_argument("--phase-1-print" ,action="store_true" ,help="print master planner (phase 1)")
- ap.add_argument("--phase-1-then-stop" ,action="store_true" ,help="stop after phase 1")
-
- # Phase-2 (inner) controls (outer forwards these to inner)
- ap.add_argument("--phase-2-print" ,action="store_true" ,help="print decoded journal (phase 2)")
- ap.add_argument("--phase-2-then-stop" ,action="store_true" ,help="stop after phase 2 decode")
-
- # Inner-only flags (not for users)
- ap.add_argument("--inner" ,action="store_true" ,help=argparse.SUPPRESS)
- ap.add_argument("--plan" ,default=None ,help=argparse.SUPPRESS)
-
- args = ap.parse_args(argv)
-
- # Inner path
- if args.inner:
- if not args.plan:
- print("error: --inner requires --plan <file>" ,file=sys.stderr)
- return 2
- return _inner_main(Path(args.plan),
- phase2_print=args.phase_2_print,
- phase2_then_stop=args.phase_2_then_stop)
-
- # Phase 0: bootstrap & stop (no scan)
- stage_root = Path(args.stage)
- try:
- filter_path = _ensure_filter_file(args.filter or None)
- except Exception as e:
- print(f"error: {e}" ,file=sys.stderr)
- return 2
-
- if not stage_root.exists():
- print(f"error: --stage not found: {stage_root}" ,file=sys.stderr)
- return 2
- if not stage_root.is_dir():
- print(f"error: --stage is not a directory: {stage_root}" ,file=sys.stderr)
- return 2
-
- if args.phase_0_then_stop:
- print(f"phase-0 OK: stage at {stage_root.resolve()} and filter at {filter_path}")
- return 0
-
- # Load acceptance function and proceed with outer
- try:
- accept_func = _load_accept_func(filter_path)
- except Exception as e:
- print(f"error: {e}" ,file=sys.stderr)
- return 2
-
- return _outer_main(stage_root ,accept_func ,args)
-
-if __name__ == "__main__":
- sys.exit(main())
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-ls_stage.py — list staged files and their header-declared install metadata.
-
-Header line format (first line of each file):
- <owner> <permissions> <write_file_name> <target_directory_path>
-
-- owner: username string (need not exist until install time)
-- permissions: four octal digits, e.g. 0644
-- write_file_name: '.' means use the read file's basename, else use the given POSIX filename
-- target_directory_path: POSIX directory path (usually absolute, e.g. /etc/unbound)
-
-Output formats:
-- list (default): "read_file_path: owner permissions write_file_name target_directory_path"
-- table: columns aligned for readability
-"""
-
-from __future__ import annotations
-
-# never write bytecode (root/sudo friendly)
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from dataclasses import dataclass
-from pathlib import Path
-import argparse
-import re
-
-# === Stage utilities (importable) ===
-
-def stage_read_file_paths(stage_root: Path)-> list[Path]:
- """Given: stage_root directory.
- Does: recursively enumerate regular files (follows symlinks to files), keep paths relative to stage_root.
- Returns: list[Path] of POSIX-order sorted relative paths (no leading slash).
- """
- rels: list[Path] = []
- for p in stage_root.rglob("*"):
- try:
- if p.is_file(): # follows symlink-to-file
- rels.append(p.relative_to(stage_root))
- except (FileNotFoundError ,RuntimeError):
- # broken link or race; skip conservatively
- continue
- return sorted(rels ,key=lambda x: x.as_posix())
-
-@dataclass
-class StageRow:
- read_rel: Path # e.g. Path("etc/unbound/unbound.conf.staged")
- owner: str # token[0]
- perm_octal_str: str # token[1], exactly as in header (validated ####)
- perm_int: int # token[1] parsed as base-8
- write_name: str # token[2] ('.' resolved to read_rel.name)
- target_dir: Path # token[3] (Path)
- header_raw: str # original header line (sans newline)
-
- # convenience
- def write_abs(self ,root: Path)-> Path:
- return (root / self.target_dir.relative_to("/")) if self.target_dir.is_absolute() else (root / self.target_dir) / self.write_name
-
-# header parsing rules
-_PERM_RE = re.compile(r"^[0-7]{4}$")
-
-def parse_stage_header_line(header: str ,read_rel: Path)-> tuple[StageRow|None ,str|None]:
- """Given: raw first line of a staged file and its stage-relative path.
- Does: parse '<owner> <perm> <write_name> <target_dir>' with max 4 tokens (target_dir may contain spaces if quoted not required).
- Returns: (StageRow, None) on success, or (None, error_message) on failure. Does NOT touch filesystem.
- """
- # strip BOM and trailing newline/spaces
- h = header.lstrip("\ufeff").strip()
- if not h:
- return None ,f"empty header line in {read_rel}"
- parts = h.split(maxsplit=3)
- if len(parts) != 4:
- return None ,f"malformed header in {read_rel}: expected 4 fields, got {len(parts)}"
- owner ,perm_s ,write_name ,target_dir_s = parts
-
- if not _PERM_RE.fullmatch(perm_s):
- return None ,f"invalid permissions '{perm_s}' in {read_rel}: must be four octal digits"
-
- # resolve '.' → basename
- resolved_write_name = read_rel.name if write_name == "." else write_name
-
- # MVP guard: write_name should be a single filename (no '/')
- if "/" in resolved_write_name:
- return None ,f"write_file_name must not contain '/': got '{resolved_write_name}' in {read_rel}"
-
- # target dir may be absolute (recommended) or relative (we treat relative as under the install root)
- target_dir = Path(target_dir_s)
-
- try:
- row = StageRow(
- read_rel = read_rel
- ,owner = owner
- ,perm_octal_str = perm_s
- ,perm_int = int(perm_s ,8)
- ,write_name = resolved_write_name
- ,target_dir = target_dir
- ,header_raw = h
- )
- return row ,None
- except Exception as e:
- return None ,f"internal parse error in {read_rel}: {e}"
-
-def read_first_line(p: Path)-> str:
- """Return the first line (sans newline). UTF-8 with BOM tolerant."""
- with open(p ,"r" ,encoding="utf-8" ,errors="replace") as fh:
- line = fh.readline()
- return line.rstrip("\n\r")
-
-def scan_stage(stage_root: Path)-> tuple[list[StageRow] ,list[str]]:
- """Given: stage_root.
- Does: enumerate files, parse each header line, collect rows and errors.
- Returns: (rows, errors)
- """
- rows: list[StageRow] = []
- errs: list[str] = []
- for rel in stage_read_file_paths(stage_root):
- abs_path = stage_root / rel
- try:
- header = read_first_line(abs_path)
- except Exception as e:
- errs.append(f"read error in {rel}: {e}")
- continue
- row ,err = parse_stage_header_line(header ,rel)
- if err:
- errs.append(err)
- else:
- rows.append(row) # type: ignore[arg-type]
- return rows ,errs
-
-# === Printers ===
-
-def print_list(rows: list[StageRow])-> None:
- """Print: 'read_file_path: owner permissions write_file_name target_directory_path' per line."""
- for r in rows:
- print(f"{r.read_rel.as_posix()}: {r.owner} {r.perm_octal_str} {r.write_name} {r.target_dir}")
-
-def print_table(rows: list[StageRow])-> None:
- """Aligned table printer (no headers, just data in columns)."""
- if not rows:
- return
- a = [r.read_rel.as_posix() for r in rows]
- b = [r.owner for r in rows]
- c = [r.perm_octal_str for r in rows]
- d = [r.write_name for r in rows]
- e = [str(r.target_dir) for r in rows]
- wa = max(len(s) for s in a)
- wb = max(len(s) for s in b)
- wc = max(len(s) for s in c)
- wd = max(len(s) for s in d)
- # e (target_dir) left ragged
- for sa ,sb ,sc ,sd ,se in zip(a ,b ,c ,d ,e):
- print(f"{sa:<{wa}} {sb:<{wb}} {sc:<{wc}} {sd:<{wd}} {se}")
-
-# === Orchestrator ===
-
-def ls_stage(stage_root: Path ,fmt: str="list")-> int:
- """Given: stage_root and output format ('list'|'table').
- Does: scan and parse staged files, print in the requested format; report syntax errors to stderr.
- Returns: 0 on success; 1 if any syntax errors were encountered.
- """
- rows ,errs = scan_stage(stage_root)
- if fmt == "table":
- print_table(rows)
- else:
- print_list(rows)
- if errs:
- print("\nerror(s):" ,file=sys.stderr)
- for e in errs:
- print(f" - {e}" ,file=sys.stderr)
- return 1
- return 0
-
-# === CLI ===
-
-def main(argv: list[str] | None=None)-> int:
- ap = argparse.ArgumentParser(
- prog="ls_stage.py"
- ,description="List staged files and their header-declared install metadata."
- )
- ap.add_argument("--stage" ,default="stage",help="stage directory (default: ./stage)")
- ap.add_argument("--format" ,choices=["list" ,"table"] ,default="list"
- ,help="output format (default: list)")
- args = ap.parse_args(argv)
- stage_root = Path(args.stage)
- if not stage_root.exists() or not stage_root.is_dir():
- print(f"error: stage directory not found or not a directory: {stage_root}" ,file=sys.stderr)
- return 2
- return ls_stage(stage_root ,fmt=args.format)
-
-if __name__ == "__main__":
- sys.exit(main())
+++ /dev/null
-table inet NO-IPV6 {
- chain input {
- type filter hook input priority raw; policy accept;
- meta nfproto ipv6 counter comment "drop all IPv6 inbound" drop
- }
-
- chain output {
- type filter hook output priority raw; policy accept;
- meta nfproto ipv6 counter comment "drop all IPv6 outbound" drop
- }
-
- chain forward {
- type filter hook forward priority raw; policy accept;
- meta nfproto ipv6 counter comment "drop all IPv6 forward" drop
- }
-}
+++ /dev/null
-table inet SUBU-DNS-REDIRECT {
- chain output {
- type nat hook output priority -100; policy accept;
-
- # Redirect DNS for the subu UIDs to local Unbound listeners
- meta skuid 2017 udp dport 53 redirect to :5301
- meta skuid 2018 udp dport 53 redirect to :5302
- meta skuid 2017 tcp dport 53 redirect to :5301
- meta skuid 2018 tcp dport 53 redirect to :5302
- }
-}
-
-table inet SUBU-PORT-EGRESS {
- chain output {
- type filter hook output priority 0; policy accept;
-
- # Always allow loopback on egress
- oifname "lo" accept
-
- # No IPv6 for subu (until you reintroduce v6)
- meta skuid {2017,2018} meta nfproto ipv6 counter comment "no IPv6 for subu" drop
-
- ##### x6 (UID 2018)
- # Block some exfil channels regardless of iface
- meta skuid 2018 tcp dport {25,465,587} counter comment "block SMTP/Submission" drop
- meta skuid 2018 udp dport {3478,5349,19302-19309} counter comment "block STUN/TURN" drop
- meta skuid 2018 tcp dport 853 counter comment "block DoT (TCP/853)" drop
-
- # (Optional) allow ICMP echo out via x6
- meta skuid 2018 oifname "x6" ip protocol icmp icmp type echo-request accept
-
- # Enforce interface binding
- meta skuid 2018 oifname "x6" accept
- meta skuid 2018 oifname != "x6" counter comment "x6 must use wg x6" drop
-
- ##### US (UID 2017)
- meta skuid 2017 tcp dport {25,465,587} counter drop comment "block SMTP/Submission"
- meta skuid 2017 udp dport {3478,5349,19302-19309} counter drop comment "block STUN/TURN"
- meta skuid 2017 tcp dport 853 counter drop comment "block DoT (TCP/853)"
-
- # (Optional) ICMP via US
- meta skuid 2017 oifname "US" ip protocol icmp icmp type echo-request accept
-
- meta skuid 2017 oifname "US" accept
- meta skuid 2017 oifname != "US" counter comment "US must use wg US" drop
- }
-}
+++ /dev/null
-[Unit]
-Description=Unbound DNS instance for %i (per-subu tunnel egress)
-After=network-online.target wg-quick@%i.service
-Requires=wg-quick@%i.service
-Wants=network-online.target
-
-[Service]
-Type=simple
-ExecStart=/usr/sbin/unbound -d -p -c /etc/unbound/unbound-%i.conf
-User=unbound
-Group=unbound
-Restart=on-failure
-RestartSec=2s
-AmbientCapabilities=CAP_NET_BIND_SERVICE
-CapabilityBoundingSet=CAP_NET_BIND_SERVICE
-NoNewPrivileges=true
-
-[Install]
-WantedBy=multi-user.target
+++ /dev/null
-server:
- username: "unbound"
- chroot: ""
- directory: "/etc/unbound"
- do-daemonize: no
- interface: 127.0.0.1@5301
- hide-identity: yes
- hide-version: yes
- harden-glue: yes
- harden-dnssec-stripped: yes
- qname-minimisation: yes
- prefetch: yes
- outgoing-interface: 10.0.0.1
-
-forward-zone:
- name: "."
- forward-addr: 1.1.1.1
- forward-addr: 1.0.0.1
+++ /dev/null
-server:
- username: "unbound"
- chroot: ""
- directory: "/etc/unbound"
- do-daemonize: no
- interface: 127.0.0.1@5302
- hide-identity: yes
- hide-version: yes
- harden-glue: yes
- harden-dnssec-stripped: yes
- qname-minimisation: yes
- prefetch: yes
- outgoing-interface: 10.8.0.2
-
-forward-zone:
- name: "."
- forward-addr: 1.1.1.1
- forward-addr: 1.0.0.1
+++ /dev/null
-#!/usr/bin/env bash
-set -euo pipefail
-echo "== DNS status =="
-systemctl --no-pager --full status DNS-redirect unbound@US unbound@x6 || true
-echo
-echo "== nftables =="
-nft list table inet NAT-DNS-REDIRECT || true
-echo
-echo "== Unbound logs (last 50 lines each) =="
-journalctl -u unbound@US -n 50 --no-pager || true
-echo
-journalctl -u unbound@x6 -n 50 --no-pager || true
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-stage_show_plan.py — run staged configs (UNPRIVILEGED) and print the plan.
-
-Given: a stage root directory.
-Does: loads Stage.py, executes each config, builds a native plan map, summarizes it.
-Returns: exit code 0 on success, non-zero on error.
-"""
-from __future__ import annotations
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from pathlib import Path
-import argparse ,importlib.util ,runpy ,socket ,getpass ,time ,hashlib
-
-# ---------- helpers ----------
-
-def _load_stage_module(stage_root_dpath: Path):
- "Given: stage root path. Does: load Stage.py as module 'Stage'. Returns: module."
- mod_fpath = stage_root_dpath/"Stage.py"
- if not mod_fpath.exists():
- raise FileNotFoundError(f"Stage.py not found at {mod_fpath}")
- spec = importlib.util.spec_from_file_location("Stage" ,str(mod_fpath))
- mod = importlib.util.module_from_spec(spec)
- sys.modules["Stage"] = mod
- assert spec and spec.loader
- spec.loader.exec_module(mod) # type: ignore
- return mod
-
-def _config_rel_fpaths(stage_root_dpath: Path)-> list[Path]:
- "Given: stage root. Does: collect *.py (excluding Stage.py) as relative file paths. Returns: list[Path]."
- rel_fpath_list: list[Path] = []
- for p in stage_root_dpath.rglob("*.py"):
- if p.name == "Stage.py": continue
- if p.is_file():
- rel_fpath_list.append(p.relative_to(stage_root_dpath))
- return sorted(rel_fpath_list ,key=lambda x: x.as_posix())
-
-def _sha256_hex(b: bytes)-> str:
- "Given: bytes. Does: sha256. Returns: hex string."
- return hashlib.sha256(b).hexdigest()
-
-# ---------- main ----------
-
-def main(argv: list[str]|None=None)-> int:
- "Given: CLI. Does: show plan. Returns: exit code."
- ap = argparse.ArgumentParser(prog="stage_show_plan.py"
- ,description="Run staged config scripts and print the resulting plan.")
- ap.add_argument("--stage",default="stage",help="stage directory (default: ./stage)")
- args = ap.parse_args(argv)
-
- stage_root_dpath = Path(args.stage)
- StageMod = _load_stage_module(stage_root_dpath)
- Stage = StageMod.Stage
- Stage._reset()
- Stage.set_meta(
- planner_user_name=getpass.getuser()
- ,planner_uid_int=os.getuid()
- ,planner_gid_int=os.getgid()
- ,host_name=socket.gethostname()
- ,created_utc_str=time.strftime("%Y-%m-%dT%H:%M:%SZ",time.gmtime())
- )
-
- for rel_fpath in _config_rel_fpaths(stage_root_dpath):
- Stage._begin(read_rel_fpath=rel_fpath ,stage_root_dpath=stage_root_dpath)
- runpy.run_path(str(stage_root_dpath/rel_fpath) ,run_name="__main__")
- Stage._end()
-
- plan_map = Stage.plan_object()
- entries_list = plan_map["entries_list"]
- print(f"Plan version: {plan_map['version_int']}")
- print(f"Planner: {plan_map['meta_map'].get('planner_user_name')}@{plan_map['meta_map'].get('host_name')} "
- f"UID:{plan_map['meta_map'].get('planner_uid_int')} GID:{plan_map['meta_map'].get('planner_gid_int')}")
- print(f"Created: {plan_map['meta_map'].get('created_utc_str')}")
- print(f"Entries: {len(entries_list)}\n")
-
- for i ,e_map in enumerate(entries_list ,1):
- op = e_map.get("op")
- dst_fpath_str = f"{e_map.get('dst_dpath')}/{e_map.get('dst_fname')}"
- if op == "copy":
- content = e_map.get("content_bytes") or b""
- sz = len(content)
- mode = e_map.get("mode_octal_str") or "????"
- owner = e_map.get("owner_name") or "?"
- h = _sha256_hex(content)
- print(f"{i:02d}. copy -> {dst_fpath_str} mode {mode} owner {owner} bytes {sz} sha256 {h[:16]}…")
- elif op == "displace":
- print(f"{i:02d}. displace -> {dst_fpath_str}")
- elif op == "delete":
- print(f"{i:02d}. delete -> {dst_fpath_str}")
- else:
- print(f"{i:02d}. ?op? -> {dst_fpath_str} ({op})")
- return 0
-
-if __name__ == "__main__":
- sys.exit(main())
+++ /dev/null
-#!/usr/bin/env -S python3 -B
-"""
-executor.py — StageHand outer/inner executor (MVP; UNPRIVILEGED for now)
-
-Phase 0 (bootstrap):
- - Ensure filter program exists (create default in CWD if --filter omitted)
- - Validate --stage exists
- - If --phase-0-then-stop: exit here (no scan, no execution)
-
-Phase 1 (outer):
- - Discover every file under --stage; acceptance filter decides which to include
- - Execute each config’s configure(prov ,planner ,WriteFileMeta) into ONE Planner
- - Optionally print the planner; optionally stop
-
-Phase 2 (inner shim in same program for now; no privilege yet):
- - Encode plan to CBOR and hand to inner path
- - Inner decodes to a Journal and can print it
-"""
-
-from __future__ import annotations
-
-# no bytecode anywhere
-import sys ,os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1")
-
-from pathlib import Path
-import argparse
-import getpass
-import tempfile
-import runpy
-import subprocess
-import datetime as _dt
-import stat
-
-# Local module: Planner.py (same directory)
-from Planner import (
- Planner
- ,PlanProvenance
- ,WriteFileMeta
- ,Journal
- ,Command
-)
-
-# -------- default filter template (written to CWD when --filter not provided) --------
-
-DEFAULT_FILTER_FILENAME = "stagehand_filter.py"
-
-DEFAULT_FILTER_SOURCE = """# StageHand acceptance filter (default template)
-# Return True to include a config file, False to skip it.
-# You receive a PlanProvenance object named `prov`.
-#
-# prov fields commonly used here:
-# prov.stage_root_dpath : Path → absolute path to the stage root
-# prov.config_abs_fpath : Path → absolute path to the candidate file
-# prov.config_rel_fpath : Path → path relative to the stage root
-# prov.read_dir_dpath : Path → directory of the candidate file
-# prov.read_fname : str → filename with trailing '.py' stripped (if present)
-#
-# Examples:
-#
-# 1) Accept everything (default behavior):
-# def accept(prov):
-# return True
-#
-# 2) Only accept configs in a 'dns/' namespace under the stage:
-# def accept(prov):
-# return prov.config_rel_fpath.as_posix().startswith("dns/")
-#
-# 3) Exclude editor backup files:
-# def accept(prov):
-# rel = prov.config_rel_fpath.as_posix()
-# return not (rel.endswith("~") or rel.endswith(".swp"))
-#
-# 4) Only accept Python files + a few non-Python names:
-# def accept(prov):
-# name = prov.config_abs_fpath.name
-# return name.endswith(".py") or name in {"hosts" ,"resolv.conf"}
-#
-# Choose ONE 'accept' definition. Below is the default:
-
-def accept(prov):
- return True
-"""
-
-# -------- utilities --------
-
-def iso_utc_now_str()-> str:
- return _dt.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
-
-def _ensure_filter_file(filter_arg: str|None)-> Path:
- """
- If --filter is provided, return that path (must exist).
- Otherwise, create ./stagehand_filter.py in the CWD if missing (writing a helpful template),
- and return its path.
- """
- if filter_arg:
- p = Path(filter_arg)
- if not p.is_file():
- raise RuntimeError(f"--filter file not found: {p}")
- return p
-
- p = Path.cwd()/DEFAULT_FILTER_FILENAME
- if not p.exists():
- try:
- p.write_text(DEFAULT_FILTER_SOURCE ,encoding="utf-8")
- print(f"(created default filter at {p})")
- except Exception as e:
- raise RuntimeError(f"failed to create default filter {p}: {e}")
- return p
-
-def _load_accept_func(filter_path: Path):
- env = runpy.run_path(str(filter_path))
- fn = env.get("accept")
- if not callable(fn):
- raise RuntimeError(f"{filter_path}: missing callable 'accept(prov)'")
- return fn
-
-def _walk_all_files(stage_root: Path):
- """
- Yield every file (regular or symlink) under stage_root recursively.
- We do not follow symlinked directories to avoid cycles.
- """
- root = stage_root.resolve()
- for dirpath ,dirnames ,filenames in os.walk(root ,followlinks=False):
- # prune symlinked dirs (files can still be symlinks)
- dirnames[:] = [d for d in dirnames if not os.path.islink(os.path.join(dirpath ,d))]
- for fname in filenames:
- p = Path(dirpath ,fname)
- try:
- st = p.lstat()
- if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
- yield p.resolve()
- except Exception:
- # unreadable/broken entries skipped
- continue
-
-def find_config_paths(stage_root: Path ,accept_func)-> list[Path]:
- """
- Return files accepted by the Python acceptance function: accept(prov) → True/False.
- Ordered breadth-first by depth, then lexicographically by relative path.
- """
- out: list[tuple[int ,str ,Path]] = []
- root = stage_root.resolve()
- for p in _walk_all_files(stage_root):
- prov = PlanProvenance(stage_root=stage_root ,config_path=p)
- try:
- if accept_func(prov):
- rel = p.resolve().relative_to(root)
- out.append((len(rel.parts) ,rel.as_posix() ,p.resolve()))
- except Exception as e:
- raise RuntimeError(f"accept() failed on {prov.config_rel_fpath.as_posix()}: {e}")
- out.sort(key=lambda t: (t[0] ,t[1])) # (depth ,name)
- return [t[2] for t in out]
-
-# --- run all configs into ONE planner ---
-
-def _run_all_configs_into_single_planner(stage_root: Path ,cfgs: list[Path])-> Planner:
- """
- Create a single Planner and execute each config's configure(prov, planner, WriteFileMeta)
- against it. Returns that single Planner containing the entire plan.
- """
- # seed with synthetic provenance; we overwrite per config before execution
- aggregate_prov = PlanProvenance(stage_root=stage_root ,config_path=stage_root/"(aggregate).py")
- planner = Planner(provenance=aggregate_prov)
-
- for cfg in cfgs:
- prov = PlanProvenance(stage_root=stage_root ,config_path=cfg)
- planner.set_provenance(prov)
-
- env = runpy.run_path(str(cfg))
- fn = env.get("configure")
- if not callable(fn):
- raise RuntimeError(f"{cfg}: missing callable configure(prov ,planner ,WriteFileMeta)")
-
- fn(prov ,planner ,WriteFileMeta)
-
- # annotate meta once, on the single planner's journal
- j = planner.journal()
- j.set_meta(
- generator_prog_str="executor.py"
- ,generated_at_utc_str=iso_utc_now_str()
- ,user_name_str=getpass.getuser()
- ,host_name_str=os.uname().nodename if hasattr(os ,"uname") else "unknown"
- ,stage_root_dpath_str=str(stage_root.resolve())
- ,configs_list=[str(p.resolve().relative_to(stage_root.resolve())) for p in cfgs]
- )
- return planner
-
-# ----- CBOR “matchbox” (simple wrapper kept local to executor) -----
-
-def _plan_to_cbor_bytes(planner: Planner)-> bytes:
- "Serialize a Planner's Journal to CBOR bytes."
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- plan_dict = planner.journal().as_dictionary()
- return cbor2.dumps(plan_dict ,canonical=True)
-
-def _journal_from_cbor_bytes(data: bytes)-> Journal:
- "Rebuild a Journal from CBOR bytes."
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- obj = cbor2.loads(data)
- if not isinstance(obj ,dict):
- raise ValueError("CBOR root must be a dict")
- return Journal(plan_dict=obj)
-
-# -------- inner executor (phase 2) --------
-
-def _inner_main(plan_path: Path ,phase2_print: bool ,phase2_then_stop: bool)-> int:
- "Inner executor path: decode CBOR → Journal; optionally print; (apply TBD)."
- try:
- data = Path(plan_path).read_bytes()
- except Exception as e:
- print(f"error: failed to read plan file: {e}" ,file=sys.stderr)
- return 2
-
- try:
- journal = _journal_from_cbor_bytes(data)
- except Exception as e:
- print(f"error: failed to decode CBOR: {e}" ,file=sys.stderr)
- return 2
-
- if phase2_print:
- journal.print()
-
- if phase2_then_stop:
- return 0
-
- # (Stage 3 apply would go here; omitted in MVP)
- return 0
-
-# -------- outer executor (phase 1 & handoff) --------
-
-def _outer_main(stage_root: Path ,accept_func ,args)-> int:
- if not stage_root.is_dir():
- print(f"error: --stage not a directory: {stage_root}" ,file=sys.stderr)
- return 2
-
- cfgs = find_config_paths(stage_root ,accept_func)
- if not cfgs:
- print("No configuration files found.")
- return 0
-
- try:
- master = _run_all_configs_into_single_planner(stage_root ,cfgs)
- except SystemExit:
- raise
- except Exception as e:
- print(f"error: executing configs: {e}" ,file=sys.stderr)
- return 2
-
- if args.phase_1_print:
- master.print()
-
- if args.phase_1_then_stop:
- return 0
-
- # Phase 2: encode CBOR and invoke inner path (same script, --inner)
- try:
- cbor_bytes = _plan_to_cbor_bytes(master)
- except Exception as e:
- print(f"error: CBOR encode failed: {e}" ,file=sys.stderr)
- return 2
-
- with tempfile.NamedTemporaryFile(prefix="stagehand_plan_" ,suffix=".cbor" ,delete=False) as tf:
- tf.write(cbor_bytes)
- plan_path = tf.name
-
- try:
- cmd = [
- sys.executable
- ,str(Path(__file__).resolve())
- ,"--inner"
- ,"--plan" ,plan_path
- ]
- if args.phase_2_print:
- cmd.append("--phase-2-print")
- if args.phase_2_then_stop:
- cmd.append("--phase-2-then-stop")
-
- proc = subprocess.run(cmd)
- return proc.returncode
- finally:
- try:
- os.unlink(plan_path)
- except Exception:
- pass
-
-# -------- CLI --------
-
-def main(argv: list[str]|None=None)-> int:
- ap = argparse.ArgumentParser(
- prog="executor.py"
- ,description="StageHand outer/inner executor (plan → CBOR → decode)."
- )
- ap.add_argument("--stage" ,default="stage"
- ,help="stage root directory (default: ./stage)")
- ap.add_argument(
- "--filter"
- ,default=""
- ,help=f"path to acceptance filter program exporting accept(prov) "
- f"(default: ./{DEFAULT_FILTER_FILENAME}; created if missing)"
- )
- ap.add_argument(
- "--phase-0-then-stop"
- ,action="store_true"
- ,help="stop after arg checks & filter bootstrap (no stage scan)"
- )
-
- # Phase-1 (outer) controls
- ap.add_argument("--phase-1-print" ,action="store_true" ,help="print master planner (phase 1)")
- ap.add_argument("--phase-1-then-stop" ,action="store_true" ,help="stop after phase 1")
-
- # Phase-2 (inner) controls (outer forwards these to inner)
- ap.add_argument("--phase-2-print" ,action="store_true" ,help="print decoded journal (phase 2)")
- ap.add_argument("--phase-2-then-stop" ,action="store_true" ,help="stop after phase 2 decode")
-
- # Inner-only flags (not for users)
- ap.add_argument("--inner" ,action="store_true" ,help=argparse.SUPPRESS)
- ap.add_argument("--plan" ,default=None ,help=argparse.SUPPRESS)
-
- args = ap.parse_args(argv)
-
- # Inner path
- if args.inner:
- if not args.plan:
- print("error: --inner requires --plan <file>" ,file=sys.stderr)
- return 2
- return _inner_main(Path(args.plan)
- ,phase2_print=args.phase_2_print
- ,phase2_then_stop=args.phase_2_then_stop)
-
- # Phase 0: bootstrap & stop (no scan)
- stage_root = Path(args.stage)
- try:
- filter_path = _ensure_filter_file(args.filter or None)
- except Exception as e:
- print(f"error: {e}" ,file=sys.stderr)
- return 2
-
- if not stage_root.exists():
- print(f"error: --stage not found: {stage_root}" ,file=sys.stderr)
- return 2
- if not stage_root.is_dir():
- print(f"error: --stage is not a directory: {stage_root}" ,file=sys.stderr)
- return 2
-
- if args.phase_0_then_stop:
- print(f"phase-0 OK: stage at {stage_root.resolve()} and filter at {filter_path}")
- return 0
-
- # Load acceptance function and proceed with outer
- try:
- accept_func = _load_accept_func(filter_path)
- except Exception as e:
- print(f"error: {e}" ,file=sys.stderr)
- return 2
-
- return _outer_main(stage_root ,accept_func ,args)
-
-if __name__ == "__main__":
- sys.exit(main())
+++ /dev/null
-
-#!/usr/bin/env -S python3 -B
-"""
-executor.py — StageHand outer/inner executor (MVP; UNPRIVILEGED for now)
-
-Phase 0 (bootstrap):
- - Ensure filter program exists (create default in CWD if --filter omitted)
- - Validate --stage exists
- - If --phase-0-then-stop: exit here (no scan, no execution)
-
-Phase 1 (outer):
- - Discover every file under --stage; acceptance filter decides which to include
- - Execute each config’s configure(prov, planner, WriteFileMeta) into ONE Planner
- - Optionally print the planner; optionally stop
-
-Phase 2 (inner shim in same program for now; no privilege yet):
- - Encode plan to CBOR and hand to inner path
- - Inner decodes to a Journal and can print it
-"""
-
-from __future__ import annotations
-
-# no bytecode anywhere
-import sys, os
-sys.dont_write_bytecode = True
-os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1")
-
-from pathlib import Path
-import argparse
-import getpass
-import tempfile
-import runpy
-import subprocess
-import datetime as _dt
-import stat
-
-# Local module: Planner.py (same directory)
-from Planner import (
- Planner, PlanProvenance, WriteFileMeta, Journal, Command,
-)
-
-# -------- default filter template (written to CWD when --filter not provided) --------
-
-DEFAULT_FILTER_FILENAME = "stagehand_filter.py"
-
-DEFAULT_FILTER_SOURCE = """# StageHand acceptance filter (default template)
-# Return True to include a config file, False to skip it.
-# You receive a PlanProvenance object named `prov`.
-#
-# prov fields commonly used here:
-# prov.stage_root_dpath : Path → absolute path to the stage root
-# prov.config_abs_fpath : Path → absolute path to the candidate file
-# prov.config_rel_fpath : Path → path relative to the stage root
-# prov.read_dir_dpath : Path → directory of the candidate file
-# prov.read_fname : str → filename with trailing '.py' stripped (if present)
-#
-# Examples:
-#
-# 1) Accept everything (default behavior):
-# def accept(prov):
-# return True
-#
-# 2) Only accept configs in a 'dns/' namespace under the stage:
-# def accept(prov):
-# return prov.config_rel_fpath.as_posix().startswith("dns/")
-#
-# 3) Exclude editor backup files:
-# def accept(prov):
-# rel = prov.config_rel_fpath.as_posix()
-# return not (rel.endswith("~") or rel.endswith(".swp"))
-#
-# 4) Only accept Python files + a few non-Python names:
-# def accept(prov):
-# name = prov.config_abs_fpath.name
-# return name.endswith(".py") or name in {"hosts", "resolv.conf"}
-#
-# Choose ONE 'accept' definition. Below is the default:
-
-def accept(prov):
- return True
-"""
-
-# -------- utilities --------
-
-def iso_utc_now_str() -> str:
- return _dt.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
-
-def _ensure_filter_file(filter_arg: str|None) -> Path:
- """
- If --filter is provided, return that path (must exist).
- Otherwise, create ./stagehand_filter.py in the CWD if missing (writing a helpful template),
- and return its path.
- """
- if filter_arg:
- p = Path(filter_arg)
- if not p.is_file():
- raise RuntimeError(f"--filter file not found: {p}")
- return p
-
- p = Path.cwd() / DEFAULT_FILTER_FILENAME
- if not p.exists():
- try:
- p.write_text(DEFAULT_FILTER_SOURCE, encoding="utf-8")
- print(f"(created default filter at {p})")
- except Exception as e:
- raise RuntimeError(f"failed to create default filter {p}: {e}")
- return p
-
-def _load_accept_func(filter_path: Path):
- env = runpy.run_path(str(filter_path))
- fn = env.get("accept")
- if not callable(fn):
- raise RuntimeError(f"{filter_path}: missing callable 'accept(prov)'")
- return fn
-
-def _walk_all_files(stage_root: Path):
- """
- Yield every file (regular or symlink) under stage_root recursively.
- We do not follow symlinked directories to avoid cycles.
- """
- root = stage_root.resolve()
- for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
- # prune symlinked dirs (files can still be symlinks)
- dirnames[:] = [d for d in dirnames if not os.path.islink(os.path.join(dirpath, d))]
- for fname in filenames:
- p = Path(dirpath, fname)
- try:
- st = p.lstat()
- if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
- yield p.resolve()
- except Exception:
- # unreadable/broken entries skipped
- continue
-
-def find_config_paths(stage_root: Path, accept_func) -> list[Path]:
- """
- Return files accepted by the Python acceptance function: accept(prov) → True/False.
- """
- out: list[Path] = []
- for p in _walk_all_files(stage_root):
- prov = PlanProvenance(stage_root=stage_root, config_path=p)
- try:
- if accept_func(prov):
- out.append(p)
- except Exception as e:
- raise RuntimeError(f"accept() failed on {prov.config_rel_fpath.as_posix()}: {e}")
- return sorted(out, key=lambda q: q.as_posix())
-
-# --- run all configs into ONE planner ---
-
-def _run_all_configs_into_single_planner(stage_root: Path, cfgs: list[Path]) -> Planner:
- """
- Create a single Planner and execute each config's configure(prov, planner, WriteFileMeta)
- against it. Returns that single Planner containing the entire plan.
- """
- # seed with synthetic provenance; we overwrite per config before execution
- aggregate_prov = PlanProvenance(stage_root=stage_root, config_path=stage_root / "(aggregate).py")
- planner = Planner(provenance=aggregate_prov)
-
- for cfg in cfgs:
- prov = PlanProvenance(stage_root=stage_root, config_path=cfg)
- planner.set_provenance(prov)
-
- env = runpy.run_path(str(cfg))
- fn = env.get("configure")
- if not callable(fn):
- raise RuntimeError(f"{cfg}: missing callable configure(prov, planner, WriteFileMeta)")
-
- fn(prov, planner, WriteFileMeta)
-
- # annotate meta once, on the single planner's journal
- j = planner.journal()
- j.set_meta(
- generator_prog_str="executor.py",
- generated_at_utc_str=iso_utc_now_str(),
- user_name_str=getpass.getuser(),
- host_name_str=os.uname().nodename if hasattr(os, "uname") else "unknown",
- stage_root_dpath_str=str(stage_root.resolve()),
- configs_list=[str(p.resolve().relative_to(stage_root.resolve())) for p in cfgs],
- )
- return planner
-
-# ----- CBOR “matchbox” (simple wrapper kept local to executor) -----
-
-def _plan_to_cbor_bytes(planner: Planner) -> bytes:
- """Serialize a Planner's Journal to CBOR bytes."""
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- plan_dict = planner.journal().as_dictionary()
- return cbor2.dumps(plan_dict, canonical=True)
-
-def _journal_from_cbor_bytes(data: bytes) -> Journal:
- """Rebuild a Journal from CBOR bytes."""
- try:
- import cbor2
- except Exception as e:
- raise RuntimeError(f"cbor2 is required: {e}")
- obj = cbor2.loads(data)
- if not isinstance(obj, dict):
- raise ValueError("CBOR root must be a dict")
- return Journal(plan_dict=obj)
-
-# -------- inner executor (phase 2) --------
-
-def _inner_main(plan_path: Path, phase2_print: bool, phase2_then_stop: bool) -> int:
- """Inner executor path: decode CBOR → Journal; optionally print; (apply TBD)."""
- try:
- data = Path(plan_path).read_bytes()
- except Exception as e:
- print(f"error: failed to read plan file: {e}", file=sys.stderr)
- return 2
-
- try:
- journal = _journal_from_cbor_bytes(data)
- except Exception as e:
- print(f"error: failed to decode CBOR: {e}", file=sys.stderr)
- return 2
-
- if phase2_print:
- journal.print()
-
- if phase2_then_stop:
- return 0
-
- # (Stage 3 apply would go here; omitted in MVP)
- return 0
-
-# -------- outer executor (phase 1 & handoff) --------
-
-def _outer_main(stage_root: Path, accept_func, args) -> int:
- if not stage_root.is_dir():
- print(f"error: --stage not a directory: {stage_root}", file=sys.stderr)
- return 2
-
- cfgs = find_config_paths(stage_root, accept_func)
- if not cfgs:
- print("No configuration files found.")
- return 0
-
- try:
- master = _run_all_configs_into_single_planner(stage_root, cfgs)
- except SystemExit:
- raise
- except Exception as e:
- print(f"error: executing configs: {e}", file=sys.stderr)
- return 2
-
- if args.phase_1_print:
- master.print()
-
- if args.phase_1_then_stop:
- return 0
-
- # Phase 2: encode CBOR and invoke inner path (same script, --inner)
- try:
- cbor_bytes = _plan_to_cbor_bytes(master)
- except Exception as e:
- print(f"error: CBOR encode failed: {e}", file=sys.stderr)
- return 2
-
- with tempfile.NamedTemporaryFile(prefix="stagehand_plan_", suffix=".cbor", delete=False) as tf:
- tf.write(cbor_bytes)
- plan_path = tf.name
-
- try:
- cmd = [
- sys.executable,
- str(Path(__file__).resolve()),
- "--inner",
- "--plan", plan_path,
- ]
- if args.phase_2_print:
- cmd.append("--phase-2-print")
- if args.phase_2_then_stop:
- cmd.append("--phase-2-then-stop")
-
- proc = subprocess.run(cmd)
- return proc.returncode
- finally:
- try:
- os.unlink(plan_path)
- except Exception:
- pass
-
-# -------- CLI --------
-
-def main(argv: list[str] | None = None) -> int:
- ap = argparse.ArgumentParser(
- prog="executor.py",
- description="StageHand outer/inner executor (plan → CBOR → decode).",
- )
- ap.add_argument("--stage", default="stage",
- help="stage root directory (default: ./stage)")
- ap.add_argument(
- "--filter",
- default="",
- help=f"path to acceptance filter program exporting accept(prov) "
- f"(default: ./{DEFAULT_FILTER_FILENAME}; created if missing)"
- )
- ap.add_argument(
- "--phase-0-then-stop",
- action="store_true",
- help="stop after arg checks & filter bootstrap (no stage scan)"
- )
-
- # Phase-1 (outer) controls
- ap.add_argument("--phase-1-print", action="store_true", help="print master planner (phase 1)")
- ap.add_argument("--phase-1-then-stop", action="store_true", help="stop after phase 1")
-
- # Phase-2 (inner) controls (outer forwards these to inner)
- ap.add_argument("--phase-2-print", action="store_true", help="print decoded journal (phase 2)")
- ap.add_argument("--phase-2-then-stop", action="store_true", help="stop after phase 2 decode")
-
- # Inner-only flags (not for users)
- ap.add_argument("--inner", action="store_true", help=argparse.SUPPRESS)
- ap.add_argument("--plan", default=None, help=argparse.SUPPRESS)
-
- args = ap.parse_args(argv)
-
- # Inner path
- if args.inner:
- if not args.plan:
- print("error: --inner requires --plan <file>", file=sys.stderr)
- return 2
- return _inner_main(Path(args.plan),
- phase2_print=args.phase_2_print,
- phase2_then_stop=args.phase_2_then_stop)
-
- # Phase 0: bootstrap & stop (no scan)
- stage_root = Path(args.stage)
- try:
- filter_path = _ensure_filter_file(args.filter or None)
- except Exception as e:
- print(f"error: {e}", file=sys.stderr)
- return 2
-
- if not stage_root.exists():
- print(f"error: --stage not found: {stage_root}", file=sys.stderr)
- return 2
- if not stage_root.is_dir():
- print(f"error: --stage is not a directory: {stage_root}", file=sys.stderr)
- return 2
-
- if args.phase_0_then_stop:
- print(f"phase-0 OK: stage at {stage_root.resolve()} and filter at {filter_path}")
- return 0
-
- # Load acceptance function and proceed with outer
- try:
- accept_func = _load_accept_func(filter_path)
- except Exception as e:
- print(f"error: {e}", file=sys.stderr)
- return 2
-
- return _outer_main(stage_root, accept_func, args)
-
-if __name__ == "__main__":
- sys.exit(main())
+++ /dev/null
-# stage_test_0/DNS/unbound_conf.py
-
-def configure(prov, planner, WriteFileMeta):
- # dpath is relative; it will be anchored to prov.read_dir_dpath,
- # so this lands in .../stage_test_0/stage_test_0_out/dns
- wfm = WriteFileMeta(
- dpath="stage_test_0_out/net",
- fname=prov.read_fname, # "unbound_conf"
- owner=prov.process_user, # current process user
- mode=0o444
- )
- planner.delete(wfm)
- planner.copy(wfm, content="server:\n verbosity: 1\n")
+++ /dev/null
-# unbound.conf (example)
-
-def configure(prov, planner, WriteFileMeta):
- wfm = WriteFileMeta(
- dpath="stage_test_0_out"
- ,fname=prov.read_fname # write file name same as read file name
- ,owner=prov.process_user
- )
- planner.displace(wfm)
- planner.copy(wfm, content="server:\n do-ip6: no\n")
+++ /dev/null
-# stage_test_0/web/site_conf.py
-
-def configure(prov, planner, WriteFileMeta):
- # This writes a faux web config to .../stage_test_0/stage_test_0_out/web/nginx.conf
- wfm = WriteFileMeta(
- dpath="stage_test_0_out/web",
- fname="nginx.conf", # explicit override (not from prov)
- owner=prov.process_user,
- mode="0644"
- )
- planner.displace(wfm)
- planner.copy(wfm, content="events {}\nhttp { server { listen 8080; } }\n")
+++ /dev/null
-# StageHand acceptance filter (default template)
-# Return True to include a config file, False to skip it.
-# You receive a PlanProvenance object named `prov`.
-#
-# prov fields commonly used here:
-# prov.stage_root_dpath : Path → absolute path to the stage root
-# prov.config_abs_fpath : Path → absolute path to the candidate file
-# prov.config_rel_fpath : Path → path relative to the stage root
-# prov.read_dir_dpath : Path → directory of the candidate file
-# prov.read_fname : str → filename with trailing '.py' stripped (if present)
-#
-# Examples:
-#
-# 1) Accept everything (default behavior):
-# def accept(prov):
-# return True
-#
-# 2) Only accept configs in a 'dns/' namespace under the stage:
-# def accept(prov):
-# return prov.config_rel_fpath.as_posix().startswith("dns/")
-#
-# 3) Exclude editor backup files:
-# def accept(prov):
-# rel = prov.config_rel_fpath.as_posix()
-# return not (rel.endswith("~") or rel.endswith(".swp"))
-#
-# 4) Only accept Python files + a few non-Python names:
-# def accept(prov):
-# name = prov.config_abs_fpath.name
-# return name.endswith(".py") or name in {"hosts", "resolv.conf"}
-#
-# Choose ONE 'accept' definition. Below is the default:
-
-def accept(prov):
- return True
--- /dev/null
+#!/usr/bin/env bash
+# disable_linger_subu — turn off systemd --user lingering for all <masu>-* users
+# Usage: sudo disable_linger_subu --masu Thomas
+
+set -euo pipefail
+MASU=""
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --masu) MASU="${2:-}"; shift 2;;
+ *) echo "unknown arg: $1" >&2; exit 2;;
+ esac
+done
+[[ -n "$MASU" ]] || { echo "usage: sudo $0 --masu <name>"; exit 2; }
+[[ $EUID -eq 0 ]] || { echo "must run as root"; exit 1; }
+
+mapfile -t SUBU_USERS < <(getent passwd | awk -F: -v pfx="^${MASU}-" '$1 ~ pfx {print $1}' | sort)
+for u in "${SUBU_USERS[@]}"; do
+ echo "loginctl disable-linger $u"
+ loginctl disable-linger "$u" || true
+done
+
+echo "Current linger files (should be empty or only intentional users):"
+ls -1 /var/lib/systemd/linger 2>/dev/null || echo "(none)"
+echo "✅ linger disabled for ${#SUBU_USERS[@]} users"
--- /dev/null
+#!/usr/bin/env bash
+# logout_subu — cleanly stop subu users, tear down bindfs, unbind /home, unmount device, close LUKS
+# Usage:
+# sudo logout_subu --masu Thomas --device Eagle [--aggressive] [--dry-run]
+#
+# Notes:
+# - Run from a directory NOT under /home/<masu> (we'll auto 'cd /' if needed).
+# - --aggressive enables pkill -KILL fallback if user@ sessions don't exit.
+# - --device is the mapname mounted at /mnt/<device> and /dev/mapper/<device>-crypt.
+
+set -euo pipefail
+
+MASU=""
+DEVICE=""
+AGGR=0
+DRY=0
+
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --masu) MASU="${2:-}"; shift 2;;
+ --device) DEVICE="${2:-}"; shift 2;;
+ --aggressive) AGGR=1; shift;;
+ --dry-run) DRY=1; shift;;
+ -h|--help)
+ grep -E '^(# |#-)' "$0" | sed 's/^# \{0,1\}//'
+ exit 0;;
+ *) echo "unknown arg: $1" >&2; exit 2;;
+ esac
+done
+
+if [[ -z "$MASU" ]]; then
+ # best guess: current sudo user or login user
+ MASU="${SUDO_USER:-${USER:-}}"
+ [[ -n "$MASU" ]] || { echo "Set --masu <name>"; exit 2; }
+fi
+
+if [[ $EUID -ne 0 ]]; then
+ echo "❌ must run as root (sudo)"; exit 1
+fi
+
+# If we’re under /home/<masu>, move away so unmount can succeed
+if [[ "$(pwd -P)" == /home/${MASU}* ]]; then
+ echo "cd / (leaving $(pwd -P) so unmounts can proceed)"
+ [[ $DRY -eq 1 ]] || cd /
+fi
+
+say() { printf '%s\n' "$*"; }
+doit() { echo "+ $*"; [[ $DRY -eq 1 ]] || eval "$@"; }
+
+# --- enumerate subu users and mountpoints
+SUBU_ROOT="/home/${MASU}/subu"
+SUBU_DATA="/home/${MASU}/subu_data"
+
+# Users of the form MASU-something that actually exist
+mapfile -t SUBU_USERS < <(getent passwd | awk -F: -v pfx="^${MASU}-" '$1 ~ pfx {print $1}' | sort)
+
+# Bindfs targets (reverse depth for unmount)
+mapfile -t SUBU_MPS < <(findmnt -Rn -S fuse.* -T "$SUBU_ROOT" -o TARGET 2>/dev/null | \
+ awk -F/ '{print NF, $0}' | sort -rn | cut -d" " -f2-)
+
+say "== stop subu systemd user managers =="
+for u in "${SUBU_USERS[@]}"; do
+ say "terminating user@ for $u"
+ doit loginctl terminate-user "$u" || true
+done
+
+# wait a moment and optionally KILL leftovers
+sleep 0.5
+for u in "${SUBU_USERS[@]}"; do
+ if loginctl list-users --no-legend | awk '{print $2}' | grep -qx "$u"; then
+ if [[ $AGGR -eq 1 ]]; then
+ uid="$(id -u "$u" 2>/dev/null || echo "")"
+ if [[ -n "$uid" ]]; then
+ say "aggressive kill of UID $uid ($u)"
+ doit pkill -KILL -u "$uid" || true
+ fi
+ else
+ say "⚠︎ $u still has a user@ manager; rerun with --aggressive to force-kill"
+ fi
+ fi
+done
+
+say "== unmount bindfs subu mounts under $SUBU_ROOT =="
+for mp in "${SUBU_MPS[@]}"; do
+ say "umount $mp"
+ if [[ $DRY -eq 1 ]]; then
+ echo "+ umount '$mp'"
+ else
+ if ! umount "$mp" 2>/dev/null; then
+ echo " (busy) trying lazy umount"
+ umount -l "$mp" || true
+ fi
+ fi
+done
+
+# Unmount the MASU home if it is a bind of /mnt/<device>/user_data/<masu>
+say "== unmount MASU home bind (if any) =="
+if findmnt -n -T "/home/${MASU}" >/dev/null 2>&1; then
+ src="$(findmnt -no SOURCE -T "/home/${MASU}")"
+ say "/home/${MASU} source: ${src}"
+ say "umount /home/${MASU}"
+ doit umount "/home/${MASU}" || true
+fi
+
+# If a device mapname was provided, unmount and close it
+if [[ -n "$DEVICE" ]]; then
+ say "== unmount /mnt/${DEVICE} and close LUKS =="
+ if findmnt -n "/mnt/${DEVICE}" >/dev/null 2>&1; then
+ say "umount /mnt/${DEVICE}"
+ doit umount "/mnt/${DEVICE}" || true
+ fi
+ if cryptsetup status "${DEVICE}-crypt" >/dev/null 2>&1; then
+ say "cryptsetup close ${DEVICE}-crypt"
+ doit cryptsetup close "${DEVICE}-crypt" || true
+ else
+ say "crypt mapping ${DEVICE}-crypt not active"
+ fi
+fi
+
+say "sync disks"
+[[ $DRY -eq 1 ]] || sync
+say "✅ done"
--- /dev/null
+#!/bin/bash
+
+# Function to bind mount a user's data to /home/<user>
+device_user_bind() {
+ local device=$1
+ local user=$2
+ local user_data_dir="/mnt/$device/user_data/$user"
+ local home_dir="/home/$user"
+
+ if [ ! -d "$user_data_dir" ]; then
+ echo "Error: $user_data_dir does not exist!"
+ return 1
+ fi
+
+ # Create the home directory if it doesn't exist
+ mkdir -p "$home_dir"
+
+ # Mount --bind the user data to the home directory
+ sudo mount --bind "$user_data_dir" "$home_dir"
+ echo "Mounted $user_data_dir -> $home_dir"
+}
+
+# Run the function with the device name and user as arguments
+device_user_bind "$1" "$2"
user=$1
# Get the list of sub-users by calling the user_list_subu_home.sh script
-subu_list=$(./user_list_subu_homedir.sh "$user")
+subu_list=$(./masu__subu_dir_llst.sh "$user")
# Check if we received any sub-users
if [ -z "$subu_list" ]; then
+++ /dev/null
-#!/bin/bash
-
-# Function to bind mount a user's data to /home/<user>
-device_user_bind() {
- local device=$1
- local user=$2
- local user_data_dir="/mnt/$device/user_data/$user"
- local home_dir="/home/$user"
-
- if [ ! -d "$user_data_dir" ]; then
- echo "Error: $user_data_dir does not exist!"
- return 1
- fi
-
- # Create the home directory if it doesn't exist
- mkdir -p "$home_dir"
-
- # Mount --bind the user data to the home directory
- sudo mount --bind "$user_data_dir" "$home_dir"
- echo "Mounted $user_data_dir -> $home_dir"
-}
-
-# Run the function with the device name and user as arguments
-device_user_bind "$1" "$2"
#!/bin/bash
+# masu_subu__map_own.sh <user> <subu> [--suid]
+#
+# Examples:
+# subu_bind Thomas developer # default (nosuid)
+# subu_bind Thomas developer --suid # enable setuid on this mount
+
+set -euo pipefail
-# Function to bind mount with UID/GID mapping
subu_bind() {
- local user=$1
- local subu=$2
+ local user="$1"
+ local subu="$2"
+ local want_suid="${3-}" # optional third arg
- # Check if bindfs is installed
- if ! command -v bindfs &> /dev/null; then
- echo "Error: bindfs is not installed!"
+ if ! command -v bindfs &>/dev/null; then
+ echo "Error: bindfs is not installed!" >&2
return 1
fi
- # Get the username and group name for the main user
- master_user_name=$user
- master_group=$user
+ # identities
+ local master_user_name="$user"
+ local master_group="$user"
+ local subu_user_name="${user}-${subu}"
+ local subu_group="${user}-${subu}"
- # Get the username and group name for the sub-user
- subu_user_name="${user}-${subu}"
- subu_group="${user}-${subu}"
+ id "$master_user_name" &>/dev/null || { echo "Error: user '$master_user_name' not found!" >&2; return 1; }
+ id "$subu_user_name" &>/dev/null || { echo "Error: sub-user '$subu_user_name' not found!" >&2; return 1; }
- # Check if the user and sub-user exist
- if ! id "$master_user_name" &>/dev/null; then
- echo "Error: User '$master_user_name' not found!"
- return 1
- fi
- if ! id "$subu_user_name" &>/dev/null; then
- echo "Error: Sub-user '${master_user_name}-${subu}' not found!"
- return 1
- fi
+ # paths
+ local subu_data_path="/home/$user/subu_data/$subu"
+ local subu_mount_point_path="/home/$user/subu/$subu"
- # Directories to be bind-mounted
- subu_data_path="/home/$user/subu_data/$subu"
- subu_mount_point_path="/home/$user/subu/$subu"
+ [[ -d "$subu_data_path" ]] || { echo "Error: source dir '$subu_data_path' does not exist!" >&2; return 1; }
+ mkdir -p "$subu_mount_point_path"
- # Check if sub-user directory exists
- if [ ! -d "$subu_data_path" ]; then
- echo "Error: Sub-user directory '$subu_data_path' does not exist!"
- return 1
+ # mount options
+ # - allow_other/default_permissions: kernel does POSIX perms
+ # - exec: allow execution on the mount
+ # - suid: only when explicitly requested (and only works when mounted as root)
+ local base_opts="allow_other,default_permissions,exec"
+ local opts="$base_opts,nosuid"
+ if [[ "$want_suid" == "--suid" ]]; then
+ opts="$base_opts,suid"
fi
- # Create the mount point if it doesn't exist
- mkdir -p "$subu_mount_point_path"
+ # The UID/GID mapping you already use
+ local map_opt="--map=${subu_user_name}/${master_user_name}:@${subu_group}/@${master_group}"
- # Perform the bind mount using bindfs with UID/GID mapping
- sudo bindfs\
- --map="$subu_user_name/$master_user_name:@$subu_group/@$master_group" \
- "$subu_data_path" \
- "$subu_mount_point_path"
+ # If already mounted, decide whether to keep or remount
+ if findmnt -n -T "$subu_mount_point_path" >/dev/null 2>&1; then
+ current_opts="$(findmnt -no OPTIONS -T "$subu_mount_point_path" || true)"
+ if [[ ",$current_opts," != *",$opts,"* ]]; then
+ echo "remounting $subu_mount_point_path with opts: $opts"
+ # clean remount: unmount then mount with new opts
+ sudo umount "$subu_mount_point_path"
+ sudo bindfs -o "$opts" $map_opt "$subu_data_path" "$subu_mount_point_path"
+ else
+ echo "already mounted with compatible options: $current_opts"
+ fi
+ else
+ echo "mounting $subu_data_path -> $subu_mount_point_path with opts: $opts"
+ sudo bindfs -o "$opts" $map_opt "$subu_data_path" "$subu_mount_point_path"
+ fi
- # Verify if the mount was successful
- if [ $? -eq 0 ]; then
- echo "Successfully bind-mounted $subu_data_path to $subu_mount_point_path with UID/GID mapping."
+ # Verify outcome
+ findmnt "$subu_mount_point_path" -o TARGET,FSTYPE,OPTIONS
+ echo "OK: $subu_data_path -> $subu_mount_point_path"
+ if [[ "$want_suid" == "--suid" ]]; then
+ echo "note: suid is enabled; setuid binaries (e.g., the gasket) can take effect on this mount."
else
- echo "Error: Failed to bind-mount $subu_data_path to $subu_mount_point_path, might already exist."
+ echo "note: nosuid (default) — setuid will NOT take effect on this mount."
fi
}
-# Call the function with user and subu as arguments
-subu_bind "$1" "$2"
+subu_bind "${1:-}" "${2:-}" "${3:-}"
--- /dev/null
+#!/bin/bash
+
+# Function to bind mount with UID/GID mapping
+subu_bind() {
+ local user=$1
+ local subu=$2
+
+ # Check if bindfs is installed
+ if ! command -v bindfs &> /dev/null; then
+ echo "Error: bindfs is not installed!"
+ return 1
+ fi
+
+ # Get the username and group name for the main user
+ master_user_name=$user
+ master_group=$user
+
+ # Get the username and group name for the sub-user
+ subu_user_name="${user}-${subu}"
+ subu_group="${user}-${subu}"
+
+ # Check if the user and sub-user exist
+ if ! id "$master_user_name" &>/dev/null; then
+ echo "Error: User '$master_user_name' not found!"
+ return 1
+ fi
+ if ! id "$subu_user_name" &>/dev/null; then
+ echo "Error: Sub-user '${master_user_name}-${subu}' not found!"
+ return 1
+ fi
+
+ # Directories to be bind-mounted
+ subu_data_path="/home/$user/subu_data/$subu"
+ subu_mount_point_path="/home/$user/subu/$subu"
+
+ # Check if sub-user directory exists
+ if [ ! -d "$subu_data_path" ]; then
+ echo "Error: Sub-user directory '$subu_data_path' does not exist!"
+ return 1
+ fi
+
+ # Create the mount point if it doesn't exist
+ mkdir -p "$subu_mount_point_path"
+
+ # Perform the bind mount using bindfs with UID/GID mapping
+ sudo bindfs\
+ --map="$subu_user_name/$master_user_name:@$subu_group/@$master_group" \
+ "$subu_data_path" \
+ "$subu_mount_point_path"
+
+ # Verify if the mount was successful
+ if [ $? -eq 0 ]; then
+ echo "Successfully bind-mounted $subu_data_path to $subu_mount_point_path with UID/GID mapping."
+ else
+ echo "Error: Failed to bind-mount $subu_data_path to $subu_mount_point_path, might already exist."
+ fi
+}
+
+# Call the function with user and subu as arguments
+subu_bind "$1" "$2"
#!/usr/bin/env bash
+# toolsmith owned developer environment file
+# see developer/tool/env for the developer's customizations
+
script_afp=$(realpath "${BASH_SOURCE[0]}")
if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
echo "$script_afp:: This script must be sourced, not executed."
exit 1
fi
+# environment shared by all roles
source tool_shared/bespoke/env
-source developer/tool/env
+export ROLE=developer
+export ENV=$ROLE
+if [[ -n "${ROLE:-}" ]] && [[ -d "$REPO_HOME/$ROLE/tool" ]]; then
+ PATH="$REPO_HOME/$ROLE/tool:$PATH"
+fi
+
+# shared tools
+# if [[ ":$PATH:" != *":$PYTHON_HOME/bin:"* ]]; then
+# export PATH="$PYTHON_HOME/bin:$PATH"
+# fi
+
+export PATH="$REPO_HOME/tool_shared/third_party/Man_In_Grey/release/shell":"$PATH"
+
+cd $ROLE
+
+# pull in developer's customization
+# source tool/env $@
+
+echo "in environment: $ENV"
#!/usr/bin/env bash
+# toolsmith-owned tester environment file
+
script_afp=$(realpath "${BASH_SOURCE[0]}")
if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
echo "$script_afp:: This script must be sourced, not executed."
exit 1
fi
+# shared, project-wide
source tool_shared/bespoke/env
-source tester/tool/env
+
+export ROLE=tester
+export ENV=$ROLE
+
+# tester-local tool dir first (if any)
+if [[ -d "$REPO_HOME/$ROLE/tool" ]]; then
+ PATH="$REPO_HOME/$ROLE/tool:$PATH"
+fi
+
+# shared Python (from toolsmith-provided venv)
+if [[ ":$PATH:" != *":$PYTHON_HOME/bin:"* ]]; then
+ PATH="$PYTHON_HOME/bin:$PATH"
+fi
+export PATH
+
+export RELEASE="$REPO_HOME/release"
+
+cd "$ROLE"
+
+# pull in tester customizations (optional file)
+[[ -f tool/env ]] && source tool/env "$@"
+
+echo "in environment: $ENV"
--- /dev/null
+*
+!/.gitignore
\ No newline at end of file
#!/usr/bin/env bash
script_afp=$(realpath "${BASH_SOURCE[0]}")
-
# vl 'vertical list'
-# e.g.:
-# > vl echo $PATH
+# Check if the command is provided
if [ -z "$1" ]; then
echo "Usage: vl <command> [args...]"
exit 1
projects / subu / commitdiff