openrtc/
scoped_streams.rs

1//! Exact-scope protected named streams. The SDK owns the bounded envelope
2//! bootstrap; product protocols receive only a scope-fenced stream pair.
3use crate::{
4    application_crypto_streams::{ApplicationStreamAdmission, PeerRecvStream, PeerSendStream},
5    native_node::{IncomingStream, IncomingStreamType},
6    stream_metadata::{decode_prefix, ChannelMetadata, DecodeDecision},
7    Client,
8};
9use anyhow::{bail, ensure, Context, Result};
10
11const MAX_ENVELOPE: usize = 8192;
12
13/// One already decrypted named envelope, with untouched product bytes behind
14/// an immutable exact-grant/crypto/physical-generation I/O fence.
15pub struct ScopedIncomingStream {
16    pub channel: ChannelMetadata,
17    pub endpoint_id: iroh::EndpointId,
18    pub transport_stable_id: u64,
19    pub send: PeerSendStream,
20    pub recv: PeerRecvStream,
21}
22
23impl Client {
24    /// Accept a named protected stream from the host's sole admitted intake.
25    /// `metadata.scope` selects current SDK authority; it never creates a grant.
26    /// Header bootstrap is private, bounded to 8 KiB and at most ten seconds.
27    /// Ciphertext inspected by the admission router is restored exactly once.
28    pub async fn accept_scoped_protected_bi(
29        &self,
30        incoming: IncomingStream,
31        timeout_ms: Option<u64>,
32    ) -> Result<ScopedIncomingStream> {
33        let timeout =
34            std::time::Duration::from_millis(timeout_ms.unwrap_or(10_000).clamp(1, 10_000));
35        tokio::time::timeout(timeout, async {
36            let IncomingStreamType::Bi(send, recv) = incoming.stream else {
37                bail!("scoped product channel requires a bidirectional stream");
38            };
39            let (send, mut recv) = self
40                .wrap_bi_with_prefix_for_admission(
41                    &incoming.endpoint_id,
42                    incoming.transport_stable_id,
43                    send,
44                    recv,
45                    &incoming.recv_prefix,
46                    ApplicationStreamAdmission::Envelope,
47                )
48                .await?;
49            let mut bytes = Vec::new();
50            let channel = loop {
51                match decode_prefix(&bytes) {
52                    DecodeDecision::Decoded(prefix) => {
53                        ensure!(
54                            prefix.consumed_len == bytes.len(),
55                            "invalid scoped channel envelope boundary"
56                        );
57                        break prefix.channel;
58                    }
59                    DecodeDecision::NotMatched => bail!("missing scoped channel envelope"),
60                    DecodeDecision::NeedMore(n) => {
61                        ensure!(
62                            n > bytes.len() && n <= MAX_ENVELOPE,
63                            "invalid scoped channel envelope length"
64                        );
65                        let start = bytes.len();
66                        bytes.resize(n, 0);
67                        let mut offset = start;
68                        while offset < n {
69                            let read = recv.read(&mut bytes[offset..]).await?;
70                            ensure!(read > 0, "scoped channel stream ended");
71                            offset += read;
72                        }
73                    }
74                }
75            };
76            let scope = channel
77                .metadata
78                .as_ref()
79                .and_then(|m| m.get("scope"))
80                .and_then(|s| s.as_str())
81                .context("missing named application stream scope")?;
82            ensure!(
83                !scope.is_empty() && scope.len() <= 1024,
84                "invalid application stream scope"
85            );
86            let local = self
87                .current_node_id()
88                .await
89                .context("local application identity is unavailable")?;
90            let id = Self::deterministic_connection_id(&local, &incoming.endpoint_id.to_string());
91            ensure!(
92                self.connection_application_crypto_is_confirmed(
93                    &id,
94                    Some(incoming.transport_stable_id)
95                ),
96                "scoped stream crypto generation is not confirmed"
97            );
98            let (send, recv) = self.bind_scoped_application_streams(&id, scope, send, recv)?;
99            let current = self
100                .get_connection(incoming.endpoint_id)
101                .await
102                .map(|c| crate::transport_generation::for_connection(&c));
103            ensure!(
104                current == Some(incoming.transport_stable_id),
105                "scoped incoming transport generation is retired"
106            );
107            // Recheck after the async physical-generation lookup, before any
108            // restored/decoded product prefix can be handed to the caller.
109            send.binding()
110                .context("scoped stream protection missing")?
111                .0
112                .check()?;
113            Ok(ScopedIncomingStream {
114                channel,
115                endpoint_id: incoming.endpoint_id,
116                transport_stable_id: incoming.transport_stable_id,
117                send,
118                recv,
119            })
120        })
121        .await
122        .context("scoped channel envelope timed out")?
123    }
124
125    // Only the bounded SDK parser can replace a bootstrap fence. This is not a
126    // public rebinding API and cannot resurrect a stream that observed denial.
127    fn bind_scoped_application_streams(
128        &self,
129        id: &str,
130        scope: &str,
131        send: PeerSendStream,
132        recv: PeerRecvStream,
133    ) -> Result<(PeerSendStream, PeerRecvStream)> {
134        let (bootstrap, key) = send
135            .binding()
136            .context("scoped stream must be application-protected")?;
137        ensure!(
138            recv.binding()
139                .is_some_and(|access| bootstrap.same_fence(access)),
140            "mismatched scoped stream halves"
141        );
142        bootstrap.check()?;
143        let access = self.application_stream_access_for_admission(
144            id,
145            key,
146            ApplicationStreamAdmission::Scope(scope.into()),
147        )?;
148        bootstrap.check()?;
149        access.check()?;
150        Ok((send.with_access(access.clone()), recv.with_access(access)))
151    }
152}
openrtc/scoped_streams.rs

openrtc/
scoped_streams.rs
