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 ride a dedicated `m4a` channel on the node's existing uplink. That channel
5//! is Noise-encrypted end to end to core and opaque to C2. Core takes the node id from
6//! the stream, never from the payload. The legacy HTTP path stays as a fallback for
7//! nodes without an uplink.
8//!
9//! Important: this channel shares the node's single TCP uplink with g4a. A handshake
10//! retry storm (or a wedged Noise session on core's m4a drain) can fill the TCP send
11//! buffer and stall PTY control. Open failures therefore back off hard.
12
13use crate::restream_session::{NodeSession, NodeUplink, UpstreamKeys};
14use hatchery_core::ChannelKind;
15use hatchery_enroll::NodeIdentity;
16use serde_json::Value;
17use std::net::SocketAddr;
18use std::sync::Arc;
19use std::time::{Duration, Instant};
20use tokio::sync::Mutex;
21
22/// Channel id of the node-call channel; core runs its `m4a` drain consumer on it.
23pub const NODE_CALL_CHANNEL: u32 = 2;
24const HANDSHAKE: Duration = Duration::from_secs(8);
25const CALL_TIMEOUT: Duration = Duration::from_secs(15);
26/// After an open/handshake failure, wait before trying again (shared uplink).
27const OPEN_BACKOFF: Duration = Duration::from_secs(60);
28
29#[derive(Default)]
30struct Link {
31    uplink: Option<Arc<NodeUplink>>,
32    session: Option<NodeSession>,
33    /// Do not attempt OPEN before this instant (protects the shared uplink).
34    open_not_before: Option<Instant>,
35}
36
37/// Shared by the door and the node-mail wake; the restream bridge publishes its uplink here.
38#[derive(Clone)]
39pub struct NodeCalls {
40    link: Arc<Mutex<Link>>,
41    node_id: String,
42    keys: Option<UpstreamKeys>,
43    legacy: Option<(SocketAddr, Arc<NodeIdentity>)>,
44    /// Enroll HTTP address used only for C2-local config (`node.config`), even when
45    /// mail/task calls go over the restream uplink (those get 410 on the HTTP path).
46    config_http: Option<(SocketAddr, Arc<NodeIdentity>)>,
47}
48
49#[derive(Debug)]
50pub enum CallError {
51    Refused,
52    Unavailable(String),
53}
54
55impl NodeCalls {
56    pub fn new(
57        node_id: String,
58        keys: Option<UpstreamKeys>,
59        legacy: Option<(SocketAddr, Arc<NodeIdentity>)>,
60        config_http: Option<(SocketAddr, Arc<NodeIdentity>)>,
61    ) -> Self {
62        Self { link: Arc::default(), node_id, keys, legacy, config_http }
63    }
64
65    /// The bridge calls this on every (re)connect; `None` when the uplink is gone.
66    pub async fn set_uplink(&self, uplink: Option<Arc<NodeUplink>>) {
67        let connected = uplink.is_some();
68        let mut l = self.link.lock().await;
69        l.uplink = uplink;
70        l.session = None;
71        // Fresh uplink: allow an immediate open attempt.
72        l.open_not_before = None;
73        drop(l);
74        if connected {
75            self.refresh_product_url().await;
76        } else {
77            crate::m4a_accounts::set_c2_product_url(None);
78        }
79    }
80
81    /// Ask C2 for its public config over enroll HTTP and stash `m4a_product_url`.
82    pub async fn refresh_product_url(&self) -> bool {
83        let Some((c2, identity)) = self.config_http.clone().or_else(|| self.legacy.clone()) else {
84            return false;
85        };
86        match crate::enroll_client::node_call(
87            c2,
88            &identity,
89            &self.node_id,
90            hatchery_enroll::call::NODE_CONFIG_OP,
91            &serde_json::json!({}),
92        )
93        .await
94        {
95            Ok(v) => {
96                let url = v
97                    .get("m4a_product_url")
98                    .and_then(|u| u.as_str())
99                    .filter(|u| !u.is_empty())
100                    .map(str::to_owned);
101                crate::m4a_accounts::set_c2_product_url(url.clone());
102                match url {
103                    Some(u) => tracing::info!(product = %u, "m4a product url from C2"),
104                    None => tracing::info!("m4a product url from C2 is empty; accounts stay local"),
105                }
106                true
107            }
108            Err(e) => {
109                tracing::warn!(error = ?e, "m4a product url from C2 failed");
110                false
111            }
112        }
113    }
114
115    /// [`Self::refresh_product_url`] with retries (5s doubling to 5 min) until C2 answers.
116    pub async fn refresh_product_url_retrying(&self) {
117        let mut wait = std::time::Duration::from_secs(5);
118        while !self.refresh_product_url().await {
119            if crate::m4a_accounts::M4aSessionAccounts::resolve_product_url(None).is_some() {
120                return;
121            }
122            tokio::time::sleep(wait).await;
123            wait = (wait * 2).min(std::time::Duration::from_secs(300));
124        }
125    }
126
127    pub async fn call(&self, op: &str, payload: &Value) -> Result<Value, CallError> {
128        if let Some(keys) = self.keys {
129            let mut l = self.link.lock().await;
130            if let Some(uplink) = l.uplink.clone() {
131                if l.session.is_none() {
132                    let gated = l.open_not_before.is_some_and(|t| Instant::now() < t);
133                    if gated {
134                        // Skip OPEN storm; fall through to legacy if any.
135                    } else {
136                        match uplink.open(&self.node_id, NODE_CALL_CHANNEL, ChannelKind::M4a, keys, HANDSHAKE).await {
137                            Ok(s) => {
138                                tracing::info!("node calls over the uplink: channel open");
139                                l.session = Some(s);
140                                l.open_not_before = None;
141                            }
142                            Err(e) => {
143                                tracing::warn!(error = %e, "node-call channel open failed; backing off {}s", OPEN_BACKOFF.as_secs());
144                                l.open_not_before = Some(Instant::now() + OPEN_BACKOFF);
145                            }
146                        }
147                    }
148                }
149                if let Some(session) = l.session.as_mut() {
150                    return match session.call(op, payload, CALL_TIMEOUT).await {
151                        Ok(v) => Ok(v),
152                        Err(e) if e.starts_with("403") => Err(CallError::Refused),
153                        Err(e) if e.chars().take(3).all(|c| c.is_ascii_digit()) && e.contains(':') => {
154                            // Application-level error from core; keep the session.
155                            Err(CallError::Unavailable(e))
156                        }
157                        Err(e) => {
158                            // Transport fault: drop session and back off so we do not
159                            // flood OPEN/handshake1 onto a shared uplink C2 is not draining.
160                            l.session = None;
161                            l.open_not_before = Some(Instant::now() + OPEN_BACKOFF);
162                            Err(CallError::Unavailable(e))
163                        }
164                    };
165                }
166            }
167        }
168        match &self.legacy {
169            Some((c2, identity)) => match crate::enroll_client::node_call(*c2, identity, &self.node_id, op, payload).await {
170                Ok(v) => Ok(v),
171                Err(crate::enroll_client::EnrollClientError::Refused) => Err(CallError::Refused),
172                Err(e) => Err(CallError::Unavailable(format!("{e:?}"))),
173            },
174            None => Err(CallError::Unavailable("no uplink".into())),
175        }
176    }
177}
hatchery_node/uplink_calls.rs

hatchery_node/
uplink_calls.rs
