hatchery_node/
uplink_calls.rs

1//! Node calls (`mail.*`, `task.*`) over the restreamer uplink.
2//!
3//! Since A2, C2 no longer forwards signed node calls to core over HTTP: it answers 410.
4//! These calls now ride a dedicated `m4a` channel on the node's existing uplink. That
5//! channel is Noise-encrypted end to end to core and opaque to C2. Core takes the node
6//! id from the stream, never from the payload. The legacy HTTP path stays as a fallback
7//! for nodes without an uplink.
8
9use crate::restream_session::{NodeSession, NodeUplink, UpstreamKeys};
10use hatchery_core::ChannelKind;
11use hatchery_enroll::NodeIdentity;
12use serde_json::Value;
13use std::net::SocketAddr;
14use std::sync::Arc;
15use std::time::Duration;
16use tokio::sync::Mutex;
17
18/// Channel id of the node-call channel; core runs its `m4a` drain consumer on it.
19pub const NODE_CALL_CHANNEL: u32 = 2;
20const HANDSHAKE: Duration = Duration::from_secs(10);
21const CALL_TIMEOUT: Duration = Duration::from_secs(20);
22
23#[derive(Default)]
24struct Link {
25    uplink: Option<Arc<NodeUplink>>,
26    session: Option<NodeSession>,
27}
28
29/// Shared by the door and the node-mail wake; the restream bridge publishes its uplink here.
30#[derive(Clone)]
31pub struct NodeCalls {
32    link: Arc<Mutex<Link>>,
33    node_id: String,
34    keys: Option<UpstreamKeys>,
35    legacy: Option<(SocketAddr, Arc<NodeIdentity>)>,
36}
37
38#[derive(Debug)]
39pub enum CallError {
40    Refused,
41    Unavailable(String),
42}
43
44impl NodeCalls {
45    pub fn new(node_id: String, keys: Option<UpstreamKeys>, legacy: Option<(SocketAddr, Arc<NodeIdentity>)>) -> Self {
46        Self { link: Arc::default(), node_id, keys, legacy }
47    }
48
49    /// The bridge calls this on every (re)connect; `None` when the uplink is gone.
50    pub async fn set_uplink(&self, uplink: Option<Arc<NodeUplink>>) {
51        let mut l = self.link.lock().await;
52        l.uplink = uplink;
53        l.session = None;
54    }
55
56    pub async fn call(&self, op: &str, payload: &Value) -> Result<Value, CallError> {
57        if let Some(keys) = self.keys {
58            let mut l = self.link.lock().await;
59            if let Some(uplink) = l.uplink.clone() {
60                if l.session.is_none() {
61                    match uplink.open(&self.node_id, NODE_CALL_CHANNEL, ChannelKind::M4a, keys, HANDSHAKE).await {
62                        Ok(s) => {
63                            tracing::info!("node calls over the uplink: channel open");
64                            l.session = Some(s);
65                        }
66                        Err(e) => tracing::warn!(error = %e, "node-call channel open failed"),
67                    }
68                }
69                if let Some(session) = l.session.as_mut() {
70                    // Calls are strictly serialized under the lock: one request, one reply.
71                    return match session.call(op, payload, CALL_TIMEOUT).await {
72                        Ok(v) => Ok(v),
73                        Err(e) if e.starts_with("403") => Err(CallError::Refused),
74                        Err(e) if e.contains(':') && e[..3].chars().all(|c| c.is_ascii_digit()) => Err(CallError::Unavailable(e)),
75                        Err(e) => {
76                            // Transport fault or lost reply: drop the session so replies cannot pair
77                            // with the wrong request; the next call opens a fresh one.
78                            l.session = None;
79                            Err(CallError::Unavailable(e))
80                        }
81                    };
82                }
83            }
84        }
85        match &self.legacy {
86            Some((c2, identity)) => match crate::enroll_client::node_call(*c2, identity, &self.node_id, op, payload).await {
87                Ok(v) => Ok(v),
88                Err(crate::enroll_client::EnrollClientError::Refused) => Err(CallError::Refused),
89                Err(e) => Err(CallError::Unavailable(format!("{e:?}"))),
90            },
91            None => Err(CallError::Unavailable("no uplink".into())),
92        }
93    }
94}
hatchery_node/uplink_calls.rs

hatchery_node/
uplink_calls.rs
