#!/usr/bin/env python3 """Install the audited Subsignal skill. Inspect this file before executing it.""" import argparse import fcntl import hashlib import json import os from pathlib import Path import re import shutil import ssl import stat import sys import tempfile from urllib.request import HTTPRedirectHandler, Request, build_opener SOURCE = "https://subsignal.dev/skills/subsignal/" FILES = {"SKILL.md", "scripts/read.py", "scripts/connect.py", "install.py"} MAX_FILE = 256 * 1024 class NoRedirects(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def fetch(name): if name not in FILES | {"manifest.json"}: raise ValueError("Unknown artifact path") request = Request(SOURCE + name, headers={"User-Agent": "subsignal-installer/1"}) with build_opener(NoRedirects()).open(request, timeout=30) as response: data = response.read(MAX_FILE + 1) if len(data) > MAX_FILE: raise ValueError("Artifact exceeds its size bound") return data def manifest(data): value = json.loads(data) if (not isinstance(value, dict) or value.get("schema") != 1 or not re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", str(value.get("version", ""))) or not isinstance(value.get("files"), dict) or set(value["files"]) != FILES): raise ValueError("Invalid release manifest") for info in value["files"].values(): if (not isinstance(info, dict) or not re.fullmatch(r"[0-9a-f]{64}", str(info.get("sha256", ""))) or type(info.get("size")) is not int or not 0 < info["size"] <= MAX_FILE): raise ValueError("Invalid release hash or size") return value def check_path(path): for part in [*reversed(path.parents), path]: if part.is_symlink(): raise ValueError("Symlink destinations are not supported") if part.exists() and not part.is_dir(): raise ValueError("Destination parent must be a directory") def check_installed(destination): # An unknown/unmanaged directory or user edits are never overwritten. for root, directories, files in os.walk(destination, followlinks=False): for name in directories + files: if (Path(root) / name).is_symlink(): raise ValueError("Existing installation contains a symlink") # Python writes __pycache__ by itself when a helper is imported; that is not a user edit. actual = {str(path.relative_to(destination)) for path in destination.rglob("*") if path.is_file() and "__pycache__" not in path.parts} if actual != FILES | {"manifest.json"}: raise ValueError("Existing installation is modified or unmanaged; preserve it and choose another destination") raw = (destination / "manifest.json").read_bytes() old = manifest(raw) for name, info in old["files"].items(): data = (destination / name).read_bytes() if len(data) != info["size"] or hashlib.sha256(data).hexdigest() != info["sha256"]: raise ValueError("Existing installation is modified; preserve it and choose another destination") return raw, old def install(destination, *, update=False, download=fetch): destination = Path(os.path.abspath(destination.expanduser())) check_path(destination) destination.parent.mkdir(parents=True, exist_ok=True) check_path(destination.parent) lock_path = destination.parent / ("." + destination.name + ".install.lock") fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: lock_info = os.fstat(fd) if not stat.S_ISREG(lock_info.st_mode) or lock_info.st_uid != os.getuid() or lock_info.st_mode & 0o077: raise ValueError("Unsafe installer lock") fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) check_path(destination) raw = download("manifest.json") release = manifest(raw) previous = None if destination.exists(): previous_raw, previous = check_installed(destination) if previous_raw == raw: return {"status": "already_installed", "version": release["version"], "destination": str(destination)} if not update: raise ValueError("A different version is installed; inspect the release and use --update") stage = Path(tempfile.mkdtemp(prefix="." + destination.name + "-", dir=destination.parent)) backup = None try: for name, info in release["files"].items(): data = download(name) if len(data) != info["size"] or hashlib.sha256(data).hexdigest() != info["sha256"]: raise ValueError("Release hash or size mismatch; nothing installed") target = stage / name target.parent.mkdir(exist_ok=True) target.write_bytes(data) target.chmod(0o644) (stage / "manifest.json").write_bytes(raw) stage.chmod(0o755) if previous: # Keep rollback skills outside the host discovery root, so its # stale SKILL.md cannot register a duplicate Subsignal skill. backups = destination.parent.parent / ".subsignal-skill-backups" check_path(backups) backups.mkdir(mode=0o700, exist_ok=True) backup_info = backups.stat() if backup_info.st_uid != os.getuid() or backup_info.st_mode & 0o077: raise ValueError("Rollback directory must be private and owned by you") backup = backups / (destination.name + ".rollback-" + previous["version"]) if backup.exists() or backup.is_symlink(): raise ValueError("Rollback destination already exists; preserve or move it before updating") os.rename(destination, backup) try: os.rename(stage, destination) except BaseException: if backup: os.rename(backup, destination) raise finally: if stage.exists(): shutil.rmtree(stage) result = {"status": "installed", "version": release["version"], "destination": str(destination)} if backup: result["rollback"] = str(backup) return result finally: os.close(fd) def failure_message(exc): if type(exc) is ValueError: return str(exc) # urllib wraps TLS failures in URLError.reason. if isinstance(exc, ssl.SSLCertVerificationError) or isinstance(getattr(exc, "reason", None), ssl.SSLCertVerificationError): version = f"{sys.version_info.major}.{sys.version_info.minor}" return ("Python could not verify HTTPS certificates. For python.org Python on macOS, run " f"'/Applications/Python {version}/Install Certificates.command', then retry. No release fallback is used.") return "Installation failed; check HTTPS release availability and local permissions. No release fallback is used." def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--destination", type=Path, default=Path.home() / ".agents/skills/subsignal") parser.add_argument("--update", action="store_true") args = parser.parse_args() try: print(json.dumps(install(args.destination, update=args.update))) return 0 except (OSError, ValueError) as exc: print(json.dumps({"error": failure_message(exc)})) return 1 if __name__ == "__main__": raise SystemExit(main())