openrtc-tauri-plugin 2.9.8

use super::*;
use iroh::{endpoint::presets::Minimal, Endpoint, RelayMode};
use std::time::Duration;

// Exercise the public host handoff with real encrypted loopback QUIC streams.
// Only the Tauri UI runtime is mocked. Admission and an installed WebView are
// separate boundaries: this fixture never claims its synthetic state admitted
// these connections, and checks that projection retains the false verdict.
#[tokio::test]
async fn custom_stream_handoff_preserves_capability_ownership_bytes_and_ack() {
    tokio::time::timeout(Duration::from_secs(20), async {
        const ALPN: &[u8] = b"openrtc/test/native-stream-projection";
        const KEY: [u8; 32] = [42; 32];
        let bind = || {
            Endpoint::builder(Minimal)
                .relay_mode(RelayMode::Disabled)
                .alpns(vec![ALPN.to_vec()])
                .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
                .unwrap()
                .bind()
        };
        let sender = bind().await.unwrap();
        let receiver = bind().await.unwrap();
        let (outgoing, incoming) = tokio::join!(sender.connect(receiver.addr(), ALPN), async {
            receiver.accept().await.unwrap().await
        });
        let outgoing = outgoing.unwrap();
        let incoming = incoming.unwrap();
        let app = tauri::test::mock_app();
        assert!(app.manage(OpenRtcTauriState::new(test_config())));
        let state = app.state::<OpenRtcTauriState>();
        state
            .capability_registry
            .lock()
            .await
            .register(
                "devices:projection-user".into(),
                "devices".into(),
                "projection-user".into(),
                false,
            )
            .unwrap();
        state
            .capability_registry
            .lock()
            .await
            .register(
                "ticket:unadmitted".into(),
                "ticket".into(),
                "unadmitted".into(),
                false,
            )
            .unwrap();
        let events = Arc::new(StdMutex::new(Vec::<serde_json::Value>::new()));
        let payload: Vec<u8> = (0..65_537).map(|i| (i % 251) as u8).collect();
        for (channel_id, capability, subscribe, projected) in [
            (
                "assistant_message_speech",
                Some("devices:projection-user"),
                true,
                true,
            ),
            (
                "openrtc.file-transfer.v1",
                Some("devices:projection-user"),
                true,
                true,
            ),
            ("host/private", None, true, false),
            (
                "assistant_message_speech",
                Some("devices:unknown"),
                true,
                false,
            ),
            (
                "assistant_message_speech",
                Some("ticket:unadmitted"),
                true,
                false,
            ),
            ("assistant_message_speech", Some(""), true, false),
            (
                "assistant_message_speech",
                Some("devices:projection-user"),
                false,
                false,
            ),
        ] {
            events.lock().unwrap().clear();
            *state.native_projection_subscription.lock().await = if subscribe {
                let sink = events.clone();
                Some(NativeProjectionSubscription {
                    webview_label: "main".into(),
                    request_id: "projection-run".into(),
                    channel: Channel::new(move |body| {
                        if let tauri::ipc::InvokeResponseBody::Json(json) = body {
                            sink.lock()
                                .unwrap()
                                .push(serde_json::from_str(&json).unwrap());
                        }
                        Ok(())
                    }),
                })
            } else {
                None
            };
            let metadata = capability.map(|owner| {
                serde_json::Map::from_iter([("openrtcCapability".into(), serde_json::json!(owner))])
            });
            let envelope =
                openrtc::stream_metadata::encode_envelope(channel_id, metadata.as_ref()).unwrap();
            let (send, recv) = outgoing.open_bi().await.unwrap();
            let mut send = PeerSendStream::encrypted(send, KEY);
            send.write_all(&envelope).await.unwrap();
            send.write_all(&payload).await.unwrap();
            send.finish().unwrap();
            let (reply, incoming_recv) = incoming.accept_bi().await.unwrap();
            let result = handoff_projected_peer_bi_stream(
                app.handle().clone(),
                None,
                sender.id().to_string(),
                incoming.stable_id() as u64,
                PeerSendStream::encrypted(reply, KEY),
                PeerRecvStream::encrypted(incoming_recv, KEY).unwrap(),
            )
            .await
            .unwrap();
            let (mut reply, mut body) = if projected {
                assert!(
                    matches!(result, ProjectedPeerBiHandoff::Handled),
                    "scoped custom channel {channel_id} was not projected"
                );
                let event = events.lock().unwrap()[0].clone();
                assert_eq!(events.lock().unwrap().len(), 1, "one intake owner");
                assert_eq!(event["kind"], "stream");
                assert_eq!(
                    event["capabilityKeys"],
                    serde_json::json!(["devices:projection-user"])
                );
                assert_eq!(event["payload"]["channel"]["channelId"], channel_id);
                assert_eq!(
                    event["payload"]["applicationAuthorized"], false,
                    "handoff must not invent admission"
                );
                let id = event["payload"]["streamId"].as_str().unwrap();
                let mut handle = state.peer_bi_streams.lock().await.remove(id).unwrap();
                let send = handle.send.lock().await.take().unwrap();
                (send, handle.recv.take().unwrap())
            } else {
                assert!(events.lock().unwrap().is_empty());
                assert!(state.peer_bi_streams.lock().await.is_empty());
                match result {
                    ProjectedPeerBiHandoff::Unhandled { send, recv } => (send, recv),
                    ProjectedPeerBiHandoff::Handled => panic!("unowned channel was consumed"),
                }
            };
            let mut received = Vec::new();
            let mut buffer = [0u8; 4096];
            loop {
                let read = body.read(&mut buffer).await.unwrap();
                if read == 0 {
                    break;
                }
                received.extend_from_slice(&buffer[..read]);
            }
            if projected {
                assert_eq!(received, payload);
            } else {
                let mut original = envelope;
                original.extend_from_slice(&payload);
                assert_eq!(
                    received, original,
                    "unhandled bytes must be restored exactly"
                );
            }
            reply.write_all(b"verified").await.unwrap();
            reply.finish().unwrap();
            let mut recv = PeerRecvStream::encrypted(recv, KEY).unwrap();
            let mut ack = [0u8; 8];
            read_peer_exact(&mut recv, &mut ack).await.unwrap();
            assert_eq!(
                &ack, b"verified",
                "sender FIN must leave the reply half alive"
            );
        }
        outgoing.close(0u32.into(), b"test complete");
        incoming.close(0u32.into(), b"test complete");
        sender.close().await;
        receiver.close().await;
    })
    .await
    .expect("native stream handoff timed out");
}