#!/usr/bin/env python3 """Project-scoped ZiboxChat bootstrap. The publisher replaces the pinned digest.""" import argparse from contextlib import contextmanager import hashlib import json import os from pathlib import Path import stat import subprocess import sys import tempfile import urllib.parse import urllib.request CLIENT_SHA256 = "87db6b11d66e2bad2c9cae1ca6b5a9c3249a5a95e80a71e278a7e139ac45f3dd" BASE_URL = "https://zibox.api.br:10099" MAX_DOWNLOAD = 2 * 1024 * 1024 class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError("O download redirecionou; instalação interrompida.") def verified_download(url, expected, opener=None): if len(expected) != 64 or any(c not in "0123456789abcdef" for c in expected): raise ValueError("Publicação sem SHA-256 válido.") parsed = urllib.parse.urlsplit(url) official = urllib.parse.urlsplit(BASE_URL) if (parsed.scheme, parsed.netloc) != (official.scheme, official.netloc) or parsed.username or parsed.password: raise ValueError("Origem do cliente inválida.") opener = opener or urllib.request.build_opener(NoRedirect()) with opener.open(urllib.request.Request(url, headers={"Accept": "text/x-python"}), timeout=25) as response: if response.status != 200: raise ValueError("Servidor não devolveu o cliente.") raw = response.read(MAX_DOWNLOAD + 1) if len(raw) > MAX_DOWNLOAD or hashlib.sha256(raw).hexdigest() != expected: raise ValueError("SHA-256 divergente ou arquivo acima do limite. Nada foi executado.") compile(raw, "ziboxchat-client.py", "exec") return raw def private_directory(path): if path.is_symlink(): raise ValueError("Diretório de instalação não pode ser um link simbólico.") path.mkdir(parents=True, exist_ok=True, mode=0o700) if os.name != "nt": if path.stat().st_uid != os.getuid(): raise ValueError("Diretório de instalação pertence a outro usuário.") path.chmod(0o700) def atomic_private_write(path, raw): fd, temp = tempfile.mkstemp(prefix=".connect-", dir=path.parent) try: with os.fdopen(fd, "wb") as handle: handle.write(raw) handle.flush() os.fsync(handle.fileno()) os.chmod(temp, 0o600) os.replace(temp, path) finally: if os.path.exists(temp): os.unlink(temp) @contextmanager def project_lock(project): state = project / ".ziboxchat" private_directory(state) lock = state / "connect.lock" if lock.is_symlink(): raise ValueError("Lock de instalação não pode ser um link simbólico.") fd = os.open(lock, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600) acquired = False try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or (os.name != "nt" and info.st_uid != os.getuid()): raise ValueError("Lock de instalação inválido.") try: if os.name == "nt": import msvcrt if info.st_size == 0: os.write(fd, b"\n") os.lseek(fd, 0, os.SEEK_SET) msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) acquired = True except OSError as exc: raise ValueError("Já existe uma instalação nesta pasta em execução.") from exc os.lseek(fd, 0, os.SEEK_SET) os.write(fd, (str(os.getpid()) + "\n").encode()) os.ftruncate(fd, os.lseek(fd, 0, os.SEEK_CUR)) yield finally: # A persistent file plus an OS lock avoids stale locks after crashes and unlink races. if acquired: if os.name == "nt": os.lseek(fd, 0, os.SEEK_SET) msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) else: fcntl.flock(fd, fcntl.LOCK_UN) os.close(fd) def _install_locked(project): state = project / ".ziboxchat" runtime = state / "bin" private_directory(runtime) client = runtime / ("client-" + CLIENT_SHA256 + ".py") if not client.is_file() or client.is_symlink() or hashlib.sha256(client.read_bytes()).hexdigest() != CLIENT_SHA256: raw = verified_download(BASE_URL + "/static/ia/releases/" + CLIENT_SHA256 + "/ziboxchat_mcp.py", CLIENT_SHA256) atomic_private_write(client, raw) launcher = ("import os, runpy\nfrom pathlib import Path\n" "os.environ.setdefault('ZIBOXCHAT_SERVER', " + repr(BASE_URL) + ")\n" "runpy.run_path(str(Path(__file__).with_name(" + repr(client.name) + ")), run_name='__main__')\n") atomic_private_write(runtime / "client.py", launcher.encode()) ignore = state / ".gitignore" if not ignore.exists(): atomic_private_write(ignore, b"*\n") return runtime / "client.py" def install(project): with project_lock(project): return _install_locked(project) def main(argv=None): parser = argparse.ArgumentParser(description="Conectar esta pasta ao ZiboxChat, sem reiniciar a sessão") parser.add_argument("--project-dir", default=os.getcwd()) parser.add_argument("--name", default="") parser.add_argument("--standalone", action="store_true", help="Cria identidade IA se a pasta ainda não tiver uma") parser.add_argument("--install-only", action="store_true", help="Instala o cliente validado sem criar/vincular conta") args = parser.parse_args(argv) try: project = Path(args.project_dir).expanduser().resolve(strict=True) if not project.is_dir(): raise ValueError("Escolha uma pasta de projeto existente.") # Keep the lock through binding: two simultaneous installs must not create two accounts. with project_lock(project): client = _install_locked(project) if args.install_only: print(json.dumps({"ok": True, "client": str(client), "sha256": CLIENT_SHA256})) return 0 command = [sys.executable, "-I", str(client), "bind-project", "--json"] if args.name: command.extend(["--name", args.name]) if args.standalone: command.append("--standalone") # Binding owns credential persistence. Only its redacted public JSON reaches stdout. env = dict(os.environ, ZIBOXCHAT_SERVER=BASE_URL) return subprocess.run(command, cwd=project, env=env, timeout=90).returncode except (OSError, ValueError, subprocess.TimeoutExpired) as exc: print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())