diff --git a/Cargo.toml b/Cargo.toml index 5e481cf..6f99815 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -77,3 +77,7 @@ rumqttc = { version = "0.25.1", default-features = false, features = ["use-nativ suppaftp = { version = "12.1.1", features = ["native-tls", "deprecated"] } zip = { version = "9.0.3", default-features = false, features = ["deflate-flate2"] } flate2 = "1.1.10" +# HTTPS GET of a presigned job-file URL for the `start_print` command (see +# src/uplink/download.rs). Features are the bare minimum: TLS via rustls, no +# openssl-sys; no json/gzip/cookies/etc, since the body is an opaque binary. +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots"] } diff --git a/src/uplink/download.rs b/src/uplink/download.rs new file mode 100644 index 0000000..2ac185c --- /dev/null +++ b/src/uplink/download.rs @@ -0,0 +1,130 @@ +//! Fetching a job file from the presigned URL a `start_print` command carries. +//! +//! The URL is attacker-influenceable in the worst case (a misconfigured or +//! compromised cloud, a bad object store) and the body is held in memory +//! because `upload_project_file` takes a `Vec`, so everything here is +//! bounded: size, time, redirects and scheme. + +use std::time::Duration; + +/// Hard ceiling on the body. A sliced `.3mf` is tens of megabytes (62 MB has +/// been seen for a multi-plate project); 256 MiB is roughly four times that, +/// so no honest job is refused, while a hostile URL cannot make a gateway on +/// small hardware (a Pi-class box is the target) buffer its way into the OOM +/// killer. Also well above anything a printer's own storage would take. +pub const MAX_JOB_FILE_BYTES: u64 = 256 * 1024 * 1024; + +/// Whole-request budget, connect through last byte. 256 MiB at the slowest +/// link worth supporting (~2 MiB/s) is just over two minutes, so 3 minutes +/// covers the cap itself; anything slower than that is a stalled transfer, not +/// a slow one. Presigned URLs live for minutes, so a longer wait would only +/// be spent on a URL about to expire. Note `dispatch` runs inside the uplink +/// loop, so this is also the longest a download can hold up heartbeats. +pub const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(180); + +/// Separate and short: an unreachable host should fail in seconds, not use +/// up the whole budget above. +pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// Downloads `url` fully into memory, or says why not. +/// +/// The error is a `String` naming the failure; the caller prefixes the stage. +pub async fn fetch_job_file(url: &str) -> Result, String> { + fetch_job_file_capped(url, MAX_JOB_FILE_BYTES).await +} + +/// Same as [`fetch_job_file`] with the cap as a parameter, so tests can +/// exercise the limit without serving 256 MiB. +pub async fn fetch_job_file_capped(url: &str, max_bytes: u64) -> Result, String> { + let parsed = reqwest::Url::parse(url).map_err(|err| format!("not a valid URL: {err}"))?; + // A presigned URL carries its authority in the query string; sending it + // over plain HTTP would expose it to anyone on the path. + if parsed.scheme() != "https" { + return Err(format!("refusing non-HTTPS URL (scheme '{}')", parsed.scheme())); + } + + let client = reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .timeout(DOWNLOAD_TIMEOUT) + // A presigned GET answers directly. Following a redirect would carry + // us somewhere the signature never vouched for. + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|err| format!("could not build HTTP client: {err}"))?; + + let mut response = client.get(parsed).send().await.map_err(|err| describe(&err))?; + + // Status before body. An expired presigned URL is a 403 with an XML + // error document, which must never be mistaken for a project file. + let status = response.status(); + if !status.is_success() { + let hint = if status == reqwest::StatusCode::FORBIDDEN { + " (the presigned URL has probably expired)" + } else { + "" + }; + return Err(format!("server answered HTTP {status}{hint}")); + } + + let expected = response.content_length(); + if let Some(len) = expected.filter(|len| *len > max_bytes) { + return Err(format!("file is {len} bytes, over the {max_bytes} byte limit")); + } + + // Stream into a buffer so the cap holds even when the server sends no + // Content-Length or lies about it. + let mut body = Vec::with_capacity(expected.unwrap_or(0).min(max_bytes) as usize); + while let Some(chunk) = response.chunk().await.map_err(|err| describe(&err))? { + if body.len() as u64 + chunk.len() as u64 > max_bytes { + return Err(format!("file exceeds the {max_bytes} byte limit")); + } + body.extend_from_slice(&chunk); + } + + // A `.3mf` is a zip: a truncated one has lost its central directory and + // is unreadable, so a short body is an error, not a smaller file. + if let Some(len) = expected.filter(|len| *len != body.len() as u64) { + return Err(format!("received {} bytes but Content-Length said {len}", body.len())); + } + if body.is_empty() { + return Err("server returned an empty body".into()); + } + + Ok(body) +} + +/// reqwest's `Display` for a transport error is just "error sending request"; +/// the useful part (timeout, connect, TLS) is in the flags and source chain. +fn describe(err: &reqwest::Error) -> String { + let kind = if err.is_timeout() { + "timed out" + } else if err.is_connect() { + "could not connect" + } else { + "transfer failed" + }; + let mut msg = format!("{kind}: {err}"); + let mut source = std::error::Error::source(err); + while let Some(cause) = source { + msg.push_str(&format!(": {cause}")); + source = cause.source(); + } + // Never echo the URL back: its query string is the credential. + match err.url() { + Some(url) => msg.replace(url.as_str(), ""), + None => msg, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Checked before any connection is made, so no server is needed. + #[tokio::test] + async fn a_non_https_url_is_refused_before_any_request() { + let err = fetch_job_file("http://example.invalid/plate.3mf?sig=abc").await.unwrap_err(); + assert!(err.contains("non-HTTPS"), "got: {err}"); + assert!(fetch_job_file("not a url").await.is_err()); + } +} diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 783b5af..d1d6e39 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -16,6 +16,7 @@ //! SQLite buffer cut in the simplification pass is the fix, and it is a //! separate piece of work - see PLAN.md). +mod download; mod protocol; pub use protocol::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage}; @@ -116,15 +117,23 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu Some(Ok(Message::Text(text))) => { match serde_json::from_str::(&text) { Ok(ServerMessage::Command { command_id, printer_id, command }) => { - let result = dispatch(fleet, &printer_id, command).await; - let msg = match result { - Ok(()) => ClientMessage::CommandResult { + // `start_print` is the one command that cannot + // be answered inline. See `spawn_long_command`. + let msg = if command.runs_for_minutes() { + spawn_long_command(fleet.clone(), printer_id.clone(), command); + ClientMessage::CommandResult { command_id, ok: true, error: None, - }, - Err(err) => { - warn!(%printer_id, %err, "command failed"); - ClientMessage::CommandResult { - command_id, ok: false, error: Some(err), + } + } else { + match dispatch(fleet, &printer_id, command).await { + Ok(()) => ClientMessage::CommandResult { + command_id, ok: true, error: None, + }, + Err(err) => { + warn!(%printer_id, %err, "command failed"); + ClientMessage::CommandResult { + command_id, ok: false, error: Some(err), + } } } }; @@ -206,10 +215,69 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec { out } +/// Runs a command that takes minutes, off this connection's task, and answers +/// the cloud immediately. +/// +/// Everything else here is an MQTT publish that returns in milliseconds, so +/// awaiting it inline is fine. `start_print` is not: it downloads tens of +/// megabytes and then transfers them to a printer over FTPS at a measured +/// 0.2 MiB/s, which is about four minutes for a 50 MB project. Awaited +/// inline, that blocks this whole `select!` loop - no heartbeats and, far +/// worse, **no telemetry** for the duration. +/// +/// No telemetry is the one that makes this mandatory rather than tidy. The +/// cloud derives a job's state from the telemetry that follows a start, so a +/// gateway that goes silent for four minutes during every job start is +/// withholding exactly the information the job depends on, while the backend +/// is also timing the command out after ten seconds and concluding the start +/// failed. +/// +/// So the `ok` this produces means "accepted, and working on it" - which is +/// all the cloud ever trusted a start reply to mean anyway. A failure after +/// this point is reported by the job never reaching `printing`, which the +/// backend reaps; it is not reported as a `CommandResult`, because answering +/// later would need an outbound channel this loop does not have yet. That is +/// the next thing to build here, and until it exists the gateway's log is +/// where the reason lives. +fn spawn_long_command(fleet: SharedFleet, printer_id: String, command: PrinterCommand) { + tokio::spawn(async move { + match dispatch(&fleet, &printer_id, command).await { + Ok(()) => info!(%printer_id, "long-running command finished"), + // Loudly, and at error level: nothing else will say why. The + // cloud sees only a job that never started printing. + Err(err) => tracing::error!( + %printer_id, %err, + "long-running command failed after the cloud was already told it was accepted" + ), + } + }); +} + /// Finds the printer by our own id and runs the command. The error is a /// `String` rather than a typed error because its only destination is a /// `CommandResult` going back to the cloud as text. async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand) -> Result<(), String> { + // Fetched before the fleet lock is taken: a download is bounded by + // minutes of network time, and holding the lock across it would stall + // telemetry for every other printer. The transfer to the printer that + // follows has to hold it (it needs `&mut` access), but that is no reason + // to hold it through this part too. A `start_print` for a printer that + // does not exist is therefore found out after the download rather than + // before; the lookup below is cheap, so it is done first for that case. + let downloaded = match &command { + PrinterCommand::StartPrint { url, .. } => { + if !fleet.lock().await.iter().any(|p| p.base().id == printer_id) { + return Err(format!("no printer with id '{printer_id}' in this gateway's fleet")); + } + Some( + download::fetch_job_file(url) + .await + .map_err(|err| format!("download failed (nothing was sent to the printer): {err}"))?, + ) + } + _ => None, + }; + let mut fleet = fleet.lock().await; let printer = fleet .iter_mut() @@ -228,5 +296,19 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand PrinterCommand::ControlPrint { control } => { printer.control_print(control).await.map_err(|err| err.to_string()) } + PrinterCommand::StartPrint { job_id, file_name, .. } => { + let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?; + let byte_count = bytes.len(); + info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer"); + // One call covers deriving the filament mapping, the FTPS + // transfer, the truncation check and the start, and reports them + // all as one error - so the stage here is "printer", not finer. + // The printer's own message says which part (mapping, upload of + // the file, or start) it was. + printer + .upload_project_file(file_name, bytes) + .await + .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}")) + } } } diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 38f1d76..72f79f0 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -118,14 +118,10 @@ pub enum ServerMessage { /// The closed set of things the cloud is allowed to ask a printer to do. /// -/// Deliberately small, and deliberately missing the obvious one: there is no -/// "start a print" variant. `BambuPrinter::start_uploaded_file` has now been -/// confirmed by hand on a real P1P (2026-10-11), so the original reason for -/// leaving it out - never having been run against hardware - is spent. What -/// is still missing is the path around it: nothing uploads a sliced file to a -/// gateway, and nothing decides which printer should run which job. A start -/// variant belongs here once there is something to start. -#[derive(Debug, Clone, Deserialize)] +/// Deliberately small. `StartPrint` is the newest and the only one that +/// reaches outside the gateway (it fetches a URL), so it is the one with +/// bounds on it - see `download.rs`. +#[derive(Clone, Deserialize)] #[serde(tag = "action", rename_all = "snake_case")] pub enum PrinterCommand { /// 0-100. Each vendor converts to whatever its own mechanism wants (for @@ -145,4 +141,63 @@ pub enum PrinterCommand { /// stop during `PREPARE`/`RUNNING` moves the printer to `FAILED`, which /// is how Bambu reports an operator-cancelled print. ControlPrint { control: PrintControl }, + /// Download a job's file, put it on the printer and start it. + /// + /// A successful result means the gateway *accepted* the job, not that + /// anything is printing: a real run on a P1P was accepted, uploaded + /// byte-perfect, reached RUNNING, then paused itself on a filament + /// fault. Job state must follow the telemetry, never this reply. + /// + /// There is no AMS or filament field on purpose. The mapping is read out + /// of the archive's own `Metadata/slice_info.config` by + /// `upload_project_file`, because only the archive knows what each plate + /// was sliced for - guessing it is what heated a P1P to 250C and paused + /// it at 1%. + /// + /// `url` is a presigned, expiring GET, minted at dispatch time (a queued + /// job may wait days). It is a URL and not the bytes so the file stays + /// off the JSON control channel and object-store credentials stay out of + /// every gateway. It is a bearer credential, which is why `Debug` below + /// hides it. `file_name` is what the file is called on the printer; + /// `.3mf` is treated as a sliced project, anything else as bare gcode. + /// `job_id` is only for lining gateway logs up against a cloud job. + StartPrint { job_id: String, file_name: String, url: String }, +} + +impl PrinterCommand { + /// Whether this command takes long enough that it must not be awaited on + /// the uplink's own task. + /// + /// The threshold is not arbitrary: everything but `StartPrint` is a single + /// MQTT publish, while `StartPrint` downloads a project and then transfers + /// it to a printer at a measured 0.2 MiB/s - minutes, during which the + /// uplink would send no telemetry at all. See `spawn_long_command`. + pub fn runs_for_minutes(&self) -> bool { + // Written as an exhaustive match rather than `matches!`, so a new + // variant has to make this decision rather than defaulting to + // "answer it inline" and quietly reintroducing the stall. + match self { + Self::StartPrint { .. } => true, + Self::SetFanSpeed { .. } | Self::SendGcodeLine { .. } | Self::ControlPrint { .. } => false, + } + } +} + +/// Hand-written so the presigned `url` never reaches a log line: `dispatch` +/// logs every command with `{:?}`, and the query string of that URL is +/// enough to download the job file until it expires. +impl std::fmt::Debug for PrinterCommand { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::SetFanSpeed { percent } => f.debug_struct("SetFanSpeed").field("percent", percent).finish(), + Self::SendGcodeLine { line } => f.debug_struct("SendGcodeLine").field("line", line).finish(), + Self::ControlPrint { control } => f.debug_struct("ControlPrint").field("control", control).finish(), + Self::StartPrint { job_id, file_name, .. } => f + .debug_struct("StartPrint") + .field("job_id", job_id) + .field("file_name", file_name) + .field("url", &"") + .finish(), + } + } } diff --git a/tests/uplink_protocol.rs b/tests/uplink_protocol.rs index 2d2ead3..60fb2f4 100644 --- a/tests/uplink_protocol.rs +++ b/tests/uplink_protocol.rs @@ -98,18 +98,59 @@ fn a_vendor_specific_control_verb_is_rejected() { } } -/// The variant that is deliberately absent - still, but no longer for the -/// original reason. The print-start path *has* now been confirmed on a real -/// P1P (2026-10-11); what is missing is the path around it, since nothing -/// uploads a sliced file to a gateway and nothing decides which printer runs -/// which job. Delete this test when there is something to start. +/// The schema's `StartPrint` variant, in the shape the backend dispatches. +/// Every field has to arrive intact: `url` is what gets downloaded and +/// `file_name` is what the printer is told to call it, so a swapped or +/// dropped one prints the wrong thing or nothing. #[test] -fn there_is_no_start_print_command() { - let raw = r#"{"type":"command","command_id":"x","printer_id":"p1p","command":{"action":"start_print","file_name":"plate_1.3mf"}}"#; - assert!( - serde_json::from_str::(raw).is_err(), - "a start-print command parsed - is there a job path to drive it yet?" - ); +fn a_start_print_command_arrives_with_its_fields_intact() { + let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"5b0e7c1a-0000-4000-8000-000000000000","file_name":"plate_1.3mf","url":"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"}}"#; + match serde_json::from_str::(raw).expect("should parse") { + ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url }, .. } => { + assert_eq!(job_id, "5b0e7c1a-0000-4000-8000-000000000000"); + assert_eq!(file_name, "plate_1.3mf"); + assert_eq!( + url, + "https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300" + ); + } + other => panic!("wrong variant: {other:?}"), + } +} + +/// The URL is a bearer credential until it expires, and `dispatch` logs +/// every command with `{:?}`. +#[test] +fn a_start_print_url_does_not_appear_in_debug_output() { + let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"j","file_name":"plate_1.3mf","url":"https://objects.example.com/x?X-Amz-Signature=deadbeef"}}"#; + let parsed = serde_json::from_str::(raw).expect("should parse"); + let shown = format!("{parsed:?}"); + assert!(!shown.contains("deadbeef"), "signature leaked into a log line: {shown}"); + assert!(shown.contains("plate_1.3mf"), "the file name should still be logged: {shown}"); +} + +/// Each field is load-bearing and none has a sensible default: a command +/// missing its URL must be rejected here, not discovered at the printer. +#[test] +fn a_start_print_missing_any_required_field_is_rejected() { + let full = [ + ("job_id", r#""job_id":"5b0e7c1a-0000-4000-8000-000000000000""#), + ("file_name", r#""file_name":"plate_1.3mf""#), + ("url", r#""url":"https://objects.example.com/jobs/abc.3mf""#), + ]; + for missing in 0..full.len() { + let fields: Vec<&str> = + full.iter().enumerate().filter(|(i, _)| *i != missing).map(|(_, (_, f))| *f).collect(); + let raw = format!( + r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"start_print",{}}}}}"#, + fields.join(",") + ); + assert!( + serde_json::from_str::(&raw).is_err(), + "start_print without '{}' parsed and should not have:\n {raw}", + full[missing].0 + ); + } } #[test]