#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import secrets import sys from pathlib import Path from typing import Any VERSION = "2026.05.28-codee-remote-owner-toolbox-v1" COMMAND_PRESETS = { "doctor": { "title": "Codee Doctor", "purpose": "Check local tools, sensitive files, manifests, and public routes.", "example": "python3 codee.py doctor .", }, "heartbeat": { "title": "Heartbeat", "purpose": "Show a one-command pulse of folder health, proof, tasks, and next steps.", "example": "python3 codee.py heartbeat . --service codeeqr", }, "scan": { "title": "Route Scan", "purpose": "Scan public routes and catch 404s before the owner promotes a link.", "example": "python3 codee.py browser-scan . --url https://clients.codee.chat/codee-chat.html", }, "portal": { "title": "Open Portal", "purpose": "Open the customer portal or print the portal route for this workspace.", "example": "python3 codee.py portal . --service codee-remote --plan pro", }, "support": { "title": "Support Packet", "purpose": "Create a support request packet with the user's exact request.", "example": "python3 codee.py support . --service codee-remote --message \"Review remote setup\"", }, "test lane": { "title": "Test Lane", "purpose": "Run one service lane proof check without changing production.", "example": "python3 codee.py test lane . --service codeeqr --route-check", }, "preview dry-run": { "title": "Preview Dry Run", "purpose": "Generate a preview plan without activating hosting or payment.", "example": "python3 codee.py preview . --service codeeqr --dry-run", }, } ALLOWLIST = set(COMMAND_PRESETS) BLOCKED_ACTIONS = { "raw shell", "secret export", "destructive cleanup", "silent screen streaming", "owner credential access", "hidden background control", } TOOLBOX = [ {"id": "pair", "title": "Pair Phone", "host_command": "python3 codee-remote-host.py pair . --write", "state": "foundation-live"}, {"id": "session-start", "title": "Start Visible Session", "host_command": "python3 codee-remote-host.py session . --session-action start --write", "state": "foundation-live"}, {"id": "session-stop", "title": "Stop Session", "host_command": "python3 codee-remote-host.py stop . --write", "state": "foundation-live"}, {"id": "enqueue-heartbeat", "title": "Queue Heartbeat", "host_command": "python3 codee-remote-host.py enqueue . --command heartbeat --write", "state": "foundation-live"}, {"id": "enqueue-doctor", "title": "Queue Doctor", "host_command": "python3 codee-remote-host.py enqueue . --command doctor --write", "state": "foundation-live"}, {"id": "proof", "title": "Export Proof Packet", "host_command": "python3 codee-remote-host.py proof . --write", "state": "foundation-live"}, {"id": "audit", "title": "Audit Events", "host_command": "python3 codee-remote-host.py audit .", "state": "foundation-live"}, ] def now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat() def write_json(path: Path, payload: dict[str, Any], *, force: bool = True) -> bool: path.parent.mkdir(parents=True, exist_ok=True) if path.exists() and not force: return False path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") return True def read_json(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: parsed = json.loads(path.read_text(encoding="utf-8")) except Exception: return {} return parsed if isinstance(parsed, dict) else {} def append_event(root: Path, event: str, payload: dict[str, Any] | None = None) -> Path: path = root / ".codee" / "remote-host" / "events.jsonl" path.parent.mkdir(parents=True, exist_ok=True) record = { "schema": "codee.remote-host.event.v1", "created_at": now(), "event": event, "host_version": VERSION, "payload": payload or {}, } with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(record, sort_keys=True) + "\n") return path def device_id(root: Path, device_name: str) -> str: seed = f"{root.resolve()}:{device_name}" return "cdev_" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16] def remote_dir(root: Path) -> Path: return root / ".codee" / "remote-host" def state_path(root: Path) -> Path: return remote_dir(root) / "host-state.json" def queue_path(root: Path) -> Path: return remote_dir(root) / "command-queue.json" def pairing_path(root: Path) -> Path: return remote_dir(root) / "pairing.json" def session_path(root: Path) -> Path: return remote_dir(root) / "session.json" def proof_path(root: Path) -> Path: return remote_dir(root) / "latest-proof.json" def normalize_command(command: str) -> str: clean = " ".join(str(command or "").strip().lower().replace("_", " ").replace("-", " ").split()) aliases = { "route scan": "scan", "site scan": "scan", "browser scan": "scan", "proof scan": "scan", "codee doctor": "doctor", "codee heartbeat": "heartbeat", "open portal": "portal", "support packet": "support", "preview": "preview dry-run", "preview dry run": "preview dry-run", } return aliases.get(clean, clean) def read_state(root: Path) -> dict[str, Any]: return read_json(state_path(root)) def write_state(root: Path, payload: dict[str, Any]) -> None: payload["updated_at"] = now() write_json(state_path(root), payload) def default_state(root: Path, device_name: str) -> dict[str, Any]: return { "ok": True, "schema": "codee.remote-host.v2", "version": VERSION, "created_at": now(), "updated_at": now(), "workspace": str(root), "device_name": device_name, "device_id": device_id(root, device_name), "allowlisted_commands": sorted(ALLOWLIST), "blocked_actions": sorted(BLOCKED_ACTIONS), "screen_streaming": "locked_until_signaling_turn_mfa_and_visible_consent", "input_control": "locked_until_explicit_desktop_consent", "session_state": "not_started", "emergency_stop": False, "visible_consent_required": True, "queue_path": ".codee/remote-host/command-queue.json", "state_path": ".codee/remote-host/host-state.json", "pairing_path": ".codee/remote-host/pairing.json", "session_path": ".codee/remote-host/session.json", "toolbox_path": ".codee/remote-host/toolbox.json", } def command_install(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() device_name = str(args.device_name or (os.uname().nodename if hasattr(os, "uname") else "Codee Desktop")).strip() payload = default_state(root, device_name) if args.write: write_state(root, payload) write_json(queue_path(root), {"schema": "codee.remote-command-queue.v2", "commands": [], "processed": []}) write_json(remote_dir(root) / "toolbox.json", {"schema": "codee.remote-host-toolbox.v1", "created_at": now(), "tools": TOOLBOX, "command_presets": COMMAND_PRESETS}) append_event(root, "RemoteHostInstalled", {"device_id": payload["device_id"], "device_name": device_name}) if args.json: print(json.dumps(payload, indent=2)) else: print("Codee Remote Host") print(f"version={VERSION}") print(f"device_id={payload['device_id']}") print(f"screen_streaming={payload['screen_streaming']}") if args.write: print(f"state={root / '.codee' / 'remote-host' / 'host-state.json'}") return 0 def command_status(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() state = read_state(root) queue = read_json(queue_path(root)) pairing = read_json(pairing_path(root)) session = read_json(session_path(root)) commands = queue.get("commands") if isinstance(queue.get("commands"), list) else [] payload = { "ok": bool(state), "schema": "codee.remote-host-status.v1", "created_at": now(), "host_version": VERSION, "state_exists": bool(state), "device_id": state.get("device_id", ""), "device_name": state.get("device_name", ""), "session_state": state.get("session_state", "not_installed"), "emergency_stop": bool(state.get("emergency_stop")), "pending_commands": len(commands), "allowlisted_commands": sorted(ALLOWLIST), "screen_streaming": state.get("screen_streaming", "not_installed"), "pairing_expires_at": pairing.get("expires_at", ""), "session_id": session.get("session_id", ""), } if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host status: ok={payload['ok']}") print(f"device_id={payload['device_id']}") print(f"session_state={payload['session_state']}") print(f"emergency_stop={payload['emergency_stop']}") print(f"pending_commands={payload['pending_commands']}") print(f"screen_streaming={payload['screen_streaming']}") return 0 if payload["ok"] else 1 def command_run_once(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() path = queue_path(root) queue = read_json(path) commands = queue.get("commands") if isinstance(queue.get("commands"), list) else [] processed: list[dict[str, Any]] = [] remaining: list[dict[str, Any]] = [] for item in commands: command = normalize_command(str(item.get("command") if isinstance(item, dict) else item).strip()) allowed = command in ALLOWLIST result = { "command": command, "allowed": allowed, "status": "accepted-dry-run" if allowed else "rejected", "preset": COMMAND_PRESETS.get(command, {}), "note": "Foundation host records queue results. Actual execution stays backend-gated and owner-visible.", "processed_at": now(), } processed.append(result) append_event(root, "RemoteHostCommandProcessed", result) if not allowed: remaining.append(item if isinstance(item, dict) else {"command": command}) write_json(path, {"schema": "codee.remote-command-queue.v2", "commands": remaining, "processed": processed[-25:]}) if args.json: print(json.dumps({"ok": True, "processed": processed, "remaining": remaining}, indent=2)) else: print(f"Codee Remote Host run-once: processed={len(processed)} remaining={len(remaining)}") return 0 def command_pair(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() state = read_state(root) if not state: device_name = str(args.device_name or (os.uname().nodename if hasattr(os, "uname") else "Codee Desktop")).strip() state = default_state(root, device_name) ttl_minutes = max(1, min(60, int(str(args.expires_minutes or "10")))) code = "-".join([secrets.token_hex(2).upper(), secrets.token_hex(2).upper(), secrets.token_hex(1).upper()]) expires_at = (dt.datetime.now(dt.timezone.utc) + dt.timedelta(minutes=ttl_minutes)).isoformat() payload = { "ok": True, "schema": "codee.remote-host-pairing.v1", "created_at": now(), "expires_at": expires_at, "ttl_minutes": ttl_minutes, "pairing_code": code, "device_id": state.get("device_id", ""), "device_name": state.get("device_name", ""), "workspace": str(root), "viewer_url": "https://clients.codee.chat/codee-remote-viewer.html", "security_note": "This pair code does not stream the screen. It only proves owner-approved pairing intent.", } state.update({"session_state": "pairing_ready", "pairing_expires_at": expires_at, "last_pairing_code_hint": code[-4:], "emergency_stop": False}) if args.write: write_state(root, state) write_json(pairing_path(root), payload) append_event(root, "RemoteHostPairingCreated", {"device_id": payload["device_id"], "expires_at": expires_at}) if args.json: print(json.dumps(payload, indent=2)) else: print("Codee Remote Host pair code") print(f"pairing_code={code}") print(f"expires_at={expires_at}") if args.write: print(f"pairing={pairing_path(root)}") return 0 def command_session(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() state = read_state(root) if not state: print("Codee Remote Host is not installed. Run `python3 codee-remote-host.py install . --write` first.") return 1 current = read_json(session_path(root)) action = str(args.session_action or "status").strip().lower() session_id = str(current.get("session_id") or ("crs_" + secrets.token_hex(8))) if action == "start": status = "waiting_for_visible_desktop_consent" state.update({"session_state": status, "emergency_stop": False}) elif action == "stop": status = "stopped_by_owner" state.update({"session_state": status, "emergency_stop": True}) else: status = str(state.get("session_state") or current.get("status") or "not_started") payload = { "ok": True, "schema": "codee.remote-host-session.v1", "created_at": current.get("created_at") or now(), "updated_at": now(), "session_id": session_id, "action": action, "status": status, "device_id": state.get("device_id", ""), "visible_consent_required": True, "screen_streaming": "locked_until_signaling_turn_mfa_and_visible_consent", "input_control": "locked_until_explicit_desktop_consent", "emergency_stop": bool(state.get("emergency_stop")), } if args.write or action in {"start", "stop"}: write_state(root, state) write_json(session_path(root), payload) append_event(root, "RemoteHostSessionUpdated", {"session_id": session_id, "action": action, "status": status}) if args.json: print(json.dumps(payload, indent=2)) else: print("Codee Remote Host session") print(f"session_id={session_id}") print(f"status={status}") print(f"emergency_stop={payload['emergency_stop']}") return 0 def command_stop(args: argparse.Namespace) -> int: args.session_action = "stop" return command_session(args) def command_enqueue(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() command = normalize_command(args.remote_command) allowed = command in ALLOWLIST item = { "id": "rcmd_" + secrets.token_hex(6), "created_at": now(), "command": command, "allowed": allowed, "preset": COMMAND_PRESETS.get(command, {}), "source": "owner-remote-viewer", "status": "queued" if allowed else "rejected", } path = queue_path(root) queue = read_json(path) commands = queue.get("commands") if isinstance(queue.get("commands"), list) else [] if allowed: commands.append(item) payload = {"ok": allowed, "schema": "codee.remote-command-enqueue.v1", "queued": item if allowed else None, "rejected": None if allowed else item, "allowlisted_commands": sorted(ALLOWLIST)} if args.write: write_json(path, {"schema": "codee.remote-command-queue.v2", "commands": commands, "processed": queue.get("processed", []) if isinstance(queue.get("processed"), list) else []}) append_event(root, "RemoteHostCommandQueued" if allowed else "RemoteHostCommandRejected", item) if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host enqueue: ok={allowed} command={command}") if not allowed: print(f"allowed={', '.join(sorted(ALLOWLIST))}") return 0 if allowed else 2 def command_toolbox(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() payload = {"ok": True, "schema": "codee.remote-host-toolbox.v1", "created_at": now(), "tools": TOOLBOX, "command_presets": COMMAND_PRESETS, "blocked_actions": sorted(BLOCKED_ACTIONS)} if args.write: write_json(remote_dir(root) / "toolbox.json", payload) append_event(root, "RemoteHostToolboxWritten", {"tool_count": len(TOOLBOX)}) if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host toolbox: tools={len(TOOLBOX)}") if args.write: print(f"toolbox={remote_dir(root) / 'toolbox.json'}") return 0 def command_proof(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() events: list[dict[str, Any]] = [] event_file = remote_dir(root) / "events.jsonl" if event_file.exists(): for line in event_file.read_text(encoding="utf-8", errors="replace").splitlines()[-100:]: try: events.append(json.loads(line)) except Exception: continue payload = { "ok": True, "schema": "codee.remote-host-proof.v1", "created_at": now(), "host_version": VERSION, "workspace": str(root), "state": read_state(root), "pairing": read_json(pairing_path(root)), "session": read_json(session_path(root)), "queue": read_json(queue_path(root)), "event_count": len(events), "latest_events": events[-25:], "public_safety": { "screen_streaming_live": False, "input_control_live": False, "reason": "Host proof is live, but active streaming/control waits for signaling, TURN, device registry, MFA/passkey, visible consent, signed host, and audit gate.", }, } if args.write: write_json(proof_path(root), payload) append_event(root, "RemoteHostProofExported", {"proof": str(proof_path(root))}) if args.json: print(json.dumps(payload, indent=2)) else: print("Codee Remote Host proof") print(f"events={len(events)}") if args.write: print(f"proof={proof_path(root)}") return 0 def command_audit(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() event_path = root / ".codee" / "remote-host" / "events.jsonl" events = [] if event_path.exists(): for line in event_path.read_text(encoding="utf-8", errors="replace").splitlines()[-100:]: try: events.append(json.loads(line)) except Exception: continue payload = {"ok": True, "schema": "codee.remote-host-audit.v1", "created_at": now(), "event_count": len(events), "latest_events": events[-10:]} write_json(root / ".codee" / "remote-host" / "latest-audit.json", payload) if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host audit: events={len(events)}") print(f"report={root / '.codee' / 'remote-host' / 'latest-audit.json'}") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Codee Remote Host foundation. No silent screen streaming or public owner credentials.") parser.add_argument("--version", action="version", version=VERSION) sub = parser.add_subparsers(dest="command", required=True) install = sub.add_parser("install", help="Install local host state and queue files.") install.add_argument("path", nargs="?", default=".") install.add_argument("--device-name", default="") install.add_argument("--write", action="store_true") install.add_argument("--json", action="store_true") install.set_defaults(func=command_install) pair = sub.add_parser("pair", help="Create an expiring owner pairing code.") pair.add_argument("path", nargs="?", default=".") pair.add_argument("--device-name", default="") pair.add_argument("--expires-minutes", default="10") pair.add_argument("--write", action="store_true") pair.add_argument("--json", action="store_true") pair.set_defaults(func=command_pair) session = sub.add_parser("session", help="Start, stop, or inspect a visible remote session packet.") session.add_argument("path", nargs="?", default=".") session.add_argument("--session-action", default="status", choices=["start", "stop", "status"]) session.add_argument("--write", action="store_true") session.add_argument("--json", action="store_true") session.set_defaults(func=command_session) stop = sub.add_parser("stop", help="Trigger the owner emergency stop state.") stop.add_argument("path", nargs="?", default=".") stop.add_argument("--write", action="store_true") stop.add_argument("--json", action="store_true") stop.set_defaults(func=command_stop) enqueue = sub.add_parser("enqueue", help="Queue one allowlisted Codee command for the owner-visible bridge.") enqueue.add_argument("path", nargs="?", default=".") enqueue.add_argument("--command", dest="remote_command", required=True) enqueue.add_argument("--write", action="store_true") enqueue.add_argument("--json", action="store_true") enqueue.set_defaults(func=command_enqueue) toolbox = sub.add_parser("toolbox", help="Write the local owner toolbox manifest.") toolbox.add_argument("path", nargs="?", default=".") toolbox.add_argument("--write", action="store_true") toolbox.add_argument("--json", action="store_true") toolbox.set_defaults(func=command_toolbox) status = sub.add_parser("status", help="Show local host status.") status.add_argument("path", nargs="?", default=".") status.add_argument("--json", action="store_true") status.set_defaults(func=command_status) run_once = sub.add_parser("run-once", help="Process one local command queue pass in dry-run mode.") run_once.add_argument("path", nargs="?", default=".") run_once.add_argument("--json", action="store_true") run_once.set_defaults(func=command_run_once) proof = sub.add_parser("proof", help="Export host pairing, session, queue, and audit proof.") proof.add_argument("path", nargs="?", default=".") proof.add_argument("--write", action="store_true") proof.add_argument("--json", action="store_true") proof.set_defaults(func=command_proof) audit = sub.add_parser("audit", help="Summarize host events.") audit.add_argument("path", nargs="?", default=".") audit.add_argument("--json", action="store_true") audit.set_defaults(func=command_audit) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) return int(args.func(args)) if __name__ == "__main__": raise SystemExit(main())