seyd_signal_client/
lib.rs

1// Copyright 2026 Anton Gravestam
2// SPDX-License-Identifier: Apache-2.0
3//! Robot-side client for the Seyd signal protocol v2.
4//!
5//! Contract: `docs/protocol/signal-v2.md`. This crate owns the WebSocket,
6//! the challenge/response authentication with the robot's Ed25519 identity,
7//! reconnection with backoff, and the re-announce after a reconnect. It knows
8//! nothing about QUIC, media or NAT — the caller supplies the `Announce` and
9//! consumes `Event`s.
10
11pub mod identity;
12pub mod messages;
13
14use futures_util::{SinkExt, StreamExt};
15use messages::*;
16use std::sync::Arc;
17use std::time::{Duration, Instant};
18use tokio::sync::{mpsc, watch, Mutex};
19use tokio_tungstenite::tungstenite::Message;
20
21pub use identity::Identity;
22
23#[derive(Debug, thiserror::Error)]
24pub enum Error {
25    #[error("websocket: {0}")]
26    Ws(Box<tokio_tungstenite::tungstenite::Error>),
27    #[error("protocol: {0}")]
28    Protocol(String),
29    #[error("denied: {0}")]
30    Denied(String),
31    #[error("json: {0}")]
32    Json(#[from] serde_json::Error),
33}
34
35impl From<tokio_tungstenite::tungstenite::Error> for Error {
36    fn from(e: tokio_tungstenite::tungstenite::Error) -> Self {
37        Error::Ws(Box::new(e))
38    }
39}
40
41/// Inbound events the agent must react to.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub enum Event {
44    /// Authenticated and ready; the client re-sends the current announce itself.
45    Connected,
46    PilotConnecting {
47        session_id: String,
48        pilot_ip: Option<String>,
49        role: Role,
50        direction: Direction,
51    },
52    Punch {
53        session_id: String,
54        pilot_ip: Option<String>,
55    },
56    SessionRevoked {
57        session_id: String,
58        reason: String,
59    },
60    /// A pilot is waiting on the cloud relay for this session (ADR 0010).
61    RelayOpen {
62        session_id: String,
63        url: String,
64        token: String,
65        role: Role,
66        pilot_ip: Option<String>,
67    },
68    /// The socket dropped; the client is reconnecting. Sessions already
69    /// established over QUIC are unaffected.
70    Disconnected,
71    /// Authentication was refused. The client keeps retrying slowly, but this
72    /// is almost always a configuration problem the operator must see.
73    Denied {
74        reason: String,
75    },
76}
77
78#[derive(Clone)]
79pub struct Config {
80    pub signal_url: String,
81    pub robot_id: String,
82    pub agent_version: String,
83    pub heartbeat: Duration,
84}
85
86/// Outbound side. Cheap to clone; safe to call from any task.
87#[derive(Clone)]
88pub struct Handle {
89    tx: mpsc::UnboundedSender<Outbound>,
90    announce: Arc<Mutex<Option<Announce>>>,
91    sessions: Arc<Mutex<Vec<String>>>,
92    status: Arc<Mutex<Option<serde_json::Value>>>,
93    connected: watch::Receiver<bool>,
94}
95
96impl Handle {
97    /// Set the current announce; sent now if connected and again after every reconnect.
98    pub async fn announce(&self, a: Announce) {
99        *self.announce.lock().await = Some(a.clone());
100        // If not connected, the post-auth re-announce delivers it; sending
101        // here too would announce twice.
102        if *self.connected.borrow() {
103            let _ = self.tx.send(Outbound::Announce(a));
104        }
105    }
106    pub async fn set_sessions(&self, ids: Vec<String>) {
107        *self.sessions.lock().await = ids;
108    }
109    pub async fn set_status(&self, status: Option<serde_json::Value>) {
110        *self.status.lock().await = status;
111    }
112    pub fn session_accepted(&self, session_id: &str, path_label: &str) {
113        let _ = self.tx.send(Outbound::SessionAccepted {
114            session_id: session_id.into(),
115            path_label: path_label.into(),
116        });
117    }
118    pub fn session_ended(&self, session_id: &str, reason: &str) {
119        let _ = self.tx.send(Outbound::SessionEnded {
120            session_id: session_id.into(),
121            reason: reason.into(),
122        });
123    }
124    pub fn is_connected(&self) -> bool {
125        *self.connected.borrow()
126    }
127}
128
129/// Start the client. Returns the outbound handle and the inbound event stream.
130/// The connection task runs until the handle and receiver are dropped.
131pub fn start(cfg: Config, identity: Identity) -> (Handle, mpsc::Receiver<Event>) {
132    let (tx, rx) = mpsc::unbounded_channel();
133    let (ev_tx, ev_rx) = mpsc::channel(64);
134    let (conn_tx, conn_rx) = watch::channel(false);
135    let handle = Handle {
136        tx,
137        announce: Arc::new(Mutex::new(None)),
138        sessions: Arc::new(Mutex::new(Vec::new())),
139        status: Arc::new(Mutex::new(None)),
140        connected: conn_rx,
141    };
142    tokio::spawn(run(cfg, identity, handle.clone(), rx, ev_tx, conn_tx));
143    (handle, ev_rx)
144}
145
146const BASE_BACKOFF: Duration = Duration::from_secs(1);
147const MAX_BACKOFF: Duration = Duration::from_secs(30);
148/// A connection that stayed up this long was healthy, not flapping, so the
149/// failure that ends it starts the backoff over. Without this a robot up for
150/// hours pays the full `MAX_BACKOFF` for a one-second blip.
151const STABLE_SESSION: Duration = Duration::from_secs(30);
152
153async fn run(
154    cfg: Config,
155    identity: Identity,
156    handle: Handle,
157    mut rx: mpsc::UnboundedReceiver<Outbound>,
158    ev_tx: mpsc::Sender<Event>,
159    conn_tx: watch::Sender<bool>,
160) {
161    let mut backoff = BASE_BACKOFF;
162    loop {
163        let started = Instant::now();
164        let outcome = session(&cfg, &identity, &handle, &mut rx, &ev_tx, &conn_tx).await;
165        // Reset before logging, so the delay reported is the one actually slept.
166        if started.elapsed() >= STABLE_SESSION {
167            backoff = BASE_BACKOFF;
168        }
169        match outcome {
170            Ok(()) => backoff = BASE_BACKOFF,
171            Err(Error::Denied(reason)) => {
172                // A bare reason code leaves an operator with nowhere to go, and
173                // these three are the ones that actually happen in the field.
174                match reason.as_str() {
175                    "unknown-robot" => tracing::error!(
176                        "signal server denied this robot: it is not enrolled. Redeem an \
177                         enrolment token from the console — `seydd --config <file> enrol \
178                         --token seyd_enr_…`, or set SEYD_ENROLMENT_TOKEN and restart."
179                    ),
180                    "key-mismatch" => tracing::error!(
181                        "signal server denied this robot: it is enrolled with a different \
182                         key. Another agent is using this robot id, or the credential file \
183                         was replaced. Delete the robot in the console and enrol again."
184                    ),
185                    _ => tracing::error!(%reason, "signal server denied this robot"),
186                }
187                let _ = ev_tx.send(Event::Denied { reason }).await;
188                backoff = MAX_BACKOFF;
189            }
190            Err(e) => {
191                tracing::warn!(error = %e, "signal connection failed; retrying in {backoff:?}")
192            }
193        }
194        let _ = conn_tx.send(false);
195        if ev_tx.send(Event::Disconnected).await.is_err() {
196            return; // consumer gone
197        }
198        tokio::time::sleep(backoff).await;
199        backoff = (backoff * 2).min(MAX_BACKOFF);
200    }
201}
202
203async fn session(
204    cfg: &Config,
205    identity: &Identity,
206    handle: &Handle,
207    rx: &mut mpsc::UnboundedReceiver<Outbound>,
208    ev_tx: &mpsc::Sender<Event>,
209    conn_tx: &watch::Sender<bool>,
210) -> Result<(), Error> {
211    let (ws, _) = tokio_tungstenite::connect_async(&cfg.signal_url).await?;
212    let (mut sink, mut stream) = ws.split();
213    tracing::info!(url = %cfg.signal_url, "signal connected");
214
215    // ── challenge / auth ─────────────────────────────────────────────────
216    let challenge: Inbound = next_json(&mut stream).await?;
217    let nonce = match challenge {
218        Inbound::Challenge { nonce } => nonce,
219        other => {
220            return Err(Error::Protocol(format!(
221                "expected challenge, got {other:?}"
222            )))
223        }
224    };
225    let nonce_bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, &nonce)
226        .map_err(|e| Error::Protocol(format!("bad nonce: {e}")))?;
227    let auth = Outbound::Auth {
228        v: 2,
229        role: "robot".into(),
230        robot_id: cfg.robot_id.clone(),
231        public_key: identity.public_key_b64(),
232        sig: identity.sign_b64(&nonce_bytes),
233        agent_version: cfg.agent_version.clone(),
234    };
235    sink.send(Message::Text(serde_json::to_string(&auth)?.into()))
236        .await?;
237    match next_json(&mut stream).await? {
238        Inbound::AuthOk { .. } => {}
239        Inbound::Denied { reason } => return Err(Error::Denied(reason)),
240        other => return Err(Error::Protocol(format!("expected auth-ok, got {other:?}"))),
241    }
242    let _ = conn_tx.send(true);
243    let _ = ev_tx.send(Event::Connected).await;
244
245    // Re-announce whatever we last knew.
246    if let Some(a) = handle.announce.lock().await.clone() {
247        sink.send(Message::Text(
248            serde_json::to_string(&Outbound::Announce(a))?.into(),
249        ))
250        .await?;
251    }
252
253    let mut heartbeat = tokio::time::interval(cfg.heartbeat);
254    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
255    heartbeat.tick().await; // first tick is immediate; skip it
256
257    loop {
258        tokio::select! {
259            msg = stream.next() => {
260                let msg = match msg {
261                    Some(Ok(Message::Text(t))) => t,
262                    Some(Ok(Message::Ping(p))) => { sink.send(Message::Pong(p)).await?; continue; }
263                    Some(Ok(Message::Close(_))) | None => return Ok(()),
264                    Some(Ok(_)) => continue,
265                    Some(Err(e)) => return Err(e.into()),
266                };
267                match serde_json::from_str::<Inbound>(&msg) {
268                    Ok(Inbound::PilotConnecting { session_id, pilot_ip, role, direction }) => {
269                        let _ = ev_tx.send(Event::PilotConnecting { session_id, pilot_ip, role, direction }).await;
270                    }
271                    Ok(Inbound::Punch { session_id, pilot_ip }) => {
272                        let _ = ev_tx.send(Event::Punch { session_id, pilot_ip }).await;
273                    }
274                    Ok(Inbound::SessionRevoked { session_id, reason }) => {
275                        let _ = ev_tx.send(Event::SessionRevoked { session_id, reason }).await;
276                    }
277                    Ok(Inbound::RelayOpen { session_id, url, token, role, pilot_ip }) => {
278                        let _ = ev_tx.send(Event::RelayOpen { session_id, url, token, role, pilot_ip }).await;
279                    }
280                    Ok(Inbound::Denied { reason }) => return Err(Error::Denied(reason)),
281                    Ok(other) => tracing::debug!(?other, "ignoring signal message"),
282                    Err(e) => tracing::debug!(error = %e, raw = %msg, "unparseable signal message"),
283                }
284            }
285            out = rx.recv() => {
286                let Some(out) = out else { return Ok(()) };
287                sink.send(Message::Text(serde_json::to_string(&out)?.into())).await?;
288            }
289            _ = heartbeat.tick() => {
290                let hb = Outbound::Heartbeat {
291                    sessions: handle.sessions.lock().await.clone(),
292                    status: handle.status.lock().await.clone(),
293                };
294                sink.send(Message::Text(serde_json::to_string(&hb)?.into())).await?;
295            }
296        }
297    }
298}
299
300async fn next_json<S>(stream: &mut S) -> Result<Inbound, Error>
301where
302    S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
303{
304    loop {
305        match stream.next().await {
306            Some(Ok(Message::Text(t))) => return Ok(serde_json::from_str(&t)?),
307            Some(Ok(Message::Close(_))) | None => {
308                return Err(Error::Protocol("closed during handshake".into()))
309            }
310            Some(Ok(_)) => continue,
311            Some(Err(e)) => return Err(e.into()),
312        }
313    }
314}
seyd_signal_client/lib.rs

seyd_signal_client/
lib.rs
