macula_rust/station_link/
call.rs

1//! Calls on a link: a CALL signed with the link's identity key, answered by
2//! the RESULT or ERROR that verifies for it: a provider reply signed by its
3//! target, or a relay error the connected station signed. A reply that does
4//! not verify is counted and ignored, and the call keeps waiting, as macula's
5//! link does. On a v4 link the liveness probe is a call too.
6
7use std::sync::{Arc, Weak};
8use std::time::Duration;
9
10use tokio::sync::oneshot;
11
12use crate::cbor::Value;
13use crate::frame::{self, RequestSpec, VerifiedRequest};
14use crate::handshake::VERSION_5;
15
16use super::confidential::{reply_outcome, sealed_request, stated, CallSeal, Seal};
17use super::{now_ms, Inner, Link, LinkError};
18use crate::seal;
19
20/// macula's default timeout for a call.
21pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(5);
22/// The longest a call waits: the far edge of a provider's deadline window.
23pub const MAX_CALL_TIMEOUT: Duration = Duration::from_secs(10 * 60);
24
25const LIVENESS_EVERY: Duration = Duration::from_secs(30);
26const LIVENESS_TIMEOUT: Duration = Duration::from_secs(30);
27const LIVENESS_PROCEDURE: &str = "_macula.ping";
28
29/// One request: the realm and procedure, the node it targets (the station
30/// itself when zero, as for `_dht.*`; the provider's node_id otherwise), its
31/// payload, how long to wait (the default when zero), a UCAN token and its
32/// delegation chain's proofs for a gated procedure, and how it is kept. A
33/// call to a provider must state `seal`: sealed to the key its verified
34/// advertisement names, or clear by the application's decision; one that
35/// states neither is refused no_signed_state before anything is sent. A call
36/// to the station is always clear.
37#[derive(Debug, Clone, PartialEq)]
38pub struct Call {
39    pub realm: [u8; 32],
40    pub procedure: String,
41    pub target: [u8; 32],
42    pub payload: Value,
43    pub timeout: Duration,
44    pub token: Option<Vec<u8>>,
45    pub proofs: Vec<Vec<u8>>,
46    pub seal: Option<Seal>,
47}
48
49impl Default for Call {
50    fn default() -> Self {
51        Call {
52            realm: [0; 32],
53            procedure: String::new(),
54            target: [0; 32],
55            payload: Value::Map(Vec::new()),
56            timeout: Duration::ZERO,
57            token: None,
58            proofs: Vec::new(),
59            seal: None,
60        }
61    }
62}
63
64/// A call waiting for its reply, with what its sealed request agreed.
65pub(super) struct Pending {
66    pub(super) request: VerifiedRequest,
67    pub(super) seal: Option<CallSeal>,
68    pub(super) outcome: oneshot::Sender<Result<Value, LinkError>>,
69}
70
71impl Link {
72    /// Signs `c` as a CALL, sends it, and waits for the RESULT or ERROR that
73    /// verifies for it.
74    pub async fn call(&self, c: Call) -> Result<Value, LinkError> {
75        call(&self.inner, c).await
76    }
77}
78
79pub(super) async fn call(inner: &Arc<Inner>, c: Call) -> Result<Value, LinkError> {
80    let timeout = if c.timeout.is_zero() {
81        DEFAULT_CALL_TIMEOUT
82    } else {
83        c.timeout.min(MAX_CALL_TIMEOUT)
84    };
85    stated(&c.target, &inner.station.node_id, &c.seal)?;
86    let target = if c.target == [0; 32] {
87        inner.station.node_id
88    } else {
89        c.target
90    };
91    let mut request_id = [0u8; 16];
92    aws_lc_rs::rand::fill(&mut request_id).map_err(|_| LinkError::Io("no randomness".into()))?;
93    let deadline = (now_ms() + timeout.as_millis() as i64) as u64;
94    let (sealed, call_seal) = match &c.seal {
95        Some(Seal::To(key)) => {
96            let (sealed, s) = sealed_request(
97                inner.profile,
98                key,
99                seal::FRAME_CALL,
100                c.realm,
101                &c.procedure,
102                inner.self_id,
103                target,
104                request_id,
105                deadline,
106                &c.payload,
107            )?;
108            (Some(sealed), Some(s))
109        }
110        _ => (None, None),
111    };
112    let signed = frame::sign_call(
113        &RequestSpec {
114            request_id,
115            realm: c.realm,
116            procedure: c.procedure,
117            target,
118            deadline,
119            payload: c.payload,
120            sealed,
121            mode: None,
122            token: c.token,
123            proofs: c.proofs,
124            source_route: None,
125            retry_budget: None,
126        },
127        &inner.key,
128    )?;
129    let request = frame::verify_request(&signed, inner.profile)?;
130    let (outcome_tx, outcome) = oneshot::channel();
131    {
132        let mut state = inner.lock();
133        if let Some(e) = &state.ended {
134            return Err(e.clone());
135        }
136        state.pending.insert(
137            request_id,
138            Pending {
139                request,
140                seal: call_seal,
141                outcome: outcome_tx,
142            },
143        );
144    }
145    let forget = || {
146        inner.lock().pending.remove(&request_id);
147    };
148    if let Err(e) = inner.write_control(&signed).await {
149        forget();
150        return Err(e);
151    }
152    let answered = tokio::time::timeout(timeout, outcome).await;
153    forget();
154    match answered {
155        Ok(Ok(outcome)) => outcome,
156        Ok(Err(_)) => Err(inner.lock().ended.clone().unwrap_or(LinkError::Closed)),
157        Err(_) => Err(LinkError::CallTimeout),
158    }
159}
160
161/// A RESULT or ERROR matched to its pending call by the ids it claims, and
162/// handed on only once it verifies for that call's request.
163pub(super) fn replied(inner: &Arc<Inner>, v: &Value) {
164    let Ok((request_id, _)) = frame::claimed_reply_ids(v) else {
165        inner.count("malformed_reply");
166        return;
167    };
168    let (request, seal) = match inner.lock().pending.get(&request_id) {
169        Some(p) => (p.request.clone(), p.seal.clone()),
170        None => {
171            inner.count("unmatched_reply");
172            return;
173        }
174    };
175    let Some(outcome) = verified_outcome(inner, v, &request, seal.as_ref()) else {
176        inner.count("unverified_reply");
177        return;
178    };
179    if let Some(p) = inner.lock().pending.remove(&request_id) {
180        let _ = p.outcome.send(outcome);
181    }
182}
183
184fn verified_outcome(
185    inner: &Inner,
186    v: &Value,
187    request: &VerifiedRequest,
188    seal: Option<&CallSeal>,
189) -> Option<Result<Value, LinkError>> {
190    if v.get("reply").is_some() {
191        let reply = frame::verify_reply(v, request, inner.profile).ok()?;
192        return Some(reply_outcome(reply, request, seal));
193    }
194    let relayed =
195        frame::verify_relay_error(v, request, inner.profile, &inner.station.node_id).ok()?;
196    Some(Err(LinkError::Relay {
197        reported_by: relayed.reported_by,
198        code: relayed.code,
199    }))
200}
201
202/// A liveness probe every 30 seconds; two misses in a row end the link. On v5
203/// the probe is a liveness_ping, answered by the station's connection with a
204/// liveness_pong of its nonce, and nothing is signed. On v4 it is a
205/// `_macula.ping` call: any verified answer counts, a provider's or a relay
206/// error from the station alike, as it proves the station is there.
207pub(super) async fn probe(link: Weak<Inner>) {
208    let mut misses = 0;
209    loop {
210        let Some(mut done) = link.upgrade().map(|l| l.done_rx.clone()) else {
211            return;
212        };
213        tokio::select! {
214            _ = done.wait_for(|ended| *ended) => return,
215            _ = tokio::time::sleep(LIVENESS_EVERY) => {}
216        }
217        let Some(inner) = link.upgrade() else { return };
218        let outcome = match inner.version {
219            VERSION_5 => probe_v5(&inner).await,
220            _ => call(
221                &inner,
222                Call {
223                    procedure: LIVENESS_PROCEDURE.to_string(),
224                    timeout: LIVENESS_TIMEOUT,
225                    ..Call::default()
226                },
227            )
228            .await
229            .map(|_| ()),
230        };
231        misses = if matches!(outcome, Err(LinkError::CallTimeout)) {
232            misses + 1
233        } else {
234            0
235        };
236        if misses >= 2 {
237            inner.end(LinkError::LivenessLost);
238            return;
239        }
240    }
241}
242
243/// One v5 liveness probe: answered when a liveness_pong of its nonce arrives
244/// within [`LIVENESS_TIMEOUT`], [`LinkError::CallTimeout`] when none does.
245async fn probe_v5(inner: &Inner) -> Result<(), LinkError> {
246    let mut nonce = [0u8; frame::LIVENESS_NONCE_SIZE];
247    aws_lc_rs::rand::fill(&mut nonce).map_err(|_| LinkError::Io("no randomness".into()))?;
248    let mut pongs = inner.pongs_rx.lock().await;
249    // A pong that arrived after its probe's deadline must not take the place
250    // of this probe's.
251    while pongs.try_recv().is_ok() {}
252    inner
253        .send_control(&frame::liveness_ping_frame(&nonce))
254        .await?;
255    let mut done = inner.done_rx.clone();
256    let answered = async {
257        while let Some(pong) = pongs.recv().await {
258            if pong == nonce {
259                return Ok(());
260            }
261        }
262        Err(LinkError::Closed)
263    };
264    tokio::select! {
265        _ = done.wait_for(|ended| *ended) => Err(LinkError::Closed),
266        outcome = tokio::time::timeout(LIVENESS_TIMEOUT, answered) => {
267            outcome.unwrap_or(Err(LinkError::CallTimeout))
268        }
269    }
270}
macula_rust/station_link/call.rs

macula_rust/station_link/
call.rs
