Add SSH remote block writer

.github/workflows/ci.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [main, master]
   pull_request:
-    branches: [main, master]
 
 env:
   CARGO_TERM_COLOR: always
@@ -13,9 +12,7 @@ permissions:
   contents: read
 
 jobs:
-  # runs first: builds the write-head binaries and leaves them as artifacts
-  # for later fls jobs (add `needs: write-head` + actions/download-artifact
-  # in a job that wants them)
+  # Build the write head before the Rust jobs embed it.
   write-head:
     name: Write head (fls-wh)
     runs-on: ubuntu-latest
@@ -70,10 +67,16 @@ jobs:
 
   lint:
     name: Lint
+    needs: write-head
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
 
+      - uses: actions/download-artifact@v4
+        with:
+          name: flswh-aarch64-qnx7
+          path: remote/
+
       - name: Install Rust toolchain
         uses: dtolnay/rust-toolchain@stable
         with:
@@ -105,10 +108,16 @@ jobs:
 
   test:
     name: Test
+    needs: write-head
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
 
+      - uses: actions/download-artifact@v4
+        with:
+          name: flswh-aarch64-qnx7
+          path: remote/
+
       - name: Install Rust toolchain
         uses: dtolnay/rust-toolchain@stable
 
@@ -135,6 +144,7 @@ jobs:
 
   build:
     name: Build (${{ matrix.target }})
+    needs: write-head
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
@@ -159,6 +169,11 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
+      - uses: actions/download-artifact@v4
+        with:
+          name: flswh-aarch64-qnx7
+          path: remote/
+
       - name: Install Rust toolchain
         uses: dtolnay/rust-toolchain@stable
         with:
@@ -210,5 +225,3 @@ jobs:
           path: ${{ matrix.artifact_name }}
           if-no-files-found: error
 
-
-

Cargo.toml

