Add bidirectional chunked file transfer for Remote Desktop

Three new message types form one transfer: FILE_BEGIN carries JSON
metadata (transfer_id, dest_path, size); FILE_CHUNK is a 36-byte
ASCII transfer id followed by raw bytes; FILE_END carries a JSON
status / error string.

Sender path (utils/remote_desktop/file_transfer.send_file) opens the
file synchronously, picks a UUID, streams 256 KiB chunks, and fires an
on_progress(transfer_id, bytes_done, total) callback per chunk. The
caller wraps in a thread for non-blocking uploads. Receiver
(FileReceiver) demultiplexes by transfer_id so multiple in-flight
files on one channel work, expanduser's ~ in dest_path, and creates
parent directories. There is no aggregate size limit and no
destination-path restriction — token holders are trusted users.

Host: set_file_receiver attaches a custom receiver (with progress /
complete callbacks); send_file_to_viewers streams a local file to
every authenticated viewer.

Viewer: send_file streams a local file to the host; set_file_receiver
attaches a receiver for files pushed from the host. Receiver callbacks
fire on the receive thread, so GUI consumers must marshal back to the
UI thread (which is what the upcoming Remote Desktop tab does via Qt
signals).

Author: JE-Chen

je_auto_control/utils/remote_desktop/__init__.py

@@ -16,6 +16,9 @@
 from je_auto_control.utils.remote_desktop.clipboard_sync import (
     ClipboardSyncError,
 )
+from je_auto_control.utils.remote_desktop.file_transfer import (
+    FileReceiver, FileSendResult, FileTransferError, send_file,
+)
 from je_auto_control.utils.remote_desktop.host import RemoteDesktopHost
 from je_auto_control.utils.remote_desktop.host_id import (
     HostIdError, format_host_id, generate_host_id, load_or_create_host_id,
@@ -46,4 +49,5 @@
     "AudioBackendError", "AudioCapture", "AudioPlayer",
     "is_audio_backend_available",
     "ClipboardSyncError",
+    "FileReceiver", "FileSendResult", "FileTransferError", "send_file",
 ]

je_auto_control/utils/remote_desktop/file_transfer.py

