Compare commits

..
Author SHA1 Message Date
octoturge b4948076a0 Trade the gateway token for an edge JWT on every connection
Windows discovery tool / windows (push) Successful in 1m24s
The proxy put its configured gateway_token straight into the upgrade's
Authorization header, but the backend only accepts a one-hour edge JWT
(scope "edge") there, minted by POST /auth/edge-token from the farm's edge
proxy token. Nothing in the proxy ever made that call, so a gateway
configured with the token the console hands out could not connect at all,
and one given a hand-minted JWT (which is what every test did) would have
been locked out an hour later on its first reconnect.

Each connection attempt now starts by trading gateway_token for a fresh JWT,
at an address derived from the socket URL so there is one to configure. A
401 is reported as what it means (token rotated or wrong), since retrying
will not fix it. Config's Debug no longer prints the token.

tests/uplink_late_result.rs now plays the cloud's exchange endpoint and
asserts the socket opens with the minted JWT, not the configured token.
.env.example and continuum.example.toml say what the token is.
2026-10-29 15:10:27 +00:00
octoturge 5fc09aef04 Test that an added printer joins the fleet and survives a restart
The OctoPrint client's connect succeeds without hardware, so it stands in
to prove the plumbing of a successful add: saved, joins the running fleet,
refused when added twice, and rebuilt from the store after a restart.
2026-10-29 15:07:44 +00:00
octoturge 9dff600160 Let the console add and remove printers; keep them in a local SQLite store
The only way to give a gateway a printer was to edit printers.toml and
restart it. A printer added from the console now arrives as a
provision_printer message, is checked by actually connecting to it (so a
mistyped access code is reported to the operator, not discovered later),
and only then is saved to the gateway's own SQLite file and joined to the
running fleet - no restart. remove_printer undoes it. Both are answered
with a command_result carrying the request id.

The access code exists only on the gateway: the cloud relays it once and
keeps nothing, which is the platform's rule for printer credentials. The
store file is created owner-only; the message type's Debug and the fleet
entry's Debug redact the credential; and an unparseable message from the
cloud is no longer logged with its payload, because the message that
fails to parse is exactly the one carrying an access code.

The cloud may not choose a certificate path or serial port on the
gateway's disk, so provision_printer cannot carry them.

