openrtc/client/

application_crypto_impl.rs

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, RwLock as StdRwLock};

use iroh::endpoint::{RecvStream, SendStream};

use crate::application_crypto::{self, APPLICATION_KEY_BYTES};
use crate::application_crypto_streams::{
    wrap_peer_streams, ApplicationStreamAccess, ApplicationStreamAdmission, PeerRecvStream,
    PeerSendStream,
};
use crate::client::Client;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct ConnectionApplicationCryptoConfirmation {
    transport_stable_id: u64,
    confirmed: bool,
}

pub(crate) struct ApplicationKeyHandshakeOutcome {
    #[cfg(not(target_arch = "wasm32"))]
    pub local_public_key: [u8; crate::key_agreement::PUBLIC_KEY_BYTES],
    #[cfg(not(target_arch = "wasm32"))]
    pub key_changed: bool,
    #[cfg(target_arch = "wasm32")]
    pub reply_action: Option<&'static str>,
}

#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
impl Client {
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn notify_connection_application_route_update(&self) {
        self.connection_application_route_updates.notify_waiters();
    }

    #[cfg(target_arch = "wasm32")]
    pub(crate) fn application_key_handshake_frame(
        &self,
        connection_id: &str,
        action: &str,
    ) -> anyhow::Result<serde_json::Value> {
        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};