@@ -0,0 +1,269 @@
+"""Chunked file transfer over the typed-message channel.
+
+Three message types form a transfer:
+
+* ``FILE_BEGIN`` — JSON ``{transfer_id, dest_path, size}`` announces a new
+  stream. ``transfer_id`` is a 36-character UUID hex string so the
+  receiver can demultiplex multiple in-flight transfers on one channel.
+* ``FILE_CHUNK`` — first 36 bytes are the ASCII transfer id, the rest is
+  raw payload. Chunks arrive in order; the receiver writes them
+  sequentially and accumulates ``bytes_done``.
+* ``FILE_END`` — JSON ``{transfer_id, status, error?}`` finalises the
+  stream. The receiver closes the file and fires ``on_complete`` with
+  success / failure info.
+
+There is no central per-host file-size limit — operators relying on
+this should keep ``trusted token holders == trusted users`` in mind, and
+treat the dropbox / destination filesystem accordingly.
+"""
+import json
+import os
+import threading
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Dict, Optional, Tuple
+
+from je_auto_control.utils.logging.logging_instance import autocontrol_logger
+from je_auto_control.utils.remote_desktop.protocol import MessageType
+
+DEFAULT_CHUNK_SIZE = 256 * 1024
+TRANSFER_ID_LEN = 36  # str(uuid.uuid4()) length
+
+ProgressCallback = Callable[[str, int, int], None]
+CompleteCallback = Callable[[str, bool, Optional[str], str], None]
+
+
+class FileTransferError(RuntimeError):
+    """Raised when a file-transfer payload is malformed."""
+
+
+def new_transfer_id() -> str:
+    """Return a fresh 36-character ASCII transfer ID."""
+    return str(uuid.uuid4())
+
+
+def encode_begin(transfer_id: str, dest_path: str, size: int) -> bytes:
+    if len(transfer_id) != TRANSFER_ID_LEN:
+        raise FileTransferError("transfer_id must be a 36-char UUID string")
+    return json.dumps({
+        "transfer_id": transfer_id,
+        "dest_path": str(dest_path),
+        "size": int(size),
+    }, ensure_ascii=False).encode("utf-8")
+
+
+def decode_begin(payload: bytes) -> Tuple[str, str, int]:
+    body = _decode_json(payload)
+    transfer_id = body.get("transfer_id")
+    dest_path = body.get("dest_path")
+    size = body.get("size")
+    if (not isinstance(transfer_id, str)
+            or len(transfer_id) != TRANSFER_ID_LEN):
+        raise FileTransferError("FILE_BEGIN missing valid transfer_id")
+    if not isinstance(dest_path, str) or not dest_path:
+        raise FileTransferError("FILE_BEGIN missing dest_path")
+    if not isinstance(size, int) or size < 0:
+        raise FileTransferError("FILE_BEGIN missing valid size")
+    return transfer_id, dest_path, size
+
+
+def encode_chunk(transfer_id: str, chunk: bytes) -> bytes:
+    if len(transfer_id) != TRANSFER_ID_LEN:
+        raise FileTransferError("transfer_id must be a 36-char UUID string")
+    return transfer_id.encode("ascii") + bytes(chunk)
+
+
+def decode_chunk(payload: bytes) -> Tuple[str, bytes]:
+    if len(payload) < TRANSFER_ID_LEN:
+        raise FileTransferError("FILE_CHUNK shorter than transfer id header")
+    transfer_id = payload[:TRANSFER_ID_LEN].decode("ascii", errors="replace")
+    return transfer_id, bytes(payload[TRANSFER_ID_LEN:])
+
+
+def encode_end(transfer_id: str, status: str = "ok",
+               error: Optional[str] = None) -> bytes:
+    if len(transfer_id) != TRANSFER_ID_LEN:
+        raise FileTransferError("transfer_id must be a 36-char UUID string")
+    body: Dict[str, Any] = {"transfer_id": transfer_id, "status": status}
+    if error is not None:
+        body["error"] = str(error)
+    return json.dumps(body, ensure_ascii=False).encode("utf-8")
+
+
+def decode_end(payload: bytes) -> Tuple[str, str, Optional[str]]:
+    body = _decode_json(payload)
+    transfer_id = body.get("transfer_id")
+    status = body.get("status", "ok")
+    if (not isinstance(transfer_id, str)
+            or len(transfer_id) != TRANSFER_ID_LEN):
+        raise FileTransferError("FILE_END missing valid transfer_id")
+    if not isinstance(status, str):
+        raise FileTransferError("FILE_END status must be a string")
+    error = body.get("error")
+    return transfer_id, status, error if isinstance(error, str) else None
+
+
+def _decode_json(payload: bytes) -> Dict[str, Any]:
+    try:
+        body = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise FileTransferError(f"invalid JSON: {error}") from error
+    if not isinstance(body, dict):
+        raise FileTransferError("payload must be a JSON object")
+    return body
+
+
+@dataclass
+class _Incoming:
+    """Per-transfer state owned by ``FileReceiver``."""
+
+    transfer_id: str
+    dest_path: Path
+    total_size: int
+    handle: Any  # file object
+    bytes_done: int = 0
+    error: Optional[str] = None
+
+
+class FileReceiver:
+    """Demultiplex incoming FILE_* messages into one or more file writes."""
+
+    def __init__(self, on_progress: Optional[ProgressCallback] = None,
+                 on_complete: Optional[CompleteCallback] = None) -> None:
+        self._on_progress = on_progress
+        self._on_complete = on_complete
+        self._active: Dict[str, _Incoming] = {}
+        self._lock = threading.Lock()
+
+    def handle_begin(self, payload: bytes) -> None:
+        transfer_id, dest_path, total_size = decode_begin(payload)
+        path = Path(os.path.expanduser(dest_path))
+        path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            handle = open(path, "wb")  # noqa: SIM115  managed manually
+        except OSError as error:
+            self._fire_complete(transfer_id, False, str(error), str(path))
+            return
+        with self._lock:
+            self._active[transfer_id] = _Incoming(
+                transfer_id=transfer_id, dest_path=path,
+                total_size=total_size, handle=handle,
+            )
+        if self._on_progress is not None:
+            self._on_progress(transfer_id, 0, total_size)
+
+    def handle_chunk(self, payload: bytes) -> None:
+        transfer_id, chunk = decode_chunk(payload)
+        with self._lock:
+            incoming = self._active.get(transfer_id)
+        if incoming is None:
+            autocontrol_logger.info(
+                "remote_desktop FILE_CHUNK for unknown transfer %s",
+                transfer_id,
+            )
+            return
+        try:
+            incoming.handle.write(chunk)
+        except OSError as error:
+            incoming.error = str(error)
+            self._abort(incoming)
+            return
+        incoming.bytes_done += len(chunk)
+        if self._on_progress is not None:
+            self._on_progress(
+                transfer_id, incoming.bytes_done, incoming.total_size,
+            )
+
+    def handle_end(self, payload: bytes) -> None:
+        transfer_id, status, error = decode_end(payload)
+        with self._lock:
+            incoming = self._active.pop(transfer_id, None)
+        if incoming is None:
+            return
+        try:
+            incoming.handle.close()
+        except OSError:
+            pass
+        ok = (status == "ok") and incoming.error is None
+        message = error or incoming.error
+        self._fire_complete(
+            transfer_id, ok, message, str(incoming.dest_path),
+        )
+
+    def _abort(self, incoming: _Incoming) -> None:
+        try:
+            incoming.handle.close()
+        except OSError:
+            pass
+        with self._lock:
+            self._active.pop(incoming.transfer_id, None)
+        self._fire_complete(
+            incoming.transfer_id, False, incoming.error,
+            str(incoming.dest_path),
+        )
+
+    def _fire_complete(self, transfer_id: str, ok: bool,
+                       error: Optional[str], dest_path: str) -> None:
+        if self._on_complete is None:
+            return
+        try:
+            self._on_complete(transfer_id, ok, error, dest_path)
+        except Exception:  # noqa: BLE001
+            autocontrol_logger.exception(
+                "remote_desktop FileReceiver.on_complete callback raised"
+            )
+
+
+@dataclass
+class FileSendResult:
+    """Outcome of one outbound transfer."""
+
+    transfer_id: str
+    success: bool
+    error: Optional[str] = None
+    bytes_sent: int = 0
+
+
+def send_file(channel, source_path: str, dest_path: str,
+              on_progress: Optional[ProgressCallback] = None,
+              chunk_size: int = DEFAULT_CHUNK_SIZE,
+              transfer_id: Optional[str] = None) -> FileSendResult:
+    """Stream ``source_path`` to ``dest_path`` over ``channel``.
+
+    Synchronous: the caller's thread does the I/O. Wrap in a thread for
+    background uploads. ``on_progress(transfer_id, bytes_done, total)``
+    fires after every chunk (and once at the start with ``bytes_done=0``).
+    """
+    transfer_id = transfer_id or new_transfer_id()
+    source = Path(os.path.expanduser(source_path))
+    if not source.is_file():
+        raise FileTransferError(f"source not found: {source}")
+    total_size = source.stat().st_size
+    channel.send_typed(MessageType.FILE_BEGIN,
+                       encode_begin(transfer_id, dest_path, total_size))
+    if on_progress is not None:
+        on_progress(transfer_id, 0, total_size)
+    bytes_sent = 0
+    try:
+        with open(source, "rb") as handle:
+            while True:
+                chunk = handle.read(int(chunk_size))
+                if not chunk:
+                    break
+                channel.send_typed(
+                    MessageType.FILE_CHUNK, encode_chunk(transfer_id, chunk),
+                )
+                bytes_sent += len(chunk)
+                if on_progress is not None:
+                    on_progress(transfer_id, bytes_sent, total_size)
+    except (OSError, ConnectionError) as error:
+        channel.send_typed(
+            MessageType.FILE_END,
+            encode_end(transfer_id, status="error", error=str(error)),
+        )
+        return FileSendResult(transfer_id=transfer_id, success=False,
+                              error=str(error), bytes_sent=bytes_sent)
+    channel.send_typed(MessageType.FILE_END, encode_end(transfer_id))
+    return FileSendResult(transfer_id=transfer_id, success=True,
+                          bytes_sent=bytes_sent)

