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

openrtc/
application_crypto_streams.rs