Startup now reads printers.toml (strict, as before; a missing file is fine)
and the store (a bad stored entry is skipped, not fatal), so the gateway
comes up with its fleet with the API unreachable. Discovery treats stored
printers as already configured. rusqlite (MIT, bundled SQLite) passes
cargo deny.
2026-10-29 15:02:48 +00:00
16 changed files with 940 additions and 34 deletions
+6 -4
View File
@@ -1,10 +1,12 @@
# Identity # Identity
CONTINUUM_GATEWAY_ID=gw-farm01-rpi01 CONTINUUM_GATEWAY_ID=gw-farm01-rpi01
CONTINUUM_FARM_ID=replace-with-farm-uuid CONTINUUM_FARM_ID=replace-with-farm-uuid
# The edge JWT from POST /auth/edge-token. Sent as an Authorization: Bearer # The farm's edge proxy token, shown once by the console's Setup page (or
# header on the uplink's WebSocket upgrade - not in a message and not in the # POST /farms/:id/edge-proxy-token). The gateway trades it for a one-hour edge
# URL - so the backend verifies it before any socket exists. # JWT (POST /auth/edge-token) on every connection, and the JWT - sent as an
CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token # Authorization: Bearer header on the WebSocket upgrade, never in a message
# or the URL - is what the backend verifies before any socket exists.
CONTINUUM_GATEWAY_TOKEN=replace-with-the-farms-edge-proxy-token
# Cloud uplink # Cloud uplink
CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1 CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1
+4
View File
@@ -14,3 +14,7 @@ go2rtc.yaml
go2rtc.yaml.tmp go2rtc.yaml.tmp
continuum.toml continuum.toml
go2rtc.yaml.pid go2rtc.yaml.pid
# The local store holds printer access codes.
continuum.db
continuum.db-*
Generated
+76
View File
@@ -208,6 +208,7 @@ dependencies = [
"openssl", "openssl",
"reqwest", "reqwest",
"rumqttc", "rumqttc",
"rusqlite",
"serde", "serde",
"serde_json", "serde_json",
"suppaftp", "suppaftp",
@@ -375,6 +376,18 @@ dependencies = [
"windows-sys 0.61.2", "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]] [[package]]
name = "fastrand" name = "fastrand"
version = "2.5.0" version = "2.5.0"
@@ -415,6 +428,12 @@ dependencies = [
"spin", "spin",
] ]
[[package]]
name = "foldhash"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
[[package]] [[package]]
name = "foreign-types" name = "foreign-types"
version = "0.3.2" version = "0.3.2"
@@ -533,6 +552,18 @@ name = "hashbrown"
version = "0.17.1" version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" 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]] [[package]]
name = "hmac" name = "hmac"
@@ -875,6 +906,17 @@ version = "0.2.191"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" 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]] [[package]]
name = "linux-raw-sys" name = "linux-raw-sys"
version = "0.12.1" version = "0.12.1"
@@ -1369,6 +1411,15 @@ dependencies = [
"windows-sys 0.52.0", "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]] [[package]]
name = "rumqttc" name = "rumqttc"
version = "0.25.1" version = "0.25.1"
@@ -1388,6 +1439,21 @@ dependencies = [
"tokio-util", "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]] [[package]]
name = "rustc-hash" name = "rustc-hash"
version = "2.1.3" version = "2.1.3"
@@ -1670,6 +1736,16 @@ dependencies = [
"lock_api", "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]] [[package]]
name = "stable_deref_trait" name = "stable_deref_trait"
version = "1.2.1" version = "1.2.1"
+6
View File
@@ -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. # `bind_ssdp`). Windows PCs routinely have VPN/Hyper-V/WSL adapters.
# MIT OR BSD-3-Clause - checked against docs/LICENSES.md before adding. # MIT OR BSD-3-Clause - checked against docs/LICENSES.md before adding.
if-addrs = "0.15" 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"] }
+2
View File
@@ -19,6 +19,8 @@
gateway_id = "REPLACE_ME" gateway_id = "REPLACE_ME"
farm_id = "REPLACE_ME" farm_id = "REPLACE_ME"
# The farm's edge proxy token from the console (Setup page). Traded for a
# short-lived edge JWT on every connection - see src/uplink/auth.rs.
gateway_token = "REPLACE_ME" gateway_token = "REPLACE_ME"
# uplink_url = "wss://api.continuum.local/ws/edge/v1" # uplink_url = "wss://api.continuum.local/ws/edge/v1"
+22 -1
View File
@@ -45,7 +45,7 @@ use std::time::Duration;
use serde::Deserialize; use serde::Deserialize;
#[derive(Debug, Clone)] #[derive(Clone)]
pub struct Config { pub struct Config {
pub gateway_id: String, pub gateway_id: String,
pub farm_id: String, pub farm_id: String,
@@ -76,6 +76,9 @@ pub struct Config {
/// `BambuPrinter::ensure_trusted`) lives entirely in that file now, /// `BambuPrinter::ensure_trusted`) lives entirely in that file now,
/// not as a separate fleet-wide env-configured policy here. /// not as a separate fleet-wide env-configured policy here.
pub printers_path: String, 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. /// How, and whether, to look for printers nobody has configured.
/// ///
/// Off unless switched on - see [`DiscoverySettings`] for why that is /// Off unless switched on - see [`DiscoverySettings`] for why that is
@@ -104,6 +107,7 @@ pub struct ProxyFile {
pub go2rtc_api_listen: Option<String>, pub go2rtc_api_listen: Option<String>,
pub stream_base_url: Option<String>, pub stream_base_url: Option<String>,
pub printers_path: Option<String>, pub printers_path: Option<String>,
pub store_path: Option<String>,
/// Only meaningful on a Proxy Box; see the module docs for why it is /// Only meaningful on a Proxy Box; see the module docs for why it is
/// exposed and not applied. /// exposed and not applied.
pub network: Option<NetworkSettings>, pub network: Option<NetworkSettings>,
@@ -283,6 +287,22 @@ impl fmt::Debug for WifiSettings {
} }
} }
/// By hand: the derived `Debug` would print `gateway_token`, the credential
/// that opens this farm's gateway socket, into any log line that formats the
/// config.
impl fmt::Debug for Config {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Config")
.field("gateway_id", &self.gateway_id)
.field("farm_id", &self.farm_id)
.field("gateway_token", &"<redacted>")
.field("uplink_url", &self.uplink_url)
.field("printers_path", &self.printers_path)
.field("store_path", &self.store_path)
.finish_non_exhaustive()
}
}
impl Config { impl Config {
/// Environment only - the behaviour from before there was a file, kept /// Environment only - the behaviour from before there was a file, kept
/// so existing callers and tests do not change. /// so existing callers and tests do not change.
@@ -326,6 +346,7 @@ impl Config {
.or_else(|| file.stream_base_url.clone()) .or_else(|| file.stream_base_url.clone())
.filter(|v| !v.is_empty()), .filter(|v| !v.is_empty()),
printers_path: or_default("CONTINUUM_PRINTERS_PATH", &file.printers_path, "printers.toml"), 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)?, discovery: discovery_settings(file.discovery.as_ref(), env)?,
}) })
} }
+20 -2
View File
@@ -32,7 +32,12 @@ use crate::uplink::ClientMessage;
/// long-running tasks. With discovery off it parks on a future that never /// long-running tasks. With discovery off it parks on a future that never
/// completes, which costs nothing and keeps the "every arm runs forever" /// completes, which costs nothing and keeps the "every arm runs forever"
/// property of that `select!` true. /// property of that `select!` true.
pub async fn run(settings: DiscoverySettings, printers_path: String, outbound: Sender<ClientMessage>) { pub async fn run(
settings: DiscoverySettings,
printers_path: String,
store: Option<std::sync::Arc<crate::store::Store>>,
outbound: Sender<ClientMessage>,
) {
if !settings.enabled { if !settings.enabled {
tracing::info!("printer discovery is off (discovery.enabled); not scanning this network"); tracing::info!("printer discovery is off (discovery.enabled); not scanning this network");
std::future::pending::<()>().await; 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 // restart - but someone who has *edited* the file and not restarted
// should not keep being offered the printer they just configured. // should not keep being offered the printer they just configured.
let configured = match crate::fleet::configured(std::path::Path::new(&printers_path)) { 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 // Fail the pass rather than report against an empty set. The
// alternative offers every printer the farm already runs as a // alternative offers every printer the farm already runs as a
// fresh find, which is worse than no report: it is a list of // fresh find, which is worse than no report: it is a list of
+65 -19
View File
@@ -6,7 +6,7 @@
use std::collections::HashSet; use std::collections::HashSet;
use std::path::Path; use std::path::Path;
use serde::Deserialize; use serde::{Deserialize, Serialize};
use crate::printer::{ use crate::printer::{
BambuPrinter, BambuProtocolPreference, BambuTls, KlipperPrinter, OctoPrintFlavour, OctoPrintPrinter, BambuPrinter, BambuProtocolPreference, BambuTls, KlipperPrinter, OctoPrintFlavour, OctoPrintPrinter,
@@ -19,10 +19,14 @@ struct FleetFile {
printers: Vec<PrinterEntry>, printers: Vec<PrinterEntry>,
} }
#[derive(Debug, Deserialize)] /// One printer's connection details, as written in `printers.toml` or stored
struct PrinterEntry { /// in the local store (src/store.rs). `Debug` is written by hand below: the
id: String, /// derived one would print `access_code` and `api_key` into any log line that
name: String, /// 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 - /// 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 /// the default, since a firmware update can move the same physical
/// printer between them), "bambu_v1"/"bambu_v2" (force a specific one), /// 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 /// 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; /// 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. /// discovery cannot tell, so it suggests `octoprint` and says so.
vendor: String, pub vendor: String,
host: Option<String>, pub host: Option<String>,
access_code: Option<String>, pub access_code: Option<String>,
/// The printer's serial number - Bambu's real MQTT topics are addressed /// The printer's serial number - Bambu's real MQTT topics are addressed
/// by serial (`device/{sn}/report`), not by our own `id`. /// by serial (`device/{sn}/report`), not by our own `id`.
sn: Option<String>, pub sn: Option<String>,
api_key: Option<String>, pub api_key: Option<String>,
/// Serial device path, e.g. `/dev/ttyACM0` on Linux. /// Serial device path, e.g. `/dev/ttyACM0` on Linux.
port: Option<String>, pub port: Option<String>,
#[serde(default = "default_baud_rate")] #[serde(default = "default_baud_rate")]
baud_rate: u32, pub baud_rate: u32,
/// Path to a printer-specific CA certificate. Leave unset to use the /// 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 /// bundled Bambu CA - only needed for a printer that doesn't chain to
/// it (P1P, at least; see certs/README.md). /// it (P1P, at least; see certs/README.md).
ca_cert_path: Option<String>, pub ca_cert_path: Option<String>,
#[serde(default = "default_true")] #[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<String>| value.as_ref().map(|_| "<redacted>");
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 { fn default_true() -> bool {
@@ -72,6 +91,26 @@ pub fn load(path: &Path) -> anyhow::Result<Vec<PrinterHandle>> {
file.printers.into_iter().map(build_printer).collect() 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<Vec<PrinterEntry>> {
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::<FleetFile>(&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<PrinterHandle> {
build_printer(entry)
}
/// Everything the fleet file already names, for telling a printer the /// Everything the fleet file already names, for telling a printer the
/// operator has dealt with from one they have never seen. /// operator has dealt with from one they have never seen.
/// ///
@@ -119,7 +158,13 @@ pub fn configured(path: &Path) -> anyhow::Result<Configured> {
}; };
let file: FleetFile = toml::from_str(&text)?; let file: FleetFile = toml::from_str(&text)?;
let mut out = Configured::default(); 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<Item = PrinterEntry>) {
let add = |set: &mut HashSet<String>, value: Option<String>| { let add = |set: &mut HashSet<String>, value: Option<String>| {
if let Some(value) = value { if let Some(value) = value {
let value = value.trim().to_lowercase(); let value = value.trim().to_lowercase();
@@ -128,11 +173,12 @@ pub fn configured(path: &Path) -> anyhow::Result<Configured> {
} }
} }
}; };
add(&mut out.hosts, entry.host); for entry in entries {
add(&mut out.serials, entry.sn); add(&mut self.hosts, entry.host);
add(&mut out.ports, entry.port); add(&mut self.serials, entry.sn);
add(&mut self.ports, entry.port);
}
} }
Ok(out)
} }
fn build_printer(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> { fn build_printer(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> {
+2
View File
@@ -16,4 +16,6 @@ pub mod config_import;
pub mod discovery; pub mod discovery;
pub mod fleet; pub mod fleet;
pub mod printer; pub mod printer;
pub mod provision;
pub mod store;
pub mod uplink; pub mod uplink;
+26 -2
View File
@@ -68,7 +68,30 @@ async fn main() -> anyhow::Result<()> {
// Loaded and connected here, once, at startup - nothing yet re-reads // Loaded and connected here, once, at startup - nothing yet re-reads
// printers.toml on a running daemon, so adding or removing a printer // printers.toml on a running daemon, so adding or removing a printer
// still means a restart. // 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; connect_fleet(&mut fleet).await;
// The fleet is now shared with the uplink rather than just parked here // The fleet is now shared with the uplink rather than just parked here
@@ -134,10 +157,11 @@ async fn main() -> anyhow::Result<()> {
_ = shutdown_requested() => { _ = shutdown_requested() => {
info!("stop requested; exiting so go2rtc is stopped with us"); 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( _ = continuum_proxy::discovery::report::run(
config.discovery.clone(), config.discovery.clone(),
config.printers_path.clone(), config.printers_path.clone(),
Some(store.clone()),
outbound.clone(), outbound.clone(),
) => {} ) => {}
_ = go2rtc::watchdog( _ = go2rtc::watchdog(
+244
View File
@@ -0,0 +1,244 @@
//! 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);
}
/// The success path, using the OctoPrint client because its `connect` is
/// the one that succeeds without a real machine (the driver is a stub).
/// What this proves is the plumbing, not a vendor: a connected printer is
/// saved, joins the running fleet, and is still there after a "restart"
/// that reopens the store and rebuilds the fleet from it.
#[tokio::test]
async fn an_added_printer_joins_the_fleet_and_survives_a_restart() {
let (store, path) = store();
let fleet = empty_fleet();
let mut req = request("octo1");
req.vendor = "octoprint".into();
req.sn = None;
req.access_code = None;
req.api_key = Some("key-123".into());
add(&fleet, &store, req.clone()).await.unwrap();
assert!(fleet.lock().await.iter().any(|p| p.base().id == "octo1"));
// Same id again: refused, and the first is untouched.
let again = add(&fleet, &store, req).await.unwrap_err();
assert!(again.contains("already has"), "{again}");
assert_eq!(fleet.lock().await.len(), 1);
// "Restart": reopen the store, rebuild what main would.
drop(store);
let reopened = Store::open(&path).unwrap();
let rebuilt: Vec<_> = reopened.printers().unwrap().into_iter().map(|e| crate::fleet::build(e).unwrap()).collect();
assert_eq!(rebuilt.len(), 1);
assert_eq!(rebuilt[0].base().id, "octo1");
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("<redacted>"), "{shown}");
assert!(!shown.contains("98765432"), "the access code reached a Debug string: {shown}");
}
}
+199
View File
@@ -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<Connection>,
}
impl Store {
pub fn open(path: &Path) -> anyhow::Result<Self> {
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<bool> {
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<Vec<PrinterEntry>> {
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());
}
}
+107
View File
@@ -0,0 +1,107 @@
//! Getting from the gateway's long-lived credential to the short-lived one
//! the socket wants.
//!
//! The cloud hands a farm's owner an *edge proxy token*, which is what goes in
//! this gateway's settings as `gateway_token`. It does not open the socket
//! directly: the upgrade wants an *edge JWT*, which the cloud signs for an
//! hour in exchange for the proxy token (`POST /auth/edge-token`). So every
//! connection attempt starts by trading, which is also why a gateway that
//! has been up for days is not locked out when the first JWT expires - each
//! reconnect gets a fresh one.
use std::time::Duration;
use anyhow::Context;
use serde::Deserialize;
use crate::config::Config;
const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(15);
const SOCKET_PATH: &str = "/ws/edge/v1";
/// Where to trade the token, derived from the socket URL so there is one
/// address to configure, not two that can disagree.
pub fn exchange_url(uplink_url: &str) -> anyhow::Result<String> {
let (scheme, rest) = if let Some(rest) = uplink_url.strip_prefix("wss://") {
("https://", rest)
} else if let Some(rest) = uplink_url.strip_prefix("ws://") {
("http://", rest)
} else {
anyhow::bail!("the uplink URL must start with ws:// or wss://: {uplink_url}");
};
let base = rest
.strip_suffix(SOCKET_PATH)
.with_context(|| format!("the uplink URL must end with {SOCKET_PATH}: {uplink_url}"))?;
Ok(format!("{scheme}{base}/auth/edge-token"))
}
#[derive(Deserialize)]
struct Exchanged {
token: String,
}
/// Trades `gateway_token` for an edge JWT.
///
/// A 401 is reported as what it means - the cloud does not accept this
/// farm/token pair - because retrying will not fix it, and the usual cause is
/// a token that was rotated in the console while this gateway kept the old
/// one.
pub async fn fetch_edge_token(config: &Config) -> anyhow::Result<String> {
let url = exchange_url(&config.uplink_url)?;
let response = reqwest::Client::builder()
.timeout(EXCHANGE_TIMEOUT)
.build()?
.post(&url)
// Built by hand: the `json` feature of reqwest is off on purpose (see
// Cargo.toml), and this is one small body.
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"farmId": config.farm_id,
"edgeProxyToken": config.gateway_token,
})
.to_string(),
)
.send()
.await
.with_context(|| format!("could not reach {url} to get an edge token"))?;
match response.status() {
status if status.is_success() => {
let body = response.bytes().await?;
Ok(serde_json::from_slice::<Exchanged>(&body).context("the cloud's edge token reply was not understood")?.token)
}
reqwest::StatusCode::UNAUTHORIZED => {
anyhow::bail!("the cloud does not accept this farm_id / gateway_token - was the token rotated in the console?")
}
status => anyhow::bail!("the cloud answered {status} when asked for an edge token"),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_exchange_address_follows_the_socket_address() {
assert_eq!(
exchange_url("wss://api.continuum.local/ws/edge/v1").unwrap(),
"https://api.continuum.local/auth/edge-token"
);
assert_eq!(
exchange_url("ws://localhost:3040/ws/edge/v1").unwrap(),
"http://localhost:3040/auth/edge-token"
);
// A prefix in front of the API survives.
assert_eq!(
exchange_url("wss://example.com/api/ws/edge/v1").unwrap(),
"https://example.com/api/auth/edge-token"
);
}
#[test]
fn an_address_it_cannot_follow_is_an_error_not_a_guess() {
assert!(exchange_url("https://api/ws/edge/v1").is_err());
assert!(exchange_url("wss://api/somewhere/else").is_err());
}
}
+74 -4
View File
@@ -26,9 +26,10 @@
//! readings for no consumer. See PLAN.md for what would actually be worth //! readings for no consumer. See PLAN.md for what would actually be worth
//! buffering. //! buffering.
mod auth;
mod download; mod download;
mod webrtc; mod webrtc;
mod protocol; pub mod protocol;
pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage}; pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};
@@ -42,6 +43,9 @@ use tokio_tungstenite::tungstenite::handshake::client::{Request, generate_key};
use tracing::{info, warn}; use tracing::{info, warn};
use crate::config::Config; use crate::config::Config;
use crate::provision;
use crate::store::Store;
use protocol::ProvisionPrinter;
use crate::printer::{GenericPrinter, PrinterHandle}; use crate::printer::{GenericPrinter, PrinterHandle};
/// The websocket, once connected. Named because `connect` has to hand it to /// The websocket, once connected. Named because `connect` has to hand it to
@@ -139,6 +143,7 @@ pub fn channel() -> (
pub async fn run( pub async fn run(
config: Config, config: Config,
fleet: SharedFleet, fleet: SharedFleet,
store: Option<Arc<Store>>,
outbound: tokio::sync::mpsc::Sender<ClientMessage>, outbound: tokio::sync::mpsc::Sender<ClientMessage>,
mut inbox: tokio::sync::mpsc::Receiver<ClientMessage>, mut inbox: tokio::sync::mpsc::Receiver<ClientMessage>,
) { ) {
@@ -156,7 +161,7 @@ pub async fn run(
// would not distinguish a connection that lived for an hour // would not distinguish a connection that lived for an hour
// from one that failed to open. // from one that failed to open.
delay = RETRY_DELAY; 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:?}"); warn!(?err, "uplink connection dropped, reconnecting in {delay:?}");
} }
} }
@@ -172,6 +177,10 @@ pub async fn run(
/// handshake from a connection that opened and later dropped; they want /// handshake from a connection that opened and later dropped; they want
/// different retry behaviour. /// different retry behaviour.
async fn connect(config: &Config) -> anyhow::Result<WsStream> { async fn connect(config: &Config) -> anyhow::Result<WsStream> {
// A fresh edge JWT per attempt: it lives an hour, so one fetched at
// startup would lock a long-running gateway out on its first reconnect.
let edge_token = auth::fetch_edge_token(config).await?;
// The edge token goes on the upgrade request, not in the first message. // The edge token goes on the upgrade request, not in the first message.
// The backend verifies it before the socket exists, so a bad token is an // The backend verifies it before the socket exists, so a bad token is an
// HTTP 401 here rather than a connection that opens and is then closed - // HTTP 401 here rather than a connection that opens and is then closed -
@@ -180,7 +189,7 @@ async fn connect(config: &Config) -> anyhow::Result<WsStream> {
// those tools already know to redact. // those tools already know to redact.
let request = Request::builder() let request = Request::builder()
.uri(&config.uplink_url) .uri(&config.uplink_url)
.header("Authorization", format!("Bearer {}", config.gateway_token)) .header("Authorization", format!("Bearer {edge_token}"))
// tokio-tungstenite fills these in for a plain URL, but not once we // tokio-tungstenite fills these in for a plain URL, but not once we
// hand it a request we built ourselves - and a handshake missing any // hand it a request we built ourselves - and a handshake missing any
// of them is rejected by the server rather than ignored. // of them is rejected by the server rather than ignored.
@@ -200,6 +209,7 @@ async fn serve(
ws_stream: WsStream, ws_stream: WsStream,
config: &Config, config: &Config,
fleet: &SharedFleet, fleet: &SharedFleet,
store: Option<&Arc<Store>>,
outbound: &tokio::sync::mpsc::Sender<ClientMessage>, outbound: &tokio::sync::mpsc::Sender<ClientMessage>,
inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>, inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
@@ -306,13 +316,25 @@ async fn serve(
outbound.clone(), 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"), Ok(other) => info!(?other, "received from cloud"),
// Loudly. This used to be an `if let Ok(..)` with // Loudly. This used to be an `if let Ok(..)` with
// no else, which is the same silent-drop bug that // no else, which is the same silent-drop bug that
// hid a real `print.push_status` failure for // hid a real `print.push_status` failure for
// weeks - and here it would mean commands // weeks - and here it would mean commands
// vanishing with the cloud never being told. // 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(()), Some(Ok(Message::Close(_))) | None => return Ok(()),
@@ -456,6 +478,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<Arc<Store>>,
request_id: String,
request: Option<ProvisionPrinter>,
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
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<Arc<Store>>,
request_id: String,
printer_id: String,
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
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<ClientMessage>) {
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. /// Answers one WebRTC offer, off this connection's task.
/// ///
/// Always answers, success or failure: the cloud is holding a browser's /// Always answers, success or failure: the cloud is holding a browser's
+39
View File
@@ -225,6 +225,45 @@ pub enum ServerMessage {
/// and checks its password before CORS, so the unauthenticated preflight /// and checks its password before CORS, so the unauthenticated preflight
/// a browser always sends is refused - measured, not assumed. /// a browser always sends is refused - measured, not assumed.
WebrtcOffer { session_id: String, camera_stream: String, sdp: String }, 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<String>,
pub access_code: Option<String>,
pub sn: Option<String>,
pub api_key: Option<String>,
}
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(|_| "<redacted>"))
.field("sn", &self.sn)
.field("api_key", &self.api_key.as_ref().map(|_| "<redacted>"))
.finish()
}
} }
/// The closed set of things the cloud is allowed to ask a printer to do. /// The closed set of things the cloud is allowed to ask a printer to do.
+48 -2
View File
@@ -18,10 +18,12 @@ use std::time::Duration;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value}; use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener; use tokio::net::TcpListener;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
use continuum_proxy::config::Config; use continuum_proxy::config::Config;
use continuum_proxy::uplink; use continuum_proxy::uplink;
@@ -44,6 +46,7 @@ fn config(port: u16) -> Config {
go2rtc_api_listen: "127.0.0.1:1984".into(), go2rtc_api_listen: "127.0.0.1:1984".into(),
stream_base_url: None, stream_base_url: None,
printers_path: "/nonexistent/printers.toml".into(), 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: // 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. // an enabled one would sweep whatever network the test host is on.
discovery: Default::default(), discovery: Default::default(),
@@ -69,6 +72,36 @@ where
} }
} }
/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that
/// the gateway sent its farm and long-lived token, and answers with a JWT.
/// Returns the request text so the caller can assert on it.
async fn answer_token_exchange(listener: &TcpListener) -> String {
let (mut stream, _) = listener.accept().await.unwrap();
let mut request = Vec::new();
let mut buf = [0u8; 4096];
loop {
let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap();
assert!(n > 0, "the gateway hung up before finishing its token request");
request.extend_from_slice(&buf[..n]);
let text = String::from_utf8_lossy(&request).into_owned();
if let Some((head, body)) = text.split_once("\r\n\r\n") {
let length = head
.lines()
.find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
.unwrap_or(0);
if body.len() >= length {
break;
}
}
}
let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#;
stream
.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
.await
.unwrap();
String::from_utf8_lossy(&request).into_owned()
}
#[tokio::test] #[tokio::test]
async fn a_long_command_that_fails_reports_its_reason_after_the_fact() { async fn a_long_command_that_fails_reports_its_reason_after_the_fact() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -76,10 +109,23 @@ async fn a_long_command_that_fails_reports_its_reason_after_the_fact() {
let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new())); let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new()));
let (outbound, inbox) = uplink::channel(); let (outbound, inbox) = uplink::channel();
tokio::spawn(uplink::run(config(port), fleet, outbound, inbox)); tokio::spawn(uplink::run(config(port), fleet, None, outbound, inbox));
// The gateway trades its long-lived token for an edge JWT first, and the
// JWT - not the token it was configured with - is what opens the socket.
let exchange = answer_token_exchange(&listener).await;
assert!(exchange.starts_with("POST /auth/edge-token "), "{exchange}");
assert!(exchange.contains(r#""farmId":"test-farm""#) && exchange.contains(r#""edgeProxyToken":"test-token""#), "{exchange}");
let (stream, _) = listener.accept().await.unwrap(); let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); let mut upgrade_auth = String::new();
let mut ws = tokio_tungstenite::accept_hdr_async(stream, |request: &Request, response: Response| {
upgrade_auth = request.headers().get("authorization").unwrap().to_str().unwrap().to_string();
Ok(response)
})
.await
.unwrap();
assert_eq!(upgrade_auth, "Bearer minted-edge-jwt", "the socket must be opened with the minted JWT");
let hello = next_json(&mut ws).await; let hello = next_json(&mut ws).await;
assert_eq!(hello["type"], "hello", "a gateway introduces itself first"); assert_eq!(hello["type"], "hello", "a gateway introduces itself first");