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