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

macula_rust/station_link/
call.rs