je_auto_control/utils/remote_desktop/host.py

@@ -20,6 +20,9 @@
 from je_auto_control.utils.remote_desktop.clipboard_sync import (
     ClipboardSyncError, decode as decode_clipboard, encode_image, encode_text,
 )
+from je_auto_control.utils.remote_desktop.file_transfer import (
+    FileReceiver, FileTransferError, send_file,
+)
 from je_auto_control.utils.remote_desktop.host_id import (
     load_or_create_host_id, validate_host_id,
 )
@@ -210,11 +213,31 @@ def _recv_loop(self) -> None:
             if msg_type is MessageType.CLIPBOARD:
                 self._handle_clipboard_payload(payload)
                 continue
+            if msg_type in (MessageType.FILE_BEGIN, MessageType.FILE_CHUNK,
+                            MessageType.FILE_END):
+                self._handle_file_payload(msg_type, payload)
+                continue
             autocontrol_logger.info(
                 "remote_desktop unexpected msg %s from %s",
                 msg_type.name, self._address,
             )
 
+    def _handle_file_payload(self, msg_type: MessageType,
+                             payload: bytes) -> None:
+        receiver = self._host._ensure_file_receiver()
+        try:
+            if msg_type is MessageType.FILE_BEGIN:
+                receiver.handle_begin(payload)
+            elif msg_type is MessageType.FILE_CHUNK:
+                receiver.handle_chunk(payload)
+            elif msg_type is MessageType.FILE_END:
+                receiver.handle_end(payload)
+        except FileTransferError as error:
+            autocontrol_logger.info(
+                "remote_desktop bad file message from %s: %r",
+                self._address, error,
+            )
+
     def _handle_clipboard_payload(self, payload: bytes) -> None:
         try:
             kind, data = decode_clipboard(payload)