        let public_key = self
            .get_or_create_connection_key_agreement(connection_id)
            .map_err(|error| {
                anyhow::anyhow!("application key agreement initialization failed: {error:?}")
            })?
            .public_key_bytes();
        Ok(serde_json::json!({
            "type": "handshake",
            "action": action,
            "capabilities": {
                "applicationKeyAgreement": true,
            },
            "applicationKeyAgreement": {
                "algorithm": crate::key_agreement::KEY_ALGORITHM,
                "publicKey": URL_SAFE_NO_PAD.encode(public_key),
            },
        }))
    }

    pub(crate) async fn accept_application_key_handshake(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        transport_stable_id: Option<u64>,
        action: Option<&str>,
        remote_public_key: [u8; crate::key_agreement::PUBLIC_KEY_BYTES],
    ) -> anyhow::Result<ApplicationKeyHandshakeOutcome> {
        if matches!(action, Some("response" | "ack")) && transport_stable_id.is_none() {
            anyhow::bail!(
                "application key confirmation is missing the Rust-owned transport stable id: action={}",
                action.unwrap_or("unknown"),
            );
        }
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("local node id is unavailable"))?;
        let (local_public_key, key, key_changed) = if let Some(transport_stable_id) =
            transport_stable_id
        {
            let remote_endpoint_id = remote_node_id
                .parse::<iroh::EndpointId>()
                .map_err(|error| anyhow::anyhow!("invalid remote endpoint id: {error}"))?;
            let node = self.iroh_node.read().await;
            let node = node
                .as_ref()
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("iroh node is unavailable"))?;
            let confirmation_transport =
                matches!(action, Some("response" | "ack")).then_some(transport_stable_id);
            let applied = node
                .with_current_connection_generation(
                    remote_endpoint_id,
                    transport_stable_id,
                    || async {
                        self.connection_manager
                            .with_current_transport_generation(
                                connection_id,
                                remote_node_id,
                                transport_stable_id,
                                |_| {
                                    self.derive_and_apply_application_key_handshake_after_owner_fence(
                                        connection_id,
                                        &local_node_id,
                                        remote_node_id,
                                        remote_public_key,
                                        confirmation_transport,
                                    )
                                },
                            )
                            .await
                    },
                )
                .await
                .flatten();
            let Some(outcome) = applied else {
                let fence = self
                    .incoming_stream_generation_fence(remote_endpoint_id, transport_stable_id)
                    .await;
                anyhow::bail!(
                    "application key confirmation rejected for stale transport generation: incoming_stable_id={} physical_stable_id={:?} managed_stable_id={:?} manager_record_exists={}",
                    fence.incoming_transport_stable_id,
                    fence.physical_transport_stable_id,
                    fence.managed_transport_stable_id,
                    fence.manager_record_exists,
                );
            };
            outcome?
        } else {
            // Initial legacy hello frames may precede manager publication and
            // cannot confirm a route. Their key state is still applied as one
            // atomic, non-routable security update.
            self.derive_and_apply_application_key_handshake_after_owner_fence(
                connection_id,
                &local_node_id,
                remote_node_id,
                remote_public_key,
                None,
            )?
        };
        eprintln!(
            "[OpenRTC][KEY-AGREEMENT] connection_id={} action={} key_changed={} confirmed={} key_fingerprint={} local_public_fingerprint={} remote_public_fingerprint={} local_node_id={} remote_node_id={}",
            connection_id,
            action.unwrap_or("hello"),
            key_changed,
            self.connection_application_crypto_is_confirmed(connection_id, transport_stable_id),
            crate::key_agreement::key_fingerprint(&key),
            crate::key_agreement::key_fingerprint(&local_public_key),
            crate::key_agreement::key_fingerprint(&remote_public_key),
            local_node_id,
            remote_node_id,
        );
        #[cfg(target_arch = "wasm32")]
        let reply_action = match action {
            Some("ack") => None,
            Some("response") => Some("ack"),
            _ => Some("response"),
        };
        Ok(ApplicationKeyHandshakeOutcome {
            #[cfg(not(target_arch = "wasm32"))]
            local_public_key,
            #[cfg(not(target_arch = "wasm32"))]
            key_changed,
            #[cfg(target_arch = "wasm32")]
            reply_action,
        })
    }

    /// Commit the derived key and its exact confirmation as one observable
    /// application-security state. The caller must already hold both the
    /// physical generation lock and the matching manager generation lock.
    fn derive_and_apply_application_key_handshake_after_owner_fence(
        &self,
        connection_id: &str,
        local_node_id: &str,
        remote_node_id: &str,
        remote_public_key: [u8; crate::key_agreement::PUBLIC_KEY_BYTES],
        confirmation_transport_stable_id: Option<u64>,
    ) -> anyhow::Result<(
        [u8; crate::key_agreement::PUBLIC_KEY_BYTES],
        [u8; APPLICATION_KEY_BYTES],
        bool,
    )> {
        let _state = self
            .connection_application_crypto_state
            .write()
            .map_err(|_| anyhow::anyhow!("application crypto state poisoned"))?;
        let mut agreements = self
            .connection_application_key_agreements
            .write()
            .map_err(|_| anyhow::anyhow!("application key agreement map poisoned"))?;
        let agreement = match agreements.get(connection_id).cloned() {
            Some(agreement) => agreement,
            None => crate::key_agreement::KeyAgreement::generate().map_err(|error| {
                anyhow::anyhow!("application key agreement initialization failed: {error:?}")
            })?,
        };
        let key = agreement
            .derive_crypto_key(&remote_public_key, local_node_id, remote_node_id)
            .map_err(|error| anyhow::anyhow!("application key derivation failed: {error:?}"))?;
        let local_public_key = agreement.public_key_bytes();

        // Lock every security component before mutating any of them. A
        // poisoned inner registry therefore fails closed without leaving a
        // partial key epoch, and rejected generations cannot recreate an
        // agreement after terminal cleanup.
        let mut keys = self
            .connection_application_crypto_keys
            .write()
            .map_err(|_| anyhow::anyhow!("application crypto key map poisoned"))?;
        let mut required = self
            .connection_application_crypto_required
            .write()
            .map_err(|_| anyhow::anyhow!("application crypto requirement map poisoned"))?;
        let mut sequences = self
            .connection_application_crypto_outbound_sequences
            .write()
            .map_err(|_| anyhow::anyhow!("application crypto sequence map poisoned"))?;
        let mut confirmed = self
            .connection_application_crypto_confirmed
            .write()
            .map_err(|_| anyhow::anyhow!("application crypto confirmation map poisoned"))?;

        agreements
            .entry(connection_id.to_string())
            .or_insert(agreement);
        let key_changed = keys.get(connection_id) != Some(&key);
        if key_changed {
            keys.insert(connection_id.to_string(), key);
        }
        required.insert(connection_id.to_string());
        if key_changed {
            sequences.insert(connection_id.to_string(), 0);
        }
        match confirmation_transport_stable_id {
            Some(transport_stable_id) => {
                confirmed.insert(
                    connection_id.to_string(),
                    ConnectionApplicationCryptoConfirmation {
                        transport_stable_id,
                        confirmed: true,
                    },
                );
            }
            None if key_changed => {
                if let Some(state) = confirmed.get_mut(connection_id) {
                    state.confirmed = false;
                }
            }
            None => {}
        }
        drop(confirmed);
        drop(sequences);
        drop(required);
        drop(keys);
        drop(agreements);
        drop(_state);
        #[cfg(not(target_arch = "wasm32"))]
        self.notify_connection_application_route_update();
        Ok((local_public_key, key, key_changed))
    }

    pub(crate) async fn assert_application_crypto_generation(
        &self,
        connection_id: &str,
        expected_endpoint: &str,
        expected_transport_stable_id: Option<u64>,
        expected_transport_generation: u64,
        expected_route_generation: u64,
    ) -> anyhow::Result<()> {
        let current = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .ok_or_else(|| {
                anyhow::anyhow!("connection {connection_id} retired during key agreement")
            })?;
        let current_endpoint = current
            .endpoint_id
            .as_deref()
            .or(current.node_id.as_deref());
        anyhow::ensure!(
            current_endpoint == Some(expected_endpoint)
                && current.transport_stable_id == expected_transport_stable_id
                && current.transport_generation == expected_transport_generation
                && current.route_generation == expected_route_generation,
            "connection {connection_id} generation changed during key agreement"
        );
        Ok(())
    }

    #[cfg_attr(not(any(target_arch = "wasm32", test)), allow(dead_code))]
    pub(crate) fn set_connection_application_crypto_key(
        &self,
        connection_id: &str,
        key: [u8; APPLICATION_KEY_BYTES],
    ) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        let key_changed = self
            .connection_application_crypto_keys
            .write()
            .map(|mut keys| {
                if keys.get(connection_id) == Some(&key) {
                    false
                } else {
                    keys.insert(connection_id.to_string(), key);
                    true
                }
            })
            .unwrap_or(false);
        if key_changed {
            if let Ok(mut confirmed) = self.connection_application_crypto_confirmed.write() {
                if let Some(state) = confirmed.get_mut(connection_id) {
                    state.confirmed = false;
                }
            }
            #[cfg(not(target_arch = "wasm32"))]
            self.notify_connection_application_route_update();
            if let Ok(mut sequences) = self
                .connection_application_crypto_outbound_sequences
                .write()
            {
                sequences.insert(connection_id.to_string(), 0);
            }
        }
    }

    pub(crate) fn set_connection_application_crypto_required(&self, connection_id: &str) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        let changed = self
            .connection_application_crypto_required
            .write()
            .map(|mut required| required.insert(connection_id.to_string()))
            .unwrap_or(false);
        #[cfg(target_arch = "wasm32")]
        let _ = changed;
        #[cfg(not(target_arch = "wasm32"))]
        if changed {
            self.notify_connection_application_route_update();
        }
    }

    /// Require reciprocal application protection for every trusted native
    /// user-device connection promoted after this call.
    ///
    /// Products with protected named channels should enable this before
    /// initializing presence or auto-connect on every participating runtime.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn require_app_crypto(&self, required: bool) {
        self.trusted_user_device_application_crypto_required
            .store(required, std::sync::atomic::Ordering::Release);
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn trusted_user_device_application_crypto_is_required(&self) -> bool {
        self.trusted_user_device_application_crypto_required
            .load(std::sync::atomic::Ordering::Acquire)
    }

    pub(crate) fn clear_connection_application_crypto_key(&self, connection_id: &str) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        if let Ok(mut keys) = self.connection_application_crypto_keys.write() {
            keys.remove(connection_id);
        }
        if let Ok(mut required) = self.connection_application_crypto_required.write() {
            required.remove(connection_id);
        }
        if let Ok(mut sequences) = self
            .connection_application_crypto_outbound_sequences
            .write()
        {
            sequences.remove(connection_id);
        }
        if let Ok(mut agreements) = self.connection_application_key_agreements.write() {
            agreements.remove(connection_id);
        }
        if let Ok(mut confirmed) = self.connection_application_crypto_confirmed.write() {
            confirmed.remove(connection_id);
        }
        #[cfg(not(target_arch = "wasm32"))]
        self.notify_connection_application_route_update();
    }

    pub fn connection_application_crypto_key(
        &self,
        connection_id: &str,
    ) -> Option<[u8; APPLICATION_KEY_BYTES]> {
        self.application_crypto_key_for_connection(Some(connection_id))
    }

    pub(crate) fn connection_requires_application_crypto(&self, connection_id: &str) -> bool {
        let Ok(_state) = self.connection_application_crypto_state.read() else {
            return false;
        };
        self.connection_application_crypto_required
            .read()
            .ok()
            .map(|required| required.contains(connection_id))
            .unwrap_or(false)
    }

    pub(crate) fn connection_requires_application_crypto_confirmation(
        &self,
        connection_id: &str,
    ) -> bool {
        self.connection_requires_application_crypto(connection_id)
            && self
                .connection_application_key_agreements
                .read()
                .ok()
                .is_some_and(|agreements| agreements.contains_key(connection_id))
    }

    #[cfg_attr(not(any(feature = "iroh-carrier-core", test)), allow(dead_code))]
    pub(crate) fn confirm_connection_application_crypto(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        let changed = self
            .connection_application_crypto_confirmed
            .write()
            .map(|mut confirmed| match confirmed.get_mut(connection_id) {
                Some(state) if state.transport_stable_id == transport_stable_id => {
                    let changed = !state.confirmed;
                    state.confirmed = true;
                    changed
                }
                Some(_) | None => false,
            })
            .unwrap_or(false);
        #[cfg(target_arch = "wasm32")]
        let _ = changed;
        #[cfg(not(target_arch = "wasm32"))]
        if changed {
            self.notify_connection_application_route_update();
        }
    }

    pub(crate) fn connection_application_crypto_is_confirmed(
        &self,
        connection_id: &str,
        transport_stable_id: Option<u64>,
    ) -> bool {
        let Ok(_state) = self.connection_application_crypto_state.read() else {
            return false;
        };
        let Some(transport_stable_id) = transport_stable_id else {
            return false;
        };
        self.connection_application_crypto_confirmed
            .read()
            .ok()
            .and_then(|confirmed| confirmed.get(connection_id).copied())
            .is_some_and(|state| {
                state.transport_stable_id == transport_stable_id && state.confirmed
            })
    }

    #[cfg(target_arch = "wasm32")]
    pub(crate) fn connection_application_crypto_confirmation_snapshot(
        &self,
        connection_id: &str,
    ) -> Option<(u64, bool)> {
        let _state = self.connection_application_crypto_state.read().ok()?;
        self.connection_application_crypto_confirmed
            .read()
            .ok()
            .and_then(|confirmed| confirmed.get(connection_id).copied())
            .map(|state| (state.transport_stable_id, state.confirmed))
    }

    #[cfg(test)]
    pub(crate) fn connection_application_crypto_has_bound_confirmation(
        &self,
        connection_id: &str,
    ) -> bool {
        let Ok(_state) = self.connection_application_crypto_state.read() else {
            return false;
        };
        self.connection_application_crypto_confirmed
            .read()
            .ok()
            .and_then(|confirmed| confirmed.get(connection_id).copied())
            .is_some_and(|state| state.confirmed)
    }

    /// Bind reciprocal key confirmation to the exact physical generation
    /// before that generation is published by the asynchronous manager. An
    /// old ACK can then never reopen the prior route during replacement.
    pub(crate) fn bind_connection_application_crypto_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        let changed = self
            .connection_application_crypto_confirmed
            .write()
            .map(|mut confirmed| match confirmed.get(connection_id) {
                Some(state) if state.transport_stable_id == transport_stable_id => false,
                _ => {
                    confirmed.insert(
                        connection_id.to_string(),
                        ConnectionApplicationCryptoConfirmation {
                            transport_stable_id,
                            confirmed: false,
                        },
                    );
                    true
                }
            })
            .unwrap_or(false);
        #[cfg(target_arch = "wasm32")]
        let _ = changed;
        #[cfg(not(target_arch = "wasm32"))]
        if changed {
            self.notify_connection_application_route_update();
        }
    }

    pub(crate) fn clear_connection_application_crypto_confirmation(&self, connection_id: &str) {
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return;
        };
        let changed = self
            .connection_application_crypto_confirmed
            .write()
            .map(|mut confirmed| {
                confirmed.get_mut(connection_id).is_some_and(|state| {
                    let changed = state.confirmed;
                    state.confirmed = false;
                    changed
                })
            })
            .unwrap_or(false);
        #[cfg(target_arch = "wasm32")]
        let _ = changed;
        #[cfg(not(target_arch = "wasm32"))]
        if changed {
            self.notify_connection_application_route_update();
        }
    }

    /// Wait for the existing admission and key-confirmation owners to commit
    /// the complete application route observed by an incoming product stream.
    /// QUIC orders bytes within one stream but not across streams, so the
    /// product stream can arrive just before a directional admission proof or
    /// the control ACK even when the sender emitted control first.
    ///
    /// This does not advance admission, confirm a key, or retry a lifecycle
    /// transition. It only waits on the typed route event and fails closed if
    /// the key, token epoch, or exact transport proof does not converge.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn wait_for_connection_application_route(
        &self,
        connection_id: &str,
        expected_transport_stable_id: u64,
        expected_key: [u8; APPLICATION_KEY_BYTES],
        expected_admission_fingerprint: Option<&str>,
        expected_security_epoch_fingerprint: Option<&str>,
        timeout: std::time::Duration,
    ) -> bool {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            // Register the waiter before checking state so a concurrent ACK
            // cannot notify between the state read and the await.
            let notified = self.connection_application_route_updates.notified();
            let same_epoch = self.connection_application_crypto_key(connection_id)
                == Some(expected_key)
                && self
                    .session_token_registry
                    .admission_fingerprint(connection_id)
                    .as_deref()
                    == expected_admission_fingerprint
                && self
                    .session_token_registry
                    .application_security_epoch_fingerprint(connection_id)
                    .as_deref()
                    == expected_security_epoch_fingerprint;
            if !same_epoch {
                return false;
            }
            if self.native_application_stream_admitted_for_transport(
                connection_id,
                expected_transport_stable_id,
            ) {
                return true;
            }
            if tokio::time::timeout_at(deadline, notified).await.is_err() {
                return false;
            }
        }
    }

    pub(crate) fn application_crypto_key_for_connection(
        &self,
        connection_id: Option<&str>,
    ) -> Option<[u8; APPLICATION_KEY_BYTES]> {
        let connection_id = connection_id?.trim();
        if connection_id.is_empty() {
            return None;
        }
        let _state = self.connection_application_crypto_state.read().ok()?;
        self.connection_application_crypto_keys
            .read()
            .ok()?
            .get(connection_id)
            .copied()
    }

    fn application_crypto_requirement_and_key(
        &self,
        connection_id: &str,
    ) -> (bool, Option<[u8; APPLICATION_KEY_BYTES]>) {
        let Ok(_state) = self.connection_application_crypto_state.read() else {
            return (false, None);
        };
        let required = self
            .connection_application_crypto_required
            .read()
            .ok()
            .is_some_and(|required| required.contains(connection_id));
        let key = self
            .connection_application_crypto_keys
            .read()
            .ok()
            .and_then(|keys| keys.get(connection_id).copied());
        (required, key)
    }

    pub(crate) fn get_or_create_connection_key_agreement(
        &self,
        connection_id: &str,
    ) -> Result<crate::key_agreement::KeyAgreement, crate::key_agreement::KeyAgreementError> {
        let connection_id = connection_id.trim();
        let Ok(_state) = self.connection_application_crypto_state.write() else {
            return crate::key_agreement::KeyAgreement::generate();
        };
        if let Some(existing) = self
            .connection_application_key_agreements
            .read()
            .ok()
            .and_then(|agreements| agreements.get(connection_id).cloned())
        {
            return Ok(existing);
        }

        let generated = crate::key_agreement::KeyAgreement::generate()?;
        if let Ok(mut agreements) = self.connection_application_key_agreements.write() {
            Ok(agreements
                .entry(connection_id.to_string())
                .or_insert_with(|| generated.clone())
                .clone())
        } else {
            Ok(generated)
        }
    }

    #[cfg_attr(
        any(target_arch = "wasm32", not(feature = "transport-webrtc")),
        allow(dead_code)
    )]
    pub(crate) fn connection_key_agreement_public_key(
        &self,
        connection_id: &str,
    ) -> Option<[u8; crate::key_agreement::PUBLIC_KEY_BYTES]> {
        let _state = self.connection_application_crypto_state.read().ok()?;
        self.connection_application_key_agreements
            .read()
            .ok()?
            .get(connection_id.trim())
            .map(|agreement| agreement.public_key_bytes())
    }

    pub(crate) async fn application_crypto_key_for_endpoint(
        &self,
        endpoint_id: &iroh::EndpointId,
    ) -> Option<[u8; APPLICATION_KEY_BYTES]> {
        let endpoint_str = endpoint_id.to_string();
        let records = self
            .connection_manager
            .get_by_endpoint_id(&endpoint_str)
            .await;
        for record in records {
            if let Some(key) =
                self.application_crypto_key_for_connection(Some(&record.connection_id))
            {
                return Some(key);
            }
        }
        None
    }

    pub(crate) async fn application_crypto_key_for_connection_or_endpoint(
        &self,
        connection_id: Option<&str>,
        endpoint_id: &iroh::EndpointId,
    ) -> Option<[u8; APPLICATION_KEY_BYTES]> {
        let endpoint_str = endpoint_id.to_string();

        if let Some(connection_id) = connection_id.map(str::trim).filter(|id| !id.is_empty()) {
            // A durable-device projection can briefly retain the prior browser
            // connection id while the gateway roster has already supplied the replacement
            // endpoint. Never bind that stale generation's crypto key to a
            // stream opened on the new endpoint; validate physical ownership
            // before preferring the projected connection id.
            if let Some(record) = self
                .connection_manager
                .get_by_connection_id(connection_id)
                .await
            {
                let record_matches_endpoint = record
                    .endpoint_id
                    .as_deref()
                    .or(record.node_id.as_deref())
                    .is_some_and(|value| value == endpoint_str);
                if record_matches_endpoint {
                    if let Some(key) =
                        self.application_crypto_key_for_connection(Some(connection_id))
                    {
                        return Some(key);
                    }
                }
            }
        }

        if let Some(record) = self
            .connection_manager
            .best_connection_for_peer(&endpoint_str)
            .await
        {
            if let Some(key) =
                self.application_crypto_key_for_connection(Some(&record.connection_id))
            {
                return Some(key);
            }
        }

        let mut records = self
            .connection_manager
            .get_by_endpoint_id(&endpoint_str)
            .await;
        records.sort_by(|left, right| {
            let left_connected = matches!(
                left.state,
                crate::connection_manager::ConnectionState::Connected
            );
            let right_connected = matches!(
                right.state,
                crate::connection_manager::ConnectionState::Connected
            );
            left_connected
                .cmp(&right_connected)
                .then_with(|| left.transport_generation.cmp(&right.transport_generation))
                .then_with(|| left.updated_at_ms.cmp(&right.updated_at_ms))
        });
        for record in records.into_iter().rev() {
            if let Some(key) =
                self.application_crypto_key_for_connection(Some(&record.connection_id))
            {
                return Some(key);
            }
        }

        None
    }

    async fn application_crypto_required_for_connection_or_endpoint(
        &self,
        connection_id: Option<&str>,
        endpoint_id: &iroh::EndpointId,
    ) -> bool {
        if connection_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .is_some_and(|value| self.connection_requires_application_crypto(value))
        {
            return true;
        }

        let endpoint_id = endpoint_id.to_string();
        self.connection_manager
            .get_by_endpoint_id(&endpoint_id)
            .await
            .iter()
            .any(|record| self.connection_requires_application_crypto(&record.connection_id))
    }

    pub(crate) async fn required_crypto_key(
        &self,
        connection_id: Option<&str>,
        endpoint_id: &iroh::EndpointId,
        operation: &str,
    ) -> anyhow::Result<Option<[u8; APPLICATION_KEY_BYTES]>> {
        if let Some(connection_id) = connection_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            let endpoint_id_text = endpoint_id.to_string();
            if self
                .connection_manager
                .get_by_connection_id(connection_id)
                .await
                .is_some_and(|record| {
                    record
                        .endpoint_id
                        .as_deref()
                        .or(record.node_id.as_deref())
                        .is_some_and(|value| value == endpoint_id_text)
                })
            {
                let (required, key) = self.application_crypto_requirement_and_key(connection_id);
                return enforce_application_crypto_requirement(required, key, operation);
            }
        }

        let key = self
            .application_crypto_key_for_connection_or_endpoint(connection_id, endpoint_id)
            .await;
        let required = self
            .application_crypto_required_for_connection_or_endpoint(connection_id, endpoint_id)
            .await;
        enforce_application_crypto_requirement(required, key, operation)
    }

    /// Wrap an incoming product stream against the exact logical connection
    /// selected by the native ingress router. Product hosts should prefer this
    /// over endpoint-only lookup so a retired same-endpoint record can never
    /// supply a stale application key.
    pub fn secure_connection_bi(
        &self,
        connection_id: &str,
        send: SendStream,
        recv: RecvStream,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        self.secure_connection_bi_with_prefix(connection_id, send, recv, &[])
    }

    /// Wrap an incoming product stream against the exact logical connection
    /// while restoring wire bytes consumed by the Rust admission classifier.
    /// Embedding hosts must pass `IncomingStream::recv_prefix` here; dropping
    /// it would corrupt the first protected frame.
    pub fn secure_connection_bi_with_prefix(
        &self,
        connection_id: &str,
        send: SendStream,
        recv: RecvStream,
        recv_prefix: &[u8],
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        let (required, key) = self.application_crypto_requirement_and_key(connection_id);
        let key =
            enforce_application_crypto_requirement(required, key, "incoming application stream")?;
        self.wrap_application_streams(connection_id, key, send, recv, recv_prefix)
    }

    /// Observe existing admission and crypto authority throughout product I/O.
    /// Weak references do not keep a retired Client alive. Same-scope bearer
    /// renewal is allowed; a changed key, scope, device or confirmation is not.
    pub(crate) fn application_stream_access(
        &self,
        connection_id: &str,
        key: [u8; APPLICATION_KEY_BYTES],
    ) -> anyhow::Result<ApplicationStreamAccess> {
        self.application_stream_access_for_admission(
            connection_id,
            key,
            ApplicationStreamAdmission::Latest,
        )
    }

    pub(crate) fn application_stream_access_for_admission(
        &self,
        connection_id: &str,
        key: [u8; APPLICATION_KEY_BYTES],
        selected: ApplicationStreamAdmission,
    ) -> anyhow::Result<ApplicationStreamAccess> {
        use crate::session_token::SessionAdmission;
        let admission = match &selected {
            ApplicationStreamAdmission::Scope(scope) => self
                .session_token_registry
                .admission_for_scope(connection_id, scope)
                .ok_or_else(|| {
                    anyhow::anyhow!("requested application stream scope is not admitted")
                })?,
            _ => self.session_admission(connection_id),
        };
        anyhow::ensure!(
            matches!(&admission, SessionAdmission::Accepted { .. })
                || (matches!(selected, ApplicationStreamAdmission::Latest)
                    && admission == SessionAdmission::Pending
                    && self.session_token_registry.is_empty()),
            "application stream is not admitted"
        );
        let confirmation = self
            .connection_application_crypto_confirmed
            .read()
            .map_err(|_| anyhow::anyhow!("application crypto state unavailable"))?
            .get(connection_id)
            .copied();
        let registry = Arc::downgrade(&self.session_token_registry);
        let state = Arc::downgrade(&self.connection_application_crypto_state);
        let keys = Arc::downgrade(&self.connection_application_crypto_keys);
        let confirmations = Arc::downgrade(&self.connection_application_crypto_confirmed);
        let id = connection_id.to_owned();
        Ok(ApplicationStreamAccess::new(move || {
            let (Some(registry), Some(state), Some(keys), Some(confirmations)) = (
                registry.upgrade(),
                state.upgrade(),
                keys.upgrade(),
                confirmations.upgrade(),
            ) else {
                return false;
            };
            let admitted = match &selected {
                ApplicationStreamAdmission::Latest => {
                    registry.admission(&id) == admission
                        && (admission != SessionAdmission::Pending || registry.is_empty())
                }
                ApplicationStreamAdmission::Scope(scope) => {
                    registry.admission_for_scope(&id, scope).as_ref() == Some(&admission)
                }
                ApplicationStreamAdmission::Envelope => {
                    matches!(registry.admission(&id), SessionAdmission::Accepted { .. })
                }
            };
            if !admitted {
                tracing::debug!(target: "openrtc::stream", event = "access_denied",
                    connection_ref = %crate::diagnostics::reference(&id), reason = "authority_changed",
                    "Protected stream authority changed");
                return false;
            }
            let Ok(_state) = state.read() else {
                return false;
            };
            let Ok(keys) = keys.read() else {
                return false;
            };
            let Ok(confirmations) = confirmations.read() else {
                return false;
            };
            let key_current = keys.get(&id) == Some(&key);
            let generation_current = confirmations.get(&id).copied() == confirmation;
            drop(confirmations);
            drop(keys);
            drop(_state);
            if !key_current || !generation_current {
                tracing::debug!(target: "openrtc::stream", event = "access_denied",
                    connection_ref = %crate::diagnostics::reference(&id),
                    key_current, generation_current, "Protected stream crypto epoch changed");
            }
            key_current && generation_current
        }))
    }

    pub(crate) fn wrap_application_streams(
        &self,
        connection_id: &str,
        key: Option<[u8; APPLICATION_KEY_BYTES]>,
        send: SendStream,
        recv: RecvStream,
        recv_prefix: &[u8],
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        self.wrap_application_streams_for_admission(
            connection_id,
            key,
            send,
            recv,
            recv_prefix,
            ApplicationStreamAdmission::Latest,
        )
    }

    pub(crate) fn wrap_application_streams_for_admission(
        &self,
        connection_id: &str,
        key: Option<[u8; APPLICATION_KEY_BYTES]>,
        send: SendStream,
        recv: RecvStream,
        recv_prefix: &[u8],
        selected: ApplicationStreamAdmission,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        match key {
            Some(key) => {
                let access = match selected {
                    ApplicationStreamAdmission::Latest => {
                        self.application_stream_access(connection_id, key)?
                    }
                    selected => {
                        self.application_stream_access_for_admission(connection_id, key, selected)?
                    }
                };
                Ok((
                    PeerSendStream::encrypted(send, key).with_access(access.clone()),
                    PeerRecvStream::encrypted_with_prefix(recv, key, recv_prefix)
                        .map_err(|error| {
                            anyhow::anyhow!(
                                "incoming application crypto stream wrap failed: {error}"
                            )
                        })?
                        .with_access(access),
                ))
            }
            None if recv_prefix.is_empty()
                && matches!(selected, ApplicationStreamAdmission::Latest) =>
            {
                Ok((PeerSendStream::plain(send), PeerRecvStream::plain(recv)))
            }
            None => anyhow::bail!(
                "incoming application stream carried an encrypted prefix without an application key"
            ),
        }
    }

    /// Wrap a native product stream against the exact admitted physical
    /// transport generation that produced it. Native host bridges should use
    /// this boundary before exposing stream bytes outside Rust so application
    /// keys remain owned by the runtime and a same-endpoint replacement cannot
    /// lend its key to a retired stream.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn wrap_incoming_bi(
        &self,
        endpoint_id: &iroh::EndpointId,
        transport_stable_id: u64,
        send: SendStream,
        recv: RecvStream,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        self.wrap_bi_with_prefix(endpoint_id, transport_stable_id, send, recv, &[])
            .await
    }

    /// Generation-fenced incoming wrapper that also restores bytes inspected
    /// by the native admission router while classifying an out-of-order
    /// encrypted product frame.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn wrap_bi_with_prefix(
        &self,
        endpoint_id: &iroh::EndpointId,
        transport_stable_id: u64,
        send: SendStream,
        recv: RecvStream,
        recv_prefix: &[u8],
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        self.wrap_bi_with_prefix_for_admission(
            endpoint_id,
            transport_stable_id,
            send,
            recv,
            recv_prefix,
            ApplicationStreamAdmission::Latest,
        )
        .await
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn wrap_bi_with_prefix_for_admission(
        &self,
        endpoint_id: &iroh::EndpointId,
        transport_stable_id: u64,
        send: SendStream,
        recv: RecvStream,
        recv_prefix: &[u8],
        selected: ApplicationStreamAdmission,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        let current_transport_stable_id = self
            .get_connection(*endpoint_id)
            .await
            .map(|connection| crate::transport_generation::for_connection(&connection));
        if current_transport_stable_id != Some(transport_stable_id) {
            anyhow::bail!(
                "incoming application stream belongs to a retired transport generation: incoming={} current={:?}",
                transport_stable_id,
                current_transport_stable_id,
            );
        }
        let local_node_id = self.current_node_id().await.ok_or_else(|| {
            anyhow::anyhow!("local node id unavailable for incoming application stream")
        })?;
        let connection_id =
            Self::deterministic_connection_id(&local_node_id, &endpoint_id.to_string());
        if !self
            .native_application_stream_admitted_for_transport(&connection_id, transport_stable_id)
        {
            anyhow::bail!("incoming application stream transport generation is not admitted");
        }
        let key = enforce_application_crypto_requirement(
            self.connection_requires_application_crypto(&connection_id),
            self.application_crypto_key_for_connection(Some(&connection_id)),
            "incoming application stream",
        )?;
        self.wrap_application_streams_for_admission(
            &connection_id,
            key,
            send,
            recv,
            recv_prefix,
            selected,
        )
    }

    /// Wrap an admitted incoming application stream with the same per-peer
    /// application-crypto boundary used by `open_peer_bi` on the sender.
    /// Product hosts should call this before parsing channel envelopes or
    /// application frames from the public incoming-stream queue.
    pub async fn secure_incoming_bi(
        &self,
        endpoint_id: &iroh::EndpointId,
        send: SendStream,
        recv: RecvStream,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        let key = self
            .required_crypto_key(None, endpoint_id, "incoming application stream")
            .await?;
        if key.is_none() {
            return wrap_peer_streams(None, send, recv).map_err(Into::into);
        }
        let connection_id = self
            .peer_session(&endpoint_id.to_string())
            .await
            .and_then(|snapshot| snapshot.active_connection_id)
            .ok_or_else(|| {
                anyhow::anyhow!("protected incoming stream has no active logical connection")
            })?;
        self.wrap_application_streams(&connection_id, key, send, recv, &[])
    }

    pub(crate) fn wrap_peer_streams_for_connection(
        &self,
        connection_id: Option<&str>,
        send: SendStream,
        recv: RecvStream,
    ) -> anyhow::Result<(PeerSendStream, PeerRecvStream)> {
        // SDK candidate/admission control deliberately does not use the product
        // access fence: it proves or restores the admission being checked.
        let key = self.application_crypto_key_for_connection(connection_id);
        wrap_peer_streams(key, send, recv)
            .map_err(|error| anyhow::anyhow!("application crypto stream wrap failed: {error}"))
    }

    pub(crate) async fn assert_raw_peer_stream_allowed(
        &self,
        endpoint_id: &iroh::EndpointId,
    ) -> anyhow::Result<()> {
        if self
            .application_crypto_key_for_endpoint(endpoint_id)
            .await
            .is_some()
        {
            anyhow::bail!(
                "raw iroh stream open is not allowed while application crypto is active for this peer; use open_peer_bi/open_peer_uni instead"
            );
        }
        Ok(())
    }

    pub(crate) fn protect_outbound_application_payload(
        &self,
        connection_id: &str,
        payload: &[u8],
    ) -> anyhow::Result<Vec<u8>> {
        self.protect_outbound_application_payload_with_type(connection_id, 0, payload)
    }

    pub(crate) fn protect_outbound_application_payload_with_type(
        &self,
        connection_id: &str,
        type_id: u8,
        payload: &[u8],
    ) -> anyhow::Result<Vec<u8>> {
        let _state = self
            .connection_application_crypto_state
            .read()
            .map_err(|_| anyhow::anyhow!("application crypto state poisoned"))?;
        let keys = self
            .connection_application_crypto_keys
            .read()
            .map_err(|_| anyhow::anyhow!("application crypto key map poisoned"))?;
        let Some(key) = keys.get(connection_id) else {
            if self
                .connection_application_crypto_required
                .read()
                .ok()
                .is_some_and(|required| required.contains(connection_id))
            {
                anyhow::bail!(
                    "application crypto required for connection {connection_id} but no key is installed"
                );
            }
            return Ok(payload.to_vec());
        };
        if application_crypto::is_encrypted_payload(payload) {
            return Ok(payload.to_vec());
        }
        let sequence = self
            .connection_application_crypto_outbound_sequences
            .write()
            .ok()
            .and_then(|mut sequences| {
                let entry = sequences.entry(connection_id.to_string()).or_insert(0);
                let current = *entry;
                *entry = current.saturating_add(1);
                Some(current)
            })
            .unwrap_or(0);
        let mut nonce = [0u8; application_crypto::APPLICATION_NONCE_BYTES];
        getrandom::getrandom(&mut nonce)
            .map_err(|_| anyhow::anyhow!("application crypto random nonce failed"))?;
        application_crypto::protect_payload_with_nonce_and_sequence(
            key, type_id, payload, &nonce, sequence,
        )
        .map_err(|error| anyhow::anyhow!("application crypto protect failed: {:?}", error))
    }

    #[cfg_attr(
        not(any(feature = "transport-webrtc", feature = "transport-moq")),
        allow(dead_code)
    )]
    pub(crate) fn open_inbound_application_payload(
        &self,
        connection_id: &str,
        payload: &[u8],
    ) -> anyhow::Result<Vec<u8>> {
        self.open_inbound_application_payload_with_type(connection_id, 0, payload)
    }

    pub(crate) fn open_inbound_application_payload_with_type(
        &self,
        connection_id: &str,
        expected_type_id: u8,
        payload: &[u8],
    ) -> anyhow::Result<Vec<u8>> {
        let Some(key) = self.application_crypto_key_for_connection(Some(connection_id)) else {
            if self.connection_requires_application_crypto(connection_id) {
                anyhow::bail!(
                    "application crypto required for connection {connection_id} but no key is installed"
                );
            }
            return Ok(payload.to_vec());
        };

        application_crypto::open_payload(&key, expected_type_id, payload, true)
            .map_err(|error| anyhow::anyhow!("application crypto open failed: {:?}", error))
    }

    /// SDK-owned default messages never depend on an embedding application's
    /// stream reader. Called only by the sole native ingress router, after
    /// admission. Unknown channels retain their original ciphertext prefix.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn route_admitted_peer_message(
        &self,
        endpoint_id: iroh::EndpointId,
        transport_stable_id: u64,
        send: SendStream,
        mut recv: RecvStream,
        mut wire_prefix: Vec<u8>,
    ) -> anyhow::Result<crate::client::BiStreamDisposition> {
        use crate::stream_metadata::{decode_prefix, DecodeDecision, DEFAULT_PEER_CHANNEL_ID};
        let remote = endpoint_id.to_string();
        let local = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("missing local identity"))?;
        let id = Self::deterministic_connection_id(&local, &remote);
        let key = self.application_crypto_key_for_connection(Some(&id));
        let mut decoder = key
            .map(|key| application_crypto::CryptoFrameDecoder::new(&key))
            .transpose()
            .map_err(|error| anyhow::anyhow!("peer stream decoder: {error:?}"))?;
        let mut plaintext = Vec::new();
        let mut decoded_wire = 0;
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        // Same per-frame ceiling as the existing crypto wrapper, plus the
        // framing prefix. The existing ingress semaphore bounds concurrency.
        let inspection_limit = application_crypto::DEFAULT_MAX_RAW_FRAME_BYTES + 4;
        let envelope_len = loop {
            if let Some(decoder) = decoder.as_mut() {
                for frame in decoder
                    .push(&wire_prefix[decoded_wire..])
                    .map_err(|error| anyhow::anyhow!("peer stream prefix: {error:?}"))?
                {
                    plaintext.extend_from_slice(&frame);
                }
            } else {
                plaintext.extend_from_slice(&wire_prefix[decoded_wire..]);
            }
            decoded_wire = wire_prefix.len();
            match decode_prefix(&plaintext) {
                DecodeDecision::Decoded(prefix)
                    if prefix.channel.channel_id == DEFAULT_PEER_CHANNEL_ID =>
                {
                    break prefix.consumed_len;
                }
                DecodeDecision::NeedMore(_) => {}
                _ => {
                    return Ok(crate::client::BiStreamDisposition::Forward {
                        send,
                        recv,
                        recv_prefix: wire_prefix,
                    })
                }
            }
            anyhow::ensure!(
                wire_prefix.len() < inspection_limit,
                "peer stream prefix exceeds inspection bound"
            );
            let mut chunk = vec![0; (inspection_limit - wire_prefix.len()).min(4096)];
            let read = tokio::time::timeout_at(deadline, recv.read(&mut chunk))
                .await??
                .unwrap_or(0);
            if read == 0 {
                // Preserve incomplete/unclaimed prefixes for the host's own
                // protocol classifier; never reinterpret them as a message.
                return Ok(crate::client::BiStreamDisposition::Forward {
                    send,
                    recv,
                    recv_prefix: wire_prefix,
                });
            }
            wire_prefix.extend_from_slice(&chunk[..read]);
        };
        anyhow::ensure!(
            self.application_crypto_key_for_connection(Some(&id)) == key,
            "peer stream security epoch changed"
        );
        let (send, mut recv) = match key {
            Some(_) => {
                self.wrap_bi_with_prefix(
                    &endpoint_id,
                    transport_stable_id,
                    send,
                    recv,
                    &wire_prefix,
                )
                .await?
            }
            None => (
                PeerSendStream::plain(send),
                PeerRecvStream::plain(recv).with_plaintext_prefix(wire_prefix),
            ),
        };
        async fn read_exact(recv: &mut PeerRecvStream, bytes: &mut [u8]) -> anyhow::Result<bool> {
            let mut offset = 0;
            while offset < bytes.len() {
                let read = recv.read(&mut bytes[offset..]).await?;
                if read == 0 {
                    anyhow::ensure!(offset == 0, "truncated peer message frame");
                    return Ok(false);
                }
                offset += read;
            }
            Ok(true)
        }
        let mut envelope = vec![0; envelope_len];
        anyhow::ensure!(
            tokio::time::timeout_at(deadline, read_exact(&mut recv, &mut envelope)).await??,
            "missing peer envelope"
        );
        loop {
            let frame_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
            let mut length = [0; 4];
            if !tokio::time::timeout_at(frame_deadline, read_exact(&mut recv, &mut length))
                .await??
            {
                break;
            }
            let length = u32::from_be_bytes(length) as usize;
            // Payload encryption adds nonce/version/tag framing; decoded
            // application bytes are checked again by open_inbound_frame.
            anyhow::ensure!(
                length > 0 && length <= crate::native_protocol::MAX_PEER_MESSAGE_BYTES + 128,
                "invalid peer message frame length {length}"
            );
            let mut frame = vec![0; length];
            anyhow::ensure!(
                tokio::time::timeout_at(frame_deadline, read_exact(&mut recv, &mut frame))
                    .await??,
                "missing peer message body"
            );
            anyhow::ensure!(
                self.application_crypto_key_for_connection(Some(&id)) == key,
                "peer message security epoch changed"
            );
            anyhow::ensure!(
                self.open_inbound_frame(
                    &id,
                    Some(&remote),
                    "iroh",
                    Some(transport_stable_id),
                    &frame
                )
                .await?,
                "invalid peer message body"
            );
        }
        let _ = send.finish();
        Ok(crate::client::BiStreamDisposition::Consumed)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub async fn open_inbound_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        transport: &str,
        expected_transport_stable_id: Option<u64>,
        frame: &[u8],
    ) -> anyhow::Result<bool> {
        let Some((&type_id, protected_payload)) = frame.split_first() else {
            return Ok(false);
        };
        if type_id != 0 {
            return Ok(false);
        }
        if !application_crypto::is_encrypted_payload(protected_payload)
            && self.connection_requires_application_crypto(connection_id)
        {
            return Ok(false);
        }

        let generation = self
            .current_native_peer_data_generation(connection_id, expected_transport_stable_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("native peer-data route generation is stale"))?;
        if self.session_registry_active() || self.offline_admission_is_required(connection_id) {
            anyhow::ensure!(
                self.native_application_stream_admitted_for_transport(
                    connection_id,
                    generation.transport_stable_id
                ),
                "peer message transport is not admitted"
            );
        }
        let payload = self.open_inbound_application_payload(connection_id, protected_payload)?;
        anyhow::ensure!(
            payload.len() <= crate::native_protocol::MAX_PEER_MESSAGE_BYTES,
            "peer message exceeds payload limit"
        );
        self.emit_native_peer_data(crate::client::NativePeerDataEvent {
            connection_id: connection_id.to_string(),
            remote_node_id: remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned),
            transport: transport.to_string(),
            transport_stable_id: generation.transport_stable_id,
            transport_generation: generation.transport_generation,
            route_generation: generation.route_generation,
            payload,
        });
        Ok(true)
    }

    pub(crate) fn connection_ids_requiring_application_crypto(
        &self,
        connection_ids: &[String],
    ) -> bool {
        let Ok(_state) = self.connection_application_crypto_state.read() else {
            return false;
        };
        self.connection_application_crypto_required
            .read()
            .ok()
            .map(|required| {
                connection_ids
                    .iter()
                    .any(|connection_id| required.contains(connection_id))
            })
            .unwrap_or(false)
    }
}

