openrtc 2.9.8

//! Exact-scope protected named streams. The SDK owns the bounded envelope
//! bootstrap; product protocols receive only a scope-fenced stream pair.
use crate::{
    application_crypto_streams::{ApplicationStreamAdmission, PeerRecvStream, PeerSendStream},
    native_node::{IncomingStream, IncomingStreamType},
    stream_metadata::{decode_prefix, ChannelMetadata, DecodeDecision},
    Client,
};
use anyhow::{bail, ensure, Context, Result};

const MAX_ENVELOPE: usize = 8192;

/// One already decrypted named envelope, with untouched product bytes behind
/// an immutable exact-grant/crypto/physical-generation I/O fence.
pub struct ScopedIncomingStream {
    pub channel: ChannelMetadata,
    pub endpoint_id: iroh::EndpointId,
    pub transport_stable_id: u64,
    pub send: PeerSendStream,
    pub recv: PeerRecvStream,
}

impl Client {
    /// Accept a named protected stream from the host's sole admitted intake.
    /// `metadata.scope` selects current SDK authority; it never creates a grant.
    /// Header bootstrap is private, bounded to 8 KiB and at most ten seconds.
    /// Ciphertext inspected by the admission router is restored exactly once.
    pub async fn accept_scoped_protected_bi(
        &self,
        incoming: IncomingStream,
        timeout_ms: Option<u64>,
    ) -> Result<ScopedIncomingStream> {
        let timeout =
            std::time::Duration::from_millis(timeout_ms.unwrap_or(10_000).clamp(1, 10_000));
        tokio::time::timeout(timeout, async {
            let IncomingStreamType::Bi(send, recv) = incoming.stream else {
                bail!("scoped product channel requires a bidirectional stream");
            };
            let (send, mut recv) = self
                .wrap_bi_with_prefix_for_admission(
                    &incoming.endpoint_id,
                    incoming.transport_stable_id,
                    send,
                    recv,
                    &incoming.recv_prefix,
                    ApplicationStreamAdmission::Envelope,
                )
                .await?;
            let mut bytes = Vec::new();
            let channel = loop {
                match decode_prefix(&bytes) {
                    DecodeDecision::Decoded(prefix) => {
                        ensure!(
                            prefix.consumed_len == bytes.len(),
                            "invalid scoped channel envelope boundary"
                        );
                        break prefix.channel;
                    }
                    DecodeDecision::NotMatched => bail!("missing scoped channel envelope"),
                    DecodeDecision::NeedMore(n) => {
                        ensure!(
                            n > bytes.len() && n <= MAX_ENVELOPE,
                            "invalid scoped channel envelope length"
                        );
                        let start = bytes.len();
                        bytes.resize(n, 0);
                        let mut offset = start;
                        while offset < n {
                            let read = recv.read(&mut bytes[offset..]).await?;
                            ensure!(read > 0, "scoped channel stream ended");
                            offset += read;
                        }
                    }
                }
            };
            let scope = channel
                .metadata
                .as_ref()
                .and_then(|m| m.get("scope"))
                .and_then(|s| s.as_str())
                .context("missing named application stream scope")?;
            ensure!(
                !scope.is_empty() && scope.len() <= 1024,
                "invalid application stream scope"
            );
            let local = self
                .current_node_id()
                .await
                .context("local application identity is unavailable")?;
            let id = Self::deterministic_connection_id(&local, &incoming.endpoint_id.to_string());
            ensure!(
                self.connection_application_crypto_is_confirmed(
                    &id,
                    Some(incoming.transport_stable_id)
                ),
                "scoped stream crypto generation is not confirmed"
            );
            let (send, recv) = self.bind_scoped_application_streams(&id, scope, send, recv)?;
            let current = self
                .get_connection(incoming.endpoint_id)
                .await
                .map(|c| crate::transport_generation::for_connection(&c));
            ensure!(
                current == Some(incoming.transport_stable_id),
                "scoped incoming transport generation is retired"
            );
            // Recheck after the async physical-generation lookup, before any
            // restored/decoded product prefix can be handed to the caller.
            send.binding()
                .context("scoped stream protection missing")?
                .0
                .check()?;
            Ok(ScopedIncomingStream {
                channel,
                endpoint_id: incoming.endpoint_id,
                transport_stable_id: incoming.transport_stable_id,
                send,
                recv,
            })
        })
        .await
        .context("scoped channel envelope timed out")?
    }

    // Only the bounded SDK parser can replace a bootstrap fence. This is not a
    // public rebinding API and cannot resurrect a stream that observed denial.
    fn bind_scoped_application_streams(
        &self,
        id: &str,
        scope: &str,
        send: PeerSendStream,
        recv: PeerRecvStream,
    ) -> Result<(PeerSendStream, PeerRecvStream)> {
        let (bootstrap, key) = send
            .binding()
            .context("scoped stream must be application-protected")?;
        ensure!(
            recv.binding()
                .is_some_and(|access| bootstrap.same_fence(access)),
            "mismatched scoped stream halves"
        );
        bootstrap.check()?;
        let access = self.application_stream_access_for_admission(
            id,
            key,
            ApplicationStreamAdmission::Scope(scope.into()),
        )?;
        bootstrap.check()?;
        access.check()?;
        Ok((send.with_access(access.clone()), recv.with_access(access)))
    }
}