orch/cluster/
network.rs

1//! Raft RPCs and forwarded writes between nodes, JSON over HTTP like `../agents-coordination`.
2//! Every request carries the cluster secret, so reaching the port is not enough to join
3
4// openraft's RPC errors are its own types, so their size is not ours to change
5#![allow(clippy::result_large_err)]
6
7use std::sync::{Arc, LazyLock};
8use std::time::Duration;
9
10use axum::extract::State;
11use axum::http::{HeaderMap, StatusCode};
12use axum::response::{IntoResponse, Response};
13use axum::routing::{get, post};
14use axum::{Json, Router};
15use openraft::error::{
16    InstallSnapshotError, NetworkError, RPCError, RaftError, RemoteError, Unreachable,
17};
18use openraft::network::{RPCOption, RaftNetwork, RaftNetworkFactory};
19use openraft::raft::{
20    AppendEntriesRequest, AppendEntriesResponse, InstallSnapshotRequest, InstallSnapshotResponse,
21    VoteRequest, VoteResponse,
22};
23use openraft::BasicNode;
24use serde::de::DeserializeOwned;
25use serde::Serialize;
26use subtle::ConstantTimeEq;
27
28use super::{NodeId, Raft, Reply, Request, TypeConfig};
29
30const SECRET_HEADER: &str = "x-orch-cluster-secret";
31const NONCE_HEADER: &str = "x-orch-nonce";
32const PROOF_HEADER: &str = "x-orch-proof";
33
34static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
35    reqwest::Client::builder()
36        .timeout(Duration::from_secs(30))
37        .build()
38        .unwrap_or_default()
39});
40
41#[derive(Clone)]
42struct App {
43    raft: Raft,
44    secret: Arc<String>,
45    me: Arc<Hello>,
46    /// The address peers dial, bound into every proof so a proof relayed through another host fails
47    addr: Arc<String>,
48}
49
50/// What a node says about itself to a peer that found it on the tailnet
51#[derive(Debug, Clone, PartialEq, Serialize, serde::Deserialize)]
52pub struct Hello {
53    pub name: String,
54    pub id: NodeId,
55    pub witness: bool,
56    #[serde(default)]
57    pub leader: Option<NodeId>,
58    #[serde(default)]
59    pub term: u64,
60    /// HMAC of the prober's nonce and this name, so the prober knows the peer holds the secret
61    #[serde(default)]
62    pub proof: String,
63}
64
65/// HMAC-SHA256 under the cluster secret, in hex. Proves the secret without sending it
66fn mac(secret: &str, parts: &[&str]) -> String {
67    use hmac::{Hmac, Mac};
68    let mut m =
69        Hmac::<sha2::Sha256>::new_from_slice(secret.as_bytes()).expect("HMAC takes any key");
70    for part in parts {
71        m.update(part.as_bytes());
72        m.update(b"\0");
73    }
74
75    m.finalize()
76        .into_bytes()
77        .iter()
78        .map(|b| format!("{b:02x}"))
79        .collect()
80}
81
82/// A node asking the leader to add it as a learner
83#[derive(Debug, Clone, Serialize, serde::Deserialize)]
84pub struct Join {
85    pub id: NodeId,
86    pub addr: String,
87}
88
89fn authorized(app: &App, headers: &HeaderMap) -> bool {
90    let given = headers
91        .get(SECRET_HEADER)
92        .map(|v| v.as_bytes())
93        .unwrap_or_default();
94    given.ct_eq(app.secret.as_bytes()).into()
95}
96
97pub(crate) fn router(raft: Raft, secret: Arc<String>, me: Hello, addr: String) -> Router {
98    Router::new()
99        .route("/raft/append", post(append))
100        .route("/raft/snapshot", post(snapshot))
101        .route("/raft/vote", post(vote))
102        .route("/cluster/write", post(write))
103        .route("/cluster/hello", get(hello))
104        .route("/cluster/join", post(join))
105        .with_state(App {
106            raft,
107            secret,
108            me: Arc::new(me),
109            addr: Arc::new(addr),
110        })
111}
112
113/// Answers only a prober that proves the secret, and proves it back, so neither side sends it
114async fn hello(State(a): State<App>, headers: HeaderMap) -> Response {
115    let header = |name: &str| {
116        headers
117            .get(name)
118            .and_then(|v| v.to_str().ok())
119            .unwrap_or_default()
120            .to_string()
121    };
122    let nonce = header(NONCE_HEADER);
123    let expected = mac(&a.secret, &["probe", &nonce, &a.addr]);
124    let proven: bool = header(PROOF_HEADER)
125        .as_bytes()
126        .ct_eq(expected.as_bytes())
127        .into();
128    if nonce.len() < 16 || !proven {
129        return StatusCode::UNAUTHORIZED.into_response();
130    }
131
132    let m = a.raft.metrics().borrow().clone();
133    Json(Hello {
134        leader: m.current_leader,
135        term: m.current_term,
136        proof: mac(&a.secret, &["hello", &nonce, &a.me.name, &a.addr]),
137        ..(*a.me).clone()
138    })
139    .into_response()
140}
141
142/// Adds the caller as a learner on the leader, or sends it on to the leader
143async fn join(State(a): State<App>, headers: HeaderMap, Json(req): Json<Join>) -> Response {
144    if !authorized(&a, &headers) {
145        return StatusCode::UNAUTHORIZED.into_response();
146    }
147
148    let node = BasicNode::new(&req.addr);
149    match a.raft.add_learner(req.id, node, false).await {
150        Ok(_) => StatusCode::OK.into_response(),
151        Err(RaftError::APIError(openraft::error::ClientWriteError::ForwardToLeader(f))) => {
152            match f.leader_node {
153                Some(leader) => match send_join(&leader.addr, &a.secret, &req).await {
154                    Ok(()) => StatusCode::OK.into_response(),
155                    Err(e) => (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
156                },
157                None => (StatusCode::SERVICE_UNAVAILABLE, "no leader yet").into_response(),
158            }
159        }
160        Err(e) => (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response(),
161    }
162}
163
164pub async fn send_join(addr: &str, secret: &str, req: &Join) -> anyhow::Result<()> {
165    let res = CLIENT
166        .post(format!("http://{addr}/cluster/join"))
167        .header(SECRET_HEADER, secret)
168        .json(req)
169        .send()
170        .await?;
171    anyhow::ensure!(
172        res.status().is_success(),
173        "join refused by {addr} ({}): {}",
174        res.status(),
175        res.text().await?
176    );
177    Ok(())
178}
179
180/// Ask a host found on the tailnet who it is. The secret never leaves this node. Both proofs cover
181/// the address dialed, and the answer must carry the name Tailscale gives that address, so a host
182/// that relays a real member's answer is rejected
183pub async fn probe(addr: &str, secret: &str, expect_name: Option<&str>) -> Option<Hello> {
184    let nonce = uuid::Uuid::new_v4().simple().to_string();
185    let res = CLIENT
186        .get(format!("http://{addr}/cluster/hello"))
187        .header(NONCE_HEADER, &nonce)
188        .header(PROOF_HEADER, mac(secret, &["probe", &nonce, addr]))
189        .timeout(Duration::from_secs(1))
190        .send()
191        .await
192        .ok()?;
193    let hello: Hello = res.error_for_status().ok()?.json().await.ok()?;
194    if expect_name.is_some_and(|name| name != hello.name) {
195        return None;
196    }
197
198    let expected = mac(secret, &["hello", &nonce, &hello.name, addr]);
199    let proven: bool = hello.proof.as_bytes().ct_eq(expected.as_bytes()).into();
200    proven.then_some(hello)
201}
202
203async fn append(
204    State(a): State<App>,
205    headers: HeaderMap,
206    Json(req): Json<AppendEntriesRequest<TypeConfig>>,
207) -> Response {
208    if !authorized(&a, &headers) {
209        return StatusCode::UNAUTHORIZED.into_response();
210    }
211
212    Json(a.raft.append_entries(req).await).into_response()
213}
214
215async fn snapshot(
216    State(a): State<App>,
217    headers: HeaderMap,
218    Json(req): Json<InstallSnapshotRequest<TypeConfig>>,
219) -> Response {
220    if !authorized(&a, &headers) {
221        return StatusCode::UNAUTHORIZED.into_response();
222    }
223
224    Json(a.raft.install_snapshot(req).await).into_response()
225}
226
227async fn vote(
228    State(a): State<App>,
229    headers: HeaderMap,
230    Json(req): Json<VoteRequest<NodeId>>,
231) -> Response {
232    if !authorized(&a, &headers) {
233        return StatusCode::UNAUTHORIZED.into_response();
234    }
235
236    Json(a.raft.vote(req).await).into_response()
237}
238
239/// A write a follower forwarded. It is proposed here and never forwarded again, so writes cannot loop
240async fn write(State(a): State<App>, headers: HeaderMap, Json(req): Json<Request>) -> Response {
241    if !authorized(&a, &headers) {
242        return StatusCode::UNAUTHORIZED.into_response();
243    }
244
245    match a.raft.client_write(req).await {
246        Ok(r) => Json(r.data).into_response(),
247        // Refused before anything was proposed, so the sender may safely try another node
248        Err(e @ RaftError::APIError(openraft::error::ClientWriteError::ForwardToLeader(_))) => {
249            (StatusCode::MISDIRECTED_REQUEST, e.to_string()).into_response()
250        }
251        Err(e) => (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response(),
252    }
253}
254
255/// Why a forwarded write failed. Only `NotApplied` is safe to send again
256#[derive(Debug)]
257pub(crate) enum ForwardError {
258    /// The connection was refused, or the node said it is not the leader
259    NotApplied(String),
260    /// It may or may not have been applied, so it must not be sent again
261    Unknown(anyhow::Error),
262}
263
264pub(crate) async fn forward(
265    addr: &str,
266    secret: &str,
267    req: &Request,
268) -> Result<Reply, ForwardError> {
269    let res = CLIENT
270        .post(format!("http://{addr}/cluster/write"))
271        .header(SECRET_HEADER, secret)
272        .json(req)
273        .send()
274        .await
275        .map_err(|e| {
276            if e.is_connect() {
277                ForwardError::NotApplied(e.to_string())
278            } else {
279                ForwardError::Unknown(e.into())
280            }
281        })?;
282    let status = res.status();
283    if status == StatusCode::MISDIRECTED_REQUEST {
284        let body = res.text().await.unwrap_or_default();
285        return Err(ForwardError::NotApplied(format!(
286            "{addr} is not the leader: {body}"
287        )));
288    }
289
290    if !status.is_success() {
291        let body = res.text().await.unwrap_or_default();
292        return Err(ForwardError::Unknown(anyhow::anyhow!(
293            "leader at {addr} refused the write ({status}): {body}"
294        )));
295    }
296
297    res.json()
298        .await
299        .map_err(|e| ForwardError::Unknown(e.into()))
300}
301
302pub(crate) struct HttpNetwork {
303    pub secret: Arc<String>,
304}
305
306impl RaftNetworkFactory<TypeConfig> for HttpNetwork {
307    type Network = Peer;
308
309    async fn new_client(&mut self, target: NodeId, node: &BasicNode) -> Peer {
310        Peer {
311            target,
312            addr: node.addr.clone(),
313            secret: self.secret.clone(),
314        }
315    }
316}
317
318pub(crate) struct Peer {
319    target: NodeId,
320    addr: String,
321    secret: Arc<String>,
322}
323
324impl Peer {
325    async fn send<Req, Resp, E>(
326        &self,
327        path: &str,
328        req: Req,
329        option: &RPCOption,
330    ) -> Result<Resp, RPCError<NodeId, BasicNode, RaftError<NodeId, E>>>
331    where
332        Req: Serialize,
333        Resp: DeserializeOwned,
334        E: std::error::Error + DeserializeOwned,
335    {
336        let res = CLIENT
337            .post(format!("http://{}/raft/{path}", self.addr))
338            .header(SECRET_HEADER, self.secret.as_str())
339            .timeout(option.hard_ttl())
340            .json(&req)
341            .send()
342            .await
343            .map_err(|e| {
344                if e.is_connect() {
345                    RPCError::Unreachable(Unreachable::new(&e))
346                } else {
347                    RPCError::Network(NetworkError::new(&e))
348                }
349            })?;
350        let res = res
351            .error_for_status()
352            .map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
353        let body: Result<Resp, RaftError<NodeId, E>> = res
354            .json()
355            .await
356            .map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
357        body.map_err(|e| RPCError::RemoteError(RemoteError::new(self.target, e)))
358    }
359}
360
361impl RaftNetwork<TypeConfig> for Peer {
362    async fn append_entries(
363        &mut self,
364        req: AppendEntriesRequest<TypeConfig>,
365        option: RPCOption,
366    ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
367        self.send("append", req, &option).await
368    }
369
370    async fn install_snapshot(
371        &mut self,
372        req: InstallSnapshotRequest<TypeConfig>,
373        option: RPCOption,
374    ) -> Result<
375        InstallSnapshotResponse<NodeId>,
376        RPCError<NodeId, BasicNode, RaftError<NodeId, InstallSnapshotError>>,
377    > {
378        self.send("snapshot", req, &option).await
379    }
380
381    async fn vote(
382        &mut self,
383        req: VoteRequest<NodeId>,
384        option: RPCOption,
385    ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
386        self.send("vote", req, &option).await
387    }
388}
orch/cluster/network.rs

orch/cluster/
network.rs