fn enforce_application_crypto_requirement(
    required: bool,
    key: Option<[u8; APPLICATION_KEY_BYTES]>,
    operation: &str,
) -> anyhow::Result<Option<[u8; APPLICATION_KEY_BYTES]>> {
    if required && key.is_none() {
        anyhow::bail!(
            "application crypto is required for {operation}, but the current connection has no installed key"
        );
    }
    Ok(key)
}

#[cfg(test)]
mod security_tests {
    use super::enforce_application_crypto_requirement;
    use crate::application_crypto::APPLICATION_KEY_BYTES;

    #[tokio::test]
    async fn scoped_application_stream_access_preserves_sibling_withdrawal_and_fences_selected_authority(
    ) {
        use crate::{
            application_crypto_streams::ApplicationStreamAdmission, client::ClientBuilder,
            session_token::AdmissionMechanism,
        };
        for change in [
            "revoke",
            "expire",
            "device",
            "reject",
            "key",
            "generation",
            "owner-drop",
        ] {
            let client = ClientBuilder::new_provider_neutral(
                "scoped-stream-access-test".into(),
                Box::new(|| None),
            )
            .build();
            let registry = client.session_token_registry.clone();
            registry.register("app-old".into(), "ticket:app", 16);
            registry
                .validate_and_consume_for_connection("app-old", Some("peer"))
                .unwrap();
            client.set_connection_application_crypto_key("peer", [7; APPLICATION_KEY_BYTES]);
            let access = client
                .application_stream_access_for_admission(
                    "peer",
                    [7; APPLICATION_KEY_BYTES],
                    ApplicationStreamAdmission::Scope("ticket:app".into()),
                )
                .unwrap();
            access.check().unwrap();
            assert!(client
                .application_stream_access_for_admission(
                    "peer",
                    [7; APPLICATION_KEY_BYTES],
                    ApplicationStreamAdmission::Scope("ticket:missing".into())
                )
                .is_err());
            registry.register("vpn".into(), "ticket:vpn", 16);
            registry
                .validate_and_consume_for_connection("vpn", Some("peer"))
                .unwrap();
            access
                .check()
                .expect("adding a sibling scope preserves selected app permission");
            registry.revoke("vpn");
            access
                .check()
                .expect("withdrawing a sibling scope preserves selected app permission");
            registry.register("app-renewed".into(), "ticket:app", 16);
            registry
                .validate_and_consume_for_connection("app-renewed", Some("peer"))
                .unwrap();
            registry.revoke("app-old");
            access
                .check()
                .expect("renewal of selected authority preserves its stream");
            registry.register("vpn-new".into(), "ticket:vpn", 16);
            registry
                .validate_and_consume_for_connection("vpn-new", Some("peer"))
                .unwrap();
            access
                .check()
                .expect("adding a sibling scope preserves selected app permission");
            match change {
                "revoke" => {
                    registry.revoke("app-renewed");
                }
                "expire" => registry.register_with_expiry_ms(
                    "app-renewed".into(),
                    "ticket:app",
                    16,
                    Some(0),
                ),
                "device" => registry.mark_accepted(
                    "peer",
                    AdmissionMechanism::SessionToken,
                    Some("ticket:app".into()),
                    Some("changed-authority".into()),
                ),
                "reject" => registry.mark_rejected("peer", "explicit rejection"),
                "key" => {
                    client.set_connection_application_crypto_key("peer", [8; APPLICATION_KEY_BYTES])
                }
                "generation" => {
                    client
                        .connection_application_crypto_confirmed
                        .write()
                        .unwrap()
                        .insert(
                            "peer".into(),
                            super::ConnectionApplicationCryptoConfirmation {
                                transport_stable_id: 2,
                                confirmed: true,
                            },
                        );
                }
                "owner-drop" => drop(client),
                _ => unreachable!(),
            }
            assert!(
                access.check().is_err(),
                "selected authority must deny after {change}"
            );
            registry.register("app-fresh".into(), "ticket:app", 16);
            let fresh = registry.validate_and_consume_for_connection("app-fresh", Some("peer"));
            if change == "reject" {
                assert!(
                    fresh.is_err(),
                    "explicit rejection cannot fall back to a retained grant"
                );
            } else {
                fresh.unwrap();
            }
            assert!(
                access.check().is_err(),
                "a denied scoped stream must stay denied after {change}"
            );
        }
    }