@@ -303,6 +326,7 @@ def __init__(self, token: str,
             frame_provider or _default_frame_provider(region, int(quality))
         )
         self._dispatch: InputDispatcher = input_dispatcher or dispatch_input
+        self._file_receiver: Optional[FileReceiver] = None
         self._audio_enabled = bool(enable_audio)
         self._audio_device = audio_device
         self._audio_sample_rate = int(audio_sample_rate)
@@ -476,6 +500,37 @@ def _broadcast_clipboard_payload(self, payload: bytes) -> int:
                 client.stop()
         return sent
 
+    def set_file_receiver(self, receiver: FileReceiver) -> None:
+        """Replace the default ``FileReceiver`` (e.g. to wire progress callbacks)."""
+        self._file_receiver = receiver
+
+    def _ensure_file_receiver(self) -> FileReceiver:
+        if self._file_receiver is None:
+            self._file_receiver = FileReceiver()
+        return self._file_receiver
+
+    def send_file_to_viewers(self, source_path: str, dest_path: str,
+                             on_progress=None) -> int:
+        """Stream ``source_path`` to every authenticated viewer.
+
+        Returns the number of viewers the transfer was attempted on.
+        Each viewer gets its own ``transfer_id`` so progress callbacks
+        can be demultiplexed in the GUI.
+        """
+        with self._clients_lock:
+            clients = [c for c in self._clients
+                       if c.authenticated and not c._shutdown.is_set()]
+        for client in clients:
+            try:
+                send_file(client._channel, source_path, dest_path,
+                          on_progress=on_progress)
+            except (OSError, ConnectionError, FileTransferError) as error:
+                autocontrol_logger.info(
+                    "remote_desktop file send to %s failed: %r",
+                    client.address, error,
+                )
+        return len(clients)
+
     def _apply_clipboard(self, kind: str, data: Any) -> None:
         """Set this host's local clipboard from a decoded CLIPBOARD payload.
 

je_auto_control/utils/remote_desktop/protocol.py

@@ -34,6 +34,9 @@ class MessageType(enum.IntEnum):
     AUDIO = 0x11  # host -> viewer: PCM audio chunk
     INPUT = 0x20  # viewer -> host: JSON input message
     CLIPBOARD = 0x21  # either way: clipboard payload (text or image)
+    FILE_BEGIN = 0x22  # either way: JSON metadata for an incoming transfer
+    FILE_CHUNK = 0x23  # either way: 36-byte transfer id + chunk bytes
+    FILE_END = 0x24  # either way: JSON status for a finished transfer
     PING = 0x30  # either way: liveness