crates/monetize-estate-side/src/pump/grpc.rs
//! **The [`Link`] over the wire**: monetize-proto's own `Fleet` client. Blocking calls over
//! a runtime handle — the puller runs on a jera job's blocking thread.
use std::time::Duration;
use monetize_embed::fleet::{SignedOrder, LONG_POLL_MAX_MS};
use monetize_proto as pb;
use tonic::transport::Channel;
use super::config::FleetConfig;
use super::puller::{Fault, Link, Signed};
/// The channel to the ONE monetize's `Fleet`: dialled at `at` when named, the URL's host
/// otherwise; TLS verified under the URL's host against the fed CA, else the public roots.
/// Plain http only to a loopback address. A fed `at` / `ca_pem` is taken only as a bench's
/// (nordisk-pairing `fleet::seam_refusal`, A7 row 23, Gunnar 8134ec6f): anything else is
/// refused by name, whether it came in a feed or from a kept enrolment.
pub fn channel(cfg: &FleetConfig) -> Result<Channel, String> {
if let Some(why) = nordisk_pairing::fleet::seam_refusal(
cfg.at.as_deref(),
cfg.ca_pem.as_deref().is_some_and(|c| !c.trim().is_empty()),
) {
return Err(why);
}
let origin: tonic::codegen::http::Uri = cfg
.url
.parse()
.map_err(|e| format!("the fleet door {}: {e}", cfg.url))?;
let name = origin.host().unwrap_or("").to_owned();
let dial = match cfg.at.as_deref() {
Some(at) => format!("{}://{at}", origin.scheme_str().unwrap_or("https")),
None => cfg.url.clone(),
};
let ep = Channel::from_shared(dial.clone())
.map_err(|e| format!("the fleet door {} at {dial}: {e}", cfg.url))?
.origin(origin)
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_millis(LONG_POLL_MAX_MS + 10_000));
let ep = if cfg.url.starts_with("https://") {
let tls = match cfg.ca_pem.as_deref() {
Some(pem) => tonic::transport::ClientTlsConfig::new()
.ca_certificate(tonic::transport::Certificate::from_pem(pem)),
None => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
};
ep.tls_config(tls.domain_name(name))
.map_err(|e| format!("the fleet door {}: {e}", cfg.url))?
} else {
let host = dial
.trim_start_matches("http://")
.split(['/', ':'])
.next()
.unwrap_or("");
let local = host == "localhost"
|| host
.trim_matches(['[', ']'])
.parse::<std::net::IpAddr>()
.map(|ip| ip.is_loopback())
.unwrap_or(false);
if !local {
return Err(format!(
"the fleet door {} is plain http to another machine: use https://",
cfg.url
));
}
ep
};
Ok(ep.connect_lazy())
}
fn fault(s: tonic::Status) -> Fault {
let code = match s.code() {
tonic::Code::Unauthenticated => "unauthenticated",
tonic::Code::InvalidArgument => "malformed",
tonic::Code::FailedPrecondition => "refused",
tonic::Code::Unavailable => "unavailable",
_ => "error",
};
Fault::new(code, s.message().to_string())
}
fn wire(s: Signed) -> pb::FleetSigned {
pb::FleetSigned {
product: s.call.product,
instance: s.call.instance,
role: s.call.role,
call: s.call.call,
body: s.call.body,
at_unix_ms: s.call.at_unix_ms,
signature: s.signature,
box_key: s.box_key,
}
}
/// A wire order as monetize-embed's [`SignedOrder`].
pub fn order_of(o: pb::FleetOrder) -> SignedOrder {
SignedOrder {
seq: o.seq,
product: o.product,
instance: o.instance,
role: o.role,
kind: o.kind,
body: o.body,
issued_unix_ms: o.issued_unix_ms,
not_after_unix_ms: o.not_after_unix_ms,
signature: o.signature,
}
}
/// The largest answer the box takes (a `fetch` ask's file: a product order's bundle).
pub const FETCH_MAX: usize = 64 << 20;
/// The generated client, driven from a blocking thread.
pub struct TonicLink {
pub client: pb::fleet_client::FleetClient<Channel>,
pub rt: tokio::runtime::Handle,
}
impl TonicLink {
pub fn new(channel: Channel, rt: tokio::runtime::Handle) -> TonicLink {
TonicLink {
client: pb::fleet_client::FleetClient::new(channel)
.max_decoding_message_size(FETCH_MAX),
rt,
}
}
}
impl Link for TonicLink {
fn report(&mut self, s: Signed) -> Result<u64, Fault> {
let mut c = self.client.clone();
self.rt
.block_on(async move { c.report(wire(s)).await })
.map(|r| r.into_inner().next_report_ms)
.map_err(fault)
}
fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
let mut c = self.client.clone();
self.rt
.block_on(async move { c.orders(wire(s)).await })
.map(|r| r.into_inner().orders.into_iter().map(order_of).collect())
.map_err(fault)
}
fn ask(&mut self, s: Signed) -> Result<super::puller::Asked, Fault> {
let mut c = self.client.clone();
self.rt
.block_on(async move { c.ask(wire(s)).await })
.map(|r| r.into_inner())
.map(|a| super::puller::Asked {
case_id: a.case_id,
words: a.words,
url: a.url,
})
.map_err(fault)
}
fn ask_bytes(&mut self, s: Signed) -> Result<(super::puller::Asked, Vec<u8>), Fault> {
let mut c = self.client.clone();
self.rt
.block_on(async move { c.ask(wire(s)).await })
.map(|r| r.into_inner())
.map(|a| {
(
super::puller::Asked {
case_id: a.case_id,
words: a.words,
url: a.url,
},
a.answer,
)
})
.map_err(fault)
}
}
#[cfg(test)]
mod seam_tests {
use super::*;
fn cfg(at: Option<&str>, ca: Option<&str>) -> FleetConfig {
FleetConfig {
url: "https://monetize.ignalina.dk".into(),
at: at.map(str::to_string),
ca_pem: ca.map(str::to_string),
instance: "acme".into(),
role: String::new(),
monetize_pubkey: String::new(),
}
}
/// ★ A7 row 23 (Gunnar 8134ec6f, ST-19): the pull channel takes a fed `fleet_at` only as a
/// bench `ip:port` and a fed `fleet_ca` only beside one (nordisk-pairing
/// `fleet::seam_refusal`); a public address, a name or a lone CA is refused by name, so a
/// release box never dials elsewhere, nor re-roots its trust, on a feed's word.
#[test]
fn the_pull_channel_takes_a_fed_seam_only_as_a_bench_address() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let _g = rt.enter();
for at in ["10.0.2.2:40443", "127.0.0.1:443", "[fd00::2]:443"] {
assert!(
channel(&cfg(Some(at), Some("PEM"))).is_ok(),
"a bench forward {at}"
);
}
for at in ["203.0.113.7:443", "monetize.example:443", "10.0.2.2"] {
let e = channel(&cfg(Some(at), None)).expect_err(at);
assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
}
let e = channel(&cfg(None, Some("PEM"))).expect_err("a CA with no bench address");
assert!(e.contains("[fleet-seam-bench-only]"), "{e}");
assert!(
channel(&cfg(None, None)).is_ok(),
"no seam: the name, under the public roots"
);
}
}