monetize_estate_side/pump/
grpc.rs

1//! **The [`Link`] over the wire**: monetize-proto's own `Fleet` client. Blocking calls over
2//! a runtime handle — the puller runs on a jera job's blocking thread.
3
4use std::time::Duration;
5
6use monetize_embed::fleet::{SignedOrder, LONG_POLL_MAX_MS};
7use monetize_proto as pb;
8use tonic::transport::Channel;
9
10use super::config::FleetConfig;
11use super::puller::{Fault, Link, Signed};
12
13/// The channel to the ONE monetize's `Fleet`: dialled at `at` when named, the URL's host
14/// otherwise; TLS verified under the URL's host against the fed CA, else the public roots.
15/// Plain http only to a loopback address. A fed `at` / `ca_pem` is taken only as a bench's
16/// (nordisk-pairing `fleet::seam_refusal`, A7 row 23, Gunnar 8134ec6f): anything else is
17/// refused by name, whether it came in a feed or from a kept enrolment.
18pub fn channel(cfg: &FleetConfig) -> Result<Channel, String> {
19    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())) {
20        return Err(why);
21    }
22    let origin: tonic::codegen::http::Uri = cfg.url.parse().map_err(|e| format!("the fleet door {}: {e}", cfg.url))?;
23    let name = origin.host().unwrap_or("").to_owned();
24    let dial = match cfg.at.as_deref() {
25        Some(at) => format!("{}://{at}", origin.scheme_str().unwrap_or("https")),
26        None => cfg.url.clone(),
27    };
28    let ep = Channel::from_shared(dial.clone())
29        .map_err(|e| format!("the fleet door {} at {dial}: {e}", cfg.url))?
30        .origin(origin)
31        .connect_timeout(Duration::from_secs(10))
32        .timeout(Duration::from_millis(LONG_POLL_MAX_MS + 10_000));
33    let ep = if cfg.url.starts_with("https://") {
34        let tls = match cfg.ca_pem.as_deref() {
35            Some(pem) => tonic::transport::ClientTlsConfig::new().ca_certificate(tonic::transport::Certificate::from_pem(pem)),
36            None => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
37        };
38        ep.tls_config(tls.domain_name(name)).map_err(|e| format!("the fleet door {}: {e}", cfg.url))?
39    } else {
40        let host = dial.trim_start_matches("http://").split(['/', ':']).next().unwrap_or("");
41        let local = host == "localhost" || host.trim_matches(['[', ']']).parse::<std::net::IpAddr>().map(|ip| ip.is_loopback()).unwrap_or(false);
42        if !local {
43            return Err(format!("the fleet door {} is plain http to another machine: use https://", cfg.url));
44        }
45        ep
46    };
47    Ok(ep.connect_lazy())
48}
49
50fn fault(s: tonic::Status) -> Fault {
51    let code = match s.code() {
52        tonic::Code::Unauthenticated => "unauthenticated",
53        tonic::Code::InvalidArgument => "malformed",
54        tonic::Code::FailedPrecondition => "refused",
55        tonic::Code::Unavailable => "unavailable",
56        _ => "error",
57    };
58    Fault::new(code, s.message().to_string())
59}
60
61fn wire(s: Signed) -> pb::FleetSigned {
62    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 }
63}
64
65/// A wire order as monetize-embed's [`SignedOrder`].
66pub fn order_of(o: pb::FleetOrder) -> SignedOrder {
67    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 }
68}
69
70/// The largest answer the box takes (a `fetch` ask's file: a product order's bundle).
71pub const FETCH_MAX: usize = 64 << 20;
72
73/// The generated client, driven from a blocking thread.
74pub struct TonicLink {
75    pub client: pb::fleet_client::FleetClient<Channel>,
76    pub rt: tokio::runtime::Handle,
77}
78
79impl TonicLink {
80    pub fn new(channel: Channel, rt: tokio::runtime::Handle) -> TonicLink {
81        TonicLink { client: pb::fleet_client::FleetClient::new(channel).max_decoding_message_size(FETCH_MAX), rt }
82    }
83}
84
85impl Link for TonicLink {
86    fn report(&mut self, s: Signed) -> Result<u64, Fault> {
87        let mut c = self.client.clone();
88        self.rt.block_on(async move { c.report(wire(s)).await }).map(|r| r.into_inner().next_report_ms).map_err(fault)
89    }
90    fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
91        let mut c = self.client.clone();
92        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)
93    }
94    fn ask(&mut self, s: Signed) -> Result<super::puller::Asked, Fault> {
95        let mut c = self.client.clone();
96        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)
97    }
98    fn ask_bytes(&mut self, s: Signed) -> Result<(super::puller::Asked, Vec<u8>), Fault> {
99        let mut c = self.client.clone();
100        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)
101    }
102}
103
104#[cfg(test)]
105mod seam_tests {
106    use super::*;
107
108    fn cfg(at: Option<&str>, ca: Option<&str>) -> FleetConfig {
109        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() }
110    }
111
112    /// ★ A7 row 23 (Gunnar 8134ec6f, ST-19): the pull channel takes a fed `fleet_at` only as a
113    /// bench `ip:port` and a fed `fleet_ca` only beside one (nordisk-pairing
114    /// `fleet::seam_refusal`); a public address, a name or a lone CA is refused by name, so a
115    /// release box never dials elsewhere, nor re-roots its trust, on a feed's word.
116    #[test]
117    fn the_pull_channel_takes_a_fed_seam_only_as_a_bench_address() {
118        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
119        let _g = rt.enter();
120        for at in ["10.0.2.2:40443", "127.0.0.1:443", "[fd00::2]:443"] {
121            assert!(channel(&cfg(Some(at), Some("PEM"))).is_ok(), "a bench forward {at}");
122        }
123        for at in ["203.0.113.7:443", "monetize.example:443", "10.0.2.2"] {
124            let e = channel(&cfg(Some(at), None)).expect_err(at);
125            assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
126        }
127        let e = channel(&cfg(None, Some("PEM"))).expect_err("a CA with no bench address");
128        assert!(e.contains("[fleet-seam-bench-only]"), "{e}");
129        assert!(channel(&cfg(None, None)).is_ok(), "no seam: the name, under the public roots");
130    }
131}
monetize_estate_side/pump/grpc.rs

monetize_estate_side/pump/
grpc.rs
