openrtc/
application_crypto_streams.rs

1//! Length-framed application-crypto adapters for iroh `SendStream` / `RecvStream`.
2//!
3//! Mirrors the TypeScript `createCryptoStreams` contract: each
4//! `write_all` produces one encrypted frame; `read` reassembles framed ciphertext
5//! from arbitrary chunk boundaries before returning plaintext.
6
7use std::collections::VecDeque;
8use std::io;
9use std::sync::{
10    atomic::{AtomicBool, Ordering},
11    Arc,
12};
13
14use iroh::endpoint::{ClosedStream, ReadError, RecvStream, SendStream, WriteError};
15use tokio::io::AsyncWriteExt;
16
17use crate::application_crypto::{
18    protect_frame, CryptoFrameDecoder, CryptoStreamError, APPLICATION_KEY_BYTES,
19};
20
21#[derive(Debug)]
22struct ApplicationCryptoStreamIoError(CryptoStreamError);
23
24impl std::fmt::Display for ApplicationCryptoStreamIoError {
25    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        write!(formatter, "{:?}", self.0)
27    }
28}
29
30impl std::error::Error for ApplicationCryptoStreamIoError {}
31
32/// Admission selected by the SDK stream factory. Envelope permission is
33/// private and used only while reading a bounded named-channel header.
34#[derive(Clone)]
35pub(crate) enum ApplicationStreamAdmission {
36    Latest,
37    Scope(crate::session_token::GrantScope),
38    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
39    Envelope,
40}
41
42/// Stream-local access fence, supplied only by the Rust admission/key owner.
43/// It observes existing authority, never schedules recovery or retires peers.
44#[derive(Clone)]
45pub(crate) struct ApplicationStreamAccess {
46    current: Arc<dyn Fn() -> bool + Send + Sync>,
47    denied: Arc<AtomicBool>,
48}
49
50impl std::fmt::Debug for ApplicationStreamAccess {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        f.write_str("ApplicationStreamAccess")
53    }
54}
55
56impl ApplicationStreamAccess {
57    pub(crate) fn new(current: impl Fn() -> bool + Send + Sync + 'static) -> Self {
58        Self {
59            current: Arc::new(current),
60            denied: Arc::new(AtomicBool::new(false)),
61        }
62    }
63
64    #[cfg(not(target_arch = "wasm32"))]
65    pub(crate) fn same_fence(&self, other: &Self) -> bool {
66        Arc::ptr_eq(&self.denied, &other.denied)
67    }
68
69    pub(crate) fn check(&self) -> io::Result<()> {
70        if !self.denied.load(Ordering::Acquire) && (self.current)() {
71            return Ok(());
72        }
73        // A stream which observed denial cannot resurrect after reauthorization.
74        self.denied.store(true, Ordering::Release);
75        Err(io::Error::new(
76            io::ErrorKind::PermissionDenied,
77            "application stream authorization expired or changed",
78        ))
79    }
80}
81
82fn check_access(access: &Option<ApplicationStreamAccess>) -> io::Result<()> {
83    access
84        .as_ref()
85        .map_or(Ok(()), ApplicationStreamAccess::check)
86}
87
88/// Recheck before every poll, including when transport backpressure wakes a
89/// pending operation. Checking only before `.await` would authorize later I/O.
90async fn with_access<T>(
91    access: &Option<ApplicationStreamAccess>,
92    operation: impl std::future::Future<Output = io::Result<T>>,
93) -> io::Result<T> {
94    let mut operation = std::pin::pin!(operation);
95    std::future::poll_fn(|cx| {
96        check_access(access)?;
97        let result = operation.as_mut().poll(cx);
98        check_access(access)?;
99        result
100    })
101    .await
102}
103
104/// Returns true only when encrypted stream decoding reached AEAD verification
105/// and rejected the frame. Transport EOF, connection loss, and incomplete
106/// shutdown frames are intentionally distinct so callers do not misreport
107/// lifecycle teardown as retired key material.
108pub fn is_crypto_auth_failure(error: &io::Error) -> bool {
109    error
110        .get_ref()
111        .and_then(|source| source.downcast_ref::<ApplicationCryptoStreamIoError>())
112        .is_some_and(|source| matches!(&source.0, CryptoStreamError::AuthenticationFailed(_)))
113}
114
115/// Return the peer-provided QUIC stop code without forcing higher layers to
116/// parse a transport error's display text.
117pub fn peer_stop_code(error: &io::Error) -> Option<u64> {
118    error
119        .get_ref()
120        .and_then(|source| source.downcast_ref::<WriteError>())
121        .and_then(|source| match source {
122            WriteError::Stopped(code) => Some(code.into_inner()),
123            _ => None,
124        })
125}
126
127#[derive(Debug)]
128pub struct CryptoSendStream {
129    inner: SendStream,
130    key: [u8; APPLICATION_KEY_BYTES],
131    access: Option<ApplicationStreamAccess>,
132}
133
134impl CryptoSendStream {
135    pub fn new(inner: SendStream, key: [u8; APPLICATION_KEY_BYTES]) -> Self {
136        Self {
137            inner,
138            key,
139            access: None,
140        }
141    }
142
143    pub async fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
144        check_access(&self.access)?;
145        let frame = protect_frame(&self.key, buf).map_err(crypto_io_error)?;
146        // Bound each authorization-to-write interval even for a large frame.
147        for chunk in frame.chunks(16 * 1024) {
148            with_access(&self.access, async {
149                self.inner.write_all(chunk).await.map_err(map_write_error)
150            })
151            .await?;
152        }
153        Ok(())
154    }
155
156    pub async fn flush(&mut self) -> io::Result<()> {
157        with_access(&self.access, self.inner.flush()).await
158    }
159
160    pub fn finish(mut self) -> io::Result<()> {
161        self.inner.finish().map_err(map_finish_error)
162    }
163
164    pub fn into_inner(self) -> SendStream {
165        self.inner
166    }
167}
168
169#[derive(Debug)]
170pub struct CryptoRecvStream {
171    inner: RecvStream,
172    decoder: CryptoFrameDecoder,
173    read_buf: Vec<u8>,
174    finished: bool,
175    access: Option<ApplicationStreamAccess>,
176}
177
178impl CryptoRecvStream {
179    pub fn new(inner: RecvStream, key: [u8; APPLICATION_KEY_BYTES]) -> io::Result<Self> {
180        Ok(Self {
181            inner,
182            decoder: CryptoFrameDecoder::new(&key).map_err(crypto_io_error)?,
183            read_buf: Vec::new(),
184            finished: false,
185            access: None,
186        })
187    }
188
189    pub fn new_with_prefix(
190        inner: RecvStream,
191        key: [u8; APPLICATION_KEY_BYTES],
192        prefix: &[u8],
193    ) -> io::Result<Self> {
194        let mut decoder = CryptoFrameDecoder::new(&key).map_err(crypto_io_error)?;
195        let mut read_buf = Vec::new();
196        if !prefix.is_empty() {
197            for frame in decoder.push(prefix).map_err(crypto_stream_io_error)? {
198                read_buf.extend_from_slice(&frame);
199            }
200        }
201        Ok(Self {
202            inner,
203            decoder,
204            read_buf,
205            finished: false,
206            access: None,
207        })
208    }
209
210    pub async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
211        check_access(&self.access)?;
212        if !self.read_buf.is_empty() {
213            let copied = self.read_buf.len().min(buf.len());
214            buf[..copied].copy_from_slice(&self.read_buf[..copied]);
215            self.read_buf.drain(..copied);
216            return Ok(copied);
217        }
218
219        if self.finished {
220            return Ok(0);
221        }
222
223        loop {
224            let mut chunk = vec![0u8; 16 * 1024];
225            let read_bytes = match with_access(&self.access, async {
226                self.inner.read(&mut chunk).await.map_err(map_read_error)
227            })
228            .await
229            {
230                Ok(Some(0)) => 0,
231                Ok(Some(read_bytes)) => read_bytes,
232                Ok(None) => 0,
233                Err(error) => return Err(error),
234            };
235            if read_bytes == 0 {
236                self.finished = true;
237                self.decoder.finish().map_err(crypto_stream_io_error)?;
238                if self.read_buf.is_empty() {
239                    return Ok(0);
240                }
241                check_access(&self.access)?;
242                let copied = self.read_buf.len().min(buf.len());
243                buf[..copied].copy_from_slice(&self.read_buf[..copied]);
244                self.read_buf.drain(..copied);
245                return Ok(copied);
246            }
247
248            chunk.truncate(read_bytes);
249            let opened = self.decoder.push(&chunk).map_err(crypto_stream_io_error)?;
250            for frame in opened {
251                self.read_buf.extend_from_slice(&frame);
252            }
253
254            if !self.read_buf.is_empty() {
255                check_access(&self.access)?;
256                let copied = self.read_buf.len().min(buf.len());
257                buf[..copied].copy_from_slice(&self.read_buf[..copied]);
258                self.read_buf.drain(..copied);
259                return Ok(copied);
260            }
261        }
262    }
263
264    pub fn into_inner(self) -> RecvStream {
265        self.inner
266    }
267}
268
269#[derive(Debug)]
270pub enum PeerSendStream {
271    Plain(SendStream),
272    Encrypted(CryptoSendStream),
273}
274
275impl PeerSendStream {
276    #[cfg(not(target_arch = "wasm32"))]
277    pub(crate) fn binding(&self) -> Option<(&ApplicationStreamAccess, [u8; APPLICATION_KEY_BYTES])> {
278        match self {
279            Self::Encrypted(stream) => stream.access.as_ref().map(|access| (access, stream.key)),
280            _ => None,
281        }
282    }
283    pub(crate) fn with_access(mut self, access: ApplicationStreamAccess) -> Self {
284        if let Self::Encrypted(stream) = &mut self {
285            stream.access = Some(access);
286        }
287        self
288    }
289    pub fn plain(inner: SendStream) -> Self {
290        Self::Plain(inner)
291    }
292
293    pub fn encrypted(inner: SendStream, key: [u8; APPLICATION_KEY_BYTES]) -> Self {
294        Self::Encrypted(CryptoSendStream::new(inner, key))
295    }
296
297    pub fn is_encrypted(&self) -> bool {
298        matches!(self, Self::Encrypted(_))
299    }
300
301    pub async fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
302        match self {
303            Self::Plain(stream) => stream.write_all(buf).await.map_err(map_write_error),
304            Self::Encrypted(stream) => stream.write_all(buf).await,
305        }
306    }
307
308    pub async fn flush(&mut self) -> io::Result<()> {
309        match self {
310            Self::Plain(stream) => stream.flush().await,
311            Self::Encrypted(stream) => stream.flush().await,
312        }
313    }
314
315    pub fn finish(self) -> io::Result<()> {
316        match self {
317            Self::Plain(mut stream) => stream.finish().map_err(map_finish_error),
318            Self::Encrypted(stream) => stream.finish(),
319        }
320    }
321
322    /// Finish a short control response and keep its QUIC send half alive until
323    /// the peer has acknowledged the buffered bytes (or the bounded wait
324    /// expires). `finish()` alone only queues the FIN; a concurrent physical-leg
325    /// replacement can otherwise close the connection before the verdict is
326    /// observable by the admission dialer.
327    pub async fn finish_and_wait_for_peer(self, timeout: std::time::Duration) -> io::Result<()> {
328        async fn finish_and_wait(
329            mut stream: SendStream,
330            timeout: std::time::Duration,
331        ) -> io::Result<()> {
332            stream.finish().map_err(map_finish_error)?;
333            #[cfg(not(target_arch = "wasm32"))]
334            match tokio::time::timeout(timeout, stream.stopped()).await {
335                Ok(Ok(_)) => Ok(()),
336                Ok(Err(error)) => Err(io::Error::other(format!(
337                    "wait for peer acknowledgement: {error}"
338                ))),
339                Err(_) => Err(io::Error::new(
340                    io::ErrorKind::TimedOut,
341                    "timed out waiting for peer acknowledgement",
342                )),
343            }
344            #[cfg(target_arch = "wasm32")]
345            {
346                use futures::FutureExt;
347
348                let stopped = stream.stopped().fuse();
349                let timeout = gloo_timers::future::sleep(timeout).fuse();
350                futures::pin_mut!(stopped, timeout);
351                futures::select! {
352                    result = stopped => result.map(|_| ()).map_err(|error| {
353                        io::Error::other(format!("wait for peer acknowledgement: {error}"))
354                    }),
355                    _ = timeout => Err(io::Error::new(
356                        io::ErrorKind::TimedOut,
357                        "timed out waiting for peer acknowledgement",
358                    )),
359                }
360            }
361        }
362
363        match self {
364            Self::Plain(stream) => finish_and_wait(stream, timeout).await,
365            Self::Encrypted(stream) => finish_and_wait(stream.into_inner(), timeout).await,
366        }
367    }
368
369    pub fn into_plain(self) -> SendStream {
370        match self {
371            Self::Plain(stream) => stream,
372            Self::Encrypted(stream) => stream.into_inner(),
373        }
374    }
375}
376
377#[derive(Debug)]
378pub enum PeerRecvStream {
379    Plain(RecvStream),
380    Encrypted(CryptoRecvStream),
381    /// Plaintext bytes read by a host-side protocol classifier, followed by
382    /// the original protected stream. This lets one native host inspect a
383    /// channel envelope without stealing or duplicating the admitted-stream
384    /// queue.
385    Prefixed {
386        prefix: VecDeque<u8>,
387        inner: Box<PeerRecvStream>,
388    },
389}
390
391impl PeerRecvStream {
392    #[cfg(not(target_arch = "wasm32"))]
393    pub(crate) fn binding(&self) -> Option<&ApplicationStreamAccess> {
394        match self {
395            Self::Encrypted(stream) => stream.access.as_ref(),
396            Self::Prefixed { inner, .. } => inner.binding(),
397            Self::Plain(_) => None,
398        }
399    }
400    pub(crate) fn with_access(mut self, access: ApplicationStreamAccess) -> Self {
401        let mut current = &mut self;
402        loop {
403            match current {
404                Self::Encrypted(stream) => {
405                    stream.access = Some(access);
406                    break;
407                }
408                Self::Prefixed { inner, .. } => current = inner.as_mut(),
409                Self::Plain(_) => break,
410            }
411        }
412        self
413    }
414
415    fn check_access(&self) -> io::Result<()> {
416        let mut current = self;
417        loop {
418            match current {
419                Self::Encrypted(stream) => return check_access(&stream.access),
420                Self::Prefixed { inner, .. } => current = inner.as_ref(),
421                Self::Plain(_) => return Ok(()),
422            }
423        }
424    }
425    pub fn plain(inner: RecvStream) -> Self {
426        Self::Plain(inner)
427    }
428
429    pub fn encrypted(inner: RecvStream, key: [u8; APPLICATION_KEY_BYTES]) -> io::Result<Self> {
430        Ok(Self::Encrypted(CryptoRecvStream::new(inner, key)?))
431    }
432
433    pub fn encrypted_with_prefix(
434        inner: RecvStream,
435        key: [u8; APPLICATION_KEY_BYTES],
436        prefix: &[u8],
437    ) -> io::Result<Self> {
438        Ok(Self::Encrypted(CryptoRecvStream::new_with_prefix(
439            inner, key, prefix,
440        )?))
441    }
442
443    pub fn is_encrypted(&self) -> bool {
444        match self {
445            Self::Encrypted(_) => true,
446            Self::Prefixed { inner, .. } => inner.is_encrypted(),
447            Self::Plain(_) => false,
448        }
449    }
450
451    /// Restore plaintext consumed during bounded host-side classification.
452    pub fn with_plaintext_prefix(self, prefix: impl IntoIterator<Item = u8>) -> Self {
453        let prefix = prefix.into_iter().collect::<VecDeque<_>>();
454        if prefix.is_empty() {
455            self
456        } else {
457            Self::Prefixed {
458                prefix,
459                inner: Box::new(self),
460            }
461        }
462    }
463
464    pub async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
465        // Restored plaintext prefixes obey the same fence as encrypted bytes.
466        self.check_access()?;
467        let mut current = self;
468        loop {
469            match current {
470                Self::Plain(stream) => return read_recv_stream(stream, buf).await,
471                Self::Encrypted(stream) => return stream.read(buf).await,
472                Self::Prefixed { prefix, inner } => {
473                    if !prefix.is_empty() {
474                        let copied = prefix.len().min(buf.len());
475                        for slot in &mut buf[..copied] {
476                            *slot = prefix.pop_front().expect("prefix length was checked");
477                        }
478                        return Ok(copied);
479                    } else {
480                        current = inner.as_mut();
481                    }
482                }
483            }
484        }
485    }
486
487    pub fn into_plain(self) -> RecvStream {
488        match self {
489            Self::Plain(stream) => stream,
490            Self::Encrypted(stream) => stream.into_inner(),
491            Self::Prefixed { inner, .. } => inner.into_plain(),
492        }
493    }
494}
495
496pub fn wrap_peer_streams(
497    key: Option<[u8; APPLICATION_KEY_BYTES]>,
498    send: SendStream,
499    recv: RecvStream,
500) -> io::Result<(PeerSendStream, PeerRecvStream)> {
501    match key {
502        Some(key) => Ok((
503            PeerSendStream::encrypted(send, key),
504            PeerRecvStream::encrypted(recv, key)?,
505        )),
506        None => Ok((PeerSendStream::plain(send), PeerRecvStream::plain(recv))),
507    }
508}
509
510pub fn wrap_send_stream(
511    key: Option<[u8; APPLICATION_KEY_BYTES]>,
512    send: SendStream,
513) -> PeerSendStream {
514    match key {
515        Some(key) => PeerSendStream::encrypted(send, key),
516        None => PeerSendStream::plain(send),
517    }
518}
519
520async fn read_recv_stream(stream: &mut RecvStream, buf: &mut [u8]) -> io::Result<usize> {
521    match stream.read(buf).await {
522        Ok(Some(0)) => Ok(0),
523        Ok(Some(read_bytes)) => Ok(read_bytes),
524        Ok(None) => Ok(0),
525        Err(error) => Err(map_read_error(error)),
526    }
527}
528
529fn map_write_error(error: WriteError) -> io::Error {
530    io::Error::other(error)
531}
532
533fn map_read_error(error: ReadError) -> io::Error {
534    io::Error::new(io::ErrorKind::Other, error.to_string())
535}
536
537fn map_finish_error(error: ClosedStream) -> io::Error {
538    io::Error::new(io::ErrorKind::NotConnected, error.to_string())
539}
540
541fn crypto_io_error(error: crate::application_crypto::CryptoError) -> io::Error {
542    io::Error::new(io::ErrorKind::InvalidData, format!("{error:?}"))
543}
544
545fn crypto_stream_io_error(error: CryptoStreamError) -> io::Error {
546    io::Error::new(
547        io::ErrorKind::InvalidData,
548        ApplicationCryptoStreamIoError(error),
549    )
550}
551
552#[cfg(test)]
553mod tests {
554    use super::*;
555
556    #[test]
557    fn resumed_backpressure_cannot_poll_io_after_access_is_lost() {
558        use std::{
559            cell::Cell,
560            future::Future,
561            task::{Context, Poll, Waker},
562        };
563        let allowed = Arc::new(AtomicBool::new(true));
564        let observed = allowed.clone();
565        let access = Some(ApplicationStreamAccess::new(move || {
566            observed.load(Ordering::Acquire)
567        }));
568        let polls = Cell::new(0);
569        let pending = std::future::poll_fn(|_| {
570            polls.set(polls.get() + 1);
571            Poll::<io::Result<()>>::Pending
572        });
573        let mut guarded = std::pin::pin!(with_access(&access, pending));
574        let mut cx = Context::from_waker(Waker::noop());
575        assert!(guarded.as_mut().poll(&mut cx).is_pending());
576        allowed.store(false, Ordering::Release);
577        assert!(
578            matches!(guarded.as_mut().poll(&mut cx), Poll::Ready(Err(error)) if error.kind() == io::ErrorKind::PermissionDenied)
579        );
580        assert_eq!(
581            polls.get(),
582            1,
583            "a resumed write must not enqueue unauthorized bytes"
584        );
585        allowed.store(true, Ordering::Release);
586        assert!(
587            access.as_ref().unwrap().check().is_err(),
588            "denied streams stay closed after reauthorization"
589        );
590    }
591
592    #[test]
593    fn access_lost_during_io_does_not_return_its_result() {
594        use std::{
595            future::Future,
596            task::{Context, Poll, Waker},
597        };
598        let allowed = Arc::new(AtomicBool::new(true));
599        let observed = allowed.clone();
600        let access = Some(ApplicationStreamAccess::new(move || {
601            observed.load(Ordering::Acquire)
602        }));
603        let operation = async {
604            allowed.store(false, Ordering::Release);
605            Ok(b"private bytes")
606        };
607        let mut guarded = std::pin::pin!(with_access(&access, operation));
608        assert!(
609            matches!(guarded.as_mut().poll(&mut Context::from_waker(Waker::noop())), Poll::Ready(Err(error)) if error.kind() == io::ErrorKind::PermissionDenied)
610        );
611    }
612
613    #[test]
614    fn distinguishes_authentication_failure_from_incomplete_shutdown_frame() {
615        let authentication = crypto_stream_io_error(CryptoStreamError::AuthenticationFailed(
616            crate::application_crypto::CryptoError::DecryptFailed,
617        ));
618        let incomplete = crypto_stream_io_error(CryptoStreamError::IncompleteFrame);
619
620        assert!(is_crypto_auth_failure(&authentication));
621        assert!(!is_crypto_auth_failure(&incomplete));
622    }
623
624    #[test]
625    fn transport_errors_are_not_crypto_authentication_failures() {
626        let transport = io::Error::new(io::ErrorKind::ConnectionReset, "connection lost");
627        assert!(!is_crypto_auth_failure(&transport));
628    }
629
630    #[test]
631    fn preserves_peer_stop_code_without_string_parsing() {
632        let stopped = map_write_error(WriteError::Stopped(iroh::endpoint::VarInt::from_u32(0)));
633        assert_eq!(peer_stop_code(&stopped), Some(0));
634
635        let lost = io::Error::other("connection lost");
636        assert_eq!(peer_stop_code(&lost), None);
637    }
638}
openrtc/application_crypto_streams.rs

openrtc/
application_crypto_streams.rs
