#!/usr/bin/env python3 """Reviewed user-local installer. Dry-run by default. No network or shell execution.""" import argparse import datetime import hashlib import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import uuid VERSION = "1.0.0" TESTED = {"2.2.6"} NAMES = ("index.ts", "rpc.ts", "tui.tsx") def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def safe_path(path): if not path.is_absolute(): raise ValueError("XDG paths must be absolute") for part in [path, *path.parents]: if part.is_symlink(): raise ValueError(f"Refusing symlinked install path: {part}") existing = next(p for p in [path, *path.parents] if p.exists()) if existing.stat().st_uid != os.getuid(): raise ValueError(f"Install destination is not owned by current user: {existing}") def validate_text(value, maximum, required=False): if (required and not value.strip()) or len(value) > maximum or re.search(r"[\x00-\x1f\x7f\[\]]", value): raise ValueError("Identity text is blank, too long, or contains brackets/control characters") return value.strip() def rollback(receipt, apply): safe_path(receipt) data = json.loads(receipt.read_text()) if data.get("recipe") != "opencode-machine-identity" or data.get("uid") != os.getuid(): raise ValueError("Receipt does not belong to this recipe/current user") config_root = Path(data["config_root"]) safe_path(config_root) allowed = {config_root / "ecosystem-field-guide/machine-identity.json"} allowed.update(config_root / "opencode/plugins/ecosystem-machine-identity" / name for name in NAMES) paths = {Path(name) for name in data["files"]} if paths != allowed: raise ValueError("Unexpected receipt file list") for path in paths: safe_path(path) if not path.is_file() or digest(path) != data["files"][str(path)]: raise ValueError(f"Refusing rollback: missing/modified file {path}") print(json.dumps({"action": "rollback", "apply": apply, "files": sorted(map(str, paths))}, indent=2)) if apply: for path in paths: path.unlink() for directory in [config_root / "opencode/plugins/ecosystem-machine-identity", config_root / "ecosystem-field-guide"]: try: directory.rmdir() except OSError: pass # Never remove unrelated files or broad parent directories. (receipt.parent / "rolled-back.json").write_text(json.dumps({"version": VERSION, "completed": True}) + "\n") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--hint", help="Approved text before dynamic hostname, e.g. 'HOME ยท'") parser.add_argument("--role", default="", help="Optional approved suffix; not inferred from management username") parser.add_argument("--color", help="Approved #RRGGBB color") parser.add_argument("--opencode", help="Path to the intended user's installed OpenCode") parser.add_argument("--accept-untested-version", action="store_true") parser.add_argument("--apply", action="store_true", help="Explicitly install or roll back; otherwise inspect only") parser.add_argument("--rollback", type=Path, help="Previously generated receipt.json") args = parser.parse_args() if os.geteuid() == 0: raise ValueError("Run as the intended OpenCode user, not root/sudo; do not change ownership of a private home") if args.rollback: rollback(args.rollback.expanduser().absolute(), args.apply) return if args.hint is None or args.color is None: parser.error("--hint and --color are required; obtain the user's explicit selection") config = {"hint": validate_text(args.hint, 80, True), "role": validate_text(args.role, 40), "color": args.color.lower()} if not re.fullmatch(r"#[0-9a-f]{6}", config["color"]): raise ValueError("Color must be #RRGGBB") binary = args.opencode or shutil.which("opencode") or str(Path.home() / ".opencode/bin/opencode") result = subprocess.run([binary, "--version"], capture_output=True, text=True, timeout=10, check=True) match = re.search(r"\bv?(\d+\.\d+\.\d+)\b", result.stdout) if not match or not match[1].startswith("2."): raise ValueError("This package requires OpenCode V2; do not apply to V1") installed_version = match[1] if installed_version not in TESTED and not args.accept_untested_version: raise ValueError(f"OpenCode {installed_version} is untested. Review V2 plugin APIs and use --accept-untested-version only with explicit approval") config_root = Path(os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))) state_root = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local/state"))) / "ecosystem-field-guide/identity-installs" safe_path(config_root) safe_path(state_root) plugin = config_root / "opencode/plugins/ecosystem-machine-identity" target = config_root / "ecosystem-field-guide/machine-identity.json" safe_path(plugin) safe_path(target) source = Path(__file__).resolve().parent / "plugin" expected = {name: (source / name).read_bytes() for name in NAMES} expected_config = (json.dumps(config, indent=2) + "\n").encode() if plugin.exists() or target.exists(): same = target.is_file() and target.read_bytes() == expected_config and plugin.is_dir() same = same and {p.name for p in plugin.iterdir()} == set(NAMES) if same and all((plugin / name).is_file() and not (plugin / name).is_symlink() and (plugin / name).read_bytes() == body for name, body in expected.items()): print(json.dumps({"status": "already-installed", "version": VERSION, "changed": False})) return raise ValueError("Existing identity installation differs; review it rather than overwriting") conflicts = [config_root / "opencode/plugins" / name for name in ("commander-machine-identity", "machine-identity")] conflicts += [config_root / "commander/machine-identity.json", config_root / "opencode/commander-identity.tsx"] if any(p.exists() or p.is_symlink() for p in conflicts): raise ValueError("Another machine identity customization exists. Reconcile it first to avoid duplicate labels") print(json.dumps({"action": "install", "apply": args.apply, "package_version": VERSION, "opencode_version": installed_version, "plugin": str(plugin), "config": str(target), "identity": config}, indent=2)) if not args.apply: return os.umask(0o077) config_root.mkdir(parents=True, exist_ok=True) state_root.mkdir(parents=True, exist_ok=True) stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8] receipt_dir = state_root / stamp receipt_dir.mkdir(mode=0o700) prior = {} for name in ("opencode/cli.json", "opencode/opencode.json", "opencode/opencode.jsonc"): path = config_root / name if path.is_file(): prior[str(path)] = digest(path) stage = Path(tempfile.mkdtemp(prefix=".identity-stage-", dir=config_root)) written = [] try: for name, body in expected.items(): (stage / name).write_bytes(body) plugin.parent.mkdir(mode=0o700, parents=True, exist_ok=True) target.parent.mkdir(mode=0o700, parents=True, exist_ok=True) with target.open("xb") as handle: handle.write(expected_config) written.append(target) # Plugin discovery sees a complete directory, not partially written code. stage.rename(plugin) written.extend(plugin / name for name in NAMES) receipt = {"recipe": "opencode-machine-identity", "version": VERSION, "uid": os.getuid(), "config_root": str(config_root), "files": {str(p): digest(p) for p in written}, "prior_config_hashes": prior} (receipt_dir / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n") except Exception: for path in reversed(written): if path.is_file() and not path.is_symlink(): path.unlink() if plugin.is_dir(): try: plugin.rmdir() except OSError: pass raise finally: if stage.is_dir(): for name in NAMES: path = stage / name if path.is_file(): path.unlink() stage.rmdir() print("Installed. Receipt: " + str(receipt_dir / "receipt.json")) print("No configuration files edited, backend restarted, or model requests sent. Reopen an old TUI if needed.") if __name__ == "__main__": try: main() except (ValueError, OSError, subprocess.SubprocessError) as error: print("Stopped: " + str(error), file=sys.stderr) sys.exit(1)