Compare commits
3
Commits
6c60bd1f70
...
b4948076a0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b4948076a0
|
||
|
|
5fc09aef04
|
||
|
|
9dff600160
|
+6
-4
@@ -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
|
||||||
|
|||||||
@@ -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
@@ -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"
|
||||||
|
|||||||
@@ -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"] }
|
||||||
|
|||||||
@@ -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
@@ -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
@@ -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
@@ -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> {
|
||||||
|
|||||||
@@ -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
@@ -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(
|
||||||
|
|||||||
@@ -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
@@ -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());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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");
|
||||||
|
|||||||
Reference in New Issue
Block a user