@@ -9,6 +9,7 @@ reqwest = { version = "0.12", default-features = false, features = ["stream", "r
 tokio = { version = "1.0", features = ["full", "process", "rt-multi-thread"] }
 futures-util = "0.3"
 bytes = "1.0"
+crc32fast = "1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 base64 = "0.22"

README.md

@@ -37,6 +37,61 @@ sudo cp target/release/fls /usr/local/bin/
 
 ## Usage
 
+### Remote devices over SSH
+
+Use `ssh://[user@]host[:port]/device` as the destination for HTTP/HTTPS or OCI images:
+
+```bash
+fls from-url "https://example.com/image.img.xz" ssh://root@board/dev/emmc0
+fls from-url "oci://quay.io/org/image:latest" ssh://root@board:2222/dev/emmc0
+# IPv6
+fls from-url "https://example.com/image.img.xz" ssh://root@[::1]/dev/emmc0
+```
+
+The legacy form `[user@]host:/device` is still accepted (no IPv6 or inline port).
+
+`fls` uploads its embedded aarch64-QNX write head over SSH, then streams framed
+commands. For another target architecture/OS, build the appropriate write head
+and use `--wh-bin /path/to/flswh`. When building from source, run
+`make remote-wh` before `cargo build` to embed the QNX binary; local-only builds
+work without it. See [remote/README.md](remote/README.md) for build instructions.
+
+SSH keys, agents, ports, and aliases use the system `ssh` and your
+`~/.ssh/config`. Set the port inline: `ssh://host:2222/dev/x`. `-p` remains
+the registry-password option. Password authentication uses `sshpass` with
+`--ssh-password-file <path>` (`FLS_SSH_PASS_FILE`) or, if no file is specified,
+the `SSHPASS` environment variable. `--ssh-compress` enables SSH compression;
+`--wh-bin` can also be supplied through `FLS_WH_BIN`.
+By default, all SSH connections use `StrictHostKeyChecking=no`,
+`UserKnownHostsFile=/dev/null`, and `LogLevel=ERROR`: host keys are not verified
+or saved to your user known-hosts file. `--strict-ssh-host-key-checking` instead
+uses `StrictHostKeyChecking=yes` with your configured known-hosts files and log
+level. In strict mode, connect once with `ssh root@board` (or
+`ssh -p <port> root@board`) to verify the key before flashing; unknown or changed
+keys fail promptly. Key/agent authentication remains non-interactive unless
+`sshpass` is selected.
+
+Before uploading, `fls` queries the remote OS, CPU architecture, and release
+using `uname` (`-p` for the QNX processor, `-m` elsewhere). The embedded binary
+is selected only for aarch64 QNX 7. Other platforms require `--wh-bin` with a
+compatible write head; automatic Linux/macOS binary embedding can be added
+when those artifacts are available.
+
+Remote zero fills use ZERO rather than transferring their contents. Adjacent
+DATA writes are coalesced into frames of up to 8 MiB, flushing before SEEK,
+FILL, and final SYNC so sparse-image fragments do not each require an ACK.
+DATA, ZERO, and SEEK frames are pipelined in order while a separate reader
+consumes progress and ACKs. `--write-buffer-size` also limits unacknowledged
+DATA bytes (default: 128 MiB); smaller windows reduce the maximum frame size.
+Small-command metadata is bounded to 1024 outstanding requests. Closing drains
+the ACKs before SYNC/QUIT, and remote errors stop the pipeline.
+The writer checks write, fill, and seek ranges against remote device capacity, and waits
+for successful SYNC and QUIT before reporting completion. `--o-direct` applies
+to local devices; remote durability uses SYNC. One flash session per host is
+supported because uploads share `/tmp/fls-wh`.
+The reported Written count is the logical device position, including sparse
+skips, matching local flashing; DONE still validates only actual written bytes.
+
 ### Basic Example
 
 Flash a compressed image from a URL to a block device:
@@ -164,6 +219,7 @@ fls from-url [OPTIONS] <URL> <DEVICE>
 - `-k, --insecure-tls` - Ignore SSL certificate verification
 - `--cacert <CACERT>` - Path to CA certificate PEM file for TLS validation
 - `--buffer-size <SIZE>` - Buffer size in MB for download buffering (default: 1024)
+- `--write-buffer-size <SIZE>` - Write queue and remote unacknowledged DATA limit (default: 128 MiB)
 - `--max-retries <NUM>` - Maximum number of retry attempts (default: 10)
 - `--retry-delay <SECONDS>` - Delay in seconds between retry attempts (default: 2)
 - `--debug` - Enable debug output
@@ -180,5 +236,3 @@ fls from-url [OPTIONS] <URL> <DEVICE>
 - Ensure the target device is not mounted
 - Verify the device path to avoid data loss
 - Use `lsblk` or `fdisk -l` to identify the correct device before flashing
-
-

build.rs

@@ -0,0 +1,12 @@
+use std::{env, fs, path::PathBuf};
+
+fn main() {
+    let src = "remote/flswh-aarch64-qnx7";
+    let out = PathBuf::from(env::var_os("OUT_DIR").unwrap()).join("flswh-aarch64-qnx7");
+    if std::path::Path::new(src).exists() {
+        fs::copy(src, out).expect("copy write head");
+    } else {
+        fs::write(out, []).expect("create empty write-head placeholder");
+    }
+    println!("cargo:rerun-if-changed={src}");
+}

src/fls/block_writer.rs

@@ -5,6 +5,9 @@ use std::os::unix::fs::OpenOptionsExt;
 use std::ptr::NonNull;
 use tokio::sync::mpsc;
 
+use super::options::FlashOptions;
+use super::ssh_writer::{parse_ssh_target, SshBlockWriter, SshTarget};
+
 const BLOCK_SIZE: usize = 1024 * 1024; // 1MB blocks for better throughput
 const ALIGNMENT: usize = 4096; // 4KB alignment for direct I/O
 
@@ -593,6 +596,61 @@ impl AsyncBlockWriter {
     }
 }
 
+pub(crate) enum DeviceWriter {
+    Local(AsyncBlockWriter),
+    Ssh(SshBlockWriter),
+}
+
+impl DeviceWriter {
+    pub(crate) fn new(
+        options: &FlashOptions,
+        progress: mpsc::UnboundedSender<u64>,
+    ) -> io::Result<Self> {
+        if let Some(target) = parse_ssh_target(&options.device)? {
+            let SshTarget { host, port, device } = target;
+            Ok(Self::Ssh(SshBlockWriter::new(
+                host, port, device, progress, options,
+            )?))
+        } else {
+            Ok(Self::Local(AsyncBlockWriter::new(
+                options.device.clone(),
+                progress,
+                options.debug,
+                options.o_direct,
+                options.write_buffer_size_mb,
+            )?))
+        }
+    }
+
+    pub(crate) async fn write(&self, data: Vec<u8>) -> io::Result<()> {
+        match self {
+            Self::Local(writer) => writer.write(data).await,
+            Self::Ssh(writer) => writer.write(data).await,
+        }
+    }
+
+    pub(crate) async fn seek(&self, offset: u64) -> io::Result<()> {
+        match self {
+            Self::Local(writer) => writer.seek(offset).await,
+            Self::Ssh(writer) => writer.seek(offset).await,
+        }
+    }
+
+    pub(crate) async fn fill(&self, pattern: [u8; 4], bytes: u64) -> io::Result<()> {
+        match self {
+            Self::Local(writer) => writer.fill(pattern, bytes).await,
+            Self::Ssh(writer) => writer.fill(pattern, bytes).await,
+        }
+    }
+
+    pub(crate) async fn close(self) -> io::Result<u64> {
+        match self {
+            Self::Local(writer) => writer.close().await,
+            Self::Ssh(writer) => writer.close().await,
+        }
+    }
+}
+
 /// Helper to write fill pattern efficiently.
 /// Alignment for O_DIRECT is handled by BlockWriter's internal buffer and flush_buffer(),
 /// which rounds up to ALIGNMENT when flushing partial blocks.

src/fls/from_url.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 
-use crate::fls::block_writer::AsyncBlockWriter;
+use crate::fls::block_writer::DeviceWriter;
 use crate::fls::byte_channel::byte_bounded_channel;
 use crate::fls::compression::Compression;
 use crate::fls::decompress::{get_compression_from_url, start_inprocess_decompressor};
@@ -42,7 +42,7 @@ use crate::fls::download_error::handle_download_retry;
 /// Execute a sequence of write commands on the block writer
 async fn execute_write_commands(
     commands: Vec<WriteCommand>,
-    writer: &AsyncBlockWriter,
+    writer: &DeviceWriter,
     error_tx: &mpsc::UnboundedSender<String>,
     debug: bool,
 ) -> io::Result<()> {
@@ -92,7 +92,7 @@ async fn execute_write_commands(
 async fn process_sparse_data(
     parser: &mut SparseParser,
     data: &[u8],
-    writer: &AsyncBlockWriter,
+    writer: &DeviceWriter,
     error_tx: &mpsc::UnboundedSender<String>,
     debug: bool,
 ) -> io::Result<()> {
@@ -107,7 +107,7 @@ async fn process_sparse_data(
 /// Write data to the block writer with error reporting
 async fn write_regular_data(
     data: Vec<u8>,
-    writer: &AsyncBlockWriter,
+    writer: &DeviceWriter,
     error_tx: &mpsc::UnboundedSender<String>,
 ) -> io::Result<()> {
     writer.write(data).await.map_err(|e| {
@@ -121,7 +121,7 @@ async fn handle_detected_format(
     format: FileFormat,
     consumed_bytes: Vec<u8>,
     remaining_data: &[u8],
-    writer: &AsyncBlockWriter,
+    writer: &DeviceWriter,
     error_tx: &mpsc::UnboundedSender<String>,
     debug: bool,
 ) -> io::Result<Option<SparseParser>> {
@@ -187,13 +187,7 @@ pub async fn flash_from_url(
     );
 
     // Create block writer
-    let block_writer = AsyncBlockWriter::new(
-        options.common.device.clone(),
-        written_progress_tx,
-        options.common.debug,
-        options.common.o_direct,
-        options.common.write_buffer_size_mb,
-    )?;
+    let block_writer = DeviceWriter::new(&options.common, written_progress_tx)?;
 
     // Create byte-bounded download buffer
     let buffer_size_mb = options.common.buffer_size_mb;
@@ -218,12 +212,29 @@ pub async fn flash_from_url(
         options.common.xz_memlimit_mb,
     )?;
 
+    // Signals the feed loop when the writer task finishes, so a full input
+    // buffer cannot deadlock it (e.g. the SSH connection breaks mid-flash).
+    let (writer_done_tx, mut writer_done_rx) = tokio::sync::watch::channel(false);
+
     // Spawn background task to read decompressed data and write to block device
     let error_tx_clone = error_tx.clone();
     let debug = options.common.debug;
     let writer_handle = {
         let writer = block_writer;
+        let done_tx = writer_done_tx;
         tokio::spawn(async move {
+            // Fire `writer_done` however this task ends (success, error, or
+            // panic) so the feed loop's blocking buffer send can always unblock.
+            struct WriterDone(Option<tokio::sync::watch::Sender<bool>>);
+            impl Drop for WriterDone {
+                fn drop(&mut self) {
+                    if let Some(tx) = self.0.take() {
+                        let _ = tx.send(true);
+                    }
+                }
+            }
+            let _done = WriterDone(Some(done_tx));
+
             let mut detector = FormatDetector::new();
             let mut parser: Option<SparseParser> = None;
             let mut format_determined = false;
@@ -368,14 +379,20 @@ pub async fn flash_from_url(
                         Ok(chunk) => {
                             let chunk_len = chunk.len() as u64;
 
-                            // Send to buffer - detect if it's blocking
+                            // Send to buffer - detect if it's blocking. Race the
+                            // send against the writer finishing: if the writer
+                            // stops (e.g. SSH breaks) while the buffer is full,
+                            // the send would otherwise block forever.
                             let send_start = std::time::Instant::now();
-                            if buffer_tx.send(chunk).await.is_err() {
-                                if writer_handle.is_finished() {
+                            let send_result = tokio::select! {
+                                result = buffer_tx.send(chunk) => result,
+                                _ = writer_done_rx.wait_for(|done| *done) => {
                                     eprintln!();
-                                    eprintln!("Writer task has terminated unexpectedly");
+                                    eprintln!("Writer task has terminated, stopping download");
                                     return Err(get_writer_error(writer_handle).await);
                                 }
+                            };
+                            if send_result.is_err() {
                                 connection_error =
                                     Some(DownloadError::Other("Buffer channel closed".to_string()));
                                 connection_broken = true;

src/fls/mod.rs

@@ -17,6 +17,7 @@ pub mod oci;
 mod options;
 mod progress;
 mod simg;
+mod ssh_writer;
 mod stream_utils;
 
 // Public re-exports