    #[tokio::test]
    async fn application_stream_access_preserves_renewal_but_fences_retirement() {
        use crate::client::ClientBuilder;
        for change in [
            "key",
            "generation",
            "scope",
            "revoke",
            "forget",
            "owner-drop",
        ] {
            let client =
                ClientBuilder::new_provider_neutral("stream-access-test".into(), Box::new(|| None))
                    .build();
            let registry = &client.session_token_registry;
            registry.register("old".into(), "room:test", 16);
            registry
                .validate_and_consume_for_connection("old", Some("peer"))
                .unwrap();
            client.set_connection_application_crypto_key("peer", [7; APPLICATION_KEY_BYTES]);
            let access = client
                .application_stream_access("peer", [7; APPLICATION_KEY_BYTES])
                .unwrap();
            access.check().unwrap();
            registry.register("renewed".into(), "room:test", 16);
            registry
                .validate_and_consume_for_connection("renewed", Some("peer"))
                .unwrap();
            registry.revoke("old");
            access
                .check()
                .expect("same-authorization renewal preserves an open stream");
            match change {
                "key" => {
                    client.set_connection_application_crypto_key("peer", [8; APPLICATION_KEY_BYTES])
                }
                "generation" => {
                    client
                        .connection_application_crypto_confirmed
                        .write()
                        .unwrap()
                        .insert(
                            "peer".into(),
                            super::ConnectionApplicationCryptoConfirmation {
                                transport_stable_id: 2,
                                confirmed: true,
                            },
                        );
                }
                "scope" => registry.mark_accepted(
                    "peer",
                    crate::session_token::AdmissionMechanism::SessionToken,
                    Some("room:other".into()),
                    None,
                ),
                "revoke" => {
                    registry.revoke("renewed");
                }
                "forget" => registry.forget_connection("peer"),
                "owner-drop" => drop(client),
                _ => unreachable!(),
            }
            assert_eq!(
                access.check().expect_err(change).kind(),
                std::io::ErrorKind::PermissionDenied
            );
        }
    }

