#!/usr/bin/env python3 """CodexPi Bridge: share one explicit folder through outbound HTTPS/HTTP requests.""" from __future__ import annotations import argparse import hashlib import json import os import platform import subprocess import socket import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from pathlib import Path, PurePosixPath MAX_TEXT_BYTES = 2_000_000 IGNORED_DIRECTORIES = {".git", ".venv", "node_modules", "__pycache__"} def request(server: str, path: str, method: str = "GET", body: dict | None = None, token: str = "") -> dict | None: url = server.rstrip("/") + path headers = {"Accept": "application/json", "User-Agent": "CodexPi-Bridge/1"} data = None if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" if token: headers["Authorization"] = f"Bearer {token}" try: with urllib.request.urlopen(urllib.request.Request(url, data=data, headers=headers, method=method), timeout=35) as response: raw = response.read() return json.loads(raw) if raw else None except urllib.error.HTTPError as exc: try: detail = json.loads(exc.read()).get("detail") except Exception: detail = None raise RuntimeError(detail or f"Erreur HTTP {exc.code}") from exc def checked_server(value: str) -> str: parsed = urllib.parse.urlsplit(value) if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password: raise ValueError("Adresse CodexPi invalide") return value.rstrip("/") def safe_path(root: Path, relative: str, allow_root: bool = False) -> Path: normalized = relative.replace("\\", "/").strip("/") parts = PurePosixPath(normalized).parts if normalized else () if any(part in {"", ".", ".."} or ":" in part for part in parts): raise ValueError("Chemin distant invalide") target = root.joinpath(*parts).resolve() try: target.relative_to(root) except ValueError as exc: raise ValueError("Accès en dehors du dossier partagé refusé") from exc if target == root and not allow_root: raise ValueError("Un fichier doit être indiqué") return target def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(131072), b""): digest.update(chunk) return digest.hexdigest() def list_files(root: Path, relative: str) -> dict: target = safe_path(root, relative, allow_root=True) if not target.is_dir(): raise ValueError("Le dossier demandé n’existe pas") entries: list[dict] = [] for current, directories, filenames in os.walk(target, followlinks=False): current_path = Path(current) directories[:] = [name for name in directories if name not in IGNORED_DIRECTORIES and not (current_path / name).is_symlink()] for name in filenames: path = current_path / name if path.is_symlink() or not path.is_file(): continue stat = path.stat() entries.append({"path": path.relative_to(root).as_posix(), "size": stat.st_size, "modified": stat.st_mtime}) if len(entries) >= 2000: return {"entries": entries, "truncated": True} return {"entries": entries, "truncated": False} def read_file(root: Path, relative: str) -> dict: target = safe_path(root, relative) if target.is_symlink() or not target.is_file(): raise ValueError("Fichier distant introuvable") data = target.read_bytes() if len(data) > MAX_TEXT_BYTES: raise ValueError("Le fichier dépasse la limite de 2 Mo") if b"\0" in data: raise ValueError("Les fichiers binaires ne sont pas pris en charge") try: content = data.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError("Le fichier n’est pas encodé en UTF-8") from exc return {"path": relative, "content": content, "sha256": hashlib.sha256(data).hexdigest(), "size": len(data)} def write_file(root: Path, relative: str, payload: dict, expected_hash: str | None) -> dict: target = safe_path(root, relative) content = payload.get("content") if not isinstance(content, str): raise ValueError("Contenu texte manquant") data = content.encode("utf-8") if len(data) > MAX_TEXT_BYTES: raise ValueError("Le fichier dépasse la limite de 2 Mo") if target.exists() and (target.is_symlink() or not target.is_file()): raise ValueError("La destination n’est pas un fichier ordinaire") current_hash = sha256(target) if target.exists() else None if expected_hash and current_hash != expected_hash: raise ValueError("Conflit : le fichier a changé sur le PC depuis sa lecture") target.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary = tempfile.mkstemp(prefix=".codexpi-", dir=target.parent) try: with os.fdopen(descriptor, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, target) finally: if os.path.exists(temporary): os.unlink(temporary) return {"path": relative, "sha256": hashlib.sha256(data).hexdigest(), "size": len(data)} def run_command(root: Path, payload: dict) -> dict: argv = payload.get("argv") if not isinstance(argv, list) or not argv or len(argv) > 100 or not all(isinstance(item, str) and 0 < len(item) <= 1000 for item in argv): raise ValueError("Commande distante invalide") timeout = min(300, max(1, int(payload.get("timeout") or 120))) try: result = subprocess.run(argv, cwd=root, capture_output=True, timeout=timeout, text=True, encoding="utf-8", errors="replace") except subprocess.TimeoutExpired as exc: raise ValueError(f"Commande interrompue après {timeout} secondes") from exc output = (result.stdout + result.stderr)[-1_000_000:] return {"argv": argv, "returncode": result.returncode, "output": output} def execute(root: Path, job: dict) -> dict: operation = job.get("operation") relative = str(job.get("path") or "") if operation == "list": return list_files(root, relative) if operation == "read": return read_file(root, relative) if operation == "write": return write_file(root, relative, job.get("payload") or {}, job.get("expected_hash")) if operation == "exec": return run_command(root, job.get("payload") or {}) raise ValueError("Opération inconnue refusée") def save_config(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary = tempfile.mkstemp(prefix="bridge-", suffix=".json", dir=path.parent) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(value, handle, ensure_ascii=False, indent=2) handle.write("\n") try: os.chmod(temporary, 0o600) except OSError: pass os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def pair(pairing_path: Path, folder: Path, name: str, config_path: Path) -> dict: pairing = json.loads(pairing_path.read_text(encoding="utf-8")) server = checked_server(str(pairing["server"])) root = folder.expanduser().resolve(strict=True) if not root.is_dir(): raise ValueError("Le dossier partagé est introuvable") result = request(server, "/api/remote/agent/pair", "POST", { "pairing_id": pairing["pairing_id"], "code": pairing["code"], "name": name, "folder_name": root.name, "platform": platform.platform()[:100], }) or {} config = {"server": server, "machine_id": result["machine_id"], "token": result["token"], "folder": str(root)} save_config(config_path, config) print(f"Machine associée. Dossier partagé : {root}") print(f"Configuration privée enregistrée dans : {config_path}") return config def run(config: dict, once: bool = False) -> None: server = checked_server(str(config["server"])) machine_id, token = str(config["machine_id"]), str(config["token"]) root = Path(config["folder"]).expanduser().resolve(strict=True) if not root.is_dir(): raise ValueError("Le dossier partagé configuré n’existe plus") print(f"CodexPi Bridge connecté à {server}") print(f"Dossier autorisé : {root}") while True: try: job = request(server, f"/api/remote/agent/jobs?machine_id={urllib.parse.quote(machine_id)}", token=token) if job: try: result = execute(root, job) response = {"machine_id": machine_id, "success": True, "result": result, "error": ""} print(f"✓ {job['operation']} {job.get('path') or '.'}") except Exception as exc: response = {"machine_id": machine_id, "success": False, "result": {}, "error": str(exc)[:2000]} print(f"✗ {job.get('operation')} : {exc}", file=sys.stderr) request(server, f"/api/remote/agent/jobs/{job['id']}", "POST", response, token) if once: return time.sleep(1 if job else 2) except (OSError, RuntimeError) as exc: if once: raise print(f"Connexion indisponible : {exc}. Nouvelle tentative dans 5 s.", file=sys.stderr) time.sleep(5) def main() -> int: default_config = Path.home() / ".codexpi" / "bridge.json" parser = argparse.ArgumentParser(description="Connecte un dossier précis à CodexPi avec fichiers et commandes contrôlées.") parser.add_argument("--pairing", type=Path, help="fichier .codexpi téléchargé depuis CodexPi") parser.add_argument("--folder", type=Path, help="dossier local à partager lors de l’association") parser.add_argument("--name", default=socket.gethostname(), help="nom affiché dans CodexPi") parser.add_argument("--config", type=Path, default=default_config, help="emplacement de la configuration privée") parser.add_argument("--once", action="store_true", help="effectue un seul contrôle puis quitte") args = parser.parse_args() try: if args.pairing: if not args.folder: parser.error("--folder est obligatoire avec --pairing") config = pair(args.pairing, args.folder, args.name, args.config) else: config = json.loads(args.config.read_text(encoding="utf-8")) run(config, args.once) return 0 except KeyboardInterrupt: print("\nConnecteur arrêté.") return 0 except Exception as exc: print(f"Erreur : {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())