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    if let Err(e) = req.cmd.check_replicable() {
246        return (StatusCode::BAD_REQUEST, e.to_string()).into_response();
247    }
248
249    match a.raft.client_write(req).await {
250        Ok(r) => Json(r.data).into_response(),
251        // Refused before anything was proposed, so the sender may safely try another node
252        Err(e @ RaftError::APIError(openraft::error::ClientWriteError::ForwardToLeader(_))) => {
253            (StatusCode::MISDIRECTED_REQUEST, e.to_string()).into_response()
254        }
255        Err(e) => (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response(),
256    }
257}
258
259/// Why a forwarded write failed. Only `NotApplied` is safe to send again
260#[derive(Debug)]
261pub(crate) enum ForwardError {
262    /// The connection was refused, or the node said it is not the leader
263    NotApplied(String),
264    /// It may or may not have been applied, so it must not be sent again
265    Unknown(anyhow::Error),
266}
267
268pub(crate) async fn forward(
269    addr: &str,
270    secret: &str,
271    req: &Request,
272) -> Result<Reply, ForwardError> {
273    let res = CLIENT
274        .post(format!("http://{addr}/cluster/write"))
275        .header(SECRET_HEADER, secret)
276        .json(req)
277        .send()
278        .await
279        .map_err(|e| {
280            if e.is_connect() {
281                ForwardError::NotApplied(e.to_string())
282            } else {
283                ForwardError::Unknown(e.into())
284            }
285        })?;
286    let status = res.status();
287    if status == StatusCode::MISDIRECTED_REQUEST {
288        let body = res.text().await.unwrap_or_default();
289        return Err(ForwardError::NotApplied(format!(
290            "{addr} is not the leader: {body}"
291        )));
292    }
293
294    if !status.is_success() {
295        let body = res.text().await.unwrap_or_default();
296        return Err(ForwardError::Unknown(anyhow::anyhow!(
297            "leader at {addr} refused the write ({status}): {body}"
298        )));
299    }
300
301    res.json()
302        .await
303        .map_err(|e| ForwardError::Unknown(e.into()))
304}
305
306pub(crate) struct HttpNetwork {
307    pub secret: Arc<String>,
308}
309
310impl RaftNetworkFactory<TypeConfig> for HttpNetwork {
311    type Network = Peer;
312
313    async fn new_client(&mut self, target: NodeId, node: &BasicNode) -> Peer {
314        Peer {
315            target,
316            addr: node.addr.clone(),
317            secret: self.secret.clone(),
318        }
319    }
320}
321
322pub(crate) struct Peer {
323    target: NodeId,
324    addr: String,
325    secret: Arc<String>,
326}
327
328impl Peer {
329    async fn send<Req, Resp, E>(
330        &self,
331        path: &str,
332        req: Req,
333        option: &RPCOption,
334    ) -> Result<Resp, RPCError<NodeId, BasicNode, RaftError<NodeId, E>>>
335    where
336        Req: Serialize,
337        Resp: DeserializeOwned,
338        E: std::error::Error + DeserializeOwned,
339    {
340        let res = CLIENT
341            .post(format!("http://{}/raft/{path}", self.addr))
342            .header(SECRET_HEADER, self.secret.as_str())
343            .timeout(option.hard_ttl())
344            .json(&req)
345            .send()
346            .await
347            .map_err(|e| {
348                if e.is_connect() {
349                    RPCError::Unreachable(Unreachable::new(&e))
350                } else {
351                    RPCError::Network(NetworkError::new(&e))
352                }
353            })?;
354        let res = res
355            .error_for_status()
356            .map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
357        let body: Result<Resp, RaftError<NodeId, E>> = res
358            .json()
359            .await
360            .map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
361        body.map_err(|e| RPCError::RemoteError(RemoteError::new(self.target, e)))
362    }
363}
364
365impl RaftNetwork<TypeConfig> for Peer {
366    async fn append_entries(
367        &mut self,
368        req: AppendEntriesRequest<TypeConfig>,
369        option: RPCOption,
370    ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
371        self.send("append", req, &option).await
372    }
373
374    async fn install_snapshot(
375        &mut self,
376        req: InstallSnapshotRequest<TypeConfig>,
377        option: RPCOption,
378    ) -> Result<
379        InstallSnapshotResponse<NodeId>,
380        RPCError<NodeId, BasicNode, RaftError<NodeId, InstallSnapshotError>>,
381    > {
382        self.send("snapshot", req, &option).await
383    }
384
385    async fn vote(
386        &mut self,
387        req: VoteRequest<NodeId>,
388        option: RPCOption,
389    ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
390        self.send("vote", req, &option).await
391    }
392}
orch/cluster/network.rs

orch/cluster/
network.rs