    #[test]
    fn required_application_streams_never_downgrade_to_plaintext() {
        let error = enforce_application_crypto_requirement(true, None, "test product stream")
            .expect_err("required crypto without a key must fail closed");
        assert!(error.to_string().contains("no installed key"));

        let key = [7_u8; APPLICATION_KEY_BYTES];
        assert_eq!(
            enforce_application_crypto_requirement(true, Some(key), "test product stream")
                .expect("installed key should satisfy the requirement"),
            Some(key),
        );
        assert_eq!(
            enforce_application_crypto_requirement(false, None, "manual low-level stream")
                .expect("manual low-level plaintext remains an explicit opt-in"),
            None,
        );
    }
}

pub(crate) fn new_connection_application_crypto_key_map(
) -> Arc<StdRwLock<HashMap<String, [u8; APPLICATION_KEY_BYTES]>>> {
    Arc::new(StdRwLock::new(HashMap::new()))
}

pub(crate) fn new_connection_application_crypto_required_set() -> Arc<StdRwLock<HashSet<String>>> {
    Arc::new(StdRwLock::new(HashSet::new()))
}

pub(crate) fn new_connection_application_crypto_confirmed_map(
) -> Arc<StdRwLock<HashMap<String, ConnectionApplicationCryptoConfirmation>>> {
    Arc::new(StdRwLock::new(HashMap::new()))
}

pub(crate) fn new_connection_application_crypto_outbound_sequences(
) -> Arc<StdRwLock<HashMap<String, u64>>> {
    Arc::new(StdRwLock::new(HashMap::new()))
}

pub(crate) fn new_connection_application_key_agreement_map(
) -> Arc<StdRwLock<HashMap<String, crate::key_agreement::KeyAgreement>>> {
    Arc::new(StdRwLock::new(HashMap::new()))
}