diff --git a/.gitignore b/.gitignore index e5e1be0..127f431 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,7 @@ go2rtc.yaml go2rtc.yaml.tmp continuum.toml go2rtc.yaml.pid + +# The local store holds printer access codes. +continuum.db +continuum.db-* diff --git a/Cargo.lock b/Cargo.lock index 6115401..b35d2ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -208,6 +208,7 @@ dependencies = [ "openssl", "reqwest", "rumqttc", + "rusqlite", "serde", "serde_json", "suppaftp", @@ -375,6 +376,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "fastrand" version = "2.5.0" @@ -415,6 +428,12 @@ dependencies = [ "spin", ] +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "foreign-types" version = "0.3.2" @@ -533,6 +552,18 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashlink" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb" +dependencies = [ + "hashbrown", +] [[package]] name = "hmac" @@ -875,6 +906,17 @@ version = "0.2.191" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" +[[package]] +name = "libsqlite3-sys" +version = "0.39.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "14a5f01b87b92705d07b433c06f96cbc764637d81a0bffb5734ce17cf0f9a2b3" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1369,6 +1411,15 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rsqlite-vfs" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "353b74f9810d5b085b6c42c9d86a5f0fdb795553235cdd48753b32e3809d18a7" +dependencies = [ + "thiserror 2.0.21", +] + [[package]] name = "rumqttc" version = "0.25.1" @@ -1388,6 +1439,21 @@ dependencies = [ "tokio-util", ] +[[package]] +name = "rusqlite" +version = "0.41.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da9bba6f3506135e120a6586df05be18b48d18896600646a4bec8f993d91a594" +dependencies = [ + "bitflags 2.13.2", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", + "sqlite-wasm-rs", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -1670,6 +1736,16 @@ dependencies = [ "lock_api", ] +[[package]] +name = "sqlite-wasm-rs" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "36bd8e29bf08189a005b2df1df99823aa0861cabe9044e5f31ed54ca292ee618" +dependencies = [ + "cc", + "rsqlite-vfs", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" diff --git a/Cargo.toml b/Cargo.toml index d425b83..df357ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -92,3 +92,9 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls- # `bind_ssdp`). Windows PCs routinely have VPN/Hyper-V/WSL adapters. # MIT OR BSD-3-Clause - checked against docs/LICENSES.md before adding. if-addrs = "0.15" +# The proxy's local store (src/store.rs): printers added from the console live +# here, so the gateway keeps its fleet with the API unreachable. `bundled` +# compiles SQLite from source so the appliance image needs no system library +# (SQLite itself is public domain). +# MIT - checked with `cargo deny check licenses` (docs/LICENSES.md policy). +rusqlite = { version = "0.41", features = ["bundled"] } diff --git a/src/config.rs b/src/config.rs index 5148b7f..ba51aee 100644 --- a/src/config.rs +++ b/src/config.rs @@ -76,6 +76,9 @@ pub struct Config { /// `BambuPrinter::ensure_trusted`) lives entirely in that file now, /// not as a separate fleet-wide env-configured policy here. pub printers_path: String, + /// The gateway's own database (src/store.rs): printers added from the + /// console, credentials included, so the file is created owner-only. + pub store_path: String, /// How, and whether, to look for printers nobody has configured. /// /// Off unless switched on - see [`DiscoverySettings`] for why that is @@ -104,6 +107,7 @@ pub struct ProxyFile { pub go2rtc_api_listen: Option, pub stream_base_url: Option, pub printers_path: Option, + pub store_path: Option, /// Only meaningful on a Proxy Box; see the module docs for why it is /// exposed and not applied. pub network: Option, @@ -326,6 +330,7 @@ impl Config { .or_else(|| file.stream_base_url.clone()) .filter(|v| !v.is_empty()), printers_path: or_default("CONTINUUM_PRINTERS_PATH", &file.printers_path, "printers.toml"), + store_path: or_default("CONTINUUM_STORE_PATH", &file.store_path, "continuum.db"), discovery: discovery_settings(file.discovery.as_ref(), env)?, }) } diff --git a/src/discovery/report.rs b/src/discovery/report.rs index 09efd75..f807e9c 100644 --- a/src/discovery/report.rs +++ b/src/discovery/report.rs @@ -32,7 +32,12 @@ use crate::uplink::ClientMessage; /// long-running tasks. With discovery off it parks on a future that never /// completes, which costs nothing and keeps the "every arm runs forever" /// property of that `select!` true. -pub async fn run(settings: DiscoverySettings, printers_path: String, outbound: Sender) { +pub async fn run( + settings: DiscoverySettings, + printers_path: String, + store: Option>, + outbound: Sender, +) { if !settings.enabled { tracing::info!("printer discovery is off (discovery.enabled); not scanning this network"); std::future::pending::<()>().await; @@ -64,7 +69,20 @@ pub async fn run(settings: DiscoverySettings, printers_path: String, outbound: S // restart - but someone who has *edited* the file and not restarted // should not keep being offered the printer they just configured. let configured = match crate::fleet::configured(std::path::Path::new(&printers_path)) { - Ok(configured) => configured, + Ok(mut configured) => { + // Printers added from the console live in the store, not the + // file, and are just as "already configured". + if let Some(store) = &store { + match store.printers() { + Ok(stored) => configured.add_entries(stored), + Err(err) => { + tracing::warn!(%err, "cannot read the local store; skipping this discovery pass"); + continue; + } + } + } + configured + } // Fail the pass rather than report against an empty set. The // alternative offers every printer the farm already runs as a // fresh find, which is worse than no report: it is a list of diff --git a/src/fleet.rs b/src/fleet.rs index 7a26aaa..eb7c359 100644 --- a/src/fleet.rs +++ b/src/fleet.rs @@ -6,7 +6,7 @@ use std::collections::HashSet; use std::path::Path; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use crate::printer::{ BambuPrinter, BambuProtocolPreference, BambuTls, KlipperPrinter, OctoPrintFlavour, OctoPrintPrinter, @@ -19,10 +19,14 @@ struct FleetFile { printers: Vec, } -#[derive(Debug, Deserialize)] -struct PrinterEntry { - id: String, - name: String, +/// One printer's connection details, as written in `printers.toml` or stored +/// in the local store (src/store.rs). `Debug` is written by hand below: the +/// derived one would print `access_code` and `api_key` into any log line that +/// formats an entry. +#[derive(Clone, Deserialize, Serialize)] +pub struct PrinterEntry { + pub id: String, + pub name: String, /// One of: "bambu" (auto-detects MQTT v5-vs-v3.1.1/BambuMQTT v2-vs-v1 - /// the default, since a firmware update can move the same physical /// printer between them), "bambu_v1"/"bambu_v2" (force a specific one), @@ -32,23 +36,38 @@ struct PrinterEntry { /// flavour - see `OctoPrintFlavour`. Pick `octoprint` if it is OctoPrint /// on a Pi and `prusa_link` if it is an MK4/XL/MINI serving its own API; /// discovery cannot tell, so it suggests `octoprint` and says so. - vendor: String, - host: Option, - access_code: Option, + pub vendor: String, + pub host: Option, + pub access_code: Option, /// The printer's serial number - Bambu's real MQTT topics are addressed /// by serial (`device/{sn}/report`), not by our own `id`. - sn: Option, - api_key: Option, + pub sn: Option, + pub api_key: Option, /// Serial device path, e.g. `/dev/ttyACM0` on Linux. - port: Option, + pub port: Option, #[serde(default = "default_baud_rate")] - baud_rate: u32, + pub baud_rate: u32, /// Path to a printer-specific CA certificate. Leave unset to use the /// bundled Bambu CA - only needed for a printer that doesn't chain to /// it (P1P, at least; see certs/README.md). - ca_cert_path: Option, + pub ca_cert_path: Option, #[serde(default = "default_true")] - require_valid_cert: bool, + pub require_valid_cert: bool, +} + +impl std::fmt::Debug for PrinterEntry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let secret = |value: &Option| value.as_ref().map(|_| ""); + f.debug_struct("PrinterEntry") + .field("id", &self.id) + .field("name", &self.name) + .field("vendor", &self.vendor) + .field("host", &self.host) + .field("access_code", &secret(&self.access_code)) + .field("sn", &self.sn) + .field("api_key", &secret(&self.api_key)) + .finish_non_exhaustive() + } } fn default_true() -> bool { @@ -72,6 +91,26 @@ pub fn load(path: &Path) -> anyhow::Result> { file.printers.into_iter().map(build_printer).collect() } +/// The entries in a printers file, or none if there is no file. +/// +/// A missing file is normal now: a gateway whose printers were all added from +/// the console has only the store. A file that exists but does not parse is +/// still an error - that is a typo to surface, not a fleet to ignore. +pub fn entries(path: &Path) -> anyhow::Result> { + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(err) => return Err(anyhow::anyhow!("failed to read {path:?}: {err}")), + }; + Ok(toml::from_str::(&text)?.printers) +} + +/// Builds the client for one entry. Public so a printer added while the +/// gateway is running goes through exactly the checks a startup one does. +pub fn build(entry: PrinterEntry) -> anyhow::Result { + build_printer(entry) +} + /// Everything the fleet file already names, for telling a printer the /// operator has dealt with from one they have never seen. /// @@ -119,7 +158,13 @@ pub fn configured(path: &Path) -> anyhow::Result { }; let file: FleetFile = toml::from_str(&text)?; let mut out = Configured::default(); - for entry in file.printers { + out.add_entries(file.printers); + Ok(out) +} + +impl Configured { + /// Folds more entries in - the file's, and the local store's. + pub fn add_entries(&mut self, entries: impl IntoIterator) { let add = |set: &mut HashSet, value: Option| { if let Some(value) = value { let value = value.trim().to_lowercase(); @@ -128,11 +173,12 @@ pub fn configured(path: &Path) -> anyhow::Result { } } }; - add(&mut out.hosts, entry.host); - add(&mut out.serials, entry.sn); - add(&mut out.ports, entry.port); + for entry in entries { + add(&mut self.hosts, entry.host); + add(&mut self.serials, entry.sn); + add(&mut self.ports, entry.port); + } } - Ok(out) } fn build_printer(entry: PrinterEntry) -> anyhow::Result { diff --git a/src/lib.rs b/src/lib.rs index bd4a29c..9c63491 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,4 +16,6 @@ pub mod config_import; pub mod discovery; pub mod fleet; pub mod printer; +pub mod provision; +pub mod store; pub mod uplink; diff --git a/src/main.rs b/src/main.rs index ab23b47..a3673d0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -68,7 +68,30 @@ async fn main() -> anyhow::Result<()> { // Loaded and connected here, once, at startup - nothing yet re-reads // printers.toml on a running daemon, so adding or removing a printer // still means a restart. - let mut fleet = fleet::load(std::path::Path::new(&config.printers_path))?; + // Two sources, one fleet: `printers.toml` (a hand-written file - strict, + // a typo stops startup) and the local store (printers added from the + // console, which were checked before they were saved). The store is + // opened first and is what lets the gateway come up with no file at all. + let store = std::sync::Arc::new(continuum_proxy::store::Store::open(std::path::Path::new(&config.store_path))?); + let mut fleet = Vec::new(); + for entry in fleet::entries(std::path::Path::new(&config.printers_path))? { + fleet.push(fleet::build(entry)?); + } + for entry in store.printers()? { + if fleet.iter().any(|printer| printer.base().id == entry.id) { + tracing::warn!(printer = %entry.id, "in both printers.toml and the local store; using printers.toml"); + continue; + } + // Skipped, not fatal: one entry the build no longer accepts should + // not keep the rest of a farm offline. + match fleet::build(entry) { + Ok(printer) => fleet.push(printer), + Err(err) => tracing::error!(%err, "a printer in the local store could not be built; skipping it"), + } + } + if fleet.is_empty() { + tracing::warn!("no printers yet - add one from the console, or list some in printers.toml"); + } connect_fleet(&mut fleet).await; // The fleet is now shared with the uplink rather than just parked here @@ -134,10 +157,11 @@ async fn main() -> anyhow::Result<()> { _ = shutdown_requested() => { info!("stop requested; exiting so go2rtc is stopped with us"); } - _ = uplink::run(config.clone(), fleet.clone(), outbound.clone(), inbox) => {} + _ = uplink::run(config.clone(), fleet.clone(), Some(store.clone()), outbound.clone(), inbox) => {} _ = continuum_proxy::discovery::report::run( config.discovery.clone(), config.printers_path.clone(), + Some(store.clone()), outbound.clone(), ) => {} _ = go2rtc::watchdog( diff --git a/src/provision.rs b/src/provision.rs new file mode 100644 index 0000000..f2881e5 --- /dev/null +++ b/src/provision.rs @@ -0,0 +1,212 @@ +//! Adding and removing printers on a running gateway, on the console's say-so. +//! +//! The access code arrives in a `provision_printer` message, is checked by +//! actually connecting, and is then written to the local store (src/store.rs). +//! It is never logged and never sent anywhere. The cloud does not keep it - +//! it relayed it - so this is the only place it exists. +//! +//! Connect first, store second: a printer that cannot be reached with the +//! code the operator typed is a typo to report now, not a row to discover +//! later in a fleet that quietly never comes up. + +use std::time::Duration; + +use tracing::info; + +use crate::fleet; +use crate::printer::GenericPrinter; +use crate::store::Store; +use crate::uplink::SharedFleet; +use crate::uplink::protocol::ProvisionPrinter; + +/// Bambu's protocol auto-detection can spend several seconds on a first +/// attempt, and an operator is watching a spinner. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); + +/// Ids end up in MQTT topics, file names (pinned certificates) and log +/// lines, so they are held to a small safe alphabet. +fn valid_id(id: &str) -> bool { + !id.is_empty() && id.len() <= 64 && id.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') +} + +pub async fn add(fleet: &SharedFleet, store: &Store, request: ProvisionPrinter) -> Result<(), String> { + if !valid_id(&request.id) { + return Err("printer id must be 1-64 letters, digits, '-' or '_'".into()); + } + if fleet.lock().await.iter().any(|p| p.base().id == request.id) { + return Err(format!("this gateway already has a printer with id '{}'", request.id)); + } + + let entry = fleet::PrinterEntry { + id: request.id, + name: request.name, + vendor: request.vendor, + host: request.host, + access_code: request.access_code, + sn: request.sn, + api_key: request.api_key, + port: None, + baud_rate: 115_200, + ca_cert_path: None, + require_valid_cert: true, + }; + + // Built (which checks the required fields are present and non-blank) and + // connected before the fleet lock is taken: connecting can take seconds, + // and the lock is what telemetry waits on. + let mut printer = fleet::build(entry.clone()).map_err(|err| err.to_string())?; + match tokio::time::timeout(CONNECT_TIMEOUT, printer.connect()).await { + Ok(Ok(())) => {} + Ok(Err(err)) => return Err(format!("could not connect to the printer: {err}")), + Err(_) => return Err("the printer did not answer in time - check the address and that it is on".into()), + } + + store.save_printer(&entry).map_err(|err| format!("connected, but could not save it locally: {err}"))?; + info!(printer = %entry.id, "printer added from the console"); + + let mut fleet = fleet.lock().await; + // Re-checked under the lock: two requests for one id can both pass the + // check above while the first is still connecting. + if fleet.iter().any(|p| p.base().id == entry.id) { + return Err(format!("this gateway already has a printer with id '{}'", entry.id)); + } + fleet.push(printer); + Ok(()) +} + +pub async fn remove(fleet: &SharedFleet, store: &Store, printer_id: &str) -> Result<(), String> { + let stored = store.remove_printer(printer_id).map_err(|err| err.to_string())?; + let mut fleet = fleet.lock().await; + let before = fleet.len(); + fleet.retain(|p| p.base().id != printer_id); + if !stored && fleet.len() == before { + return Err(format!("no printer with id '{printer_id}' on this gateway")); + } + info!(printer = %printer_id, "printer removed from the console"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ids_are_a_small_safe_alphabet() { + assert!(valid_id("p1p")); + assert!(valid_id("Printer_2-b")); + assert!(!valid_id("")); + assert!(!valid_id("../etc/passwd")); + assert!(!valid_id("has space")); + assert!(!valid_id(&"a".repeat(65))); + } +} + +#[cfg(test)] +mod behaviour { + use std::sync::Arc; + + use super::*; + + fn request(id: &str) -> ProvisionPrinter { + ProvisionPrinter { + id: id.into(), + name: "Bench P1".into(), + vendor: "bambu".into(), + // Nothing listens here, so a connect attempt is refused at once. + host: Some("127.0.0.1".into()), + access_code: Some("12345678".into()), + sn: Some("01P00A000000001".into()), + api_key: None, + } + } + + fn store() -> (Store, std::path::PathBuf) { + let path = std::env::temp_dir().join(format!("continuum-provision-test-{}-{}.db", std::process::id(), rand_suffix())); + (Store::open(&path).unwrap(), path) + } + + fn rand_suffix() -> u128 { + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() + } + + fn empty_fleet() -> SharedFleet { + Arc::new(tokio::sync::Mutex::new(Vec::new())) + } + + #[tokio::test] + async fn a_bad_id_is_refused_before_anything_happens() { + let (store, path) = store(); + let err = add(&empty_fleet(), &store, request("../x")).await.unwrap_err(); + assert!(err.contains("printer id"), "{err}"); + assert!(store.printers().unwrap().is_empty()); + let _ = std::fs::remove_file(path); + } + + #[tokio::test] + async fn a_blank_credential_is_named_and_nothing_is_saved() { + let (store, path) = store(); + let mut req = request("p1"); + req.access_code = Some(" ".into()); + let err = add(&empty_fleet(), &store, req).await.unwrap_err(); + assert!(err.contains("access_code"), "{err}"); + assert!(store.printers().unwrap().is_empty()); + let _ = std::fs::remove_file(path); + } + + #[tokio::test] + async fn a_printer_that_cannot_be_reached_is_not_saved() { + let (store, path) = store(); + let fleet = empty_fleet(); + let err = add(&fleet, &store, request("p1")).await.unwrap_err(); + assert!(err.contains("could not connect") || err.contains("did not answer"), "{err}"); + assert!(store.printers().unwrap().is_empty(), "a failed add must not leave a stored printer"); + assert!(fleet.lock().await.is_empty()); + let _ = std::fs::remove_file(path); + } + + #[tokio::test] + async fn the_error_never_contains_the_access_code() { + let (store, path) = store(); + let err = add(&empty_fleet(), &store, request("p1")).await.unwrap_err(); + assert!(!err.contains("12345678"), "{err}"); + let _ = std::fs::remove_file(path); + } + + #[tokio::test] + async fn removing_forgets_a_printer_in_both_places() { + let (store, path) = store(); + let entry = crate::fleet::PrinterEntry { + id: "p1".into(), + name: "Bench".into(), + vendor: "bambu".into(), + host: Some("127.0.0.1".into()), + access_code: Some("12345678".into()), + sn: Some("01P00A000000001".into()), + api_key: None, + port: None, + baud_rate: 115_200, + ca_cert_path: None, + require_valid_cert: true, + }; + store.save_printer(&entry).unwrap(); + let fleet = empty_fleet(); + fleet.lock().await.push(crate::fleet::build(entry).unwrap()); + + remove(&fleet, &store, "p1").await.unwrap(); + assert!(fleet.lock().await.is_empty()); + assert!(store.printers().unwrap().is_empty()); + + let err = remove(&fleet, &store, "p1").await.unwrap_err(); + assert!(err.contains("no printer"), "{err}"); + let _ = std::fs::remove_file(path); + } + + #[test] + fn the_wire_message_parses_and_its_debug_hides_the_credential() { + let json = r#"{"type":"provision_printer","request_id":"r1","printer":{"id":"p1","name":"Bench","vendor":"bambu","host":"10.0.0.5","access_code":"98765432","sn":"SN1"}}"#; + let message: crate::uplink::ServerMessage = serde_json::from_str(json).unwrap(); + let shown = format!("{message:?}"); + assert!(shown.contains("p1") && shown.contains(""), "{shown}"); + assert!(!shown.contains("98765432"), "the access code reached a Debug string: {shown}"); + } +} diff --git a/src/store.rs b/src/store.rs new file mode 100644 index 0000000..082b84f --- /dev/null +++ b/src/store.rs @@ -0,0 +1,199 @@ +//! The gateway's own database: printers an operator added from the console. +//! +//! Why the gateway keeps these itself and not the cloud: a printer's access +//! code is a credential for a machine on someone's LAN, and the platform's +//! rule is that those never rest in the cloud. The console sends the code +//! once, over the uplink, and this is where it lands. It also means the +//! fleet survives the API being unreachable - the gateway starts, connects +//! its printers and keeps reporting locally from this alone. +//! +//! SQLite, one file. The file holds credentials, so on Unix it is created +//! owner-only (0600) before SQLite touches it; SQLite gives its journal files +//! the same mode as the main one. +//! +//! `printers.toml` still works and is read alongside this; see `main.rs`. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::Context; +use rusqlite::{Connection, params}; + +use crate::fleet::PrinterEntry; + +pub struct Store { + // rusqlite's `Connection` is not `Sync`, so it cannot be shared between + // tasks directly. Wrapping it in a `Mutex` is the plain fix: one caller + // at a time, which is plenty for a few writes a day. + conn: Mutex, +} + +impl Store { + pub fn open(path: &Path) -> anyhow::Result { + if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) { + std::fs::create_dir_all(dir).with_context(|| format!("creating {dir:?} for the local store"))?; + } + create_private(path)?; + let conn = Connection::open(path).with_context(|| format!("opening the local store at {path:?}"))?; + // Each printer is a JSON document keyed by its id. The id is the + // gateway's own printer id - the one telemetry is reported under - + // so "which printer" has a single answer everywhere. + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS printers ( + id TEXT PRIMARY KEY, + json TEXT NOT NULL + );", + )?; + Ok(Self { conn: Mutex::new(conn) }) + } + + /// Adds a printer, or replaces the one with the same id. + pub fn save_printer(&self, entry: &PrinterEntry) -> anyhow::Result<()> { + let json = serde_json::to_string(entry)?; + let conn = self.conn.lock().expect("store lock poisoned"); + conn.execute( + "INSERT INTO printers (id, json) VALUES (?1, ?2) + ON CONFLICT(id) DO UPDATE SET json = excluded.json", + params![entry.id, json], + )?; + Ok(()) + } + + /// Forgets a printer. True if there was one. + pub fn remove_printer(&self, id: &str) -> anyhow::Result { + let conn = self.conn.lock().expect("store lock poisoned"); + Ok(conn.execute("DELETE FROM printers WHERE id = ?1", params![id])? > 0) + } + + /// Every stored printer, oldest first. + pub fn printers(&self) -> anyhow::Result> { + let conn = self.conn.lock().expect("store lock poisoned"); + let mut statement = conn.prepare("SELECT json FROM printers ORDER BY rowid")?; + let rows = statement.query_map([], |row| row.get::<_, String>(0))?; + let mut out = Vec::new(); + for json in rows { + out.push(serde_json::from_str(&json?).context("a stored printer no longer parses")?); + } + Ok(out) + } +} + +/// Creates the file empty and owner-only if it does not exist yet. +#[cfg(unix)] +fn create_private(path: &Path) -> anyhow::Result<()> { + use std::os::unix::fs::OpenOptionsExt; + std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(false) + .mode(0o600) + .open(path) + .with_context(|| format!("creating the local store {path:?}"))?; + Ok(()) +} + +/// Windows: the file inherits its folder's permissions, which for the +/// per-user data directory is already that user's. Nothing to set here. +#[cfg(not(unix))] +fn create_private(_path: &Path) -> anyhow::Result<()> { + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(id: &str, access_code: &str) -> PrinterEntry { + PrinterEntry { + id: id.into(), + name: format!("Printer {id}"), + vendor: "bambu".into(), + host: Some("192.168.0.50".into()), + access_code: Some(access_code.into()), + sn: Some("01P00A000000000".into()), + api_key: None, + port: None, + baud_rate: 115_200, + ca_cert_path: None, + require_valid_cert: true, + } + } + + /// A throwaway directory, removed when dropped. Hand-rolled rather than + /// pulling in a crate for four tests. + struct TempDir(std::path::PathBuf); + + impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + fn temp_db() -> (TempDir, std::path::PathBuf) { + use std::sync::atomic::{AtomicU32, Ordering}; + static NEXT: AtomicU32 = AtomicU32::new(0); + let dir = std::env::temp_dir().join(format!( + "continuum-store-test-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + let path = dir.join("nested").join("continuum.db"); + (TempDir(dir), path) + } + + #[test] + fn what_was_saved_comes_back_after_reopening() { + let (_dir, path) = temp_db(); + Store::open(&path).unwrap().save_printer(&entry("a", "11111111")).unwrap(); + + let reopened = Store::open(&path).unwrap(); + let printers = reopened.printers().unwrap(); + assert_eq!(printers.len(), 1); + assert_eq!(printers[0].id, "a"); + assert_eq!(printers[0].access_code.as_deref(), Some("11111111")); + assert_eq!(printers[0].sn.as_deref(), Some("01P00A000000000")); + } + + #[test] + fn saving_the_same_id_replaces_it_and_order_is_oldest_first() { + let (_dir, path) = temp_db(); + let store = Store::open(&path).unwrap(); + store.save_printer(&entry("a", "old")).unwrap(); + store.save_printer(&entry("b", "bbb")).unwrap(); + store.save_printer(&entry("a", "new")).unwrap(); + + let printers = store.printers().unwrap(); + let ids: Vec<_> = printers.iter().map(|p| p.id.as_str()).collect(); + assert_eq!(ids, ["a", "b"]); + assert_eq!(printers[0].access_code.as_deref(), Some("new")); + } + + #[test] + fn removing_reports_whether_anything_was_there() { + let (_dir, path) = temp_db(); + let store = Store::open(&path).unwrap(); + store.save_printer(&entry("a", "x")).unwrap(); + assert!(store.remove_printer("a").unwrap()); + assert!(!store.remove_printer("a").unwrap()); + assert!(store.printers().unwrap().is_empty()); + } + + #[cfg(unix)] + #[test] + fn the_file_holding_credentials_is_owner_only() { + use std::os::unix::fs::PermissionsExt; + let (_dir, path) = temp_db(); + let store = Store::open(&path).unwrap(); + store.save_printer(&entry("a", "secret")).unwrap(); + let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o600); + } + + #[test] + fn a_corrupt_file_is_an_error_not_an_empty_fleet() { + let (_dir, path) = temp_db(); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(&path, b"this is not a sqlite database at all, just text").unwrap(); + assert!(Store::open(&path).is_err()); + } +} diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index ec3aec9..30165f8 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -28,7 +28,7 @@ mod download; mod webrtc; -mod protocol; +pub mod protocol; pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage}; @@ -42,6 +42,9 @@ use tokio_tungstenite::tungstenite::handshake::client::{Request, generate_key}; use tracing::{info, warn}; use crate::config::Config; +use crate::provision; +use crate::store::Store; +use protocol::ProvisionPrinter; use crate::printer::{GenericPrinter, PrinterHandle}; /// The websocket, once connected. Named because `connect` has to hand it to @@ -139,6 +142,7 @@ pub fn channel() -> ( pub async fn run( config: Config, fleet: SharedFleet, + store: Option>, outbound: tokio::sync::mpsc::Sender, mut inbox: tokio::sync::mpsc::Receiver, ) { @@ -156,7 +160,7 @@ pub async fn run( // would not distinguish a connection that lived for an hour // from one that failed to open. delay = RETRY_DELAY; - if let Err(err) = serve(ws_stream, &config, &fleet, &outbound, &mut inbox).await { + if let Err(err) = serve(ws_stream, &config, &fleet, store.as_ref(), &outbound, &mut inbox).await { warn!(?err, "uplink connection dropped, reconnecting in {delay:?}"); } } @@ -200,6 +204,7 @@ async fn serve( ws_stream: WsStream, config: &Config, fleet: &SharedFleet, + store: Option<&Arc>, outbound: &tokio::sync::mpsc::Sender, inbox: &mut tokio::sync::mpsc::Receiver, ) -> anyhow::Result<()> { @@ -306,13 +311,25 @@ async fn serve( outbound.clone(), ); } + // Spawned: connecting to the new printer can take + // many seconds, and awaiting it here would stall + // heartbeats and telemetry for the whole farm. + Ok(ServerMessage::ProvisionPrinter { request_id, printer }) => { + spawn_provision(fleet.clone(), store.cloned(), request_id, Some(printer), outbound.clone()); + } + Ok(ServerMessage::RemovePrinter { request_id, printer_id }) => { + spawn_remove(fleet.clone(), store.cloned(), request_id, printer_id, outbound.clone()); + } Ok(other) => info!(?other, "received from cloud"), // Loudly. This used to be an `if let Ok(..)` with // no else, which is the same silent-drop bug that // hid a real `print.push_status` failure for // weeks - and here it would mean commands // vanishing with the cloud never being told. - Err(err) => warn!(%err, payload = %text, "unparseable message from cloud, ignoring"), + // The payload is NOT logged. A `provision_printer` message + // carries an access code, and one that failed to parse is + // exactly the message that would end up here. + Err(err) => warn!(%err, bytes = text.len(), "unparseable message from cloud, ignoring"), } } Some(Ok(Message::Close(_))) | None => return Ok(()), @@ -456,6 +473,54 @@ fn spawn_long_command( }); } +/// Adds a printer off the connection's task and always reports how it went: +/// the console is holding a form open waiting for exactly this. +fn spawn_provision( + fleet: SharedFleet, + store: Option>, + request_id: String, + request: Option, + outbound: tokio::sync::mpsc::Sender, +) { + tokio::spawn(async move { + let result = match (store, request) { + (Some(store), Some(request)) => provision::add(&fleet, &store, request).await, + (None, _) => Err("this gateway has no local store, so it cannot add printers".to_string()), + (_, None) => Err("no printer in the request".to_string()), + }; + report(request_id, result, &outbound).await; + }); +} + +fn spawn_remove( + fleet: SharedFleet, + store: Option>, + request_id: String, + printer_id: String, + outbound: tokio::sync::mpsc::Sender, +) { + tokio::spawn(async move { + let result = match store { + Some(store) => provision::remove(&fleet, &store, &printer_id).await, + None => Err("this gateway has no local store".to_string()), + }; + report(request_id, result, &outbound).await; + }); +} + +async fn report(request_id: String, result: Result<(), String>, outbound: &tokio::sync::mpsc::Sender) { + let msg = match result { + Ok(()) => ClientMessage::CommandResult { command_id: request_id, ok: true, error: None }, + Err(error) => { + warn!(%error, "printer request failed"); + ClientMessage::CommandResult { command_id: request_id, ok: false, error: Some(error) } + } + }; + if outbound.send(msg).await.is_err() { + warn!("uplink is shutting down; the result was not reported"); + } +} + /// Answers one WebRTC offer, off this connection's task. /// /// Always answers, success or failure: the cloud is holding a browser's diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 5957c1b..3c609bb 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -225,6 +225,45 @@ pub enum ServerMessage { /// and checks its password before CORS, so the unauthenticated preflight /// a browser always sends is refused - measured, not assumed. WebrtcOffer { session_id: String, camera_stream: String, sdp: String }, + /// Add a printer to this gateway. Carries the printer's credential, sent + /// once by the console and kept only here - see `provision`. Answered + /// with a `CommandResult` carrying the same id. + ProvisionPrinter { request_id: String, printer: ProvisionPrinter }, + /// Forget a printer. Answered like `ProvisionPrinter`. + RemovePrinter { request_id: String, printer_id: String }, +} + +/// What the console may tell a gateway about a printer to add. +/// +/// Deliberately narrower than `fleet::PrinterEntry`: no `ca_cert_path` and no +/// serial `port`, because both name something on the gateway's own disk or +/// USB bus, and the cloud has no business choosing those. +/// +/// `Debug` is by hand so the credential cannot reach a log line - the +/// `Ok(other) => info!(?other, ...)` arm in the uplink formats whole messages. +#[derive(Clone, Deserialize)] +pub struct ProvisionPrinter { + pub id: String, + pub name: String, + pub vendor: String, + pub host: Option, + pub access_code: Option, + pub sn: Option, + pub api_key: Option, +} + +impl std::fmt::Debug for ProvisionPrinter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ProvisionPrinter") + .field("id", &self.id) + .field("name", &self.name) + .field("vendor", &self.vendor) + .field("host", &self.host) + .field("access_code", &self.access_code.as_ref().map(|_| "")) + .field("sn", &self.sn) + .field("api_key", &self.api_key.as_ref().map(|_| "")) + .finish() + } } /// The closed set of things the cloud is allowed to ask a printer to do. diff --git a/tests/uplink_late_result.rs b/tests/uplink_late_result.rs index 6295c0b..c2d5206 100644 --- a/tests/uplink_late_result.rs +++ b/tests/uplink_late_result.rs @@ -44,6 +44,7 @@ fn config(port: u16) -> Config { go2rtc_api_listen: "127.0.0.1:1984".into(), stream_base_url: None, printers_path: "/nonexistent/printers.toml".into(), + store_path: "/nonexistent/continuum.db".into(), // Off, which is the default and also the only safe thing in a test: // an enabled one would sweep whatever network the test host is on. discovery: Default::default(), @@ -76,7 +77,7 @@ async fn a_long_command_that_fails_reports_its_reason_after_the_fact() { let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new())); let (outbound, inbox) = uplink::channel(); - tokio::spawn(uplink::run(config(port), fleet, outbound, inbox)); + tokio::spawn(uplink::run(config(port), fleet, None, outbound, inbox)); let (stream, _) = listener.accept().await.unwrap(); let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();