coven_domain/joining/
pairing_transport.rs

1use super::{
2    DevicePairingError, DevicePairingOffer, DevicePairingRequest, SealedDevicePairingRequest,
3};
4use base64::engine::general_purpose::URL_SAFE_NO_PAD;
5use base64::Engine;
6use coven_keys::keys::UserKeypair;
7use coven_replication::sync::store::DeviceJoinTransportTiming;
8use serde::{Deserialize, Serialize};
9use std::sync::{Arc, Mutex};
10use tokio::io::{AsyncReadExt, AsyncWriteExt};
11use tokio::net::{TcpListener, TcpStream};
12use tokio::sync::watch;
13use tracing::debug;
14
15const MAX_PAIRING_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
16
17#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(rename_all = "snake_case", deny_unknown_fields)]
19enum HostResponse {
20    AwaitingApproval,
21    Invited(Vec<u8>),
22    Cancelling(Vec<u8>),
23    Cancelled,
24}
25
26#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(deny_unknown_fields)]
28struct HostState {
29    request: Option<DevicePairingRequest>,
30    response: HostResponse,
31}
32
33#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(deny_unknown_fields)]
35struct PersistedPairingHost {
36    offer: DevicePairingOffer,
37    pairing_key: String,
38    state: HostState,
39}
40
41#[derive(Clone)]
42struct PairingJournal {
43    path: std::path::PathBuf,
44}
45
46impl PairingJournal {
47    fn create(
48        path: std::path::PathBuf,
49        offer: DevicePairingOffer,
50        pairing_key: &UserKeypair,
51    ) -> Result<(Self, PersistedPairingHost), DevicePairingTransportError> {
52        let journal = Self { path };
53        if coven_foundation::atomic_file::AtomicFile::new(journal.path.clone())
54            .read_optional()?
55            .is_some()
56        {
57            return Err(DevicePairingTransportError::SessionAlreadyExists);
58        }
59        let persisted = PersistedPairingHost {
60            offer,
61            pairing_key: URL_SAFE_NO_PAD.encode(pairing_key.to_keypair_bytes()),
62            state: HostState::awaiting_request(),
63        };
64        journal.replace(&persisted)?;
65        Ok((journal, persisted))
66    }
67
68    fn open(
69        path: std::path::PathBuf,
70        now_unix_seconds: i64,
71    ) -> Result<(Self, PersistedPairingHost, UserKeypair), DevicePairingTransportError> {
72        let journal = Self { path };
73        let bytes = coven_foundation::atomic_file::AtomicFile::new(journal.path.clone())
74            .read_optional()?
75            .ok_or(DevicePairingTransportError::SessionMissing)?;
76        Self::open_bytes(journal, &bytes, now_unix_seconds)
77    }
78
79    fn open_bytes(
80        journal: Self,
81        bytes: &[u8],
82        now_unix_seconds: i64,
83    ) -> Result<(Self, PersistedPairingHost, UserKeypair), DevicePairingTransportError> {
84        let persisted: PersistedPairingHost = serde_json::from_slice(bytes)?;
85        if persisted.offer.expires_at_unix_seconds() <= now_unix_seconds {
86            return Err(DevicePairingTransportError::Expired);
87        }
88        let key_bytes = URL_SAFE_NO_PAD.decode(&persisted.pairing_key)?;
89        let key_bytes: [u8; 64] = key_bytes
90            .try_into()
91            .map_err(|_| DevicePairingTransportError::PairingKeyLength)?;
92        let pairing_key = UserKeypair::from_signing_key_bytes(&key_bytes)?;
93        if coven_keys::keys::public_key_hex(&pairing_key) != persisted.offer.pairing_public_key() {
94            return Err(DevicePairingTransportError::PairingKeyMismatch);
95        }
96        Ok((journal, persisted, pairing_key))
97    }
98
99    fn replace(&self, state: &PersistedPairingHost) -> Result<(), DevicePairingTransportError> {
100        let bytes = serde_json::to_vec(state)?;
101        coven_foundation::atomic_file::AtomicFile::new(self.path.clone()).replace(&bytes)?;
102        Ok(())
103    }
104
105    fn remove(&self) -> Result<(), DevicePairingTransportError> {
106        coven_foundation::atomic_file::AtomicFile::new(self.path.clone()).remove()?;
107        coven_foundation::atomic_file::sync_parent_dir_blocking(&self.path)?;
108        Ok(())
109    }
110}
111
112impl HostState {
113    fn awaiting_request() -> Self {
114        Self {
115            request: None,
116            response: HostResponse::AwaitingApproval,
117        }
118    }
119}
120
121struct DevicePairingHostInner {
122    offer: DevicePairingOffer,
123    state: Arc<Mutex<PersistedPairingHost>>,
124    journal: PairingJournal,
125    request_tx: watch::Sender<Option<DevicePairingRequest>>,
126    server: Mutex<Option<tokio::task::JoinHandle<()>>>,
127}
128
129impl Drop for DevicePairingHostInner {
130    fn drop(&mut self) {
131        if let Some(server) = self.server.lock().expect("lock pairing server task").take() {
132            server.abort();
133        }
134    }
135}
136
137/// A listener behind the one QR code shown by the existing device. It accepts
138/// one exact signed identity, survives client reconnects, and returns the same
139/// sealed invitation on every retry.
140#[derive(Clone)]
141pub struct DevicePairingHost {
142    inner: Arc<DevicePairingHostInner>,
143}
144
145impl DevicePairingHost {
146    pub async fn start(
147        listener: TcpListener,
148        offer: DevicePairingOffer,
149        pairing_key: UserKeypair,
150        journal_path: std::path::PathBuf,
151        clock: coven_foundation::clock::ClockRef,
152    ) -> Result<Self, DevicePairingTransportError> {
153        let (journal, persisted) = PairingJournal::create(journal_path, offer, &pairing_key)?;
154        Self::start_persisted(listener, journal, persisted, pairing_key, clock).await
155    }
156
157    pub async fn resume(
158        listener: TcpListener,
159        journal_path: std::path::PathBuf,
160        clock: coven_foundation::clock::ClockRef,
161    ) -> Result<Self, DevicePairingTransportError> {
162        let (journal, persisted, pairing_key) =
163            PairingJournal::open(journal_path, clock.now().timestamp())?;
164        Self::start_persisted(listener, journal, persisted, pairing_key, clock).await
165    }
166
167    pub async fn start_or_resume(
168        listener: TcpListener,
169        offer: DevicePairingOffer,
170        pairing_key: UserKeypair,
171        journal_path: std::path::PathBuf,
172        clock: coven_foundation::clock::ClockRef,
173    ) -> Result<Self, DevicePairingTransportError> {
174        let journal = PairingJournal {
175            path: journal_path.clone(),
176        };
177        match coven_foundation::atomic_file::AtomicFile::new(journal_path).read_optional()? {
178            Some(bytes) => {
179                match PairingJournal::open_bytes(journal.clone(), &bytes, clock.now().timestamp()) {
180                    Ok((journal, persisted, pairing_key)) => {
181                        Self::start_persisted(listener, journal, persisted, pairing_key, clock)
182                            .await
183                    }
184                    Err(DevicePairingTransportError::Expired) => {
185                        journal.remove()?;
186                        Self::start(listener, offer, pairing_key, journal.path, clock).await
187                    }
188                    Err(error) => Err(error),
189                }
190            }
191            None => Self::start(listener, offer, pairing_key, journal.path, clock).await,
192        }
193    }
194
195    async fn start_persisted(
196        listener: TcpListener,
197        journal: PairingJournal,
198        persisted: PersistedPairingHost,
199        pairing_key: UserKeypair,
200        clock: coven_foundation::clock::ClockRef,
201    ) -> Result<Self, DevicePairingTransportError> {
202        let offer = persisted.offer.clone();
203        let state = Arc::new(Mutex::new(persisted));
204        let initial_request = state
205            .lock()
206            .expect("lock pairing host state")
207            .state
208            .request
209            .clone();
210        let request_tx = watch::channel(initial_request).0;
211        let server_state = Arc::clone(&state);
212        let server_request_tx = request_tx.clone();
213        let server_offer = offer.clone();
214        let server_journal = journal.clone();
215        let server_clock = clock;
216        let server = tokio::spawn(async move {
217            loop {
218                let (stream, peer) = match listener.accept().await {
219                    Ok(accepted) => accepted,
220                    Err(error) => {
221                        debug!(%error, "device pairing listener stopped accepting connections");
222                        return;
223                    }
224                };
225                let state = Arc::clone(&server_state);
226                let request_tx = server_request_tx.clone();
227                let offer = server_offer.clone();
228                let pairing_key = pairing_key.clone();
229                let journal = server_journal.clone();
230                let clock = server_clock.clone();
231                tokio::spawn(async move {
232                    if let Err(error) = handle_connection(
233                        stream,
234                        &offer,
235                        &pairing_key,
236                        state,
237                        journal,
238                        clock,
239                        request_tx,
240                    )
241                    .await
242                    {
243                        debug!(%peer, %error, "device pairing connection refused");
244                    }
245                });
246            }
247        });
248        Ok(Self {
249            inner: Arc::new(DevicePairingHostInner {
250                offer,
251                state,
252                journal,
253                request_tx,
254                server: Mutex::new(Some(server)),
255            }),
256        })
257    }
258
259    pub fn offer(&self) -> &DevicePairingOffer {
260        &self.inner.offer
261    }
262
263    pub fn subscribe_request(&self) -> watch::Receiver<Option<DevicePairingRequest>> {
264        self.inner.request_tx.subscribe()
265    }
266
267    pub async fn wait_for_request(
268        &self,
269    ) -> Result<DevicePairingRequest, DevicePairingTransportError> {
270        let mut receiver = self.subscribe_request();
271        loop {
272            if let Some(request) = receiver.borrow().clone() {
273                return Ok(request);
274            }
275            receiver
276                .changed()
277                .await
278                .map_err(|_| DevicePairingTransportError::HostStopped)?;
279        }
280    }
281
282    pub fn deliver_invitation(
283        &self,
284        request: &DevicePairingRequest,
285        invitation: Vec<u8>,
286    ) -> Result<(), DevicePairingTransportError> {
287        let mut persisted = self.inner.state.lock().expect("lock pairing host state");
288        if persisted.state.request.as_ref() != Some(request) {
289            return Err(DevicePairingTransportError::RequestMismatch);
290        }
291        let mut next = persisted.clone();
292        match &next.state.response {
293            HostResponse::AwaitingApproval => {
294                next.state.response = HostResponse::Invited(invitation);
295            }
296            HostResponse::Invited(existing) if existing == &invitation => {}
297            HostResponse::Invited(_) | HostResponse::Cancelling(_) | HostResponse::Cancelled => {
298                return Err(DevicePairingTransportError::ResponseConflict)
299            }
300        }
301        self.inner.journal.replace(&next)?;
302        *persisted = next;
303        drop(persisted);
304        Ok(())
305    }
306
307    pub fn invitation(
308        &self,
309        request: &DevicePairingRequest,
310    ) -> Result<Option<Vec<u8>>, DevicePairingTransportError> {
311        let persisted = self.inner.state.lock().expect("lock pairing host state");
312        if persisted.state.request.as_ref() != Some(request) {
313            return Err(DevicePairingTransportError::RequestMismatch);
314        }
315        match &persisted.state.response {
316            HostResponse::AwaitingApproval => Ok(None),
317            HostResponse::Invited(invitation) => Ok(Some(invitation.clone())),
318            HostResponse::Cancelling(_) | HostResponse::Cancelled => {
319                Err(DevicePairingTransportError::Cancelled)
320            }
321        }
322    }
323
324    /// Return the exact Store attempt retained by a durable cancellation so a
325    /// restarted owner can finish its signed unwind instead of approving it.
326    pub fn cancellation_invitation(
327        &self,
328        request: &DevicePairingRequest,
329    ) -> Result<Option<Vec<u8>>, DevicePairingTransportError> {
330        let persisted = self.inner.state.lock().expect("lock pairing host state");
331        if persisted.state.request.as_ref() != Some(request) {
332            return Err(DevicePairingTransportError::RequestMismatch);
333        }
334        match &persisted.state.response {
335            HostResponse::Cancelling(invitation) => Ok(Some(invitation.clone())),
336            HostResponse::AwaitingApproval | HostResponse::Invited(_) | HostResponse::Cancelled => {
337                Ok(None)
338            }
339        }
340    }
341
342    /// Persist cancellation and return the delivered invitation, when one
343    /// exists, so the owner can unwind the exact Store attempt it started.
344    pub fn cancel(&self) -> Result<Option<Vec<u8>>, DevicePairingTransportError> {
345        let mut persisted = self.inner.state.lock().expect("lock pairing host state");
346        let mut next = persisted.clone();
347        let invitation = match &next.state.response {
348            HostResponse::AwaitingApproval => None,
349            HostResponse::Invited(invitation) => Some(invitation.clone()),
350            HostResponse::Cancelling(invitation) => Some(invitation.clone()),
351            HostResponse::Cancelled => None,
352        };
353        next.state.response = match &invitation {
354            Some(invitation) => HostResponse::Cancelling(invitation.clone()),
355            None => HostResponse::Cancelled,
356        };
357        self.inner.journal.replace(&next)?;
358        *persisted = next;
359        drop(persisted);
360        Ok(invitation)
361    }
362
363    pub fn finish(&self) -> Result<(), DevicePairingTransportError> {
364        let persisted = self.inner.state.lock().expect("lock pairing host state");
365        if matches!(persisted.state.response, HostResponse::AwaitingApproval) {
366            return Err(DevicePairingTransportError::ResponseConflict);
367        }
368        self.inner.journal.remove()
369    }
370}
371
372#[derive(Serialize, Deserialize)]
373#[serde(deny_unknown_fields)]
374struct PairingWireRequest {
375    request: SealedDevicePairingRequest,
376}
377
378#[derive(Serialize, Deserialize)]
379#[serde(rename_all = "snake_case", deny_unknown_fields)]
380enum PairingWireResponse {
381    AwaitingApproval,
382    Invited { invitation: String },
383    Cancelled,
384    SessionClaimed,
385    Expired,
386}
387
388async fn handle_connection(
389    mut stream: TcpStream,
390    offer: &DevicePairingOffer,
391    pairing_key: &UserKeypair,
392    state: Arc<Mutex<PersistedPairingHost>>,
393    journal: PairingJournal,
394    clock: coven_foundation::clock::ClockRef,
395    request_tx: watch::Sender<Option<DevicePairingRequest>>,
396) -> Result<(), DevicePairingTransportError> {
397    let wire: PairingWireRequest = read_frame(&mut stream).await?;
398    if clock.now().timestamp() >= offer.expires_at_unix_seconds() {
399        write_frame(&mut stream, &PairingWireResponse::Expired).await?;
400        return Ok(());
401    }
402    let request = wire.request.open(offer, pairing_key)?;
403    let response = {
404        let mut persisted = state.lock().expect("lock pairing host state");
405        match &persisted.state.request {
406            None => {
407                let mut next = persisted.clone();
408                next.state.request = Some(request.clone());
409                journal.replace(&next)?;
410                *persisted = next;
411                request_tx.send_replace(Some(request));
412                response_for(&persisted.state.response)
413            }
414            Some(existing) if existing == &request => response_for(&persisted.state.response),
415            Some(_) => PairingWireResponse::SessionClaimed,
416        }
417    };
418    write_frame(&mut stream, &response).await?;
419    Ok(())
420}
421
422fn response_for(response: &HostResponse) -> PairingWireResponse {
423    match response {
424        HostResponse::AwaitingApproval => PairingWireResponse::AwaitingApproval,
425        HostResponse::Invited(invitation) => PairingWireResponse::Invited {
426            invitation: URL_SAFE_NO_PAD.encode(invitation),
427        },
428        // Once a Store attempt exists, the joining device needs that exact
429        // invitation before it can observe and complete the signed unwind.
430        HostResponse::Cancelling(invitation) => PairingWireResponse::Invited {
431            invitation: URL_SAFE_NO_PAD.encode(invitation),
432        },
433        HostResponse::Cancelled => PairingWireResponse::Cancelled,
434    }
435}
436
437/// Submit the same signed request until the owner approves, cancels, or the
438/// caller's deadline expires. Each retry may use another endpoint from the QR;
439/// the host accepts the exact request idempotently and refuses a competing one.
440pub async fn receive_device_invitation(
441    offer: &DevicePairingOffer,
442    request: &SealedDevicePairingRequest,
443    timing: DeviceJoinTransportTiming,
444    clock: coven_foundation::clock::ClockRef,
445    cancel: &watch::Receiver<bool>,
446) -> Result<Vec<u8>, DevicePairingTransportError> {
447    let deadline = clock.now()
448        + chrono::Duration::from_std(timing.deadline)
449            .map_err(|_| DevicePairingTransportError::DeadlineOutOfRange)?;
450    let mut cancellation = cancel.clone();
451    let wire = PairingWireRequest {
452        request: request.clone(),
453    };
454    let mut failures = Vec::new();
455    loop {
456        if *cancel.borrow() {
457            return Err(DevicePairingTransportError::Cancelled);
458        }
459        if clock.now().timestamp() >= offer.expires_at_unix_seconds() {
460            return Err(DevicePairingTransportError::Expired);
461        }
462        if clock.now() >= deadline {
463            return Err(DevicePairingTransportError::Unavailable(failures));
464        }
465        failures.clear();
466        for endpoint in offer.endpoints() {
467            match exchange(*endpoint, &wire).await {
468                Ok(PairingWireResponse::AwaitingApproval) => break,
469                Ok(PairingWireResponse::Invited { invitation }) => {
470                    return URL_SAFE_NO_PAD
471                        .decode(invitation)
472                        .map_err(DevicePairingTransportError::Ciphertext)
473                }
474                Ok(PairingWireResponse::Cancelled) => {
475                    return Err(DevicePairingTransportError::Cancelled)
476                }
477                Ok(PairingWireResponse::SessionClaimed) => {
478                    return Err(DevicePairingTransportError::SessionClaimed)
479                }
480                Ok(PairingWireResponse::Expired) => {
481                    return Err(DevicePairingTransportError::Expired)
482                }
483                Err(error) => {
484                    debug!(%endpoint, %error, "device pairing endpoint unavailable");
485                    failures.push(format!("{endpoint}: {error}"));
486                }
487            }
488        }
489        tokio::select! {
490            _ = tokio::time::sleep(timing.poll) => {}
491            changed = cancellation.changed() => {
492                changed.map_err(|_| DevicePairingTransportError::CancellationChannelClosed)?;
493            }
494        }
495    }
496}
497
498async fn exchange(
499    endpoint: std::net::SocketAddr,
500    request: &PairingWireRequest,
501) -> Result<PairingWireResponse, DevicePairingTransportError> {
502    let mut stream = TcpStream::connect(endpoint).await?;
503    write_frame(&mut stream, request).await?;
504    read_frame(&mut stream).await
505}
506
507async fn write_frame<T: Serialize>(
508    stream: &mut TcpStream,
509    value: &T,
510) -> Result<(), DevicePairingTransportError> {
511    let bytes = serde_json::to_vec(value)?;
512    if bytes.len() > MAX_PAIRING_MESSAGE_BYTES {
513        return Err(DevicePairingTransportError::MessageTooLarge(bytes.len()));
514    }
515    stream.write_u32(bytes.len() as u32).await?;
516    stream.write_all(&bytes).await?;
517    stream.shutdown().await?;
518    Ok(())
519}
520
521async fn read_frame<T: for<'de> Deserialize<'de>>(
522    stream: &mut TcpStream,
523) -> Result<T, DevicePairingTransportError> {
524    let length = stream.read_u32().await? as usize;
525    if length > MAX_PAIRING_MESSAGE_BYTES {
526        return Err(DevicePairingTransportError::MessageTooLarge(length));
527    }
528    let mut bytes = vec![0; length];
529    stream.read_exact(&mut bytes).await?;
530    Ok(serde_json::from_slice(&bytes)?)
531}
532
533#[derive(Debug, thiserror::Error)]
534pub enum DevicePairingTransportError {
535    #[error("pairing protocol: {0}")]
536    Pairing(#[from] DevicePairingError),
537    #[error("pairing network: {0}")]
538    Network(#[from] std::io::Error),
539    #[error("pairing message JSON: {0}")]
540    Json(#[from] serde_json::Error),
541    #[error("pairing journal: {0}")]
542    Journal(#[from] coven_foundation::atomic_file::FileError),
543    #[error("pairing journal key: {0}")]
544    PairingKey(#[from] coven_keys::keys::KeyError),
545    #[error("pairing response ciphertext: {0}")]
546    Ciphertext(base64::DecodeError),
547    #[error("pairing journal key encoding: {0}")]
548    PairingKeyEncoding(#[from] base64::DecodeError),
549    #[error("pairing message contains {0} bytes")]
550    MessageTooLarge(usize),
551    #[error("a pairing session is already durable at this path")]
552    SessionAlreadyExists,
553    #[error("the durable pairing session is absent")]
554    SessionMissing,
555    #[error("the pairing session expired")]
556    Expired,
557    #[error("the durable pairing key is not 64 bytes")]
558    PairingKeyLength,
559    #[error("the durable pairing key does not match the displayed offer")]
560    PairingKeyMismatch,
561    #[error("another joining identity already claimed this pairing session")]
562    SessionClaimed,
563    #[error("pairing request does not match the accepted identity")]
564    RequestMismatch,
565    #[error("pairing session already has another terminal response")]
566    ResponseConflict,
567    #[error("pairing was cancelled")]
568    Cancelled,
569    #[error("pairing cancellation channel closed")]
570    CancellationChannelClosed,
571    #[error("pairing host stopped")]
572    HostStopped,
573    #[error("pairing deadline cannot be represented by the injected clock")]
574    DeadlineOutOfRange,
575    #[error("no pairing endpoint responded before the deadline: {0:?}")]
576    Unavailable(Vec<String>),
577}
578
579#[cfg(test)]
580mod tests {
581    use super::*;
582    use coven_foundation::config::CloudProvider;
583    use std::time::Duration;
584
585    async fn host() -> (DevicePairingHost, UserKeypair, tempfile::TempDir) {
586        let listener = TcpListener::bind("127.0.0.1:0")
587            .await
588            .expect("bind pairing listener");
589        let endpoint = listener.local_addr().expect("pairing endpoint");
590        let pairing_key = UserKeypair::generate();
591        let offer = DevicePairingOffer::new(
592            &pairing_key,
593            vec![endpoint],
594            "Transport Test Store".to_string(),
595            CloudProvider::S3,
596            1_900_000_000,
597        )
598        .expect("pairing offer");
599        let journal = tempfile::tempdir().expect("pairing journal directory");
600        (
601            DevicePairingHost::start(
602                listener,
603                offer,
604                pairing_key,
605                journal.path().join("pairing.json"),
606                Arc::new(coven_foundation::clock::SystemClock),
607            )
608            .await
609            .expect("start pairing host"),
610            UserKeypair::generate(),
611            journal,
612        )
613    }
614
615    fn timing() -> DeviceJoinTransportTiming {
616        DeviceJoinTransportTiming {
617            poll: Duration::from_millis(2),
618            deadline: Duration::from_secs(2),
619        }
620    }
621
622    #[tokio::test]
623    async fn one_request_reconnects_until_the_owner_returns_its_invitation() {
624        let (host, joining_identity, _journal) = host().await;
625        let request = DevicePairingRequest::signed(host.offer(), &joining_identity, None);
626        let sealed = SealedDevicePairingRequest::new(host.offer(), &request).expect("seal request");
627        let (_cancel_tx, cancel) = watch::channel(false);
628        let receiving = tokio::spawn({
629            let offer = host.offer().clone();
630            let sealed = sealed.clone();
631            async move {
632                receive_device_invitation(
633                    &offer,
634                    &sealed,
635                    timing(),
636                    Arc::new(coven_foundation::clock::SystemClock),
637                    &cancel,
638                )
639                .await
640            }
641        });
642
643        let observed = host.wait_for_request().await.expect("receive request");
644        assert_eq!(observed, request);
645        host.deliver_invitation(&request, b"sealed invitation".to_vec())
646            .expect("deliver invitation");
647
648        assert_eq!(
649            receiving
650                .await
651                .expect("join client task")
652                .expect("invitation"),
653            b"sealed invitation",
654        );
655    }
656
657    #[tokio::test]
658    async fn a_second_identity_cannot_replace_the_request_the_owner_is_reviewing() {
659        let (host, first_identity, _journal) = host().await;
660        let first = DevicePairingRequest::signed(host.offer(), &first_identity, None);
661        let first_sealed =
662            SealedDevicePairingRequest::new(host.offer(), &first).expect("seal first request");
663        let (_first_cancel_tx, first_cancel) = watch::channel(false);
664        let first_receive = tokio::spawn({
665            let offer = host.offer().clone();
666            async move {
667                receive_device_invitation(
668                    &offer,
669                    &first_sealed,
670                    timing(),
671                    Arc::new(coven_foundation::clock::SystemClock),
672                    &first_cancel,
673                )
674                .await
675            }
676        });
677        assert_eq!(host.wait_for_request().await.expect("first request"), first);
678
679        let second = DevicePairingRequest::signed(host.offer(), &UserKeypair::generate(), None);
680        let second_sealed =
681            SealedDevicePairingRequest::new(host.offer(), &second).expect("seal second request");
682        let (_second_cancel_tx, second_cancel) = watch::channel(false);
683        assert!(matches!(
684            receive_device_invitation(
685                host.offer(),
686                &second_sealed,
687                timing(),
688                Arc::new(coven_foundation::clock::SystemClock),
689                &second_cancel,
690            )
691            .await,
692            Err(DevicePairingTransportError::SessionClaimed)
693        ));
694
695        host.deliver_invitation(&first, b"first invitation".to_vec())
696            .expect("finish first request");
697        assert_eq!(
698            first_receive
699                .await
700                .expect("first client task")
701                .expect("first invitation"),
702            b"first invitation",
703        );
704    }
705
706    #[tokio::test]
707    async fn an_owner_restart_resumes_the_exact_request_and_response() {
708        let listener = TcpListener::bind("127.0.0.1:0")
709            .await
710            .expect("bind first pairing listener");
711        let endpoint = listener.local_addr().expect("pairing endpoint");
712        let pairing_key = UserKeypair::generate();
713        let offer = DevicePairingOffer::new(
714            &pairing_key,
715            vec![endpoint],
716            "Restart Test Store".to_string(),
717            CloudProvider::S3,
718            1_900_000_000,
719        )
720        .expect("pairing offer");
721        let journal = tempfile::tempdir().expect("pairing journal directory");
722        let journal_path = journal.path().join("pairing.json");
723        let clock: coven_foundation::clock::ClockRef =
724            Arc::new(coven_foundation::clock::SystemClock);
725        let host = DevicePairingHost::start(
726            listener,
727            offer.clone(),
728            pairing_key,
729            journal_path.clone(),
730            clock.clone(),
731        )
732        .await
733        .expect("start first pairing host");
734        let request = DevicePairingRequest::signed(&offer, &UserKeypair::generate(), None);
735        let sealed = SealedDevicePairingRequest::new(&offer, &request).expect("seal request");
736        let (_cancel_tx, cancel) = watch::channel(false);
737        let receiving = tokio::spawn({
738            let offer = offer.clone();
739            let clock = clock.clone();
740            async move { receive_device_invitation(&offer, &sealed, timing(), clock, &cancel).await }
741        });
742        assert_eq!(
743            host.wait_for_request().await.expect("first request"),
744            request
745        );
746
747        drop(host);
748        tokio::task::yield_now().await;
749        let listener = TcpListener::bind(endpoint)
750            .await
751            .expect("rebind pairing listener");
752        let resumed = DevicePairingHost::resume(listener, journal_path, clock)
753            .await
754            .expect("resume pairing host");
755        assert_eq!(
756            resumed.wait_for_request().await.expect("durable request"),
757            request,
758        );
759        resumed
760            .deliver_invitation(&request, b"resumed invitation".to_vec())
761            .expect("persist invitation after restart");
762        assert_eq!(
763            receiving
764                .await
765                .expect("joining task")
766                .expect("resumed invitation"),
767            b"resumed invitation",
768        );
769    }
770
771    #[tokio::test]
772    async fn cancellation_is_durable_and_reaches_the_exact_waiting_identity() {
773        let (host, joining_identity, journal) = host().await;
774        let request = DevicePairingRequest::signed(host.offer(), &joining_identity, None);
775        let sealed = SealedDevicePairingRequest::new(host.offer(), &request).expect("seal request");
776        let (_cancel_tx, cancel) = watch::channel(false);
777        let receiving = tokio::spawn({
778            let offer = host.offer().clone();
779            async move {
780                receive_device_invitation(
781                    &offer,
782                    &sealed,
783                    timing(),
784                    Arc::new(coven_foundation::clock::SystemClock),
785                    &cancel,
786                )
787                .await
788            }
789        });
790        assert_eq!(host.wait_for_request().await.expect("request"), request);
791        host.cancel().expect("persist cancellation");
792
793        assert!(matches!(
794            receiving.await.expect("joining task"),
795            Err(DevicePairingTransportError::Cancelled)
796        ));
797        assert!(journal.path().join("pairing.json").exists());
798    }
799
800    #[tokio::test]
801    async fn cancellation_after_invitation_retains_the_attempt_needed_to_unwind() {
802        let (host, joining_identity, journal) = host().await;
803        let request = DevicePairingRequest::signed(host.offer(), &joining_identity, None);
804        let sealed = SealedDevicePairingRequest::new(host.offer(), &request).expect("seal request");
805        let (_cancel_tx, cancel) = watch::channel(false);
806        let receiving = tokio::spawn({
807            let offer = host.offer().clone();
808            async move {
809                receive_device_invitation(
810                    &offer,
811                    &sealed,
812                    timing(),
813                    Arc::new(coven_foundation::clock::SystemClock),
814                    &cancel,
815                )
816                .await
817            }
818        });
819        assert_eq!(host.wait_for_request().await.expect("request"), request);
820        host.deliver_invitation(&request, b"sealed invitation".to_vec())
821            .expect("deliver invitation");
822        assert_eq!(
823            host.cancel().expect("cancel invited pairing"),
824            Some(b"sealed invitation".to_vec()),
825        );
826        assert_eq!(
827            receiving.await.expect("joining task").expect("invitation"),
828            b"sealed invitation",
829        );
830        drop(host);
831
832        let listener = TcpListener::bind("127.0.0.1:0")
833            .await
834            .expect("bind resumed pairing listener");
835        let resumed = DevicePairingHost::resume(
836            listener,
837            journal.path().join("pairing.json"),
838            Arc::new(coven_foundation::clock::SystemClock),
839        )
840        .await
841        .expect("resume cancelled pairing");
842        assert_eq!(
843            resumed.cancel().expect("resume cancellation"),
844            Some(b"sealed invitation".to_vec()),
845        );
846    }
847}
coven_domain/joining/pairing_transport.rs

coven_domain/joining/
pairing_transport.rs
