Preserve raw terminal output for TUIs

.gitignore

@@ -1,3 +1,4 @@
 /target/
 **/*.rs.bk
 .DS_Store
+vscode-extension/*.vsix

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "0.1.4"
+version = "0.1.8"
 dependencies = [
  "anyhow",
  "base64",
@@ -447,6 +447,7 @@ dependencies = [
  "crossterm",
  "dirs",
  "ed25519-dalek",
+ "filetime",
  "hkdf",
  "hmac",
  "libc",
@@ -577,6 +578,16 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "filetime"
+version = "0.2.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759"
+dependencies = [
+ "cfg-if",
+ "libc",
+]
+
 [[package]]
 name = "foldhash"
 version = "0.1.5"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "0.1.4"
+version = "0.1.8"
 edition = "2024"
 license = "MIT"
 
@@ -14,6 +14,7 @@ clap = { version = "4.5", features = ["derive"] }
 crossterm = "0.28"
 dirs = "5.0"
 ed25519-dalek = "2.1"
+filetime = "0.2"
 hkdf = "0.12"
 hmac = "0.12"
 libc = "0.2"

Makefile

@@ -1,4 +1,4 @@
-.PHONY: build test fmt install package-release package-release-linux package-release-windows publish-release bench-report bench-local bench-local-json bench-docker-ssh bench-docker-mosh fuzz-smoke fuzz-deep soak-local tui-harness
+.PHONY: build test fmt clippy release-check install package-release package-release-linux package-release-windows publish-release bench-report bench-local bench-local-json bench-docker-ssh bench-docker-mosh fuzz-smoke fuzz-deep soak-local tui-harness
 
 build:
 	cargo build --release
@@ -9,6 +9,16 @@ test:
 fmt:
 	cargo fmt
 
+clippy:
+	cargo clippy --all-targets -- -D warnings
+
+release-check:
+	cargo fmt --check
+	sh -n install.sh scripts/*.sh
+	cargo clippy --all-targets -- -D warnings
+	cargo test
+	$(MAKE) tui-harness
+
 install:
 	sh install.sh --from-current
 

README.md

@@ -1,32 +1,31 @@
 # Dosh
 
-Dosh is an encrypted remote terminal for fast reconnecting shells. It is meant
+Dosh is an encrypted remote terminal for fast reconnecting shells.
-to replace Mosh and day-to-day interactive SSH sessions.
 
-It runs a `dosh-server` on the remote machine and a `dosh` client locally. The
+It runs a `dosh-server` on a Unix-like host and a `dosh` client on macOS,
-first setup can use SSH. After that, Dosh can attach over encrypted UDP with
+Linux, or Windows. Setup can use SSH, then Dosh connects over encrypted UDP,
-cached credentials, keep terminal sessions alive, reconnect after network
+keeps sessions alive across disconnects, supports terminal apps, and can carry
-changes, and forward TCP ports.
+TCP forwarding, file copy, and VS Code Remote-SSH streams.
 
 ## Support
 
 - Client: macOS, Linux, Windows
-- Server: Unix-like systems with PTYs
+- Server: Linux and other Unix-like systems with PTYs
+- Default UDP port: `50000`
 - Windows: client only
-- UDP port: one configured server port, default `50000`
 
-Dosh is not an SCP/SFTP client and does not implement X11 forwarding. Windows
+Dosh is meant to replace Mosh and everyday interactive SSH sessions. It does
-is client-only.
+not currently include SFTP, X11 forwarding, or a Windows server.
 
 ## Install
 
-Server and client on Unix/macOS:
+Unix/macOS server and client:
 
 ```sh
 curl -fsSL https://git.palav.dev/Palav/dosh/raw/branch/main/install.sh | sh -s -- both --repo https://git.palav.dev/Palav/dosh.git --port 50000
 ```
 
-Client only on Unix/macOS:
+Unix/macOS client only:
 
 ```sh
 curl -fsSL https://git.palav.dev/Palav/dosh/raw/branch/main/install.sh | sh -s -- client --repo https://git.palav.dev/Palav/dosh.git
@@ -38,30 +37,48 @@ Windows client:
 irm https://git.palav.dev/Palav/dosh/raw/branch/main/install.ps1 | iex
 ```
 
-## Commands
+## Use
 
 ```sh
-dosh setup HOST          # import SSH config and trust the Dosh host key
+dosh setup HOST
-dosh HOST                # connect
+dosh HOST
-dosh HOST COMMAND        # connect and run a command
+dosh HOST COMMAND
-dosh update              # update Dosh
+dosh update
-dosh status HOST         # show remote tmux sessions and Dosh service status
-dosh restart HOST        # restart dosh-server and show service status
-dosh doctor HOST         # check config and connectivity
-dosh recover HOST        # clear cached attach state and re-check
 ```
 
-Examples:
+Useful commands:
 
 ```sh
-dosh palav
+dosh exec HOST COMMAND
-dosh palav tm
+dosh cp SRC DST
-dosh forward palav -L 8080:127.0.0.1:80
+dosh ls host:path
-dosh forward palav -D 1080
+dosh cat host:path
-dosh forward palav -R 2222:127.0.0.1:22
+dosh mkdir host:path
+dosh rm [-r] host:path
+dosh forward HOST -L 8080:127.0.0.1:80
+dosh forward HOST -D 1080
+dosh forward HOST -R 2222:127.0.0.1:22
+dosh status HOST
+dosh doctor HOST
+dosh recover HOST
+dosh restart HOST
 ```
 
-Agent forwarding is opt-in with `-A` and must be enabled on the server.
+Agent forwarding is opt-in with `-A` and must also be enabled on the server.
+File copy must be enabled by the server config.
+
+## VS Code
+
+Dosh can carry VS Code Remote-SSH through its transport:
+
+```sh
+dosh vscode setup HOST
+dosh vscode HOST /remote/path
+```
+
+This creates a managed SSH config entry using `ProxyCommand dosh proxy-stdio`.
+VS Code still uses Remote-SSH and its normal remote server; Dosh carries the
+SSH byte stream.
 
 ## Config
 
@@ -75,8 +92,21 @@ Common client settings:
 
 ```toml
 default_session = "new"
-auth_preference = "native,ssh"
 predict = true
 cache_attach_tickets = true
 disconnect_status = true
 ```
+
+## Rust Library
+
+Dosh exposes a Rust transport for encrypted, reconnecting application streams.
+Use `dosh::client::DoshClient` and `dosh::server::DoshServer` for the normal
+native-auth path. Use `dosh::transport::DoshTransport` only when you already
+own session setup and authentication.
+
+Runnable examples:
+
+```text
+examples/sdk_echo_client.rs
+examples/sdk_echo_server.rs
+```

examples/sdk_echo_client.rs

@@ -0,0 +1,50 @@
+use anyhow::{Result, bail};
+use dosh::client::DoshClient;
+use dosh::transport::{SessionEvent, TransportEvent};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let mut args = std::env::args().skip(1);
+    let host = args.next().unwrap_or_else(|| "server".to_string());
+    let message = args.collect::<Vec<_>>().join(" ");
+    let message = if message.is_empty() {
+        "hello from dosh".to_string()
+    } else {
+        message
+    };
+
+    let client = DoshClient::load()?;
+    let mut transport = client
+        .connect(host)
+        .service("echo")
+        .connect()
+        .await?
+        .into_transport();
+    let stream = transport.open_service("echo").await?;
+
+    loop {
+        match transport.recv().await? {
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. })
+                if stream_id == stream =>
+            {
+                transport.send(stream, message.as_bytes().to_vec()).await?;
+            }
+            SessionEvent::Stream(TransportEvent::Data(data)) if data.stream_id == stream => {
+                for chunk in data.chunks {
+                    print!("{}", String::from_utf8_lossy(&chunk));
+                }
+                println!();
+                transport.close(stream).await?;
+                return Ok(());
+            }
+            SessionEvent::Stream(TransportEvent::OpenReject { reason, .. }) => {
+                bail!("server rejected echo stream: {reason}");
+            }
+            SessionEvent::Stream(_)
+            | SessionEvent::Ping
+            | SessionEvent::Pong
+            | SessionEvent::Ignored => {}
+        }
+        transport.maintenance().await?;
+    }
+}

examples/sdk_echo_server.rs

@@ -0,0 +1,37 @@
+use anyhow::Result;
+use dosh::server::{DoshServer, DoshServerConfig, DoshServerEvent};
+use dosh::transport::{SessionEvent, TransportEvent};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let config = DoshServerConfig::default().service("echo")?;
+    let mut server = DoshServer::bind(config).await?;
+    eprintln!("listening on {}", server.local_addr()?);
+
+    loop {
+        match server.recv().await? {
+            DoshServerEvent::Accepted(client) => {
+                eprintln!(
+                    "accepted user={} session={} conn={:?}",
+                    client.user, client.session, client.conn_id
+                );
+            }
+            DoshServerEvent::Session {
+                conn_id,
+                event: SessionEvent::Stream(TransportEvent::Open(open)),
+            } => {
+                server.accept_stream(conn_id, open.stream_id).await?;
+            }
+            DoshServerEvent::Session {
+                conn_id,
+                event: SessionEvent::Stream(TransportEvent::Data(data)),
+            } => {
+                for chunk in data.chunks {
+                    server.send(conn_id, data.stream_id, chunk).await?;
+                }
+            }
+            DoshServerEvent::Session { .. } | DoshServerEvent::Ignored => {}
+        }
+        server.maintenance().await?;
+    }
+}

install.ps1

@@ -1,7 +1,7 @@
 param(
     [ValidateSet("client")]
     [string]$Role = $(if ($env:DOSH_ROLE) { $env:DOSH_ROLE } else { "client" }),
-    [string]$Repo = $env:DOSH_REPO,
+    [string]$Repo = $(if ($env:DOSH_REPO) { $env:DOSH_REPO } else { "https://git.palav.dev/Palav/dosh.git" }),
     [string]$Server = $env:DOSH_SERVER,
     [string]$DoshHost = $(if ($env:DOSH_HOST) { $env:DOSH_HOST } else { $env:DOSH_DOSH_HOST }),
     [int]$Port = $(if ($env:DOSH_PORT) { [int]$env:DOSH_PORT } else { 50000 }),

install.sh

@@ -12,11 +12,7 @@ start_server=1
 force_config=0
 update_cache="${DOSH_UPDATE_CACHE:-$HOME/.cache/dosh/source}"
 quiet="${DOSH_UPDATE_QUIET:-0}"
-if [ "${DOSH_UPDATE_QUIET:-0}" = "1" ] && [ "${DOSH_BINARY_REQUIRED:-0}" != "1" ]; then
+use_prebuilt="${DOSH_USE_PREBUILT:-1}"
-  use_prebuilt=0
-else
-  use_prebuilt="${DOSH_USE_PREBUILT:-1}"
-fi
 binary_url="${DOSH_BINARY_URL:-}"
 binary_base="${DOSH_BINARY_BASE:-}"
 binary_name="${DOSH_BINARY_NAME:-}"
@@ -410,7 +406,7 @@ write_systemd_service() {
   else
     cat >"$systemd_user_dir/dosh-server.service" <<EOF
 [Unit]
-Description=Dosh server
+Description=dosh dormant shell server
 After=network-online.target
 Wants=network-online.target
 
@@ -420,6 +416,15 @@ ExecStart=$bindir/dosh-server serve
 Restart=on-failure
 RestartSec=1
 KillMode=process
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=read-only
+ReadWritePaths=$config_dir $data_dir
+RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
+RestrictRealtime=true
+LockPersonality=true
+MemoryDenyWriteExecute=true
 
 [Install]
 WantedBy=default.target
@@ -479,7 +484,8 @@ EOF
     mkdir -p "$systemd_user_dir"
     write_systemd_service
     if [ "$start_server" -eq 1 ] && systemctl --user daemon-reload >/dev/null 2>&1; then
-      systemctl --user enable --now dosh-server.service
+      systemctl --user enable dosh-server.service
+      systemctl --user restart dosh-server.service
     fi
   elif [ "$start_server" -eq 1 ]; then
     nohup "$bindir/dosh-server" serve >"$data_dir/dosh-server.log" 2>&1 &
@@ -547,8 +553,8 @@ EOF
     fi
     cat >"$hosts_config" <<EOF
 # Example:
-# [homelab]
+# [server]
-# ssh = "homelab"
+# ssh = "server"
 # dosh_host = "server.example.com"
 # port = 50000
 # default_command = "tm"

scripts/publish-gitea-release.sh

@@ -99,6 +99,21 @@ fi
 
 web_repo="${base%/}/${repo}"
 failed=0
+artifact_version() {
+  artifact="$1"
+  case "$artifact" in
+    *.tar.gz)
+      tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
+      ;;
+    *.zip)
+      unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
+      ;;
+    *)
+      return 1
+      ;;
+  esac
+}
+
 for artifact in target/dosh-release/dosh-linux-*.tar.gz target/dosh-release/dosh-macos-*.tar.gz target/dosh-release/dosh-windows-*.zip; do
   [ -f "$artifact" ] || continue
   name="$(basename "$artifact")"
@@ -107,6 +122,11 @@ for artifact in target/dosh-release/dosh-linux-*.tar.gz target/dosh-release/dosh
       continue
       ;;
   esac
+  actual="$(artifact_version "$artifact" || true)"
+  if [ "$actual" != "$version" ]; then
+    echo "skipping stale artifact $name: version ${actual:-unknown}, expected $version" >&2
+    continue
+  fi
   url="$web_repo/releases/download/$tag/$name"
   if curl -fsSI "$url" >/dev/null; then
     echo "verified $url"

scripts/tui-harness.sh

@@ -8,6 +8,7 @@ for test_name in \
   live_output_forwards_terminal_control_sequences \
   tui_control_sequences_survive_transport_verbatim \
   unicode_output_survives_transport \
+  tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution \
   large_tui_paint_is_delivered_in_mtu_safe_frames \
   resume_snapshot_preserves_alternate_screen_mode
 do

scripts/upload-gitea-release.sh

@@ -32,6 +32,21 @@ token="${GITEA_TOKEN:-}"
 title="${GITEA_TITLE:-$tag}"
 expected_version="${tag#v}"
 
+artifact_version() {
+  artifact="$1"
+  case "$artifact" in
+    *.tar.gz)
+      tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
+      ;;
+    *.zip)
+      unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
+      ;;
+    *)
+      return 1
+      ;;
+  esac
+}
+
 if [ -z "$token" ] && [ -f "$HOME/.config/dosh/gitea-token" ]; then
   token="$(tr -d '\r\n' <"$HOME/.config/dosh/gitea-token")"
 fi
@@ -52,6 +67,11 @@ if [ "$#" -eq 0 ]; then
         continue
         ;;
     esac
+    actual="$(artifact_version "$artifact" || true)"
+    if [ "$actual" != "$expected_version" ]; then
+      echo "skipping $artifact: version ${actual:-unknown}, expected $expected_version" >&2
+      continue
+    fi
     if [ -f "$artifact.sha256" ]; then
       set -- "$@" "$artifact" "$artifact.sha256"
     else
@@ -103,21 +123,6 @@ delete_existing_asset() {
     done
 }
 
-artifact_version() {
-  artifact="$1"
-  case "$artifact" in
-    *.tar.gz)
-      tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
-      ;;
-    *.zip)
-      unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
-      ;;
-    *)
-      return 1
-      ;;
-  esac
-}
-
 verify_release_artifact_version() {
   artifact="$1"
   name="$(basename "$artifact")"

src/bin/dosh-server.rs

@@ -1,4 +1,4 @@
-use anyhow::{Context, Result, anyhow};
+use anyhow::{Context, Result, anyhow, bail};
 use clap::{Parser, Subcommand};
 use dosh::auth::{
     build_attach_ticket, build_bootstrap, encode_bootstrap, load_or_create_server_secret, now_secs,
@@ -6,6 +6,14 @@ use dosh::auth::{
 };
 use dosh::config::{ServerConfig, expand_tilde, load_server_config};
 use dosh::crypto;
+use dosh::exec_service::{
+    EXEC_STREAM_SENTINEL, ExecResponse, FrameDecoder as ExecFrameDecoder,
+    encode_response as encode_exec_response,
+};
+use dosh::file_transfer::{
+    CHUNK_SIZE, FILE_STREAM_SENTINEL, FileEntry, FileKind, FileMeta, FileRequest, FileResponse,
+    FrameDecoder, clean_remote_path, encode_response,
+};
 use dosh::native::{
     EnvVar, ForwardingKind, ForwardingRequest, NativeAuthOk, NativeServerHello,
     derive_native_session_key, generate_native_ephemeral, host_public_key, host_public_key_line,
@@ -20,16 +28,21 @@ use dosh::protocol::{
     TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
 };
 use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
+use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
+use std::fs;
+use std::io::{Read, Seek, SeekFrom, Write};
 use std::net::{IpAddr, SocketAddr};
 use std::os::unix::fs::PermissionsExt;
 use std::os::unix::net::UnixStream as StdUnixStream;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixListener};
+use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
@@ -219,19 +232,6 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
         }
     });
 
-    let pacing_state = Arc::clone(&state);
-    let pacing_socket = Arc::clone(&socket);
-    let pacing_interval_ms = config.output_frame_interval_ms.max(1);
-    tokio::spawn(async move {
-        let mut interval = tokio::time::interval(Duration::from_millis(pacing_interval_ms));
-        loop {
-            interval.tick().await;
-            if let Err(err) = flush_paced_snapshots(&pacing_state, &pacing_socket).await {
-                eprintln!("paced snapshot error: {err:#}");
-            }
-        }
-    });
-
     let cleanup_state = Arc::clone(&state);
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(5));
@@ -387,8 +387,6 @@ struct ClientState {
     rows: u16,
     last_seen: Instant,
     pending: VecDeque<PendingFrame>,
-    last_frame_sent: Instant,
-    paced_snapshot_due: bool,
     allowed_forwardings: Vec<ForwardingRequest>,
     stream_writers: HashMap<u64, mpsc::Sender<Vec<u8>>>,
     opened_streams: HashSet<u64>,
@@ -1017,6 +1015,8 @@ async fn handle_native_user_auth(
                 allow_tcp_forwarding: locked.config.allow_tcp_forwarding,
                 allow_remote_forwarding: locked.config.allow_remote_forwarding,
                 allow_agent_forwarding: locked.config.allow_agent_forwarding,
+                allow_file_transfer: locked.config.allow_file_transfer,
+                allow_exec_command: locked.config.allow_exec_command,
                 policy_flags: Vec::new(),
                 server_version: env!("CARGO_PKG_VERSION").to_string(),
             })
@@ -1152,8 +1152,6 @@ async fn handle_native_user_auth(
                     rows,
                     last_seen: Instant::now(),
                     pending: VecDeque::new(),
-                    last_frame_sent: Instant::now() - Duration::from_secs(3600),
-                    paced_snapshot_due: false,
                     allowed_forwardings: req.auth.requested_forwardings.clone(),
                     stream_writers: HashMap::new(),
                     opened_streams: HashSet::new(),
@@ -1296,8 +1294,6 @@ async fn handle_bootstrap_attach(
                     rows: req.rows,
                     last_seen: Instant::now(),
                     pending: VecDeque::new(),
-                    last_frame_sent: Instant::now() - Duration::from_secs(3600),
-                    paced_snapshot_due: false,
                     allowed_forwardings: Vec::new(),
                     stream_writers: HashMap::new(),
                     opened_streams: HashSet::new(),
@@ -1432,8 +1428,6 @@ async fn handle_ticket_attach(
                     rows: req.rows,
                     last_seen: Instant::now(),
                     pending: VecDeque::new(),
-                    last_frame_sent: Instant::now() - Duration::from_secs(3600),
-                    paced_snapshot_due: false,
                     allowed_forwardings: Vec::new(),
                     stream_writers: HashMap::new(),
                     opened_streams: HashSet::new(),
@@ -1801,6 +1795,88 @@ async fn handle_stream_open(
         }
     }
 
+    if open.target_host == FILE_STREAM_SENTINEL {
+        let (writer_tx, writer_rx) = mpsc::channel::<Vec<u8>>(1024);
+        register_opened_stream(
+            state,
+            &session_name,
+            packet.header.conn_id,
+            open.stream_id,
+            writer_tx,
+        )?;
+        send_stream_open_ok(state, socket, packet.header.conn_id, open.stream_id).await?;
+
+        let state = Arc::clone(state);
+        let socket = Arc::clone(socket);
+        let client_id = packet.header.conn_id;
+        let stream_id = open.stream_id;
+        tokio::spawn(async move {
+            if let Err(err) = run_file_stream_service(
+                state.clone(),
+                socket.clone(),
+                client_id,
+                stream_id,
+                writer_rx,
+            )
+            .await
+            {
+                let _ = send_file_response_to_client(
+                    &state,
+                    &socket,
+                    client_id,
+                    stream_id,
+                    FileResponse::Error {
+                        message: err.to_string(),
+                    },
+                )
+                .await;
+            }
+            let _ = send_stream_close_to_client(&state, &socket, client_id, stream_id).await;
+        });
+        return Ok(());
+    }
+
+    if open.target_host == EXEC_STREAM_SENTINEL {
+        let (writer_tx, writer_rx) = mpsc::channel::<Vec<u8>>(1024);
+        register_opened_stream(
+            state,
+            &session_name,
+            packet.header.conn_id,
+            open.stream_id,
+            writer_tx,
+        )?;
+        send_stream_open_ok(state, socket, packet.header.conn_id, open.stream_id).await?;
+
+        let state = Arc::clone(state);
+        let socket = Arc::clone(socket);
+        let client_id = packet.header.conn_id;
+        let stream_id = open.stream_id;
+        tokio::spawn(async move {
+            if let Err(err) = run_exec_stream_service(
+                state.clone(),
+                socket.clone(),
+                client_id,
+                stream_id,
+                writer_rx,
+            )
+            .await
+            {
+                let _ = send_exec_response_to_client(
+                    &state,
+                    &socket,
+                    client_id,
+                    stream_id,
+                    ExecResponse::Error {
+                        message: err.to_string(),
+                    },
+                )
+                .await;
+            }
+            let _ = send_stream_close_to_client(&state, &socket, client_id, stream_id).await;
+        });
+        return Ok(());
+    }
+
     let target = format!("{}:{}", open.target_host, open.target_port);
     let stream = match TcpStream::connect((open.target_host.as_str(), open.target_port)).await {
         Ok(stream) => stream,
@@ -1817,30 +1893,13 @@ async fn handle_stream_open(
     };
     let (mut reader, mut writer) = stream.into_split();
     let (writer_tx, mut writer_rx) = mpsc::channel::<Vec<u8>>(1024);
-    {
+    register_opened_stream(
-        let mut locked = state.lock().expect("server state poisoned");
+        state,
-        let session = locked
+        &session_name,
-            .sessions
+        packet.header.conn_id,
-            .get_mut(&session_name)
+        open.stream_id,
-            .ok_or_else(|| anyhow!("unknown session"))?;
+        writer_tx,
-        let client = session
+    )?;
-            .clients
-            .get_mut(&packet.header.conn_id)
-            .ok_or_else(|| anyhow!("unknown client"))?;
-        client.stream_writers.insert(open.stream_id, writer_tx);
-        client.opened_streams.insert(open.stream_id);
-        client
-            .stream_send_credit
-            .insert(open.stream_id, STREAM_INITIAL_WINDOW);
-        client
-            .stream_next_send_offset
-            .entry(open.stream_id)
-            .or_insert(0);
-        client
-            .stream_next_recv_offset
-            .entry(open.stream_id)
-            .or_insert(0);
-    }
 
     tokio::spawn(async move {
         while let Some(bytes) = writer_rx.recv().await {
@@ -1884,6 +1943,32 @@ async fn handle_stream_open(
     Ok(())
 }
 
+fn register_opened_stream(
+    state: &Arc<Mutex<ServerState>>,
+    session_name: &str,
+    client_id: [u8; 16],
+    stream_id: u64,
+    writer_tx: mpsc::Sender<Vec<u8>>,
+) -> Result<()> {
+    let mut locked = state.lock().expect("server state poisoned");
+    let session = locked
+        .sessions
+        .get_mut(session_name)
+        .ok_or_else(|| anyhow!("unknown session"))?;
+    let client = session
+        .clients
+        .get_mut(&client_id)
+        .ok_or_else(|| anyhow!("unknown client"))?;
+    client.stream_writers.insert(stream_id, writer_tx);
+    client.opened_streams.insert(stream_id);
+    client
+        .stream_send_credit
+        .insert(stream_id, STREAM_INITIAL_WINDOW);
+    client.stream_next_send_offset.entry(stream_id).or_insert(0);
+    client.stream_next_recv_offset.entry(stream_id).or_insert(0);
+    Ok(())
+}
+
 async fn handle_stream_data(
     state: &Arc<Mutex<ServerState>>,
     socket: &Arc<UdpSocket>,
@@ -2320,6 +2405,26 @@ fn stream_open_allowed(
     client: &ClientState,
     open: &StreamOpen,
 ) -> Result<()> {
+    if open.target_host == FILE_STREAM_SENTINEL {
+        anyhow::ensure!(config.allow_file_transfer, "file transfer disabled");
+        let allowed = client.allowed_forwardings.iter().any(|forward| {
+            forward.kind == ForwardingKind::Local
+                && forward.target_host.as_deref() == Some(FILE_STREAM_SENTINEL)
+                && forward.target_port == Some(0)
+        });
+        anyhow::ensure!(allowed, "file transfer was not requested during auth");
+        return Ok(());
+    }
+    if open.target_host == EXEC_STREAM_SENTINEL {
+        anyhow::ensure!(config.allow_exec_command, "exec command disabled");
+        let allowed = client.allowed_forwardings.iter().any(|forward| {
+            forward.kind == ForwardingKind::Local
+                && forward.target_host.as_deref() == Some(EXEC_STREAM_SENTINEL)
+                && forward.target_port == Some(0)
+        });
+        anyhow::ensure!(allowed, "exec command was not requested during auth");
+        return Ok(());
+    }
     anyhow::ensure!(config.allow_tcp_forwarding, "TCP forwarding disabled");
     let allowed = client
         .allowed_forwardings
@@ -2343,6 +2448,512 @@ fn stream_open_allowed(
     Ok(())
 }
 
+struct UploadState {
+    final_path: PathBuf,
+    temp_path: Option<PathBuf>,
+    file: fs::File,
+    hasher: Sha256,
+    written: u64,
+    expected_size: u64,
+    modified_secs: Option<u64>,
+    atomic: bool,
+}
+
+async fn run_file_stream_service(
+    state: Arc<Mutex<ServerState>>,
+    socket: Arc<UdpSocket>,
+    client_id: [u8; 16],
+    stream_id: u64,
+    mut rx: mpsc::Receiver<Vec<u8>>,
+) -> Result<()> {
+    let home = std::env::var_os("HOME")
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from("."));
+    let mut decoder = FrameDecoder::default();
+    let mut upload: Option<UploadState> = None;
+    while let Some(bytes) = rx.recv().await {
+        for frame in decoder.push(&bytes)? {
+            let request = match dosh::file_transfer::decode_request(&frame) {
+                Ok(request) => request,
+                Err(err) => {
+                    send_file_response_to_client(
+                        &state,
+                        &socket,
+                        client_id,
+                        stream_id,
+                        FileResponse::Error {
+                            message: err.to_string(),
+                        },
+                    )
+                    .await?;
+                    continue;
+                }
+            };
+            if let Err(err) = handle_file_request(
+                &state,
+                &socket,
+                client_id,
+                stream_id,
+                &home,
+                &mut upload,
+                request,
+            )
+            .await
+            {
+                send_file_response_to_client(
+                    &state,
+                    &socket,
+                    client_id,
+                    stream_id,
+                    FileResponse::Error {
+                        message: err.to_string(),
+                    },
+                )
+                .await?;
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn handle_file_request(
+    state: &Arc<Mutex<ServerState>>,
+    socket: &Arc<UdpSocket>,
+    client_id: [u8; 16],
+    stream_id: u64,
+    home: &Path,
+    upload: &mut Option<UploadState>,
+    request: FileRequest,
+) -> Result<()> {
+    match request {
+        FileRequest::Stat { path } => {
+            let path = clean_remote_path(&path, home)?;
+            let meta = file_meta(&path, home)?;
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::Stat { meta },
+            )
+            .await
+        }
+        FileRequest::List { path } => {
+            let path = clean_remote_path(&path, home)?;
+            let mut entries = Vec::new();
+            for entry in fs::read_dir(&path).with_context(|| format!("list {}", path.display()))? {
+                let entry = entry?;
+                let name = entry.file_name().to_string_lossy().to_string();
+                entries.push(FileEntry {
+                    name,
+                    meta: file_meta(&entry.path(), home)?,
+                });
+            }
+            entries.sort_by(|a, b| a.name.cmp(&b.name));
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::List { entries },
+            )
+            .await
+        }
+        FileRequest::Mkdir { path, mode } => {
+            let path = clean_remote_path(&path, home)?;
+            fs::create_dir_all(&path).with_context(|| format!("create {}", path.display()))?;
+            if let Some(mode) = mode {
+                set_mode(&path, mode)?;
+            }
+            send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
+                .await
+        }
+        FileRequest::Remove { path, recursive } => {
+            let path = clean_remote_path(&path, home)?;
+            let metadata =
+                fs::symlink_metadata(&path).with_context(|| format!("stat {}", path.display()))?;
+            if metadata.is_dir() {
+                anyhow::ensure!(
+                    recursive,
+                    "{} is a directory; pass -r to remove directories",
+                    path.display()
+                );
+                fs::remove_dir_all(&path).with_context(|| format!("remove {}", path.display()))?;
+            } else {
+                fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
+            }
+            send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
+                .await
+        }
+        FileRequest::PutStart {
+            path,
+            size,
+            mode,
+            modified_secs,
+            overwrite,
+            resume,
+        } => {
+            if upload.is_some() {
+                bail!("upload already in progress");
+            }
+            let final_path = clean_remote_path(&path, home)?;
+            if let Some(parent) = final_path.parent() {
+                fs::create_dir_all(parent)
+                    .with_context(|| format!("create {}", parent.display()))?;
+            }
+            if final_path.exists() && !overwrite && !resume {
+                bail!("destination exists: {}", final_path.display());
+            }
+            let mut hasher = Sha256::new();
+            let (file, temp_path, written, atomic) = if resume && final_path.exists() {
+                let mut file = fs::OpenOptions::new()
+                    .read(true)
+                    .append(true)
+                    .open(&final_path)
+                    .with_context(|| format!("open {}", final_path.display()))?;
+                let written = file.metadata()?.len().min(size);
+                if written > 0 {
+                    let mut prefix = fs::File::open(&final_path)?;
+                    hash_exact_prefix(&mut prefix, written, &mut hasher)?;
+                }
+                file.seek(SeekFrom::Start(written))?;
+                (file, None, written, false)
+            } else {
+                let temp_path = upload_temp_path(&final_path, stream_id);
+                let file = fs::OpenOptions::new()
+                    .create(true)
+                    .truncate(true)
+                    .write(true)
+                    .open(&temp_path)
+                    .with_context(|| format!("create {}", temp_path.display()))?;
+                (file, Some(temp_path), 0, true)
+            };
+            *upload = Some(UploadState {
+                final_path,
+                temp_path,
+                file,
+                hasher,
+                written,
+                expected_size: size,
+                modified_secs,
+                atomic,
+            });
+            if let Some(mode) = mode {
+                let target = upload
+                    .as_ref()
+                    .and_then(|state| state.temp_path.as_ref())
+                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
+                    .clone();
+                set_mode(&target, mode)?;
+            }
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::Resume { offset: written },
+            )
+            .await
+        }
+        FileRequest::PutChunk { offset, bytes } => {
+            let Some(active) = upload.as_mut() else {
+                bail!("no upload in progress");
+            };
+            anyhow::ensure!(
+                offset == active.written,
+                "upload offset mismatch: got {offset}, expected {}",
+                active.written
+            );
+            active.file.write_all(&bytes)?;
+            active.hasher.update(&bytes);
+            active.written = active.written.saturating_add(bytes.len() as u64);
+            Ok(())
+        }
+        FileRequest::PutFinish { sha256 } => {
+            let Some(mut active) = upload.take() else {
+                bail!("no upload in progress");
+            };
+            active.file.flush()?;
+            active.file.sync_all()?;
+            anyhow::ensure!(
+                active.written == active.expected_size,
+                "upload size mismatch: got {}, expected {}",
+                active.written,
+                active.expected_size
+            );
+            let digest: [u8; 32] = active.hasher.finalize().into();
+            if digest != sha256 {
+                if let Some(temp_path) = &active.temp_path {
+                    let _ = fs::remove_file(temp_path);
+                }
+                bail!("upload checksum mismatch");
+            }
+            drop(active.file);
+            if active.atomic
+                && let Some(temp_path) = active.temp_path.take()
+            {
+                fs::rename(&temp_path, &active.final_path).with_context(|| {
+                    format!(
+                        "rename {} to {}",
+                        temp_path.display(),
+                        active.final_path.display()
+                    )
+                })?;
+            }
+            if let Some(modified_secs) = active.modified_secs {
+                filetime::set_file_mtime(
+                    &active.final_path,
+                    filetime::FileTime::from_unix_time(modified_secs as i64, 0),
+                )
+                .with_context(|| format!("set mtime {}", active.final_path.display()))?;
+            }
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::Done {
+                    sha256: digest,
+                    bytes: active.written,
+                },
+            )
+            .await
+        }
+        FileRequest::Get { path, offset } => {
+            let path = clean_remote_path(&path, home)?;
+            let meta = file_meta(&path, home)?;
+            anyhow::ensure!(
+                meta.kind == FileKind::File,
+                "remote path is not a file: {}",
+                meta.path
+            );
+            let mut file =
+                fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
+            file.seek(SeekFrom::Start(offset))?;
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::Start {
+                    meta: meta.clone(),
+                    offset,
+                },
+            )
+            .await?;
+            let mut hasher = Sha256::new();
+            if offset > 0 {
+                let mut prefix = fs::File::open(&path)?;
+                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
+            }
+            let mut sent = offset;
+            let mut buf = vec![0u8; CHUNK_SIZE];
+            loop {
+                let n = file.read(&mut buf)?;
+                if n == 0 {
+                    break;
+                }
+                hasher.update(&buf[..n]);
+                send_file_response_to_client(
+                    state,
+                    socket,
+                    client_id,
+                    stream_id,
+                    FileResponse::Chunk {
+                        offset: sent,
+                        bytes: buf[..n].to_vec(),
+                    },
+                )
+                .await?;
+                sent = sent.saturating_add(n as u64);
+            }
+            let digest: [u8; 32] = hasher.finalize().into();
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::Done {
+                    sha256: digest,
+                    bytes: sent,
+                },
+            )
+            .await
+        }
+    }
+}
+
+async fn send_file_response_to_client(
+    state: &Arc<Mutex<ServerState>>,
+    socket: &Arc<UdpSocket>,
+    client_id: [u8; 16],
+    stream_id: u64,
+    response: FileResponse,
+) -> Result<()> {
+    send_stream_data_to_client(
+        state,
+        socket,
+        client_id,
+        stream_id,
+        encode_response(&response)?,
+    )
+    .await
+}
+
+async fn run_exec_stream_service(
+    state: Arc<Mutex<ServerState>>,
+    socket: Arc<UdpSocket>,
+    client_id: [u8; 16],
+    stream_id: u64,
+    mut rx: mpsc::Receiver<Vec<u8>>,
+) -> Result<()> {
+    let mut decoder = ExecFrameDecoder::default();
+    while let Some(bytes) = rx.recv().await {
+        for frame in decoder.push(&bytes)? {
+            let request = match dosh::exec_service::decode_request(&frame) {
+                Ok(request) => request,
+                Err(err) => {
+                    send_exec_response_to_client(
+                        &state,
+                        &socket,
+                        client_id,
+                        stream_id,
+                        ExecResponse::Error {
+                            message: err.to_string(),
+                        },
+                    )
+                    .await?;
+                    continue;
+                }
+            };
+            let output = TokioCommand::new("sh")
+                .arg("-lc")
+                .arg(&request.command)
+                .stdin(Stdio::null())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .output()
+                .await
+                .with_context(|| format!("run command {:?}", request.command))?;
+            for chunk in output.stdout.chunks(CHUNK_SIZE) {
+                send_exec_response_to_client(
+                    &state,
+                    &socket,
+                    client_id,
+                    stream_id,
+                    ExecResponse::Stdout(chunk.to_vec()),
+                )
+                .await?;
+            }
+            for chunk in output.stderr.chunks(CHUNK_SIZE) {
+                send_exec_response_to_client(
+                    &state,
+                    &socket,
+                    client_id,
+                    stream_id,
+                    ExecResponse::Stderr(chunk.to_vec()),
+                )
+                .await?;
+            }
+            send_exec_response_to_client(
+                &state,
+                &socket,
+                client_id,
+                stream_id,
+                ExecResponse::Exit {
+                    code: output.status.code(),
+                },
+            )
+            .await?;
+        }
+    }
+    Ok(())
+}
+
+async fn send_exec_response_to_client(
+    state: &Arc<Mutex<ServerState>>,
+    socket: &Arc<UdpSocket>,
+    client_id: [u8; 16],
+    stream_id: u64,
+    response: ExecResponse,
+) -> Result<()> {
+    send_stream_data_to_client(
+        state,
+        socket,
+        client_id,
+        stream_id,
+        encode_exec_response(&response)?,
+    )
+    .await
+}
+
+fn file_meta(path: &Path, home: &Path) -> Result<FileMeta> {
+    let metadata =
+        fs::symlink_metadata(path).with_context(|| format!("stat {}", path.display()))?;
+    let file_type = metadata.file_type();
+    let kind = if file_type.is_file() {
+        FileKind::File
+    } else if file_type.is_dir() {
+        FileKind::Directory
+    } else if file_type.is_symlink() {
+        FileKind::Symlink
+    } else {
+        FileKind::Other
+    };
+    let modified_secs = metadata
+        .modified()
+        .ok()
+        .and_then(|mtime| mtime.duration_since(std::time::UNIX_EPOCH).ok())
+        .map(|duration| duration.as_secs());
+    let display_path = path
+        .strip_prefix(home)
+        .ok()
+        .and_then(|relative| relative.to_str())
+        .filter(|relative| !relative.is_empty())
+        .map(|relative| format!("~/{relative}"))
+        .unwrap_or_else(|| path.display().to_string());
+    Ok(FileMeta {
+        path: display_path,
+        kind,
+        len: metadata.len(),
+        mode: Some(metadata.permissions().mode() & 0o7777),
+        modified_secs,
+    })
+}
+
+fn set_mode(path: &Path, mode: u32) -> Result<()> {
+    let mut permissions = fs::metadata(path)?.permissions();
+    permissions.set_mode(mode & 0o7777);
+    fs::set_permissions(path, permissions).with_context(|| format!("chmod {}", path.display()))
+}
+
+fn upload_temp_path(final_path: &Path, stream_id: u64) -> PathBuf {
+    let name = final_path
+        .file_name()
+        .and_then(|name| name.to_str())
+        .unwrap_or("upload");
+    final_path.with_file_name(format!(
+        ".{name}.dosh-upload-{}-{stream_id}",
+        std::process::id()
+    ))
+}
+
+fn hash_exact_prefix(file: &mut fs::File, bytes: u64, hasher: &mut Sha256) -> Result<()> {
+    file.seek(SeekFrom::Start(0))?;
+    let mut remaining = bytes;
+    let mut buf = vec![0u8; CHUNK_SIZE];
+    while remaining > 0 {
+        let want = remaining.min(buf.len() as u64) as usize;
+        let n = file.read(&mut buf[..want])?;
+        if n == 0 {
+            bail!("file ended while hashing resume prefix");
+        }
+        hasher.update(&buf[..n]);
+        remaining -= n as u64;
+    }
+    Ok(())
+}
+
 async fn send_stream_open_ok(
     state: &Arc<Mutex<ServerState>>,
     socket: &Arc<UdpSocket>,
@@ -2726,8 +3337,6 @@ async fn broadcast_output(
         let mut locked = state.lock().expect("server state poisoned");
         let scrollback = locked.config.scrollback;
         let retransmit_window = locked.config.retransmit_window;
-        let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
-        let now = Instant::now();
         let session = locked
             .sessions
             .get_mut(&output.session)
@@ -2762,15 +3371,6 @@ async fn broadcast_output(
         }
         let mut sends = Vec::new();
         for (client_id, client) in session.clients.iter_mut() {
-            let terminal_control =
-                output.bytes.contains(&0x1b) || session.parser.screen().alternate_screen();
-            if !terminal_control
-                && !frame_interval.is_zero()
-                && now.duration_since(client.last_frame_sent) < frame_interval
-            {
-                client.paced_snapshot_due = true;
-                continue;
-            }
             client.send_seq += 1;
             // Count traffic toward the packet-count rekey trigger (spec §11).
             client.epoch_packets = client.epoch_packets.saturating_add(1);
@@ -2809,8 +3409,6 @@ async fn broadcast_output(
                 last_sent: Instant::now(),
                 attempts: 0,
             });
-            client.last_frame_sent = now;
-            client.paced_snapshot_due = false;
             sends.push((client.endpoint, packet));
         }
         sends
@@ -2994,76 +3592,6 @@ async fn retransmit_pending(
     Ok(())
 }
 
-async fn flush_paced_snapshots(
-    state: &Arc<Mutex<ServerState>>,
-    socket: &Arc<UdpSocket>,
-) -> Result<()> {
-    let sends = {
-        let mut locked = state.lock().expect("server state poisoned");
-        let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
-        if frame_interval.is_zero() {
-            return Ok(());
-        }
-        let retransmit_window = locked.config.retransmit_window;
-        let now = Instant::now();
-        let mut sends = Vec::new();
-        for (session_name, session) in locked.sessions.iter_mut() {
-            let due = session.clients.values().any(|client| {
-                client.paced_snapshot_due
-                    && now.duration_since(client.last_frame_sent) >= frame_interval
-            });
-            if !due {
-                continue;
-            }
-            let snapshot = screen_snapshot(session.parser.screen());
-            let output_seq = session.output_seq;
-            for (client_id, client) in session.clients.iter_mut() {
-                if !client.paced_snapshot_due
-                    || now.duration_since(client.last_frame_sent) < frame_interval
-                {
-                    continue;
-                }
-                client.send_seq += 1;
-                client.epoch_packets = client.epoch_packets.saturating_add(1);
-                let frame = Frame {
-                    session: session_name.clone(),
-                    output_seq,
-                    bytes: snapshot.clone(),
-                    snapshot: true,
-                    closed: false,
-                };
-                let body = protocol::to_body(&frame)?;
-                let packet = protocol::encode_encrypted(
-                    PacketKind::Frame,
-                    *client_id,
-                    client.send_seq,
-                    client.last_acked,
-                    &client.session_key,
-                    SERVER_TO_CLIENT,
-                    &body,
-                )?;
-                while client.pending.len() >= retransmit_window {
-                    client.pending.pop_front();
-                }
-                client.pending.push_back(PendingFrame {
-                    output_seq,
-                    packet: packet.clone(),
-                    last_sent: now,
-                    attempts: 0,
-                });
-                client.last_frame_sent = now;
-                client.paced_snapshot_due = false;
-                sends.push((client.endpoint, packet));
-            }
-        }
-        sends
-    };
-    for (endpoint, packet) in sends {
-        socket.send_to(&packet, endpoint).await?;
-    }
-    Ok(())
-}
-
 /// How long the previous epoch's key is retained after a rekey so in-flight
 /// pre-rekey packets still decrypt instead of being dropped as fatal.
 const REKEY_PREVIOUS_KEY_GRACE_SECS: u64 = 5;
@@ -3445,8 +3973,6 @@ mod tests {
             rows: 24,
             last_seen: Instant::now(),
             pending: VecDeque::new(),
-            last_frame_sent: Instant::now() - Duration::from_secs(3600),
-            paced_snapshot_due: false,
             allowed_forwardings: Vec::new(),
             stream_writers: HashMap::new(),
             opened_streams: HashSet::new(),
@@ -3763,8 +4289,6 @@ mod tests {
                         rows: 24,
                         last_seen: Instant::now(),
                         pending: VecDeque::new(),
-                        last_frame_sent: Instant::now() - Duration::from_secs(3600),
-                        paced_snapshot_due: false,
                         allowed_forwardings: Vec::new(),
                         stream_writers: HashMap::new(),
                         opened_streams: HashSet::new(),
@@ -3859,8 +4383,6 @@ mod tests {
                         rows: 24,
                         last_seen: Instant::now(),
                         pending: VecDeque::new(),
-                        last_frame_sent: Instant::now() - Duration::from_secs(3600),
-                        paced_snapshot_due: false,
                         allowed_forwardings: Vec::new(),
                         stream_writers: HashMap::new(),
                         opened_streams: HashSet::from([42]),
@@ -3930,7 +4452,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn output_pacing_coalesces_fast_frames_into_snapshot() {
+    async fn live_terminal_output_is_never_replaced_by_paced_snapshots() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
         let config = ServerConfig {
             output_frame_interval_ms: 1000,
@@ -3982,25 +4504,6 @@ mod tests {
         let frame: Frame = protocol::from_body(&plain).unwrap();
         assert!(!frame.snapshot);
         assert_eq!(frame.bytes, b"first");
-        assert!(
-            tokio::time::timeout(Duration::from_millis(30), receiver.recv_from(&mut buf))
-                .await
-                .is_err(),
-            "second frame should be coalesced"
-        );
-
-        {
-            let mut locked = state.lock().unwrap();
-            let client = locked
-                .sessions
-                .get_mut("test")
-                .unwrap()
-                .clients
-                .get_mut(&client_id)
-                .unwrap();
-            client.last_frame_sent = Instant::now() - Duration::from_secs(2);
-        }
-        flush_paced_snapshots(&state, &sender).await.unwrap();
         let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
             .await
             .unwrap()
@@ -4008,8 +4511,9 @@ mod tests {
         let packet = protocol::decode(&buf[..n]).unwrap();
         let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
         let frame: Frame = protocol::from_body(&plain).unwrap();
-        assert!(frame.snapshot);
+        assert!(!frame.snapshot);
         assert_eq!(frame.output_seq, 2);
+        assert_eq!(frame.bytes, b"second");
     }
 
     #[test]

src/client.rs

@@ -0,0 +1,442 @@
+use crate::config::{
+    ClientConfig, HostConfig, expand_tilde, load_client_config, load_hosts_config,
+};
+use crate::crypto;
+use crate::native::{
+    self, EnvVar, ForwardingKind, ForwardingRequest, KnownHostStatus, NativeClientHello,
+    derive_native_session_key, generate_native_ephemeral, sign_user_auth_with_private_key,
+    supported_user_key_algorithms, trust_host, verify_known_host, verify_server_hello,
+};
+use crate::protocol::{
+    self, AttachReject, CLIENT_TO_SERVER, NativeAuthOkBody, NativeClientHelloBody,
+    NativeServerHelloBody, NativeUserAuthBody, PacketKind, SERVER_TO_CLIENT,
+};
+use crate::ssh_agent;
+use crate::transport::{DoshTransport, SessionRole, SessionTransportConfig, TransportConfig};
+use anyhow::{Context, Result, anyhow, bail};
+use std::net::{SocketAddr, ToSocketAddrs};
+use std::path::PathBuf;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use tokio::net::UdpSocket;
+
+#[derive(Debug, Clone)]
+pub struct DoshClient {
+    config: ClientConfig,
+    hosts: crate::config::HostsConfig,
+}
+
+impl DoshClient {
+    pub fn load() -> Result<Self> {
+        Ok(Self {
+            config: load_client_config(None)?,
+            hosts: load_hosts_config(None)?,
+        })
+    }
+
+    pub fn load_from_paths(
+        client_config: Option<PathBuf>,
+        hosts_config: Option<PathBuf>,
+    ) -> Result<Self> {
+        Ok(Self {
+            config: load_client_config(client_config)?,
+            hosts: load_hosts_config(hosts_config)?,
+        })
+    }
+
+    pub fn with_config(config: ClientConfig, hosts: crate::config::HostsConfig) -> Self {
+        Self { config, hosts }
+    }
+
+    pub fn connect(&self, host: impl Into<String>) -> DoshClientBuilder {
+        DoshClientBuilder::new(self.clone(), host.into())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct DoshClientBuilder {
+    client: DoshClient,
+    host: String,
+    services: Vec<String>,
+    identity_files: Vec<PathBuf>,
+    session: Option<String>,
+    user: Option<String>,
+    udp_host: Option<String>,
+    udp_port: Option<u16>,
+    trust_on_first_use: Option<bool>,
+    use_ssh_agent: Option<bool>,
+    timeout: Option<Duration>,
+    env: Vec<EnvVar>,
+}
+
+impl DoshClientBuilder {
+    pub fn new(client: DoshClient, host: String) -> Self {
+        Self {
+            client,
+            host,
+            services: Vec::new(),
+            identity_files: Vec::new(),
+            session: None,
+            user: None,
+            udp_host: None,
+            udp_port: None,
+            trust_on_first_use: None,
+            use_ssh_agent: None,
+            timeout: None,
+            env: Vec::new(),
+        }
+    }
+
+    pub fn service(mut self, name: impl Into<String>) -> Self {
+        self.services.push(name.into());
+        self
+    }
+
+    pub fn services(mut self, names: impl IntoIterator<Item = impl Into<String>>) -> Self {
+        self.services.extend(names.into_iter().map(Into::into));
+        self
+    }
+
+    pub fn identity_file(mut self, path: impl Into<PathBuf>) -> Self {
+        self.identity_files.push(path.into());
+        self
+    }
+
+    pub fn session(mut self, session: impl Into<String>) -> Self {
+        self.session = Some(session.into());
+        self
+    }
+
+    pub fn user(mut self, user: impl Into<String>) -> Self {
+        self.user = Some(user.into());
+        self
+    }
+
+    pub fn udp_host(mut self, host: impl Into<String>) -> Self {
+        self.udp_host = Some(host.into());
+        self
+    }
+
+    pub fn udp_port(mut self, port: u16) -> Self {
+        self.udp_port = Some(port);
+        self
+    }
+
+    pub fn trust_on_first_use(mut self, trust: bool) -> Self {
+        self.trust_on_first_use = Some(trust);
+        self
+    }
+
+    pub fn use_ssh_agent(mut self, value: bool) -> Self {
+        self.use_ssh_agent = Some(value);
+        self
+    }
+
+    pub fn timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = Some(timeout);
+        self
+    }
+
+    pub fn env(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
+        self.env.push(EnvVar {
+            name: name.into(),
+            value: value.into(),
+        });
+        self
+    }
+
+    pub async fn connect(self) -> Result<ConnectedDoshClient> {
+        let host_config = self
+            .client
+            .hosts
+            .hosts
+            .get(&self.host)
+            .cloned()
+            .unwrap_or_default();
+        let raw_server = host_config.ssh.clone().unwrap_or_else(|| self.host.clone());
+        let requested_user = self
+            .user
+            .clone()
+            .or_else(|| host_config.user.clone())
+            .or_else(|| user_from_destination(&raw_server))
+            .or_else(|| std::env::var("USER").ok())
+            .unwrap_or_else(|| "unknown".to_string());
+        let udp_host = self
+            .udp_host
+            .clone()
+            .or_else(|| host_config.dosh_host.clone())
+            .or_else(|| self.client.config.dosh_host.clone())
+            .unwrap_or_else(|| destination_host(&raw_server));
+        let udp_port = self
+            .udp_port
+            .or(host_config.port)
+            .unwrap_or(self.client.config.dosh_port);
+        let peer_addr = resolve_addr(&udp_host, udp_port)?;
+        let timeout = self.timeout.unwrap_or_else(|| {
+            Duration::from_millis(self.client.config.native_auth_timeout_ms.max(1))
+        });
+        let session = self.session.unwrap_or_else(default_sdk_session);
+        let requested_forwardings = self
+            .services
+            .iter()
+            .map(|service| {
+                Ok(ForwardingRequest {
+                    kind: ForwardingKind::Local,
+                    bind_host: None,
+                    listen_port: 0,
+                    target_host: Some(crate::transport::service_target(service)?),
+                    target_port: Some(0),
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        let (client_secret, client_public) = generate_native_ephemeral();
+        let hello = NativeClientHello {
+            protocol_version: native::NATIVE_PROTOCOL_VERSION,
+            client_random: crypto::random_32(),
+            client_ephemeral_public: client_public,
+            requested_host: self.host.clone(),
+            requested_user,
+            requested_session: session.clone(),
+            requested_mode: "forward-only".to_string(),
+            terminal_size: (80, 24),
+            supported_aead: vec!["chacha20poly1305".to_string()],
+            supported_user_key_algorithms: supported_user_key_algorithms(),
+            cached_host_key_fingerprint: None,
+            attach_ticket_envelope: None,
+            requested_env: self.env,
+        };
+        let packet = protocol::encode_plain(
+            PacketKind::NativeClientHello,
+            [0u8; 16],
+            1,
+            0,
+            &protocol::to_body(&NativeClientHelloBody {
+                hello: hello.clone(),
+            })?,
+        )?;
+        socket.send_to(&packet, peer_addr).await?;
+
+        let mut buf = vec![0u8; 65535];
+        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        let packet = protocol::decode(&buf[..n])?;
+        if packet.header.kind != PacketKind::NativeServerHello {
+            if packet.header.kind == PacketKind::AttachReject {
+                let reject: AttachReject = protocol::from_body(&packet.body)?;
+                bail!("native auth rejected: {}", reject.reason);
+            }
+            bail!("native auth received unexpected server response");
+        }
+        let server_hello: NativeServerHelloBody = protocol::from_body(&packet.body)?;
+        verify_server_hello(&hello, &server_hello.hello)?;
+        verify_or_trust_host(
+            &self.client.config,
+            &self.host,
+            &server_hello.hello.host_key,
+            self.trust_on_first_use,
+        )?;
+
+        let session_key = derive_native_session_key(
+            &client_secret,
+            server_hello.hello.server_ephemeral_public,
+            &hello,
+            &server_hello.hello,
+        )?;
+        let auth = sign_auth(
+            &self.client.config,
+            &host_config,
+            &hello,
+            &server_hello.hello,
+            requested_forwardings,
+            self.identity_files,
+            self.use_ssh_agent,
+        )?;
+        let mut pending_id = [0u8; 16];
+        pending_id.copy_from_slice(&server_hello.hello.auth_challenge[..16]);
+        let auth_packet = protocol::encode_encrypted(
+            PacketKind::NativeUserAuth,
+            pending_id,
+            2,
+            1,
+            &session_key,
+            CLIENT_TO_SERVER,
+            &protocol::to_body(&NativeUserAuthBody { auth })?,
+        )?;
+        socket.send_to(&auth_packet, peer_addr).await?;
+        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        let packet = protocol::decode(&buf[..n])?;
+        if packet.header.kind != PacketKind::NativeAuthOk {
+            if packet.header.kind == PacketKind::AttachReject {
+                let reject: AttachReject = protocol::from_body(&packet.body)?;
+                bail!("native auth rejected: {}", reject.reason);
+            }
+            bail!("native auth received unexpected auth response");
+        }
+        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT)?;
+        let ok: NativeAuthOkBody = protocol::from_body(&plain)?;
+        let transport = DoshTransport::new_owned(
+            socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id: ok.ok.client_id,
+                session_key: ok.ok.session_key,
+                peer_addr,
+                initial_send_seq: 2,
+                initial_ack: ok.ok.initial_seq,
+                stream: TransportConfig::default(),
+            },
+        );
+        Ok(ConnectedDoshClient {
+            host: self.host,
+            session: ok.ok.session,
+            transport,
+        })
+    }
+}
+
+pub struct ConnectedDoshClient {
+    pub host: String,
+    pub session: String,
+    pub transport: DoshTransport,
+}
+
+impl ConnectedDoshClient {
+    pub fn into_transport(self) -> DoshTransport {
+        self.transport
+    }
+}
+
+fn verify_or_trust_host(
+    config: &ClientConfig,
+    host: &str,
+    host_key: &native::HostPublicKey,
+    trust_override: Option<bool>,
+) -> Result<()> {
+    let known_hosts = expand_tilde(&config.known_hosts);
+    match verify_known_host(&known_hosts, host, host_key)? {
+        KnownHostStatus::Trusted => Ok(()),
+        KnownHostStatus::Unknown if trust_override.unwrap_or(config.trust_on_first_use) => {
+            trust_host(&known_hosts, host, host_key, "sdk-tofu", false)?;
+            Ok(())
+        }
+        KnownHostStatus::Unknown => Err(anyhow!(
+            "Dosh host key for {host} is not trusted; run `dosh trust {host}` first or enable trust_on_first_use"
+        )),
+        KnownHostStatus::Mismatch { expected, actual } => Err(anyhow!(
+            "Dosh host key mismatch for {host}: expected {expected}, got {actual}"
+        )),
+    }
+}
+
+fn sign_auth(
+    config: &ClientConfig,
+    host_config: &HostConfig,
+    hello: &NativeClientHello,
+    server_hello: &native::NativeServerHello,
+    requested_forwardings: Vec<ForwardingRequest>,
+    explicit_identity_files: Vec<PathBuf>,
+    use_ssh_agent: Option<bool>,
+) -> Result<native::NativeUserAuth> {
+    let use_agent = use_ssh_agent.unwrap_or(config.use_ssh_agent);
+    let mut errors = Vec::new();
+    if use_agent {
+        match ssh_agent::sign_user_auth_with_agent(
+            hello,
+            server_hello,
+            requested_forwardings.clone(),
+        ) {
+            Ok(auth) => return Ok(auth),
+            Err(err) => errors.push(format!("ssh-agent: {err:#}")),
+        }
+    }
+
+    let mut paths = explicit_identity_files;
+    if paths.is_empty() {
+        paths.extend(config.identity_files.iter().map(|path| expand_tilde(path)));
+    }
+    if paths.is_empty() {
+        paths.extend(default_identity_paths());
+    }
+    for path in paths {
+        match native::load_native_identity(&path).and_then(|identity| {
+            sign_user_auth_with_private_key(
+                &identity,
+                hello,
+                server_hello,
+                requested_forwardings.clone(),
+            )
+        }) {
+            Ok(auth) => return Ok(auth),
+            Err(err) => errors.push(format!("{}: {err:#}", path.display())),
+        }
+    }
+    let _ = host_config;
+    Err(anyhow!(
+        "native auth found no usable identity: {}",
+        errors.join("; ")
+    ))
+}
+
+fn default_identity_paths() -> Vec<PathBuf> {
+    ["~/.ssh/id_ed25519", "~/.ssh/id_ecdsa", "~/.ssh/id_rsa"]
+        .into_iter()
+        .map(expand_tilde)
+        .collect()
+}
+
+fn resolve_addr(host: &str, port: u16) -> Result<SocketAddr> {
+    (host, port)
+        .to_socket_addrs()
+        .with_context(|| format!("resolve UDP target {host}:{port}"))?
+        .next()
+        .ok_or_else(|| anyhow!("no UDP address resolved for {host}:{port}"))
+}
+
+fn destination_host(destination: &str) -> String {
+    destination
+        .rsplit('@')
+        .next()
+        .unwrap_or(destination)
+        .split(':')
+        .next()
+        .unwrap_or(destination)
+        .to_string()
+}
+
+fn user_from_destination(destination: &str) -> Option<String> {
+    destination
+        .rsplit_once('@')
+        .map(|(user, _)| user.to_string())
+        .filter(|user| !user.is_empty())
+}
+
+fn default_sdk_session() -> String {
+    let millis = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis();
+    format!("sdk-{millis}-{}", std::process::id())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_user_and_host_from_destination() {
+        assert_eq!(
+            user_from_destination("palav@example.com").as_deref(),
+            Some("palav")
+        );
+        assert_eq!(destination_host("palav@example.com"), "example.com");
+        assert_eq!(destination_host("example.com:2222"), "example.com");
+    }
+
+    #[test]
+    fn default_identity_paths_are_expanded() {
+        assert!(
+            default_identity_paths()
+                .iter()
+                .any(|path| path.ends_with(".ssh/id_ed25519"))
+        );
+    }
+}

src/config.rs

@@ -46,6 +46,10 @@ pub struct ServerConfig {
     pub allow_remote_non_loopback_bind: bool,
     #[serde(default)]
     pub allow_agent_forwarding: bool,
+    #[serde(default = "default_true")]
+    pub allow_file_transfer: bool,
+    #[serde(default = "default_true")]
+    pub allow_exec_command: bool,
     #[serde(default = "default_accept_env")]
     pub accept_env: Vec<String>,
     /// Run terminal sessions under per-session holder processes so shells
@@ -82,6 +86,8 @@ impl Default for ServerConfig {
             allow_remote_forwarding: false,
             allow_remote_non_loopback_bind: false,
             allow_agent_forwarding: false,
+            allow_file_transfer: true,
+            allow_exec_command: true,
             accept_env: default_accept_env(),
             persist_sessions: true,
         }

src/exec_service.rs

@@ -0,0 +1,71 @@
+use anyhow::{Context, Result, bail};
+use serde::{Deserialize, Serialize};
+
+pub const EXEC_STREAM_SENTINEL: &str = "@dosh-exec";
+pub const FRAME_MAX_LEN: usize = 1024 * 1024;
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ExecRequest {
+    pub command: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum ExecResponse {
+    Stdout(Vec<u8>),
+    Stderr(Vec<u8>),
+    Exit { code: Option<i32> },
+    Error { message: String },
+}
+
+#[derive(Default)]
+pub struct FrameDecoder {
+    buf: Vec<u8>,
+}
+
+impl FrameDecoder {
+    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<Vec<u8>>> {
+        self.buf.extend_from_slice(bytes);
+        let mut frames = Vec::new();
+        loop {
+            if self.buf.len() < 4 {
+                break;
+            }
+            let len = u32::from_be_bytes(self.buf[..4].try_into().unwrap()) as usize;
+            if len > FRAME_MAX_LEN {
+                bail!("exec protocol frame too large: {len} bytes");
+            }
+            if self.buf.len() < 4 + len {
+                break;
+            }
+            frames.push(self.buf[4..4 + len].to_vec());
+            self.buf.drain(..4 + len);
+        }
+        Ok(frames)
+    }
+}
+
+pub fn encode_request(request: &ExecRequest) -> Result<Vec<u8>> {
+    encode_frame(&bincode::serialize(request).context("encode exec request")?)
+}
+
+pub fn encode_response(response: &ExecResponse) -> Result<Vec<u8>> {
+    encode_frame(&bincode::serialize(response).context("encode exec response")?)
+}
+
+pub fn decode_request(frame: &[u8]) -> Result<ExecRequest> {
+    bincode::deserialize(frame).context("decode exec request")
+}
+
+pub fn decode_response(frame: &[u8]) -> Result<ExecResponse> {
+    bincode::deserialize(frame).context("decode exec response")
+}
+
+fn encode_frame(payload: &[u8]) -> Result<Vec<u8>> {
+    if payload.len() > FRAME_MAX_LEN {
+        bail!("exec protocol frame too large: {} bytes", payload.len());
+    }
+    let mut out = Vec::with_capacity(4 + payload.len());
+    out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
+    out.extend_from_slice(payload);
+    Ok(out)
+}

src/file_transfer.rs

@@ -0,0 +1,250 @@
+use anyhow::{Context, Result, anyhow, bail};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+
+pub const FILE_STREAM_SENTINEL: &str = "@dosh-file";
+pub const FRAME_MAX_LEN: usize = 1024 * 1024;
+pub const CHUNK_SIZE: usize = 32 * 1024;
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum FileRequest {
+    Stat {
+        path: String,
+    },
+    List {
+        path: String,
+    },
+    Mkdir {
+        path: String,
+        mode: Option<u32>,
+    },
+    Remove {
+        path: String,
+        recursive: bool,
+    },
+    PutStart {
+        path: String,
+        size: u64,
+        mode: Option<u32>,
+        modified_secs: Option<u64>,
+        overwrite: bool,
+        resume: bool,
+    },
+    PutChunk {
+        offset: u64,
+        bytes: Vec<u8>,
+    },
+    PutFinish {
+        sha256: [u8; 32],
+    },
+    Get {
+        path: String,
+        offset: u64,
+    },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum FileResponse {
+    Ok,
+    Resume { offset: u64 },
+    Stat { meta: FileMeta },
+    List { entries: Vec<FileEntry> },
+    Start { meta: FileMeta, offset: u64 },
+    Chunk { offset: u64, bytes: Vec<u8> },
+    Done { sha256: [u8; 32], bytes: u64 },
+    Error { message: String },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FileMeta {
+    pub path: String,
+    pub kind: FileKind,
+    pub len: u64,
+    pub mode: Option<u32>,
+    pub modified_secs: Option<u64>,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+pub enum FileKind {
+    File,
+    Directory,
+    Symlink,
+    Other,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FileEntry {
+    pub name: String,
+    pub meta: FileMeta,
+}
+
+#[derive(Default)]
+pub struct FrameDecoder {
+    buf: Vec<u8>,
+}
+
+impl FrameDecoder {
+    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<Vec<u8>>> {
+        self.buf.extend_from_slice(bytes);
+        let mut frames = Vec::new();
+        loop {
+            if self.buf.len() < 4 {
+                break;
+            }
+            let len = u32::from_be_bytes(self.buf[..4].try_into().unwrap()) as usize;
+            if len > FRAME_MAX_LEN {
+                bail!("file protocol frame too large: {len} bytes");
+            }
+            if self.buf.len() < 4 + len {
+                break;
+            }
+            frames.push(self.buf[4..4 + len].to_vec());
+            self.buf.drain(..4 + len);
+        }
+        Ok(frames)
+    }
+}
+
+pub fn encode_request(request: &FileRequest) -> Result<Vec<u8>> {
+    encode_frame(&bincode::serialize(request).context("encode file request")?)
+}
+
+pub fn encode_response(response: &FileResponse) -> Result<Vec<u8>> {
+    encode_frame(&bincode::serialize(response).context("encode file response")?)
+}
+
+pub fn decode_request(frame: &[u8]) -> Result<FileRequest> {
+    bincode::deserialize(frame).context("decode file request")
+}
+
+pub fn decode_response(frame: &[u8]) -> Result<FileResponse> {
+    bincode::deserialize(frame).context("decode file response")
+}
+
+pub fn encode_frame(payload: &[u8]) -> Result<Vec<u8>> {
+    if payload.len() > FRAME_MAX_LEN {
+        bail!("file protocol frame too large: {} bytes", payload.len());
+    }
+    let mut out = Vec::with_capacity(4 + payload.len());
+    out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
+    out.extend_from_slice(payload);
+    Ok(out)
+}
+
+pub fn parse_copy_endpoint(raw: &str) -> CopyEndpoint {
+    if looks_like_windows_path(raw) {
+        return CopyEndpoint::Local(PathBuf::from(raw));
+    }
+    let Some(index) = raw.find(':') else {
+        return CopyEndpoint::Local(PathBuf::from(raw));
+    };
+    let host = &raw[..index];
+    let path = &raw[index + 1..];
+    if host.is_empty() || host.contains('/') || host.contains('\\') {
+        return CopyEndpoint::Local(PathBuf::from(raw));
+    }
+    CopyEndpoint::Remote {
+        host: host.to_string(),
+        path: if path.is_empty() {
+            ".".to_string()
+        } else {
+            path.to_string()
+        },
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CopyEndpoint {
+    Local(PathBuf),
+    Remote { host: String, path: String },
+}
+
+pub fn remote_join(parent: &str, child: &str) -> String {
+    if parent.is_empty() || parent == "." {
+        return child.to_string();
+    }
+    format!("{}/{}", parent.trim_end_matches('/'), child)
+}
+
+pub fn remote_basename(path: &str) -> String {
+    path.trim_end_matches('/')
+        .rsplit('/')
+        .next()
+        .filter(|value| !value.is_empty())
+        .unwrap_or(path)
+        .to_string()
+}
+
+pub fn clean_remote_path(path: &str, home: &Path) -> Result<PathBuf> {
+    if path.as_bytes().contains(&0) {
+        bail!("remote path contains NUL");
+    }
+    let expanded = if path == "~" {
+        home.to_path_buf()
+    } else if let Some(rest) = path.strip_prefix("~/") {
+        home.join(rest)
+    } else {
+        let raw = Path::new(path);
+        if raw.is_absolute() {
+            raw.to_path_buf()
+        } else {
+            home.join(raw)
+        }
+    };
+    Ok(expanded)
+}
+
+pub fn ensure_relative_child(path: &Path) -> Result<String> {
+    let name = path
+        .file_name()
+        .and_then(|name| name.to_str())
+        .ok_or_else(|| anyhow!("path has no final component: {}", path.display()))?;
+    if name.is_empty() || name == "." || name == ".." {
+        bail!("invalid path final component: {}", path.display());
+    }
+    Ok(name.to_string())
+}
+
+fn looks_like_windows_path(raw: &str) -> bool {
+    let bytes = raw.as_bytes();
+    bytes.len() >= 3
+        && bytes[0].is_ascii_alphabetic()
+        && bytes[1] == b':'
+        && (bytes[2] == b'\\' || bytes[2] == b'/')
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn frames_round_trip_incrementally() {
+        let first = encode_response(&FileResponse::Ok).unwrap();
+        let second = encode_response(&FileResponse::Error {
+            message: "nope".to_string(),
+        })
+        .unwrap();
+        let mut decoder = FrameDecoder::default();
+        assert!(decoder.push(&first[..2]).unwrap().is_empty());
+        let frames = decoder.push(&first[2..]).unwrap();
+        assert_eq!(frames.len(), 1);
+        assert_eq!(decode_response(&frames[0]).unwrap(), FileResponse::Ok);
+        let frames = decoder.push(&second).unwrap();
+        assert_eq!(frames.len(), 1);
+    }
+
+    #[test]
+    fn copy_endpoint_parses_scp_style_but_not_windows_drive() {
+        assert_eq!(
+            parse_copy_endpoint("palav:tmp/file"),
+            CopyEndpoint::Remote {
+                host: "palav".to_string(),
+                path: "tmp/file".to_string()
+            }
+        );
+        assert_eq!(
+            parse_copy_endpoint("C:\\Users\\palav\\x"),
+            CopyEndpoint::Local(PathBuf::from("C:\\Users\\palav\\x"))
+        );
+    }
+}

src/lib.rs

@@ -1,11 +1,25 @@
+//! Dosh is an encrypted UDP transport for remote terminals and embeddable Rust
+//! application streams.
+//!
+//! Use [`client::DoshClient`] and [`server::DoshServer`] when you want Dosh to
+//! handle native authentication, host key checks, roaming, keepalives, reliable
+//! streams, and service routing. Use [`transport::DoshTransport`] only when an
+//! application already owns session establishment and wants direct access to the
+//! encrypted stream transport.
+
 pub mod auth;
 pub mod build_info;
+pub mod client;
 pub mod config;
 pub mod crypto;
+pub mod exec_service;
+pub mod file_transfer;
 pub mod native;
 #[cfg(unix)]
 pub mod persist;
 pub mod protocol;
 #[cfg(unix)]
 pub mod pty;
+pub mod server;
 pub mod ssh_agent;
+pub mod transport;

src/protocol.rs

@@ -357,6 +357,10 @@ pub struct NativeAuthCheckOkBody {
     pub allow_tcp_forwarding: bool,
     pub allow_remote_forwarding: bool,
     pub allow_agent_forwarding: bool,
+    #[serde(default)]
+    pub allow_file_transfer: bool,
+    #[serde(default)]
+    pub allow_exec_command: bool,
     pub policy_flags: Vec<String>,
     #[serde(default)]
     pub server_version: String,

src/server.rs

@@ -0,0 +1,591 @@
+use crate::config::{ServerConfig, load_server_config};
+use crate::crypto;
+use crate::native::{
+    self, ForwardingKind, ForwardingRequest, NativeAuthOk, NativeServerHello,
+    derive_native_session_key, generate_native_ephemeral, host_public_key, load_or_create_host_key,
+    sign_server_hello, verify_native_user_auth_from_config,
+};
+use crate::protocol::{
+    self, AttachReject, CLIENT_TO_SERVER, NativeAuthOkBody, NativeClientHelloBody,
+    NativeServerHelloBody, NativeUserAuthBody, PacketKind, SERVER_TO_CLIENT,
+};
+use crate::transport::{
+    DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
+    service_name_from_target,
+};
+use anyhow::{Context, Result, anyhow, bail};
+use ed25519_dalek::SigningKey;
+use std::collections::{HashMap, HashSet};
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::net::UdpSocket;
+
+#[derive(Debug, Clone)]
+pub struct DoshServerConfig {
+    pub server: ServerConfig,
+    pub bind_addr: Option<SocketAddr>,
+    pub services: HashSet<String>,
+    pub transport: TransportConfig,
+    pub require_current_user: bool,
+    pub auth_timeout: Duration,
+}
+
+impl DoshServerConfig {
+    pub fn new(server: ServerConfig) -> Self {
+        Self {
+            server,
+            bind_addr: None,
+            services: HashSet::new(),
+            transport: TransportConfig::default(),
+            require_current_user: true,
+            auth_timeout: Duration::from_secs(30),
+        }
+    }
+
+    pub fn bind_addr(mut self, addr: SocketAddr) -> Self {
+        self.bind_addr = Some(addr);
+        self
+    }
+
+    pub fn service(mut self, name: impl Into<String>) -> Result<Self> {
+        self.services.insert(validate_service_name(name.into())?);
+        Ok(self)
+    }
+
+    pub fn services(mut self, names: impl IntoIterator<Item = impl Into<String>>) -> Result<Self> {
+        for name in names {
+            self.services.insert(validate_service_name(name.into())?);
+        }
+        Ok(self)
+    }
+
+    pub fn require_current_user(mut self, value: bool) -> Self {
+        self.require_current_user = value;
+        self
+    }
+
+    pub fn transport(mut self, transport: TransportConfig) -> Self {
+        self.transport = transport;
+        self
+    }
+}
+
+impl Default for DoshServerConfig {
+    fn default() -> Self {
+        Self::new(ServerConfig::default())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct DoshAccepted {
+    pub conn_id: [u8; 16],
+    pub user: String,
+    pub session: String,
+    pub services: Vec<String>,
+    pub peer_addr: SocketAddr,
+}
+
+#[derive(Debug, Clone)]
+pub enum DoshServerEvent {
+    Accepted(DoshAccepted),
+    Session {
+        conn_id: [u8; 16],
+        event: SessionEvent,
+    },
+    Ignored,
+}
+
+struct PendingServerAuth {
+    client: native::NativeClientHello,
+    server: NativeServerHello,
+    session_key: [u8; 32],
+    peer: SocketAddr,
+    created_at: Instant,
+}
+
+pub struct DoshServer {
+    socket: Arc<UdpSocket>,
+    config: DoshServerConfig,
+    host_signing: SigningKey,
+    pending: HashMap<[u8; 16], PendingServerAuth>,
+    transports: HashMap<[u8; 16], DoshTransport>,
+    accepted: HashMap<[u8; 16], DoshAccepted>,
+}
+
+impl DoshServer {
+    pub async fn load() -> Result<Self> {
+        Self::bind(DoshServerConfig::new(load_server_config(None)?)).await
+    }
+
+    pub async fn load_from_path(path: Option<PathBuf>) -> Result<Self> {
+        Self::bind(DoshServerConfig::new(load_server_config(path)?)).await
+    }
+
+    pub async fn bind(config: DoshServerConfig) -> Result<Self> {
+        let bind_addr = match config.bind_addr {
+            Some(addr) => addr,
+            None => format!("{}:{}", config.server.bind, config.server.port)
+                .parse()
+                .with_context(|| {
+                    format!(
+                        "parse Dosh bind address {}:{}",
+                        config.server.bind, config.server.port
+                    )
+                })?,
+        };
+        let host_signing = load_or_create_host_key(&config.server)?;
+        let socket = Arc::new(UdpSocket::bind(bind_addr).await?);
+        Ok(Self {
+            socket,
+            config,
+            host_signing,
+            pending: HashMap::new(),
+            transports: HashMap::new(),
+            accepted: HashMap::new(),
+        })
+    }
+
+    pub fn local_addr(&self) -> Result<SocketAddr> {
+        Ok(self.socket.local_addr()?)
+    }
+
+    pub fn connection(&self, conn_id: &[u8; 16]) -> Option<&DoshAccepted> {
+        self.accepted.get(conn_id)
+    }
+
+    pub fn transport(&self, conn_id: &[u8; 16]) -> Option<&DoshTransport> {
+        self.transports.get(conn_id)
+    }
+
+    pub fn transport_mut(&mut self, conn_id: &[u8; 16]) -> Option<&mut DoshTransport> {
+        self.transports.get_mut(conn_id)
+    }
+
+    pub async fn recv(&mut self) -> Result<DoshServerEvent> {
+        let mut buf = vec![0u8; 65535];
+        loop {
+            self.expire_pending();
+            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let packet = match protocol::decode(&buf[..n]) {
+                Ok(packet) => packet,
+                Err(_) => continue,
+            };
+            if packet.header.kind == PacketKind::NativeClientHello {
+                self.handle_client_hello(peer, packet.body).await?;
+                return Ok(DoshServerEvent::Ignored);
+            }
+            if packet.header.kind == PacketKind::NativeUserAuth {
+                return self.handle_user_auth(peer, &packet).await;
+            }
+            if let Some(transport) = self.transports.get_mut(&packet.header.conn_id) {
+                let event = transport.handle_datagram(&buf[..n], peer).await?;
+                return Ok(DoshServerEvent::Session {
+                    conn_id: packet.header.conn_id,
+                    event,
+                });
+            }
+            self.send_reject(peer, packet.header.conn_id, "unknown Dosh connection")
+                .await?;
+            return Ok(DoshServerEvent::Ignored);
+        }
+    }
+
+    pub async fn accept_stream(&mut self, conn_id: [u8; 16], stream_id: u64) -> Result<()> {
+        self.transport_mut(&conn_id)
+            .ok_or_else(|| anyhow!("unknown Dosh connection"))?
+            .accept_stream(stream_id)
+            .await
+    }
+
+    pub async fn reject_stream(
+        &mut self,
+        conn_id: [u8; 16],
+        stream_id: u64,
+        reason: impl Into<String>,
+    ) -> Result<()> {
+        self.transport_mut(&conn_id)
+            .ok_or_else(|| anyhow!("unknown Dosh connection"))?
+            .reject_stream(stream_id, reason)
+            .await
+    }
+
+    pub async fn send(
+        &mut self,
+        conn_id: [u8; 16],
+        stream_id: u64,
+        bytes: impl Into<Vec<u8>>,
+    ) -> Result<()> {
+        self.transport_mut(&conn_id)
+            .ok_or_else(|| anyhow!("unknown Dosh connection"))?
+            .send(stream_id, bytes)
+            .await
+    }
+
+    pub async fn close(&mut self, conn_id: [u8; 16], stream_id: u64) -> Result<()> {
+        self.transport_mut(&conn_id)
+            .ok_or_else(|| anyhow!("unknown Dosh connection"))?
+            .close(stream_id)
+            .await
+    }
+
+    pub async fn maintenance(&mut self) -> Result<()> {
+        for transport in self.transports.values_mut() {
+            transport.maintenance().await?;
+        }
+        Ok(())
+    }
+
+    async fn handle_client_hello(&mut self, peer: SocketAddr, body: Vec<u8>) -> Result<()> {
+        let req: NativeClientHelloBody = protocol::from_body(&body)?;
+        if let Err(err) =
+            native::check_native_protocol_version(req.hello.protocol_version, "client")
+        {
+            self.send_reject(peer, [0u8; 16], &err.to_string()).await?;
+            return Ok(());
+        }
+        let result = self.build_server_hello(req.hello, peer);
+        let (pending_id, hello) = match result {
+            Ok(value) => value,
+            Err(err) => {
+                self.send_reject(peer, [0u8; 16], &err.to_string()).await?;
+                return Ok(());
+            }
+        };
+        let body = protocol::to_body(&NativeServerHelloBody { hello })?;
+        let out = protocol::encode_plain(PacketKind::NativeServerHello, pending_id, 1, 0, &body)?;
+        self.socket.send_to(&out, peer).await?;
+        Ok(())
+    }
+
+    fn build_server_hello(
+        &mut self,
+        client: native::NativeClientHello,
+        peer: SocketAddr,
+    ) -> Result<([u8; 16], NativeServerHello)> {
+        if !self.config.server.native_auth {
+            bail!("native auth disabled");
+        }
+        if !client
+            .supported_aead
+            .iter()
+            .any(|algorithm| algorithm == "chacha20poly1305")
+        {
+            bail!("native auth requires chacha20poly1305");
+        }
+        if !client
+            .supported_user_key_algorithms
+            .iter()
+            .any(|algorithm| native::is_supported_user_signature_algorithm(algorithm))
+        {
+            bail!("native auth requires a supported user key algorithm");
+        }
+        if self.config.require_current_user {
+            let current_user = std::env::var("USER").unwrap_or_else(|_| "unknown".to_string());
+            if client.requested_user != current_user {
+                bail!("native auth user mismatch");
+            }
+        }
+
+        let (server_secret, server_public) = generate_native_ephemeral();
+        let mut server = NativeServerHello {
+            protocol_version: native::NATIVE_PROTOCOL_VERSION,
+            server_random: crypto::random_32(),
+            server_ephemeral_public: server_public,
+            host_key: host_public_key(&self.host_signing),
+            chosen_aead: "chacha20poly1305".to_string(),
+            server_key_epoch: 1,
+            auth_challenge: crypto::random_32(),
+            rate_limit_remaining: None,
+            host_signature: Vec::new(),
+        };
+        sign_server_hello(&self.host_signing, &client, &mut server)?;
+        let session_key = derive_native_session_key(
+            &server_secret,
+            client.client_ephemeral_public,
+            &client,
+            &server,
+        )?;
+        let mut pending_id = [0u8; 16];
+        pending_id.copy_from_slice(&server.auth_challenge[..16]);
+        self.pending.insert(
+            pending_id,
+            PendingServerAuth {
+                client,
+                server: server.clone(),
+                session_key,
+                peer,
+                created_at: Instant::now(),
+            },
+        );
+        Ok((pending_id, server))
+    }
+
+    async fn handle_user_auth(
+        &mut self,
+        peer: SocketAddr,
+        packet: &protocol::Packet,
+    ) -> Result<DoshServerEvent> {
+        let Some(pending) = self.pending.remove(&packet.header.conn_id) else {
+            self.send_reject(peer, packet.header.conn_id, "unknown native auth")
+                .await?;
+            return Ok(DoshServerEvent::Ignored);
+        };
+        if pending.peer != peer {
+            self.send_reject(peer, packet.header.conn_id, "native auth peer changed")
+                .await?;
+            return Ok(DoshServerEvent::Ignored);
+        }
+        if pending.created_at.elapsed() > self.config.auth_timeout {
+            self.send_reject(peer, packet.header.conn_id, "native auth expired")
+                .await?;
+            return Ok(DoshServerEvent::Ignored);
+        }
+
+        let body = protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER)?;
+        let req: NativeUserAuthBody = protocol::from_body(&body)?;
+        let services = match self.verify_auth_and_services(&pending, &req.auth) {
+            Ok(services) => services,
+            Err(err) => {
+                self.send_reject(peer, packet.header.conn_id, &format!("{err:#}"))
+                    .await?;
+                return Ok(DoshServerEvent::Ignored);
+            }
+        };
+
+        let conn_id = crypto::random_16();
+        let key_id = protocol::session_key_id(&pending.session_key);
+        let ok = NativeAuthOk {
+            client_id: conn_id,
+            session: pending.client.requested_session.clone(),
+            mode: pending.client.requested_mode.clone(),
+            session_key: pending.session_key,
+            session_key_id: key_id,
+            attach_ticket: Vec::new(),
+            attach_ticket_psk: crypto::random_32(),
+            initial_seq: 1,
+            snapshot: Vec::new(),
+            policy_flags: services
+                .iter()
+                .map(|service| format!("service:{service}"))
+                .collect(),
+        };
+        let body = protocol::to_body(&NativeAuthOkBody { ok })?;
+        let out = protocol::encode_encrypted(
+            PacketKind::NativeAuthOk,
+            conn_id,
+            1,
+            packet.header.seq,
+            &pending.session_key,
+            SERVER_TO_CLIENT,
+            &body,
+        )?;
+        self.socket.send_to(&out, peer).await?;
+
+        let transport = DoshTransport::new(
+            Arc::clone(&self.socket),
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: pending.session_key,
+                peer_addr: peer,
+                initial_send_seq: 2,
+                initial_ack: packet.header.seq,
+                stream: self.config.transport.clone(),
+            },
+        );
+        let accepted = DoshAccepted {
+            conn_id,
+            user: pending.client.requested_user,
+            session: pending.client.requested_session,
+            services,
+            peer_addr: peer,
+        };
+        self.transports.insert(conn_id, transport);
+        self.accepted.insert(conn_id, accepted.clone());
+        Ok(DoshServerEvent::Accepted(accepted))
+    }
+
+    fn verify_auth_and_services(
+        &self,
+        pending: &PendingServerAuth,
+        auth: &native::NativeUserAuth,
+    ) -> Result<Vec<String>> {
+        verify_native_user_auth_from_config(
+            &self.config.server,
+            &pending.client,
+            &pending.server,
+            auth,
+            Some(pending.peer.ip()),
+        )
+        .context("verify native user auth")?;
+        validate_requested_services(&self.config.services, &auth.requested_forwardings)
+    }
+
+    async fn send_reject(&self, peer: SocketAddr, conn_id: [u8; 16], reason: &str) -> Result<()> {
+        let body = protocol::to_body(&AttachReject {
+            reason: reason.to_string(),
+        })?;
+        let out = protocol::encode_plain(PacketKind::AttachReject, conn_id, 0, 0, &body)?;
+        self.socket.send_to(&out, peer).await?;
+        Ok(())
+    }
+
+    fn expire_pending(&mut self) {
+        let timeout = self.config.auth_timeout;
+        self.pending
+            .retain(|_, pending| pending.created_at.elapsed() <= timeout);
+    }
+}
+
+fn validate_requested_services(
+    allowed_services: &HashSet<String>,
+    requests: &[ForwardingRequest],
+) -> Result<Vec<String>> {
+    let mut services = Vec::new();
+    for request in requests {
+        if request.kind != ForwardingKind::Local {
+            bail!("embedded Dosh services only accept local service requests");
+        }
+        if request.listen_port != 0 || request.target_port != Some(0) {
+            bail!("embedded Dosh service requests must use port 0");
+        }
+        let target = request
+            .target_host
+            .as_deref()
+            .ok_or_else(|| anyhow!("embedded Dosh service request is missing target host"))?;
+        let service = service_name_from_target(target)
+            .ok_or_else(|| anyhow!("target {target:?} is not a Dosh service"))?;
+        if !allowed_services.contains(service) {
+            bail!("Dosh service {service:?} is not registered");
+        }
+        if !services.iter().any(|existing| existing == service) {
+            services.push(service.to_string());
+        }
+    }
+    Ok(services)
+}
+
+fn validate_service_name(name: String) -> Result<String> {
+    crate::transport::service_target(&name)?;
+    Ok(name)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::client::DoshClient;
+    use crate::config::{ClientConfig, HostsConfig};
+    use crate::transport::TransportEvent;
+    use ed25519_dalek::SigningKey;
+
+    #[tokio::test]
+    async fn sdk_client_and_server_exchange_service_stream() {
+        let dir = tempfile::tempdir().unwrap();
+        let host_key = dir.path().join("host_key");
+        let authorized_keys = dir.path().join("authorized_keys");
+        let identity = dir.path().join("id_ed25519");
+        let known_hosts = dir.path().join("known_hosts");
+
+        let signing = SigningKey::from_bytes(&[91u8; 32]);
+        let keypair = ssh_key::private::Ed25519Keypair::from(&signing);
+        let private =
+            ssh_key::PrivateKey::new(ssh_key::private::KeypairData::from(keypair), "").unwrap();
+        private
+            .write_openssh_file(&identity, ssh_key::LineEnding::LF)
+            .unwrap();
+        std::fs::write(
+            &authorized_keys,
+            format!("{}\n", private.public_key().to_openssh().unwrap()),
+        )
+        .unwrap();
+
+        let server_config = ServerConfig {
+            host_key: host_key.to_string_lossy().to_string(),
+            authorized_keys: vec![authorized_keys.to_string_lossy().to_string()],
+            ..ServerConfig::default()
+        };
+        let server_config = DoshServerConfig::new(server_config)
+            .bind_addr("127.0.0.1:0".parse().unwrap())
+            .service("echo")
+            .unwrap()
+            .require_current_user(false);
+        let mut server = DoshServer::bind(server_config).await.unwrap();
+        let server_port = server.local_addr().unwrap().port();
+
+        let client_config = ClientConfig {
+            dosh_port: server_port,
+            trust_on_first_use: true,
+            known_hosts: known_hosts.to_string_lossy().to_string(),
+            identity_files: vec![identity.to_string_lossy().to_string()],
+            use_ssh_agent: false,
+            ..ClientConfig::default()
+        };
+        let client = DoshClient::with_config(client_config, HostsConfig::default());
+        let connect = client
+            .connect("127.0.0.1")
+            .user("sdk-user")
+            .service("echo")
+            .connect();
+        let accept = async {
+            loop {
+                if let DoshServerEvent::Accepted(accepted) = server.recv().await.unwrap() {
+                    break accepted;
+                }
+            }
+        };
+        let (connected, accepted) = tokio::join!(connect, accept);
+        let mut client_transport = connected.unwrap().into_transport();
+        let conn_id = accepted.conn_id;
+        assert_eq!(accepted.services, vec!["echo".to_string()]);
+
+        let stream_id = client_transport.open_service("echo").await.unwrap();
+        match server.recv().await.unwrap() {
+            DoshServerEvent::Session {
+                event: SessionEvent::Stream(TransportEvent::Open(open)),
+                ..
+            } => {
+                assert_eq!(open.stream_id, stream_id);
+                server.accept_stream(conn_id, open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected event {other:?}"),
+        }
+        assert!(matches!(
+            client_transport.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { .. })
+        ));
+
+        client_transport
+            .send(stream_id, b"ping".to_vec())
+            .await
+            .unwrap();
+        loop {
+            match server.recv().await.unwrap() {
+                DoshServerEvent::Session {
+                    event: SessionEvent::Stream(TransportEvent::Data(data)),
+                    ..
+                } => {
+                    assert_eq!(data.chunks, vec![b"ping".to_vec()]);
+                    server
+                        .send(conn_id, data.stream_id, b"pong".to_vec())
+                        .await
+                        .unwrap();
+                    break;
+                }
+                DoshServerEvent::Session { .. } | DoshServerEvent::Ignored => {}
+                other => panic!("unexpected event {other:?}"),
+            }
+        }
+        loop {
+            match client_transport.recv().await.unwrap() {
+                SessionEvent::Stream(TransportEvent::Data(data)) => {
+                    assert_eq!(data.chunks, vec![b"pong".to_vec()]);
+                    break;
+                }
+                SessionEvent::Stream(_) | SessionEvent::Ignored => {}
+                other => panic!("unexpected event {other:?}"),
+            }
+        }
+    }
+}

tests/integration_smoke.rs

@@ -98,6 +98,20 @@ persist_sessions = false
     config
 }
 
+fn write_server_config_with_output_interval(
+    dir: &tempfile::TempDir,
+    port: u16,
+    output_frame_interval_ms: u64,
+) -> std::path::PathBuf {
+    let config = write_server_config(dir, port);
+    let mut raw = fs::read_to_string(&config).unwrap();
+    raw.push_str(&format!(
+        "output_frame_interval_ms = {output_frame_interval_ms}\n"
+    ));
+    fs::write(&config, raw).unwrap();
+    config
+}
+
 fn start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child {
     let server = env!("CARGO_BIN_EXE_dosh-server");
     let child = Command::new(server)
@@ -760,7 +774,9 @@ fn native_doctor_checks_auth_without_opening_terminal() {
     assert!(output.status.success(), "stdout={stdout}\nstderr={stderr}");
     assert!(stdout.contains("[ok] native udp"), "stdout={stdout}");
     assert!(stdout.contains("[ok] native auth"), "stdout={stdout}");
-    assert!(stdout.contains("[ok] forwarding policy"), "stdout={stdout}");
+    assert!(stdout.contains("[ok] server policy"), "stdout={stdout}");
+    assert!(stdout.contains("file_transfer=true"), "stdout={stdout}");
+    assert!(stdout.contains("exec_command=true"), "stdout={stdout}");
 }
 
 #[test]
@@ -868,6 +884,177 @@ fn native_local_forward_background_smoke() {
     assert_eq!(&buf[..n], b"dosh-background-forward-ping");
 }
 
+#[test]
+fn native_file_copy_recursive_round_trip() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let src = dir.path().join("srcdir");
+    fs::create_dir_all(src.join("nested")).unwrap();
+    fs::write(src.join("root.txt"), b"root file\n").unwrap();
+    fs::write(src.join("nested/child.txt"), b"child file\n").unwrap();
+
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+    let upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("-r")
+        .arg(&src)
+        .arg("local:remote-copy")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        upload.status.success(),
+        "upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&upload.stdout),
+        String::from_utf8_lossy(&upload.stderr)
+    );
+
+    let remote_copy = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("-r")
+        .arg("local:remote-copy")
+        .arg("local:remote-copy-2")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        remote_copy.status.success(),
+        "remote copy failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&remote_copy.stdout),
+        String::from_utf8_lossy(&remote_copy.stderr)
+    );
+
+    let list = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("ls")
+        .arg("local:remote-copy-2")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        list.status.success(),
+        "ls failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&list.stdout),
+        String::from_utf8_lossy(&list.stderr)
+    );
+    let list_stdout = String::from_utf8_lossy(&list.stdout);
+    assert!(list_stdout.contains("root.txt"), "stdout={list_stdout}");
+    assert!(list_stdout.contains("nested"), "stdout={list_stdout}");
+
+    let cat = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cat")
+        .arg("local:remote-copy-2/root.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        cat.status.success(),
+        "cat failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&cat.stdout),
+        String::from_utf8_lossy(&cat.stderr)
+    );
+    assert_eq!(String::from_utf8_lossy(&cat.stdout), "root file\n");
+
+    let downloaded = dir.path().join("downloaded");
+    let download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("-r")
+        .arg("local:remote-copy-2")
+        .arg(&downloaded)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        download.status.success(),
+        "download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&download.stdout),
+        String::from_utf8_lossy(&download.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(downloaded.join("root.txt")).unwrap(),
+        "root file\n"
+    );
+    assert_eq!(
+        fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
+        "child file\n"
+    );
+
+    let remove = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("rm")
+        .arg("-r")
+        .arg("local:remote-copy")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        remove.status.success(),
+        "rm failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&remove.stdout),
+        String::from_utf8_lossy(&remove.stderr)
+    );
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
+#[test]
+fn native_exec_command_smoke() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+    let output = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf out; printf err >&2")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    let _ = server.kill();
+    let _ = server.wait();
+
+    assert!(
+        output.status.success(),
+        "exec failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert_eq!(String::from_utf8_lossy(&output.stdout), "out");
+    assert_eq!(String::from_utf8_lossy(&output.stderr), "err");
+}
+
 #[test]
 fn native_local_forward_bulk_load_does_not_delay_interactive_terminal() {
     use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
@@ -1466,6 +1653,47 @@ fn unicode_output_survives_transport() {
     );
 }
 
+#[test]
+fn tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config_with_output_interval(&dir, port, 1000);
+    let mut server = start_server(&dir, &config);
+    let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write");
+
+    let graph = "DOSH_GRAPH ⣀⣤⣶⣿ █▇▆▅▄▃▂▁ ╭╮╯╰";
+    let input = Input {
+        bytes: format!(
+            "printf '\\033[?1049h\\033[?25l'; for i in 1 2 3 4 5; do printf '\\033[H{} %s\\n' \"$i\"; done; printf '\\033[?25h\\033[?1049l'\n",
+            graph
+        )
+        .into_bytes(),
+    };
+    send_encrypted(
+        &socket,
+        port,
+        PacketKind::Input,
+        ok.client_id,
+        2,
+        0,
+        &bootstrap.session_key,
+        &protocol::to_body(&input).unwrap(),
+    );
+    let text = collect_frame_text(&socket, &bootstrap.session_key, 3000);
+
+    let _ = server.kill();
+    let _ = server.wait();
+
+    assert!(
+        text.contains(graph) && text.matches("DOSH_GRAPH").count() >= 5,
+        "expected repeated raw graph glyph frames, got {text:?}"
+    );
+    assert!(
+        text.contains("\x1b[?25l") && text.contains("\x1b[?25h"),
+        "expected cursor visibility controls to survive around graph repaint, got {text:?}"
+    );
+}
+
 #[test]
 fn large_tui_paint_is_delivered_in_mtu_safe_frames() {
     let dir = tempfile::tempdir().unwrap();

tests/release_scripts.rs

@@ -0,0 +1,20 @@
+#[test]
+fn quiet_update_keeps_prebuilt_enabled_by_default() {
+    let install = include_str!("../install.sh");
+    assert!(install.contains("use_prebuilt=\"${DOSH_USE_PREBUILT:-1}\""));
+    assert!(
+        !install.contains("use_prebuilt=0"),
+        "quiet updates must not force source builds"
+    );
+}
+
+#[test]
+fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
+    let upload = include_str!("../scripts/upload-gitea-release.sh");
+    assert!(upload.contains("skipping $artifact: version"));
+    assert!(upload.contains("expected_version"));
+
+    let publish = include_str!("../scripts/publish-gitea-release.sh");
+    assert!(publish.contains("skipping stale artifact"));
+    assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\""));
+}

vscode-extension/README.md

@@ -0,0 +1,15 @@
+# Dosh Remote
+
+Open VS Code Remote-SSH windows through Dosh.
+
+The extension writes a managed SSH config entry like:
+
+```sshconfig
+Host dosh-palav
+    HostName 127.0.0.1
+    HostKeyAlias palav
+    ProxyCommand dosh proxy-stdio palav %h %p
+```
+
+VS Code still uses Remote-SSH, so server install, extensions, terminals, and
+normal SSH behavior stay intact. Dosh carries the SSH TCP byte stream.

vscode-extension/extension.js

@@ -0,0 +1,170 @@
+const vscode = require('vscode');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+function activate(context) {
+  context.subscriptions.push(
+    vscode.commands.registerCommand('dosh.openRemote', openRemote),
+    vscode.commands.registerCommand('dosh.configureHost', configureHostCommand),
+    vscode.commands.registerCommand('dosh.showSshConfig', showSshConfig)
+  );
+}
+
+async function openRemote() {
+  const configured = await configureHost();
+  if (!configured) {
+    return;
+  }
+  const remotePath = await vscode.window.showInputBox({
+    title: 'Remote path',
+    prompt: 'Path to open on the remote host',
+    value: '~'
+  });
+  if (remotePath === undefined) {
+    return;
+  }
+  const suffix = remotePath ? `/${remotePath.replace(/^\/+/, '')}` : '';
+  await vscode.commands.executeCommand(
+    'vscode.openFolder',
+    vscode.Uri.parse(`vscode-remote://ssh-remote+${configured.alias}${suffix}`),
+    { forceNewWindow: true }
+  );
+}
+
+async function configureHostCommand() {
+  const configured = await configureHost();
+  if (configured) {
+    vscode.window.showInformationMessage(`Dosh Remote-SSH host ready: ${configured.alias}`);
+  }
+}
+
+async function configureHost() {
+  const host = await vscode.window.showInputBox({
+    title: 'Dosh host',
+    prompt: 'Dosh host alias, e.g. palav',
+    ignoreFocusOut: true
+  });
+  if (!host) {
+    return undefined;
+  }
+  const user = await vscode.window.showInputBox({
+    title: 'Remote SSH user',
+    prompt: 'Optional. Leave blank to let SSH config decide.',
+    ignoreFocusOut: true
+  });
+  const config = vscode.workspace.getConfiguration('dosh');
+  const alias = `dosh-${safeAlias(host)}`;
+  const block = sshBlock({
+    alias,
+    host,
+    user: user || undefined,
+    executable: config.get('executable') || 'dosh',
+    targetHost: config.get('targetHost') || '127.0.0.1',
+    targetPort: Number(config.get('targetPort') || 22),
+    doshPort: Number(config.get('doshPort') || 0)
+  });
+  const sshDir = path.join(os.homedir(), '.ssh');
+  fs.mkdirSync(sshDir, { recursive: true });
+  const generatedPath = config.get('generatedSshConfig') || path.join(sshDir, 'config.dosh');
+  const mainConfig = path.join(sshDir, 'config');
+  ensureInclude(mainConfig, path.basename(generatedPath));
+  upsertBlock(generatedPath, alias, block);
+  return { alias, generatedPath };
+}
+
+function sshBlock(options) {
+  let proxy = shellQuote(options.executable);
+  if (options.doshPort > 0) {
+    proxy += ` --dosh-port ${options.doshPort}`;
+  }
+  proxy += ` proxy-stdio ${shellQuote(options.host)} %h %p`;
+  const lines = [
+    `# BEGIN DOSH ${options.alias}`,
+    `Host ${options.alias}`,
+    `    HostName ${options.targetHost}`,
+    `    Port ${options.targetPort}`,
+    `    HostKeyAlias ${options.host}`,
+    options.user ? `    User ${options.user}` : undefined,
+    '    ClearAllForwardings yes',
+    '    ServerAliveInterval 15',
+    '    ServerAliveCountMax 3',
+    `    ProxyCommand ${proxy}`,
+    `# END DOSH ${options.alias}`,
+    ''
+  ].filter(Boolean);
+  return lines.join('\n');
+}
+
+function ensureInclude(configPath, includeFile) {
+  const raw = fs.existsSync(configPath) ? fs.readFileSync(configPath, 'utf8') : '';
+  if (raw.split(/\r?\n/).some((line) => line.trim().toLowerCase() === `include ${includeFile}`.toLowerCase())) {
+    return;
+  }
+  fs.writeFileSync(configPath, `Include ${includeFile}\n${raw ? `\n${raw}` : ''}`);
+}
+
+function upsertBlock(configPath, alias, block) {
+  const raw = fs.existsSync(configPath) ? fs.readFileSync(configPath, 'utf8') : '';
+  const begin = `# BEGIN DOSH ${alias}`;
+  const end = `# END DOSH ${alias}`;
+  const lines = raw.split(/\r?\n/);
+  const out = [];
+  let skipping = false;
+  let replaced = false;
+  for (const line of lines) {
+    if (line.trim() === begin) {
+      out.push(block.trimEnd());
+      skipping = true;
+      replaced = true;
+      continue;
+    }
+    if (skipping) {
+      if (line.trim() === end) {
+        skipping = false;
+      }
+      continue;
+    }
+    if (line.length > 0 || out.length > 0) {
+      out.push(line);
+    }
+  }
+  if (!replaced) {
+    if (out.length > 0 && out[out.length - 1] !== '') {
+      out.push('');
+    }
+    out.push(block.trimEnd());
+  }
+  fs.writeFileSync(configPath, `${out.join('\n').trimEnd()}\n`);
+}
+
+async function showSshConfig() {
+  const config = vscode.workspace.getConfiguration('dosh');
+  const sshDir = path.join(os.homedir(), '.ssh');
+  const generatedPath = config.get('generatedSshConfig') || path.join(sshDir, 'config.dosh');
+  if (!fs.existsSync(generatedPath)) {
+    vscode.window.showWarningMessage('No generated Dosh SSH config yet.');
+    return;
+  }
+  const doc = await vscode.workspace.openTextDocument(generatedPath);
+  await vscode.window.showTextDocument(doc);
+}
+
+function safeAlias(value) {
+  const alias = value.replace(/[^a-zA-Z0-9_-]+/g, '-').replace(/^-+|-+$/g, '');
+  return alias || 'host';
+}
+
+function shellQuote(value) {
+  if (/^[a-zA-Z0-9_./:-]+$/.test(value)) {
+    return value;
+  }
+  return `'${value.replace(/'/g, `'\\''`)}'`;
+}
+
+function deactivate() {}
+
+module.exports = {
+  activate,
+  deactivate
+};

vscode-extension/package.json

@@ -0,0 +1,69 @@
+{
+  "name": "dosh-vscode",
+  "displayName": "Dosh Remote",
+  "description": "Open VS Code Remote-SSH windows through Dosh.",
+  "version": "0.1.0",
+  "publisher": "palav",
+  "license": "MIT",
+  "engines": {
+    "vscode": "^1.90.0"
+  },
+  "categories": [
+    "Other"
+  ],
+  "extensionDependencies": [
+    "ms-vscode-remote.remote-ssh"
+  ],
+  "activationEvents": [
+    "onCommand:dosh.openRemote",
+    "onCommand:dosh.configureHost",
+    "onCommand:dosh.showSshConfig"
+  ],
+  "main": "./extension.js",
+  "contributes": {
+    "commands": [
+      {
+        "command": "dosh.openRemote",
+        "title": "Dosh: Open Remote Folder"
+      },
+      {
+        "command": "dosh.configureHost",
+        "title": "Dosh: Configure Remote-SSH Host"
+      },
+      {
+        "command": "dosh.showSshConfig",
+        "title": "Dosh: Show Generated SSH Config"
+      }
+    ],
+    "configuration": {
+      "title": "Dosh Remote",
+      "properties": {
+        "dosh.executable": {
+          "type": "string",
+          "default": "dosh",
+          "description": "Path to the local dosh executable."
+        },
+        "dosh.targetHost": {
+          "type": "string",
+          "default": "127.0.0.1",
+          "description": "Host opened from the Dosh server side for VS Code's SSH connection."
+        },
+        "dosh.targetPort": {
+          "type": "number",
+          "default": 22,
+          "description": "SSH port opened from the Dosh server side."
+        },
+        "dosh.doshPort": {
+          "type": "number",
+          "default": 0,
+          "description": "Optional Dosh UDP port override. 0 uses Dosh config."
+        },
+        "dosh.generatedSshConfig": {
+          "type": "string",
+          "default": "",
+          "description": "Path for generated SSH config. Empty means ~/.ssh/config.dosh."
+        }
+      }
+    }
+  }
+}