macula_rust/
station_link.rs

1//! A client's link to one macula station, as macula_station_link and
2//! macula-go's stationlink are: a QUIC connection dialed to the station its
3//! target pins, one bidirectional control stream, the connection handshake on
4//! it (v5, or v4 once after a station refuses v5), then status statements
5//! both ways and every frame of the session.
6//!
7//! After HELLO the link sends its own status statement at every reissue of
8//! the node's statement issuer, and ends when the station's statement is five
9//! minutes past its expiry or the station's TLS binding reaches its
10//! not_after. On a v5 link no frame carries a neighbour signature: the
11//! session proofs authenticated the station once, and a neighbour-signed
12//! frame ends the link. On a v4 link, in pq_hybrid, every control frame is
13//! neighbour-signed with a sequence number per direction, from 0 after HELLO;
14//! a frame out of sequence ends the link. A liveness probe every 30 seconds
15//! ends the link after two misses in a row: on v5 a liveness_ping, on v4 a
16//! `_macula.ping` call.
17
18mod admission;
19mod authorize;
20mod call;
21mod confidential;
22mod dht;
23mod framing;
24mod pubsub;
25mod report;
26mod serve;
27mod stream;
28mod versions;
29
30pub use admission::{Admission, AdmissionLimits};
31pub use call::{Call, DEFAULT_CALL_TIMEOUT, MAX_CALL_TIMEOUT};
32pub use confidential::{
33    is_clear_refusal, Confidentiality, ConfidentialityError, ConfidentialityReason, Seal,
34};
35pub use pubsub::{Event, EventDedup, Publication, PublicationSeq, SignedPublication, Subscription};
36pub use report::{Report, ReportError};
37pub use serve::{handler, BoxFuture, Handler, Offer, Request, Served, StreamOffer};
38pub(crate) use stream::Reseal;
39pub use stream::{
40    stream_handler, Stream, StreamCall, StreamEvent, StreamHandler, DEFAULT_STREAM_DEADLINE,
41};
42pub use versions::{forget_v5_peer, handshake_counters};
43
44use std::collections::HashMap;
45use std::fmt;
46use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
47use std::sync::{Arc, Mutex, MutexGuard};
48use std::time::Duration;
49
50use sha2::{Digest, Sha384};
51use tokio::sync::watch;
52
53use crate::cbor::{self, Value};
54use crate::frame::{self, FrameError, Liveness, NeighbourLink, NeighbourPeer};
55use crate::handshake::{
56    self, ClientSession, Exporter, HandshakeError, Peer, RefusalCode, Station, VERSION, VERSION_5,
57};
58use crate::node_key::NodeKey;
59use crate::profile::Profile;
60use crate::record::RecordError;
61use crate::seal::Keyring;
62use crate::statement_issuer::{
63    ConnectMaterial, IssuerError, StatementIssuer, StatementSubscription,
64};
65use crate::transport::{self, DialError, Target};
66
67use framing::{read_frame, FrameWriter, HANDSHAKE_FRAME_BYTES, MAX_FRAME_BYTES};
68
69/// How long the handshake may take, as macula's 30 seconds.
70pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
71
72/// How long Close waits, after its GOODBYE, for the station to close the
73/// connection before closing it itself.
74const CLOSE_LINGER: Duration = Duration::from_secs(1);
75
76/// How long past a statement's expiry the link keeps a station whose next
77/// statement has not arrived: macula's five minutes.
78const STATUS_GRACE_MS: i64 = 5 * 60 * 1000;
79
80/// Why a link or one of its operations failed.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub enum LinkError {
83    /// The configuration cannot be dialed: a key of another profile than the
84    /// target's, or admission limits macula would not start with.
85    InvalidConfig(String),
86    /// The dial failed.
87    Dial(String),
88    /// The handshake refused, or was refused.
89    Handshake(HandshakeError),
90    /// The handshake did not finish within [`HANDSHAKE_TIMEOUT`].
91    HandshakeTimeout,
92    /// The issuer had no CONNECT material in force.
93    Issuer(IssuerError),
94    /// The QUIC connection or a stream failed.
95    Io(String),
96    /// A frame longer than the cap in force.
97    FrameTooLarge(usize),
98    /// A frame that could not be built, or one received that does not
99    /// verify.
100    Frame(FrameError),
101    /// A record that could not be built, or one found that does not verify.
102    Record(RecordError),
103    /// The station's status statement lapsed past the grace.
104    StatusExpired,
105    /// The station's TLS binding reached its not_after.
106    BindingExpired,
107    /// The link was closed by its owner.
108    Closed,
109    /// The station ended the link with a GOODBYE, for this reason.
110    Goodbye(String),
111    /// The station missed two liveness probes in a row.
112    LivenessLost,
113    /// A station this process completed handshake v5 with that now answers
114    /// v4, or a v4 link to it that a v5 completion ended: refused, not
115    /// retried, until [`forget_v5_peer`].
116    V5DowngradeRefused,
117    /// No verified reply answered the call within its timeout.
118    CallTimeout,
119    /// A provider's ERROR for the call.
120    Provider {
121        responded_by: [u8; 32],
122        code: String,
123        detail: Option<String>,
124    },
125    /// A relay error the connected station reported for the call.
126    Relay { reported_by: [u8; 32], code: String },
127    /// A find_record the station answered not_found.
128    RecordNotFound,
129    /// A station reply to a DHT call in a shape that call never answers with.
130    UnexpectedReply(String),
131    /// An offer without exactly one handler, or an org procedure's without
132    /// the realm key.
133    InvalidOffer,
134    /// A procedure without a namespace, which nobody can authorize.
135    NoOrg,
136    /// A procedure already served on this link.
137    AlreadyServed,
138    /// A required-confidential procedure on a link whose `kem_advertise` is
139    /// off: it would name no key and refuse every clear call.
140    KemAdvertiseDisabled,
141    /// A served procedure withdrawn by its owner.
142    Stopped,
143    /// A stream ended by a STREAM_ERROR: the peer's, the station's relay
144    /// error (`relay`), or this side's own abort.
145    Stream {
146        code: String,
147        message: String,
148        relay: bool,
149    },
150    /// A stream that ended normally.
151    EndOfStream,
152    /// A send on a stream this side has ended.
153    StreamClosed,
154    /// A STREAM_OPEN over the 1 MiB a peer reads of one.
155    StreamOpenTooLarge(usize),
156    /// A call or an open that could not be kept confidential, so it was not
157    /// made, or failed rather than be taken in the clear.
158    Confidentiality(ConfidentialityError),
159    /// The provider's sealed_refused: it could not open the request. `named`
160    /// is the key it holds now, `None` when it holds none.
161    SealedRefused { named: Option<[u8; 8]> },
162    /// A clear answer to a sealed request that nothing clear may give: not a
163    /// relay error, not a refusal from the closed set, not sealed_refused.
164    /// Refused, never taken as the answer.
165    ClearAnswerToSealed,
166    /// A provider's sealed stream that has sealed all the frames GCM's
167    /// bound allows under random nonces: another would pass it.
168    SealedFramesExhausted,
169}
170
171impl fmt::Display for LinkError {
172    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
173        match self {
174            LinkError::Provider {
175                code,
176                detail: Some(d),
177                ..
178            } => write!(f, "the provider answered {code}: {d}"),
179            LinkError::Provider { code, .. } => write!(f, "the provider answered {code}"),
180            LinkError::Relay { code, .. } => {
181                write!(f, "the station could not relay the call: {code}")
182            }
183            LinkError::Stream { code, message, .. } if !message.is_empty() => {
184                write!(f, "stream error {code}: {message}")
185            }
186            LinkError::Stream { code, .. } => write!(f, "stream error {code}"),
187            LinkError::Handshake(e) => write!(f, "handshake: {e}"),
188            LinkError::Frame(e) => write!(f, "frame: {e}"),
189            LinkError::Record(e) => write!(f, "record: {e}"),
190            LinkError::Issuer(e) => write!(f, "{e}"),
191            LinkError::Goodbye(reason) => write!(f, "the station said goodbye: {reason}"),
192            LinkError::Confidentiality(e) => write!(f, "{e}"),
193            LinkError::SealedRefused { named: Some(id) } => write!(
194                f,
195                "the provider could not open the request; it holds key {}",
196                id.iter().map(|b| format!("{b:02x}")).collect::<String>()
197            ),
198            LinkError::SealedRefused { named: None } => {
199                f.write_str("the provider opens no sealed payload")
200            }
201            LinkError::ClearAnswerToSealed => f.write_str("a clear answer to a sealed request"),
202            LinkError::KemAdvertiseDisabled => {
203                f.write_str("a required-confidential procedure needs kem_advertise on")
204            }
205            LinkError::SealedFramesExhausted => {
206                f.write_str("this sealed stream has sealed all the frames it may")
207            }
208            other => write!(f, "{other:?}"),
209        }
210    }
211}
212
213impl std::error::Error for LinkError {}
214
215impl From<FrameError> for LinkError {
216    fn from(e: FrameError) -> Self {
217        LinkError::Frame(e)
218    }
219}
220
221impl From<RecordError> for LinkError {
222    fn from(e: RecordError) -> Self {
223        LinkError::Record(e)
224    }
225}
226
227impl From<HandshakeError> for LinkError {
228    fn from(e: HandshakeError) -> Self {
229        LinkError::Handshake(e)
230    }
231}
232
233impl From<DialError> for LinkError {
234    fn from(e: DialError) -> Self {
235        LinkError::Dial(e.to_string())
236    }
237}
238
239/// What a link is dialed with: the station to reach, the node's identity
240/// key, the statement issuer that holds its CONNECT key and statements, and
241/// the realm membership endorsement to present, empty for none. Links of one
242/// node share their publication seq, admission and dedup; a link given none
243/// makes its own.
244pub struct Config {
245    pub target: Target,
246    pub identity: Arc<NodeKey>,
247    pub issuer: StatementIssuer,
248    pub member_endorsement: Vec<u8>,
249    pub publication_seq: Option<Arc<PublicationSeq>>,
250    pub admission: Option<Arc<Admission>>,
251    pub dedup: Option<Arc<EventDedup>>,
252    /// This link's place in the admission: the station it dialed, host:port,
253    /// when `None`.
254    pub share: Option<String>,
255    /// This identity's KEM keys (macula 13, E2E design amendment A1): with
256    /// them the link opens sealed requests; without them it refuses them,
257    /// holding no key. Links of one identity share one.
258    pub keyring: Option<Arc<Keyring>>,
259    /// Names the keyring's current key in the advertisements of procedures
260    /// served confidentially ([`Confidentiality::Preferred`] or
261    /// [`Confidentiality::Required`]). Off by default: switch it on only once
262    /// every station runs macula 12.11 or later, which stores and routes a
263    /// keyed advertisement, and every caller runs 13. It needs a keyring.
264    pub kem_advertise: bool,
265}
266
267impl Config {
268    /// A link's configuration with none of the shared parts.
269    pub fn new(target: Target, identity: Arc<NodeKey>, issuer: StatementIssuer) -> Config {
270        Config {
271            target,
272            identity,
273            issuer,
274            member_endorsement: Vec::new(),
275            publication_seq: None,
276            admission: None,
277            dedup: None,
278            share: None,
279            keyring: None,
280            kem_advertise: false,
281        }
282    }
283}
284
285/// A handshaked link to one station. Cloning it shares the link.
286#[derive(Clone)]
287pub struct Link {
288    inner: Arc<Inner>,
289}
290
291struct Inner {
292    /// Distinct for every link a process dials.
293    serial: u64,
294    connection: quinn::Connection,
295    _endpoint: quinn::Endpoint,
296    control: FrameWriter,
297    profile: Profile,
298    key: Arc<NodeKey>,
299    self_id: [u8; 32],
300    station: Station,
301    station_capabilities: u64,
302    connection_hash: [u8; 48],
303    /// The handshake version the link completed: 4 or 5.
304    version: i64,
305    /// v5 liveness answers, for the probe.
306    pongs: tokio::sync::mpsc::Sender<[u8; frame::LIVENESS_NONCE_SIZE]>,
307    /// A liveness_pong is being written (macula-rust#9): at most one is in
308    /// flight, so a station's pings cannot pile up tasks on this link.
309    pong_in_flight: AtomicBool,
310    pongs_rx: tokio::sync::Mutex<tokio::sync::mpsc::Receiver<[u8; frame::LIVENESS_NONCE_SIZE]>>,
311    /// Orders a neighbour signature's seq with its write.
312    send_seq: tokio::sync::Mutex<u64>,
313    status_deadline: AtomicI64,
314    publication_seq: Arc<PublicationSeq>,
315    admission: Arc<Admission>,
316    dedup: Arc<EventDedup>,
317    share: String,
318    keyring: Option<Arc<Keyring>>,
319    kem_advertise: bool,
320    state: Mutex<State>,
321    done_tx: watch::Sender<bool>,
322    done_rx: watch::Receiver<bool>,
323}
324
325struct State {
326    ended: Option<LinkError>,
327    /// The owner is closing the link: however it then ends, it ends
328    /// [`LinkError::Closed`].
329    closing: bool,
330    unrouted: HashMap<String, u64>,
331    pending: HashMap<[u8; 16], call::Pending>,
332    subs: HashMap<([u8; 32], String), Vec<pubsub::SubscriberSlot>>,
333    served: HashMap<([u8; 32], String), serve::ServedEntry>,
334    streams: Vec<std::sync::Weak<stream::StreamInner>>,
335}
336
337impl Link {
338    /// Dials `cfg.target`, runs the handshake as a client, and returns the
339    /// link once the station's HELLO accepts it. It dials with version 5
340    /// unless the station refused v5 in the last 10 minutes; a station never
341    /// seen on v5 that refuses v5 with `unsupported_version` is dialled once
342    /// more, on a new connection, with v4. A station seen on v5 that answers
343    /// v4 is [`LinkError::V5DowngradeRefused`]. Each handshake is bounded by
344    /// [`HANDSHAKE_TIMEOUT`].
345    pub async fn dial(cfg: Config) -> Result<Link, LinkError> {
346        if cfg.identity.profile() != cfg.target.profile {
347            return Err(LinkError::InvalidConfig(
348                "the identity key is of another profile than the target's".into(),
349            ));
350        }
351        if let Some(admission) = &cfg.admission {
352            admission.limits().validate()?;
353        }
354        if cfg
355            .keyring
356            .as_ref()
357            .is_some_and(|k| k.profile() != cfg.target.profile)
358        {
359            return Err(LinkError::InvalidConfig(
360                "the keyring is of another profile than the target's".into(),
361            ));
362        }
363        if cfg.kem_advertise && cfg.keyring.is_none() {
364            return Err(LinkError::InvalidConfig(
365                "kem_advertise names a key: it needs a keyring".into(),
366            ));
367        }
368        let node_id = cfg.target.expected_node_id;
369        let version = versions::dial_version(&node_id, std::time::Instant::now());
370        let linked = dial_once(&cfg, version).await;
371        let v5_refused = matches!(
372            linked,
373            Err(LinkError::Handshake(HandshakeError::Refused(
374                RefusalCode::UnsupportedVersion
375            )))
376        ) && version == VERSION_5;
377        if !v5_refused {
378            return linked;
379        }
380        if !versions::unsupported_version(&node_id, std::time::Instant::now()) {
381            return Err(LinkError::V5DowngradeRefused);
382        }
383        dial_once(&cfg, VERSION).await
384    }
385
386    /// The handshake version the link completed: 4 or 5.
387    pub fn handshake_version(&self) -> i64 {
388        self.inner.version
389    }
390
391    /// The node_id of the station the link reached.
392    pub fn station_node_id(&self) -> [u8; 32] {
393        self.inner.station.node_id
394    }
395
396    /// A number distinct for every link this process dials.
397    pub fn serial(&self) -> u64 {
398        self.inner.serial
399    }
400
401    /// The node_id this link connected as.
402    pub fn node_id(&self) -> [u8; 32] {
403        self.inner.self_id
404    }
405
406    /// The capability bits the station's HELLO announced.
407    pub fn station_capabilities(&self) -> u64 {
408        self.inner.station_capabilities
409    }
410
411    /// The profile the link runs.
412    pub fn profile(&self) -> Profile {
413        self.inner.profile
414    }
415
416    /// Why the link ended, or `None` while it runs.
417    pub fn error(&self) -> Option<LinkError> {
418        self.inner.lock().ended.clone()
419    }
420
421    /// Waits until the link has ended, and says why.
422    pub async fn done(&self) -> LinkError {
423        let mut done = self.inner.done_rx.clone();
424        let _ = done.wait_for(|ended| *ended).await;
425        self.error().unwrap_or(LinkError::Closed)
426    }
427
428    /// The frames received that nothing on this link handles, by what they
429    /// were.
430    pub fn unrouted(&self) -> HashMap<String, u64> {
431        self.inner.lock().unrouted.clone()
432    }
433
434    /// Sends a GOODBYE with `reason`, closes the control stream's sending
435    /// side, waits up to a second for the station to close the connection,
436    /// and ends the link. Closing an ended link does nothing.
437    pub async fn close(&self, reason: &str) -> Result<(), LinkError> {
438        if !self.inner.mark_closing() {
439            return Ok(());
440        }
441        let goodbye = frame::goodbye_frame(reason, None)?;
442        let sent = self.inner.send_control(&goodbye).await;
443        if sent.is_ok() {
444            self.inner.control.finish().await;
445            let _ = tokio::time::timeout(CLOSE_LINGER, self.inner.connection.closed()).await;
446        }
447        self.inner.end(LinkError::Closed);
448        sent
449    }
450}
451
452/// One QUIC connection and one handshake in `version`, within
453/// [`HANDSHAKE_TIMEOUT`]. A failed handshake closes its connection.
454async fn dial_once(cfg: &Config, version: i64) -> Result<Link, LinkError> {
455    tokio::time::timeout(HANDSHAKE_TIMEOUT, async {
456        let dialed = transport::dial_target(&cfg.target).await?;
457        let connection = dialed.connection.clone();
458        let linked = handshaken(cfg, dialed, version).await;
459        if let Err(e) = &linked {
460            versions::count_refusal(e);
461            connection.close(0u32.into(), b"handshake failed");
462        }
463        linked
464    })
465    .await
466    .map_err(|_| LinkError::HandshakeTimeout)
467    .and_then(|linked| linked)
468}
469
470/// Runs the client's side of the handshake, in `version`, on a new control
471/// stream of `dialed`.
472async fn handshaken(
473    cfg: &Config,
474    dialed: transport::Dialed,
475    version: i64,
476) -> Result<Link, LinkError> {
477    let export = keying_exporter(dialed.connection.clone());
478    let export: &Exporter = &export;
479    let (send, mut recv) = dialed
480        .connection
481        .open_bi()
482        .await
483        .map_err(|e| LinkError::Io(format!("open the control stream: {e}")))?;
484    let control = FrameWriter::new(send);
485    control
486        .write(&handshake::opener(), HANDSHAKE_FRAME_BYTES)
487        .await?;
488    let challenge = read_frame(&mut recv, HANDSHAKE_FRAME_BYTES).await?;
489    let material = cfg.issuer.connect_material().map_err(LinkError::Issuer)?;
490    let (connect, station) = handshake::answer_challenge(
491        &challenge,
492        &client_session(cfg, &dialed.leaf, &material, version, export),
493    )?;
494    control.write(&connect, HANDSHAKE_FRAME_BYTES).await?;
495    let hello = read_frame(&mut recv, HANDSHAKE_FRAME_BYTES).await?;
496    let capabilities = handshake::read_hello(&hello, &station)?;
497    let self_id = cfg
498        .identity
499        .node_id()
500        .map_err(|e| LinkError::InvalidConfig(e.to_string()))?;
501    let statements = cfg
502        .issuer
503        .subscribe(&material.binding)
504        .map_err(LinkError::Issuer)?;
505    let inner = new_inner(
506        cfg,
507        dialed,
508        control,
509        station,
510        capabilities,
511        &challenge,
512        self_id,
513    );
514    let station_node_id = inner.station.node_id;
515    versions::completed(&station_node_id, &inner)?;
516    spawn_link_tasks(&inner, statements, recv);
517    Ok(Link { inner })
518}
519
520/// The TLS exporter of `connection`, as the v5 handshake asks for keying
521/// material.
522fn keying_exporter(
523    connection: quinn::Connection,
524) -> impl Fn(&str, &[u8], usize) -> Option<Vec<u8>> + Send + Sync {
525    move |label: &str, context: &[u8], length: usize| {
526        let mut out = vec![0; length];
527        connection
528            .export_keying_material(&mut out, label.as_bytes(), context)
529            .ok()?;
530        Some(out)
531    }
532}
533
534/// What the client brings to the handshake in `version`: the configured
535/// identity and target, the leaf it received, and the issuer's CONNECT
536/// material, at the time now.
537fn client_session<'a>(
538    cfg: &Config,
539    leaf: &'a [u8],
540    material: &'a ConnectMaterial,
541    version: i64,
542    export: &'a Exporter,
543) -> ClientSession<'a> {
544    ClientSession {
545        profile: cfg.target.profile,
546        expected_node_id: cfg.target.expected_node_id,
547        leaf,
548        identity_key: cfg.identity.public_key(),
549        connect_key: &material.key,
550        connect_binding: &material.binding,
551        connect_status: &material.status,
552        capabilities: 0,
553        now_ms: now_ms(),
554        member_endorsement: cfg.member_endorsement.clone(),
555        version,
556        export: Some(export),
557    }
558}
559
560/// The state of a link whose handshake completed with `station` on the
561/// control stream `control` of `dialed`, numbered with the next serial.
562fn new_inner(
563    cfg: &Config,
564    dialed: transport::Dialed,
565    control: FrameWriter,
566    station: Station,
567    capabilities: u64,
568    challenge: &[u8],
569    self_id: [u8; 32],
570) -> Arc<Inner> {
571    let (done_tx, done_rx) = watch::channel(false);
572    let share = cfg
573        .share
574        .clone()
575        .unwrap_or_else(|| format!("{}:{}", cfg.target.host, cfg.target.port));
576    let (pongs, pongs_rx) = tokio::sync::mpsc::channel(1);
577    static SERIALS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
578    Arc::new(Inner {
579        serial: SERIALS.fetch_add(1, Ordering::Relaxed),
580        connection: dialed.connection,
581        _endpoint: dialed.endpoint,
582        control,
583        profile: cfg.target.profile,
584        key: cfg.identity.clone(),
585        self_id,
586        status_deadline: AtomicI64::new(station.status_expires_at + STATUS_GRACE_MS),
587        station_capabilities: capabilities,
588        connection_hash: Sha384::digest(challenge).into(),
589        version: station.version,
590        pongs,
591        pong_in_flight: AtomicBool::new(false),
592        pongs_rx: tokio::sync::Mutex::new(pongs_rx),
593        station,
594        send_seq: tokio::sync::Mutex::new(0),
595        publication_seq: cfg.publication_seq.clone().unwrap_or_default(),
596        admission: cfg
597            .admission
598            .clone()
599            .unwrap_or_else(|| Arc::new(Admission::new(AdmissionLimits::default()))),
600        dedup: cfg.dedup.clone().unwrap_or_default(),
601        share,
602        keyring: cfg.keyring.clone(),
603        kem_advertise: cfg.kem_advertise,
604        state: Mutex::new(State {
605            ended: None,
606            closing: false,
607            unrouted: HashMap::new(),
608            pending: HashMap::new(),
609            subs: HashMap::new(),
610            served: HashMap::new(),
611            streams: Vec::new(),
612        }),
613        done_tx,
614        done_rx,
615    })
616}
617
618/// Starts the tasks a linked link runs: its statements, its control
619/// stream's reader, the streams it accepts, its liveness probe, and the
620/// watches on its expiries and its connection.
621fn spawn_link_tasks(
622    inner: &Arc<Inner>,
623    statements: StatementSubscription,
624    recv: quinn::RecvStream,
625) {
626    tokio::spawn(send_statements(Arc::downgrade(inner), statements));
627    tokio::spawn(read_control(inner.clone(), recv));
628    tokio::spawn(stream::accept_streams(Arc::downgrade(inner)));
629    tokio::spawn(call::probe(Arc::downgrade(inner)));
630    tokio::spawn(watch_expiries(Arc::downgrade(inner)));
631    tokio::spawn(watch_connection(Arc::downgrade(inner)));
632}
633
634impl Inner {
635    fn lock(&self) -> MutexGuard<'_, State> {
636        self.state
637            .lock()
638            .unwrap_or_else(|poisoned| poisoned.into_inner())
639    }
640
641    fn count(&self, what: &str) {
642        *self.lock().unrouted.entry(what.to_string()).or_default() += 1;
643    }
644
645    /// Marks the link closing, so it ends as closed: false when it has
646    /// already ended.
647    fn mark_closing(&self) -> bool {
648        let mut state = self.lock();
649        if state.ended.is_some() {
650            return false;
651        }
652        state.closing = true;
653        true
654    }
655
656    /// A version-2 frame on the control stream: on v5 as it is, on v4
657    /// neighbour-signed with the next seq when the profile signs its type.
658    async fn send_control(&self, v: &Value) -> Result<(), LinkError> {
659        if self.version == VERSION_5 {
660            return self.write_control(v).await;
661        }
662        let mut seq = self.send_seq.lock().await;
663        let signed = frame::sign_neighbour(
664            v,
665            &self.key,
666            &NeighbourLink {
667                connection: self.connection_hash,
668                seq: *seq,
669            },
670        )?;
671        if frame::neighbour_signed(self.profile, &frame_type_of(v)) {
672            *seq += 1;
673        }
674        self.write_control(&signed).await
675    }
676
677    /// A frame on the control stream as it is: a data frame, which carries
678    /// its own end-to-end signature and no neighbour signature.
679    async fn write_control(&self, v: &Value) -> Result<(), LinkError> {
680        let encoded =
681            cbor::encode(v).map_err(|e| LinkError::Frame(FrameError::Payload(e.to_string())))?;
682        self.control.write(&encoded, MAX_FRAME_BYTES).await
683    }
684
685    /// Ends the link once, with `err`: pending calls, subscriptions, served
686    /// procedures and streams end with it, and the connection closes.
687    fn end(&self, err: LinkError) {
688        let mut state = self.lock();
689        let Some(err) = state.mark_ended(err) else {
690            return;
691        };
692        let pending = std::mem::take(&mut state.pending);
693        let subs = std::mem::take(&mut state.subs);
694        let served = std::mem::take(&mut state.served);
695        let streams = std::mem::take(&mut state.streams);
696        drop(state);
697        for (_, p) in pending {
698            let _ = p.outcome.send(Err(err.clone()));
699        }
700        drop(subs);
701        for s in served.into_values() {
702            s.end(err.clone());
703        }
704        for s in streams.into_iter().filter_map(|w| w.upgrade()) {
705            stream::StreamInner::end(&s, Some(err.clone()));
706        }
707        self.connection.close(0u32.into(), b"link ended");
708        if self.version == VERSION {
709            versions::v4_ended(&self.station.node_id, self.serial);
710        }
711        let _ = self.done_tx.send_replace(true);
712    }
713}
714
715impl State {
716    /// Marks the link ended once, with `err`, or as closed when its owner was
717    /// closing it: the error it ended with, or `None` when it had already
718    /// ended.
719    fn mark_ended(&mut self, err: LinkError) -> Option<LinkError> {
720        if self.ended.is_some() {
721            return None;
722        }
723        let err = if self.closing { LinkError::Closed } else { err };
724        self.ended = Some(err.clone());
725        Some(err)
726    }
727}
728
729/// Sends each statement the issuer reissues as a STATUS, until the link
730/// ends. A STATUS carries no neighbour signature and takes no seq.
731async fn send_statements(link: std::sync::Weak<Inner>, mut statements: StatementSubscription) {
732    loop {
733        let Some(done) = link.upgrade().map(|l| l.done_rx.clone()) else {
734            return;
735        };
736        let mut done = done;
737        let statement = tokio::select! {
738            _ = done.wait_for(|ended| *ended) => return,
739            statement = statements.recv() => statement,
740        };
741        let (Some(statement), Some(inner)) = (statement, link.upgrade()) else {
742            return;
743        };
744        if let Err(e) = inner
745            .control
746            .write(&handshake::status_frame(&statement), MAX_FRAME_BYTES)
747            .await
748        {
749            inner.end(e);
750            return;
751        }
752    }
753}
754
755/// Reads the station's frames until the link ends.
756async fn read_control(inner: Arc<Inner>, mut recv: quinn::RecvStream) {
757    let mut recv_seq = 0u64;
758    let mut done = inner.done_rx.clone();
759    loop {
760        let payload = tokio::select! {
761            _ = done.wait_for(|ended| *ended) => return,
762            payload = read_frame(&mut recv, MAX_FRAME_BYTES) => payload,
763        };
764        let outcome = match payload {
765            Ok(payload) => received(&inner, &payload, &mut recv_seq),
766            Err(e) => Err(e),
767        };
768        if let Err(e) = outcome {
769            inner.end(e);
770            return;
771        }
772    }
773}
774
775/// One frame from the station: a STATUS renews the station's statement, a
776/// liveness frame is answered or handed to the probe, every other frame is
777/// opened (on v4 from its neighbour signature at the next seq), and a GOODBYE
778/// ends the link.
779fn received(inner: &Arc<Inner>, payload: &[u8], recv_seq: &mut u64) -> Result<(), LinkError> {
780    let v = cbor::decode(payload).map_err(|_| LinkError::Frame(FrameError::Malformed))?;
781    let frame_type = frame_type_of(&v);
782    if frame_type == "status" {
783        let expires_at = handshake::read_status(
784            payload,
785            &Peer {
786                profile: inner.profile,
787                identity_key: inner.station.identity_key.clone(),
788                binding: inner.station.tls_binding.clone(),
789                now_ms: now_ms(),
790            },
791        )?;
792        inner
793            .status_deadline
794            .store(expires_at + STATUS_GRACE_MS, Ordering::SeqCst);
795        return Ok(());
796    }
797    if let Some((kind, nonce)) = liveness_frame(inner.version, &v)? {
798        return liveness(inner, kind, nonce);
799    }
800    let opened = opened(inner, &v, &frame_type, recv_seq)?;
801    match frame_type.as_str() {
802        "event" => pubsub::evented(inner, &opened),
803        "result" | "error" => call::replied(inner, &opened),
804        "call" => serve::called(inner, &opened),
805        "goodbye" => {
806            let reason = match opened.get("reason") {
807                Some(Value::Text(r)) => r.clone(),
808                _ => String::new(),
809            };
810            return Err(LinkError::Goodbye(reason));
811        }
812        _ => inner.count(&frame_type),
813    }
814    Ok(())
815}
816
817/// A received frame as the link reads it: on v5 with no neighbour signature,
818/// on v4 opened from its neighbour signature at the next seq.
819fn opened(
820    inner: &Inner,
821    v: &Value,
822    frame_type: &str,
823    recv_seq: &mut u64,
824) -> Result<Value, LinkError> {
825    if inner.version == VERSION_5 {
826        return Ok(frame::verify_session_frame(v)?);
827    }
828    let opened = frame::verify_neighbour(
829        v,
830        &NeighbourPeer {
831            profile: inner.profile,
832            peer_key: inner.station.identity_key.clone(),
833            connection: inner.connection_hash,
834            seq: *recv_seq,
835        },
836    )?;
837    if frame::neighbour_signed(inner.profile, frame_type) {
838        *recv_seq += 1;
839    }
840    Ok(opened)
841}
842
843/// A liveness frame as a link of `version` reads it: its kind and nonce, or
844/// `None` for any other frame. It exists only on v5, and there it is read as
845/// every v5 frame is, so one that carries a neighbour signature ends the link
846/// as malformed, as on v4 any liveness frame does.
847fn liveness_frame(
848    version: i64,
849    v: &Value,
850) -> Result<Option<(Liveness, [u8; frame::LIVENESS_NONCE_SIZE])>, LinkError> {
851    let Some(found) = frame::liveness_nonce(v) else {
852        return Ok(None);
853    };
854    if version != VERSION_5 {
855        return Err(LinkError::Frame(FrameError::Malformed));
856    }
857    frame::verify_session_frame(v)?;
858    Ok(Some(found))
859}
860
861/// Answers the station's liveness_ping with a liveness_pong of the same
862/// nonce, and hands a liveness_pong to the probe. Neither goes further.
863fn liveness(
864    inner: &Arc<Inner>,
865    kind: Liveness,
866    nonce: [u8; frame::LIVENESS_NONCE_SIZE],
867) -> Result<(), LinkError> {
868    match kind {
869        // A ping while a pong is still being written is not answered again:
870        // the station's probe needs one answer per interval, not one per
871        // ping, and a burst of pings starts one task, not one each.
872        Liveness::Ping if !pong_slot(&inner.pong_in_flight) => {}
873        Liveness::Ping => {
874            tokio::spawn(send_pong(inner.clone(), nonce));
875        }
876        Liveness::Pong => {
877            let _ = inner.pongs.try_send(nonce);
878        }
879    }
880    Ok(())
881}
882
883/// Writes the liveness_pong of `nonce`, gives back the pong slot, and ends
884/// the link when the write failed.
885async fn send_pong(inner: Arc<Inner>, nonce: [u8; frame::LIVENESS_NONCE_SIZE]) {
886    let sent = inner
887        .send_control(&frame::liveness_pong_frame(&nonce))
888        .await;
889    inner.pong_in_flight.store(false, Ordering::Release);
890    if let Err(e) = sent {
891        inner.end(e);
892    }
893}
894
895/// Takes the one pong slot: true when no pong was in flight, and it is now.
896fn pong_slot(in_flight: &AtomicBool) -> bool {
897    !in_flight.swap(true, Ordering::AcqRel)
898}
899
900/// Ends the link when the station's statement lapses past the grace, or its
901/// TLS binding reaches its not_after.
902async fn watch_expiries(link: std::sync::Weak<Inner>) {
903    loop {
904        let Some(inner) = link.upgrade() else { return };
905        let now = now_ms();
906        if now >= inner.station.binding_not_after {
907            inner.end(LinkError::BindingExpired);
908            return;
909        }
910        let status_deadline = inner.status_deadline.load(Ordering::SeqCst);
911        if now >= status_deadline {
912            inner.end(LinkError::StatusExpired);
913            return;
914        }
915        let wait = (status_deadline.min(inner.station.binding_not_after) - now).clamp(1, 60_000);
916        let mut done = inner.done_rx.clone();
917        drop(inner);
918        tokio::select! {
919            _ = done.wait_for(|ended| *ended) => return,
920            _ = tokio::time::sleep(Duration::from_millis(wait as u64)) => {}
921        }
922    }
923}
924
925/// Ends the link when its QUIC connection closes.
926async fn watch_connection(link: std::sync::Weak<Inner>) {
927    let Some(connection) = link.upgrade().map(|l| l.connection.clone()) else {
928        return;
929    };
930    let cause = connection.closed().await;
931    if let Some(inner) = link.upgrade() {
932        inner.end(LinkError::Io(cause.to_string()));
933    }
934}
935
936fn frame_type_of(v: &Value) -> String {
937    match v.get("frame_type") {
938        Some(Value::Text(t)) => t.clone(),
939        _ => String::new(),
940    }
941}
942
943fn now_ms() -> i64 {
944    crate::uuid_v7::now_ms() as i64
945}
946
947#[cfg(test)]
948mod tests {
949    use super::*;
950
951    fn with_neighbour(v: Value) -> Value {
952        let Value::Map(mut pairs) = v else {
953            panic!("a frame is a map");
954        };
955        pairs.push((Value::text("neighbour"), Value::Map(Vec::new())));
956        Value::Map(pairs)
957    }
958
959    /// macula-rust#9: a burst of pings takes one pong slot until its pong is
960    /// written.
961    #[test]
962    fn a_burst_of_pings_starts_one_pong_at_a_time() {
963        let in_flight = AtomicBool::new(false);
964        let taken = (0..1000).filter(|_| pong_slot(&in_flight)).count();
965        assert_eq!(taken, 1);
966        in_flight.store(false, Ordering::Release);
967        assert!(pong_slot(&in_flight), "free again once the pong is written");
968    }
969
970    /// macula-go 2294f25: liveness frames are read after the session-mode
971    /// check, so a neighbour-signed one ends a v5 link.
972    #[test]
973    fn a_liveness_frame_is_read_as_every_v5_frame_and_exists_only_on_v5() {
974        let nonce = [9; frame::LIVENESS_NONCE_SIZE];
975        let ping = frame::liveness_ping_frame(&nonce);
976        let pong = frame::liveness_pong_frame(&nonce);
977        assert_eq!(
978            liveness_frame(VERSION_5, &ping),
979            Ok(Some((Liveness::Ping, nonce)))
980        );
981        assert_eq!(
982            liveness_frame(VERSION_5, &pong),
983            Ok(Some((Liveness::Pong, nonce)))
984        );
985        for signed in [with_neighbour(ping.clone()), with_neighbour(pong.clone())] {
986            assert_eq!(
987                liveness_frame(VERSION_5, &signed),
988                Err(LinkError::Frame(FrameError::Malformed))
989            );
990        }
991        assert_eq!(
992            liveness_frame(VERSION, &ping),
993            Err(LinkError::Frame(FrameError::Malformed))
994        );
995        let goodbye = frame::goodbye_frame("bye", None).unwrap();
996        assert_eq!(liveness_frame(VERSION_5, &goodbye), Ok(None));
997    }
998}
macula_rust/station_link.rs

macula_rust/
station_link.rs
