fix(messaging): validate MessagePort state in author transfer order

This commit is contained in:
ldm0
2026-10-20 05:21:33 +08:00
committed by Donough Liu
parent ebae4ca6ca
commit 8433ac01e2
8 changed files with 471 additions and 179 deletions
@@ -287,7 +287,7 @@ pub(crate) fn message_port_id_from_object<'s>(
scope: &mut v8::PinScope<'s, '_>,
port: v8::Local<'s, v8::Object>,
) -> Option<MessagePortId> {
let value = get_private_value(scope, port, MESSAGE_PORT_ID_SLOT)?;
let value = message_port_slot_value(scope, port, MESSAGE_PORT_ID_SLOT)?;
if let Ok(big) = v8::Local::<v8::BigInt>::try_from(value) {
let (n, lossless) = big.u64_value();
return lossless.then_some(n);
@@ -329,7 +329,7 @@ pub(crate) fn detach_transferred_message_port<'s>(
return;
};
forget_message_port_wrapper(scope, port_id);
set_private_value(
set_message_port_slot_value(
scope,
port,
MESSAGE_PORT_ID_SLOT,
@@ -349,7 +349,7 @@ pub(in crate::context_bootstrap) fn close_message_port_object<'s>(
forget_message_port_wrapper(scope, port_id);
}
}
set_private_value(
set_message_port_slot_value(
scope,
port,
MESSAGE_PORT_ID_SLOT,
@@ -590,6 +590,7 @@ fn message_port_slot_value<'s>(
port: v8::Local<'s, v8::Object>,
slot: &str,
) -> Option<v8::Local<'s, v8::Value>> {
let port = moli_webapi_declare::web_api_object_target(scope, port)?;
get_private_value(scope, port, slot)
}
@@ -599,6 +600,9 @@ fn set_message_port_slot_value<'s>(
slot: &str,
value: v8::Local<'s, v8::Value>,
) {
let Some(port) = moli_webapi_declare::web_api_object_target(scope, port) else {
return;
};
set_private_value(scope, port, slot, value);
}
@@ -4,8 +4,8 @@ use crate::{
message_port_id_from_object,
},
structured_clone::{
RuntimeMessageAgentCluster, V8StructuredClonePayload, deserialize_from_wire,
deserialize_message_event_from_wire, serialize_for_wire_for_runtime,
RuntimeMessageAgentCluster, V8StructuredClonePayload, WireTransfer, WireTransfers,
deserialize_from_wire, deserialize_message_event_from_wire, serialize_for_wire_for_runtime,
serialize_for_wire_for_runtime_message, serialize_for_wire_for_runtime_with_transfers,
serialize_for_wire_for_storage,
},
@@ -15,28 +15,23 @@ use crate::{
};
pub(in crate::context_bootstrap) struct PostMessageTransferList<'s> {
array_buffers: Vec<v8::Local<'s, v8::ArrayBuffer>>,
message_ports: Vec<v8::Local<'s, v8::Object>>,
readable_streams: Vec<v8::Local<'s, v8::Object>>,
writable_streams: Vec<v8::Local<'s, v8::Object>>,
transform_streams: Vec<v8::Local<'s, v8::Object>>,
entries: Vec<WireTransfer<'s>>,
}
impl<'s> PostMessageTransferList<'s> {
fn wire_transfers(&self) -> WireTransfers<'_, 's> {
WireTransfers {
entries: &self.entries,
}
}
pub(in crate::context_bootstrap) fn serialize(
self,
scope: &mut v8::PinScope<'s, '_>,
value: v8::Local<'s, v8::Value>,
) -> Option<V8StructuredClonePayload> {
let mut payload = serialize_for_wire_for_runtime_message(
scope,
value,
&self.array_buffers,
&self.message_ports,
&self.readable_streams,
&self.writable_streams,
&self.transform_streams,
)?;
let mut payload =
serialize_for_wire_for_runtime_message(scope, value, &self.wire_transfers())?;
attach_runtime_message_source(&mut payload, RuntimeMessageSourceSecurity::current(scope));
Some(payload)
}
@@ -181,26 +176,15 @@ fn structured_serialize_value_for_window_post_message_transfers<'s>(
transfers: PostMessageTransferList<'s>,
source_security: RuntimeMessageSourceSecurity,
) -> Option<V8StructuredClonePayload> {
let mut payload = serialize_for_wire_for_runtime_message(
scope,
value,
&transfers.array_buffers,
&transfers.message_ports,
&transfers.readable_streams,
&transfers.writable_streams,
&transfers.transform_streams,
)?;
let mut payload =
serialize_for_wire_for_runtime_message(scope, value, &transfers.wire_transfers())?;
attach_runtime_message_source(&mut payload, source_security);
Some(payload)
}
fn empty_post_message_transfer_list<'s>() -> PostMessageTransferList<'s> {
PostMessageTransferList {
array_buffers: Vec::new(),
message_ports: Vec::new(),
readable_streams: Vec::new(),
writable_streams: Vec::new(),
transform_streams: Vec::new(),
entries: Vec::new(),
}
}
@@ -321,15 +305,7 @@ pub(crate) fn structured_serialize_value_with_options<'s>(
options: v8::Local<'s, v8::Value>,
) -> Option<V8StructuredClonePayload> {
let transfers = parse_structured_clone_options_transfer_list(scope, options)?;
serialize_for_wire_for_runtime_with_transfers(
scope,
value,
&transfers.array_buffers,
&transfers.message_ports,
&transfers.readable_streams,
&transfers.writable_streams,
&transfers.transform_streams,
)
serialize_for_wire_for_runtime_with_transfers(scope, value, &transfers.wire_transfers())
}
pub(crate) fn structured_clone_value_for_storage<'s>(
@@ -423,11 +399,7 @@ fn parse_transfer_values<'s>(
operation: TransferListOperation<'_>,
source_port_id: Option<MessagePortId>,
) -> Option<PostMessageTransferList<'s>> {
let mut array_buffers = Vec::with_capacity(sequence.len());
let mut message_ports = Vec::new();
let mut readable_streams = Vec::new();
let mut writable_streams = Vec::new();
let mut transform_streams = Vec::new();
let mut entries = Vec::with_capacity(sequence.len());
let mut seen: Vec<v8::Local<'s, v8::Value>> = Vec::new();
for candidate in sequence {
if candidate.is_null_or_undefined() {
@@ -445,14 +417,7 @@ fn parse_transfer_values<'s>(
return None;
}
if let Ok(buffer) = v8::Local::<v8::ArrayBuffer>::try_from(candidate) {
if buffer.was_detached() {
operation.throw_data_clone_error(
scope,
"transfer list contains a detached ArrayBuffer.",
);
return None;
}
if !buffer.is_detachable() {
if !buffer.was_detached() && !buffer.is_detachable() {
operation.throw_type_error(
scope,
"transfer list contains a non-transferable ArrayBuffer.",
@@ -460,38 +425,51 @@ fn parse_transfer_values<'s>(
return None;
}
seen.push(buffer.into());
array_buffers.push(buffer);
entries.push(WireTransfer::ArrayBuffer(buffer));
continue;
}
if let Ok(object) = v8::Local::<v8::Object>::try_from(candidate) {
match moli_webapi_declare::web_api_object_type(scope, object).map(|kind| kind.name()) {
Some("MessagePort") => {
if let Some(port_id) = message_port_id_from_object(scope, object) {
if Some(port_id) == source_port_id {
operation.throw_data_clone_error(
scope,
"transfer list contains the source MessagePort.",
);
return None;
}
seen.push(object.into());
message_ports.push(object);
continue;
let target = moli_webapi_declare::web_api_object_target(scope, object)?;
if seen
.iter()
.any(|existing| existing.strict_equals(target.into()))
{
operation.throw_data_clone_error(
scope,
"transfer list contains duplicate transferable entries.",
);
return None;
}
if source_port_id.is_some()
&& message_port_id_from_object(scope, target) == source_port_id
{
operation.throw_data_clone_error(
scope,
"transfer list contains the source MessagePort.",
);
return None;
}
// A missing endpoint is a detached port, checked by the
// transfer steps after the message's getters have run.
seen.push(target.into());
entries.push(WireTransfer::MessagePort(target));
continue;
}
Some("ReadableStream") => {
seen.push(object.into());
readable_streams.push(object);
entries.push(WireTransfer::ReadableStream(object));
continue;
}
Some("WritableStream") => {
seen.push(object.into());
writable_streams.push(object);
entries.push(WireTransfer::WritableStream(object));
continue;
}
Some("TransformStream") => {
seen.push(object.into());
transform_streams.push(object);
entries.push(WireTransfer::TransformStream(object));
continue;
}
_ => {}
@@ -501,13 +479,7 @@ fn parse_transfer_values<'s>(
.throw_data_clone_error(scope, "transfer list contains a non-transferable object.");
return None;
}
Some(PostMessageTransferList {
array_buffers,
message_ports,
readable_streams,
writable_streams,
transform_streams,
})
Some(PostMessageTransferList { entries })
}
fn resolve_post_message_transfer_value<'s>(
@@ -0,0 +1,158 @@
async (owner, callee, label) => {
const rows = [];
const assert = (ok, message = 'assertion failed') => { if (!ok) throw new Error(message); };
const throws = run => {
try { run(); } catch (error) {
assert(error instanceof callee.DOMException && error.name === 'DataCloneError', String(error));
return;
}
throw new Error('expected DataCloneError');
};
const next = port => new Promise(resolve => { port.onmessage = resolve; });
for (const mode of ['clone', 'port']) {
const transport = new owner.MessageChannel();
const send = (value, transfer) => mode === 'clone'
? callee.structuredClone(value, {transfer})
: Reflect.apply(callee.MessagePort.prototype.postMessage, transport.port1, [value, {transfer}]);
const check = async (name, run) => {
try { await run(); rows.push({name: `${label}: ${mode}: ${name}`, passed: true}); }
catch (error) { rows.push({name: `${label}: ${mode}: ${name}`, passed: false, error: String(error)}); }
};
const detach = (port, how) => {
if (how === 'closed') port.close();
else callee.structuredClone(port, {transfer: [port]}).close();
};
try {
await check('graph identity, port index and entanglement', async () => {
const source = new owner.MessageChannel(), unused = new owner.MessageChannel();
let copy, extras = [];
try {
const value = {first: source.port1, second: source.port1};
const buffer = new owner.ArrayBuffer(3);
if (mode === 'clone') copy = send(value, [buffer, source.port1, unused.port1]);
else {
const received = next(transport.port2);
send(value, [buffer, source.port1, unused.port1]);
const event = await received;
copy = event.data; extras = Array.from(event.ports);
assert(extras.length === 2 && extras[0] === copy.first, 'MessageEvent.ports identity');
}
assert(copy.first === copy.second && buffer.byteLength === 0);
assert(Object.getPrototypeOf(copy.first) === (mode === 'clone' ? callee : owner).MessagePort.prototype);
const received = next(copy.first);
source.port2.postMessage('live');
assert((await received).data === 'live');
throws(() => callee.structuredClone(1, {transfer: [source.port1]}));
throws(() => callee.structuredClone(1, {transfer: [unused.port1]}));
} finally {
source.port1.close(); source.port2.close(); unused.port1.close(); unused.port2.close();
copy?.first.close(); extras.forEach(port => port.close());
}
});
await check('unreferenced transfer detaches sender', () => {
const source = new owner.MessageChannel();
try {
send(7, [source.port1]);
throws(() => callee.structuredClone(1, {transfer: [source.port1]}));
} finally { source.port1.close(); source.port2.close(); }
});
await check('duplicate transfer rejects before message getters', () => {
const source = new owner.MessageChannel(); let reads = 0;
try {
throws(() => send({get value() {reads++;}}, [source.port1, source.port1]));
assert(reads === 0);
callee.structuredClone(source.port1, {transfer: [source.port1]}).close();
} finally { source.port1.close(); source.port2.close(); }
});
await check('port outside transfer list cannot be cloned', () => {
const source = new owner.MessageChannel();
try { throws(() => send(source.port1, [])); }
finally { source.port1.close(); source.port2.close(); }
});
for (const how of ['closed', 'transferred']) {
for (const referenced of [false, true]) {
await check(`${how} port defers validation, referenced=${referenced}`, () => {
const source = new owner.MessageChannel(), sentinel = new Error('getter wins');
detach(source.port1, how);
let caught, reads = 0;
try {
send({port: referenced ? source.port1 : null, get value() {reads++; throw sentinel;}}, [source.port1]);
} catch (error) { caught = error; }
finally { source.port1.close(); source.port2.close(); }
assert(caught === sentinel && reads === 1, 'getter exception identity');
});
}
await check(`${how} port rejects after successful serialization`, () => {
const source = new owner.MessageChannel(); detach(source.port1, how);
try { throws(() => send({port: source.port1}, [source.port1])); }
finally { source.port2.close(); }
});
for (const bufferFirst of [false, true]) {
await check(`${how} port preserves transfer order, buffer first=${bufferFirst}`, () => {
const source = new owner.MessageChannel(), buffer = new owner.ArrayBuffer(3);
detach(source.port1, how);
try {
throws(() => send({port: source.port1}, bufferFirst ? [buffer, source.port1] : [source.port1, buffer]));
assert(buffer.byteLength === (bufferFirst ? 0 : 3));
} finally { source.port2.close(); }
});
}
}
await check('getter closes referenced port before transfer steps', () => {
const source = new owner.MessageChannel(), buffer = new owner.ArrayBuffer(3);
try {
throws(() => send({port: source.port1, get value() {source.port1.close();}}, [source.port1, buffer]));
assert(buffer.byteLength === 3);
} finally { source.port1.close(); source.port2.close(); }
});
for (const referenced of [false, true]) {
await check(`getter transfers port before transfer steps, referenced=${referenced}`, () => {
const source = new owner.MessageChannel(), buffer = new owner.ArrayBuffer(3);
try {
throws(() => send({port: referenced ? source.port1 : null, get value() {
callee.structuredClone(source.port1, {transfer: [source.port1]}).close();
}}, [source.port1, buffer]));
assert(buffer.byteLength === 3);
} finally { source.port1.close(); source.port2.close(); }
});
}
await check('serialization failure preserves transferable resources', () => {
const source = new owner.MessageChannel(), buffer = new owner.ArrayBuffer(3);
try {
throws(() => send({port: source.port1, bad() {}}, [source.port1, buffer]));
assert(buffer.byteLength === 3);
callee.structuredClone(source.port1, {transfer: [source.port1]}).close();
} finally { source.port1.close(); source.port2.close(); }
});
await check('author proxies and forged receivers do not run traps', () => {
const source = new owner.MessageChannel(); let traps = 0;
const revoked = owner.Proxy.revocable(source.port1, {}); revoked.revoke();
try {
for (const value of [new owner.Proxy(source.port1, {get() {traps++;}}), revoked.proxy,
Object.create(source.port1), Object.create(owner.MessagePort.prototype)]) {
throws(() => send(1, [value]));
}
assert(traps === 0);
} finally { source.port1.close(); source.port2.close(); }
});
await check('native port identity ignores author metadata and constructor changes', async () => {
const source = new owner.MessageChannel(), prototype = callee.MessagePort.prototype;
const constructor = callee.MessagePort;
Object.defineProperty(source.port1, 'endpoint', {get() {throw Error('author endpoint');}});
Object.freeze(source.port1);
try {
callee.MessagePort = function() {throw Error('author constructor');};
const copy = callee.structuredClone(source.port1, {transfer: [source.port1]});
assert(Object.getPrototypeOf(copy) === prototype);
const received = next(copy);
source.port2.postMessage('still usable');
assert((await received).data === 'still usable'); copy.close();
} finally {
callee.MessagePort = constructor;
source.port1.close(); source.port2.close();
}
});
} finally { transport.port1.close(); transport.port2.close(); }
}
return rows;
}
@@ -0,0 +1,85 @@
use super::*;
use std::time::{Duration, Instant};
#[tokio::test(flavor = "current_thread")]
async fn message_port_transfer_checks_live_state_after_serialization_across_realms() {
let mut vm = new_storage_page_task_executor_test_vm("https://port-transfer.test/");
let loader = ResourceRequestClient::new(&moli_fetch::FetchConfig::default()).unwrap();
vm.eval("document.body.innerHTML='<iframe id=child></iframe>'")
.unwrap();
vm.eval(&format!(
r#"(async()=>{{
const probe={},realms=[globalThis,document.getElementById('child').contentWindow],rows=[];
for(let owner=0;owner<2;owner++)for(let callee=0;callee<2;callee++)
rows.push(...await probe(realms[owner],realms[callee],owner+'-'+callee));
globalThis.__portTransfers=rows;
}})().catch(error=>globalThis.__portTransfers=String(error));"#,
include_str!("message_port_transfer.js")
))
.unwrap();
let deadline = Instant::now() + Duration::from_secs(30);
while vm.eval("typeof __portTransfers").unwrap() == "undefined" {
vm.run_one_oldest_ready_page_task_executor_turn(&loader)
.await
.unwrap();
assert!(
Instant::now() < deadline,
"MessagePort transfer contract timed out"
);
tokio::time::sleep(Duration::from_millis(1)).await;
}
assert_eq!(
vm.eval("Array.isArray(__portTransfers)").unwrap(),
"true",
"{}",
vm.eval("JSON.stringify(__portTransfers)").unwrap()
);
assert_eq!(vm.eval("__portTransfers.length").unwrap(), "160");
assert_eq!(
vm.eval("JSON.stringify(__portTransfers.filter(row=>!row.passed))")
.unwrap(),
"[]"
);
}
#[test]
fn message_port_transfer_respects_registered_native_proxy_identity() {
let mut vm = new_storage_page_task_executor_test_vm("https://port-transfer.test/");
vm.eval("globalThis.channel=new MessageChannel();globalThis.original=channel.port1")
.unwrap();
let context_ptr = &vm.page_default_runtime.context as *const _;
vm.with_context_scope_by_ptr_and_checkpoint_for_test(context_ptr, |scope, _| {
let global = scope.get_current_context().global(scope);
let key = crate::util::v8str(scope, "original");
let target =
v8::Local::<v8::Object>::try_from(global.get(scope, key.into()).unwrap()).unwrap();
let handler = crate::util::new_null_prototype_object(scope);
let proxy = v8::Proxy::new(scope, target, handler).unwrap();
moli_webapi_declare::register_web_api_proxy(scope, proxy).unwrap();
let key = crate::util::v8str(scope, "nativeProxy");
assert_eq!(
global.create_data_property(scope, key.into(), proxy.into()),
Some(true)
);
Ok(())
})
.unwrap();
vm.eval("channel.port2.onmessage=event=>globalThis.__nativePortMessage=event.data;nativeProxy.postMessage('native proxy');").unwrap();
assert_eq!(
vm.eval_after_selected_page_tasks("__nativePortMessage")
.unwrap(),
"native proxy"
);
assert_eq!(vm.eval(r#"(() => {
for(const list of [[original,nativeProxy],[nativeProxy,original]]) {
let reads=0,error;
try{structuredClone({get value(){reads++;}},{transfer:list});}catch(caught){error=caught;}
if(!(error instanceof DOMException)||error.name!=='DataCloneError'||reads!==0)return false;
}
const copy=structuredClone({port:original},{transfer:[nativeProxy]});
nativeProxy.close();
let error;try{structuredClone(1,{transfer:[original]});}catch(caught){error=caught;}
copy.port.close();channel.port2.close();
return copy.port instanceof MessagePort&&error instanceof DOMException&&error.name==='DataCloneError';
})()"#).unwrap(),"true");
}
@@ -3,6 +3,7 @@ mod credential_container;
mod credential_json_options;
mod cross_origin_navigation;
mod current_pass_css_fonts;
mod message_port_transfer;
mod picture_in_picture_interfaces;
mod visual_viewport_receiver;
mod writable_stream_native_receivers;
+126 -102
View File
@@ -46,6 +46,7 @@ const HOST_OBJECT_TAG_QUOTA_EXCEEDED_ERROR: u32 = 8;
const HOST_OBJECT_TAG_WRITABLE_STREAM: u32 = 9;
const HOST_OBJECT_TAG_TRANSFORM_STREAM: u32 = 10;
const HOST_OBJECT_TAG_GEOMETRY: u32 = 11;
const HOST_OBJECT_TAG_TRANSFERRED_MESSAGE_PORT: u32 = 22;
const GEOMETRY_KIND_DOM_RECT_READONLY: u32 = 2;
const GEOMETRY_KIND_DOM_RECT: u32 = 3;
@@ -306,7 +307,7 @@ pub(crate) struct StructuredCloneMetadata {
}
struct WireSerializer {
allowed_message_port_ids: HashSet<MessagePortId>,
transferred_message_ports: Vec<v8::Global<v8::Object>>,
allowed_readable_streams: Vec<v8::Global<v8::Object>>,
allowed_writable_streams: Vec<v8::Global<v8::Object>>,
allowed_transform_streams: Vec<v8::Global<v8::Object>>,
@@ -359,18 +360,21 @@ impl v8::ValueSerializerImpl for WireSerializer {
) -> Option<bool> {
match moli_webapi_declare::web_api_object_type(scope, object).map(|kind| kind.name()) {
Some("MessagePort") => {
if let Some(port_id) = message_port_id_from_object(scope, object) {
if !self.allowed_message_port_ids.contains(&port_id) {
throw_data_clone_exception(
scope,
"MessagePort must be listed in the postMessage transfer list.",
);
return None;
}
serializer.write_uint32(HOST_OBJECT_TAG_MESSAGE_PORT);
serializer.write_uint64(port_id);
return Some(true);
}
let target = moli_webapi_declare::web_api_object_target(scope, object)?;
let Some(index) = self
.transferred_message_ports
.iter()
.position(|port| v8::Local::new(scope, port).strict_equals(target.into()))
else {
throw_data_clone_exception(
scope,
"MessagePort must be listed in the postMessage transfer list.",
);
return None;
};
serializer.write_uint32(HOST_OBJECT_TAG_TRANSFERRED_MESSAGE_PORT);
serializer.write_uint32(index as u32);
return Some(true);
}
Some("DOMRect" | "DOMRectReadOnly") => {
if let Some((mutable, values)) = dom_rect_clone_data(scope, object) {
@@ -566,6 +570,7 @@ impl v8::ValueSerializerImpl for WireSerializer {
}
struct WireDeserializer {
transferred_message_ports: Vec<MessagePortId>,
wasm_modules: HashMap<u32, ClonedWasmModule>,
readable_streams: HashMap<u32, ReadableStreamClonePayload>,
writable_streams: HashMap<u32, WritableStreamClonePayload>,
@@ -590,6 +595,19 @@ impl v8::ValueDeserializerImpl for WireDeserializer {
return None;
}
match tag {
HOST_OBJECT_TAG_TRANSFERRED_MESSAGE_PORT => {
let port_id = read_u32(deserializer)
.and_then(|index| self.transferred_message_ports.get(index as usize).copied());
port_id
.and_then(|id| ensure_message_port_wrapper_for_id(scope, id))
.or_else(|| {
throw_data_clone_exception(
scope,
"Failed to deserialize transferred MessagePort.",
);
None
})
}
HOST_OBJECT_TAG_MESSAGE_PORT => {
let mut port_id = 0;
if !deserializer.read_uint64(&mut port_id) {
@@ -1054,6 +1072,21 @@ fn copy_array_buffer_bytes(buffer: v8::Local<'_, v8::ArrayBuffer>) -> Vec<u8> {
unsafe { std::slice::from_raw_parts(data.as_ptr().cast::<u8>(), length) }.to_vec()
}
/// Transfer steps retain the observable order of the author's transfer list.
#[derive(Clone, Copy)]
pub(crate) enum WireTransfer<'s> {
ArrayBuffer(v8::Local<'s, v8::ArrayBuffer>),
MessagePort(v8::Local<'s, v8::Object>),
ReadableStream(v8::Local<'s, v8::Object>),
WritableStream(v8::Local<'s, v8::Object>),
TransformStream(v8::Local<'s, v8::Object>),
}
#[derive(Default)]
pub(crate) struct WireTransfers<'a, 's> {
pub entries: &'a [WireTransfer<'s>],
}
pub(crate) fn serialize_for_wire_for_runtime<'s>(
scope: &mut v8::PinScope<'s, '_>,
value: v8::Local<'s, v8::Value>,
@@ -1061,11 +1094,7 @@ pub(crate) fn serialize_for_wire_for_runtime<'s>(
serialize_for_wire_with_policy(
scope,
value,
&[],
&[],
&[],
&[],
&[],
&WireTransfers::default(),
StructuredClonePolicy::Runtime,
)
}
@@ -1073,22 +1102,9 @@ pub(crate) fn serialize_for_wire_for_runtime<'s>(
pub(crate) fn serialize_for_wire_for_runtime_with_transfers<'s>(
scope: &mut v8::PinScope<'s, '_>,
value: v8::Local<'s, v8::Value>,
array_buffer_transfers: &[v8::Local<'s, v8::ArrayBuffer>],
message_port_transfers: &[v8::Local<'s, v8::Object>],
readable_stream_transfers: &[v8::Local<'s, v8::Object>],
writable_stream_transfers: &[v8::Local<'s, v8::Object>],
transform_stream_transfers: &[v8::Local<'s, v8::Object>],
transfers: &WireTransfers<'_, 's>,
) -> Option<V8StructuredClonePayload> {
serialize_for_wire_with_policy(
scope,
value,
array_buffer_transfers,
message_port_transfers,
readable_stream_transfers,
writable_stream_transfers,
transform_stream_transfers,
StructuredClonePolicy::Runtime,
)
serialize_for_wire_with_policy(scope, value, transfers, StructuredClonePolicy::Runtime)
}
pub(crate) fn serialize_for_wire_for_storage<'s>(
@@ -1098,11 +1114,7 @@ pub(crate) fn serialize_for_wire_for_storage<'s>(
serialize_for_wire_with_policy(
scope,
value,
&[],
&[],
&[],
&[],
&[],
&WireTransfers::default(),
StructuredClonePolicy::Storage,
)
}
@@ -1110,20 +1122,12 @@ pub(crate) fn serialize_for_wire_for_storage<'s>(
pub(crate) fn serialize_for_wire_for_runtime_message<'s>(
scope: &mut v8::PinScope<'s, '_>,
value: v8::Local<'s, v8::Value>,
array_buffer_transfers: &[v8::Local<'s, v8::ArrayBuffer>],
message_port_transfers: &[v8::Local<'s, v8::Object>],
readable_stream_transfers: &[v8::Local<'s, v8::Object>],
writable_stream_transfers: &[v8::Local<'s, v8::Object>],
transform_stream_transfers: &[v8::Local<'s, v8::Object>],
transfers: &WireTransfers<'_, 's>,
) -> Option<V8StructuredClonePayload> {
serialize_for_wire_with_policy(
scope,
value,
array_buffer_transfers,
message_port_transfers,
readable_stream_transfers,
writable_stream_transfers,
transform_stream_transfers,
transfers,
StructuredClonePolicy::RuntimeMessage,
)
}
@@ -1131,25 +1135,34 @@ pub(crate) fn serialize_for_wire_for_runtime_message<'s>(
fn serialize_for_wire_with_policy<'s>(
scope: &mut v8::PinScope<'s, '_>,
value: v8::Local<'s, v8::Value>,
array_buffer_transfers: &[v8::Local<'s, v8::ArrayBuffer>],
message_port_transfers: &[v8::Local<'s, v8::Object>],
readable_stream_transfers: &[v8::Local<'s, v8::Object>],
writable_stream_transfers: &[v8::Local<'s, v8::Object>],
transform_stream_transfers: &[v8::Local<'s, v8::Object>],
transfers: &WireTransfers<'_, 's>,
policy: StructuredClonePolicy,
) -> Option<V8StructuredClonePayload> {
let mut array_buffer_transfers = Vec::new();
let mut message_port_transfers = Vec::new();
let mut readable_stream_transfers = Vec::new();
let mut writable_stream_transfers = Vec::new();
let mut transform_stream_transfers = Vec::new();
for transfer in transfers.entries {
match *transfer {
WireTransfer::ArrayBuffer(buffer) => array_buffer_transfers.push(buffer),
WireTransfer::MessagePort(port) => message_port_transfers.push(port),
WireTransfer::ReadableStream(stream) => readable_stream_transfers.push(stream),
WireTransfer::WritableStream(stream) => writable_stream_transfers.push(stream),
WireTransfer::TransformStream(stream) => transform_stream_transfers.push(stream),
}
}
let context = scope.get_current_context();
let transferred_message_ports: Vec<MessagePortId> = message_port_transfers
.iter()
.filter_map(|port| message_port_id_from_object(scope, *port))
.collect();
let wasm_modules = Rc::new(RefCell::new(ClonedWasmModuleStore::default()));
let blobs = Rc::new(RefCell::new(ClonedBlobStore::default()));
let file_system_handles = Rc::new(RefCell::new(ClonedFileSystemHandleStore::default()));
let serializer = v8::ValueSerializer::new(
scope,
Box::new(WireSerializer {
allowed_message_port_ids: transferred_message_ports.iter().copied().collect(),
transferred_message_ports: message_port_transfers
.iter()
.map(|port| v8::Global::new(scope, *port))
.collect(),
allowed_readable_streams: readable_stream_transfers
.iter()
.map(|stream| v8::Global::new(scope, *stream))
@@ -1169,7 +1182,6 @@ fn serialize_for_wire_with_policy<'s>(
}),
);
serializer.write_header();
let mut transferred_array_buffers = Vec::with_capacity(array_buffer_transfers.len());
for (index, buffer) in array_buffer_transfers.iter().enumerate() {
if buffer.get_backing_store().is_resizable_by_user_javascript() {
// V8's inline wire representation preserves a resizable backing
@@ -1179,56 +1191,67 @@ fn serialize_for_wire_with_policy<'s>(
}
let transfer_id = index as u32 + 1;
serializer.transfer_array_buffer(transfer_id, *buffer);
transferred_array_buffers.push(TransferredArrayBuffer {
transfer_id,
bytes: copy_array_buffer_bytes(*buffer),
});
}
let Some(true) = serializer.write_value(context, value) else {
return None;
};
for buffer in array_buffer_transfers {
if buffer.detach(None) != Some(true) {
throw_data_clone_exception(scope, "Failed to transfer ArrayBuffer.");
return None;
}
}
for port in message_port_transfers {
let Some(port_id) = message_port_id_from_object(scope, *port) else {
continue;
};
detach_message_port_owner_for_transfer(scope, port_id);
detach_transferred_message_port(scope, *port);
}
let mut transferred_array_buffers = Vec::with_capacity(array_buffer_transfers.len());
let mut transferred_message_ports = Vec::with_capacity(message_port_transfers.len());
let mut readable_streams = Vec::with_capacity(readable_stream_transfers.len());
for (clone_id, stream) in readable_stream_transfers.iter().enumerate() {
let prepared = prepare_readable_stream_transfer(scope, *stream)?;
let clone_id =
u32::try_from(clone_id).expect("a JavaScript transfer sequence length fits in u32");
readable_streams.push(ClonedReadableStream {
clone_id,
payload: prepared.commit(scope),
});
}
let mut writable_streams = Vec::with_capacity(writable_stream_transfers.len());
for (clone_id, stream) in writable_stream_transfers.iter().enumerate() {
let prepared = prepare_writable_stream_transfer(scope, *stream)?;
let clone_id =
u32::try_from(clone_id).expect("a JavaScript transfer sequence length fits in u32");
writable_streams.push(ClonedWritableStream {
clone_id,
payload: prepared.commit(scope),
});
}
let mut transform_streams = Vec::with_capacity(transform_stream_transfers.len());
for (clone_id, stream) in transform_stream_transfers.iter().enumerate() {
let prepared = prepare_transform_stream_transfer(scope, *stream)?;
let clone_id =
u32::try_from(clone_id).expect("a JavaScript transfer sequence length fits in u32");
transform_streams.push(ClonedTransformStream {
clone_id,
payload: prepared.commit(scope),
});
let mut array_buffer_index = 0;
for transfer in transfers.entries {
match *transfer {
WireTransfer::ArrayBuffer(buffer) => {
array_buffer_index += 1;
if buffer.was_detached() {
throw_data_clone_exception(scope, "A transferred ArrayBuffer is detached.");
return None;
}
// Getters in the message may have changed the backing bytes.
if !buffer.get_backing_store().is_resizable_by_user_javascript() {
transferred_array_buffers.push(TransferredArrayBuffer {
transfer_id: array_buffer_index,
bytes: copy_array_buffer_bytes(buffer),
});
}
if buffer.detach(None) != Some(true) {
throw_data_clone_exception(scope, "Failed to transfer ArrayBuffer.");
return None;
}
}
WireTransfer::MessagePort(port) => {
let Some(port_id) = message_port_id_from_object(scope, port) else {
throw_data_clone_exception(scope, "A transferred MessagePort is detached.");
return None;
};
detach_message_port_owner_for_transfer(scope, port_id);
detach_transferred_message_port(scope, port);
transferred_message_ports.push(port_id);
}
WireTransfer::ReadableStream(stream) => {
let prepared = prepare_readable_stream_transfer(scope, stream)?;
readable_streams.push(ClonedReadableStream {
clone_id: readable_streams.len() as u32,
payload: prepared.commit(scope),
});
}
WireTransfer::WritableStream(stream) => {
let prepared = prepare_writable_stream_transfer(scope, stream)?;
writable_streams.push(ClonedWritableStream {
clone_id: writable_streams.len() as u32,
payload: prepared.commit(scope),
});
}
WireTransfer::TransformStream(stream) => {
let prepared = prepare_transform_stream_transfer(scope, stream)?;
transform_streams.push(ClonedTransformStream {
clone_id: transform_streams.len() as u32,
payload: prepared.commit(scope),
});
}
}
}
let wasm_modules = wasm_modules.borrow().modules.clone();
let blobs = blobs.borrow().blobs.clone();
@@ -1331,6 +1354,7 @@ fn deserialize_from_wire_impl<'s>(
let deserializer = v8::ValueDeserializer::new(
scope,
Box::new(WireDeserializer {
transferred_message_ports: payload.base.transferred_message_ports.clone(),
wasm_modules,
readable_streams,
writable_streams,
@@ -0,0 +1,47 @@
use super::*;
#[tokio::test]
async fn worker_message_port_transfer_validates_after_serialization_across_global_kinds() {
ensure_v8();
let storage_key = moli_storage_key::MoliStorageKey::new(
"https://port-transfer.test".to_owned(),
"https://port-transfer.test".to_owned(),
None,
moli_storage_key::StoragePartitionRelation::FirstParty,
);
for kind in [
super::super::WorkerGlobalKind::Dedicated {
name: String::new(),
},
super::super::WorkerGlobalKind::Shared {
name: "port-transfer".to_owned(),
storage_key,
},
super::super::WorkerGlobalKind::Service {
registration_id: ServiceWorkerRegistrationId::from_u64_for_test(1),
version_id: ServiceWorkerVersionId::from_u64_for_test(1),
scope_url: url::Url::parse("https://port-transfer.test/").unwrap(),
},
] {
let source = format!(
"const probe={};probe(globalThis,globalThis,'worker').then(rows=>console.log(JSON.stringify({{total:rows.length,failures:rows.filter(row=>!row.passed)}})));",
include_str!("../../../script_vm/tests/message_port_transfer.js")
);
let mut handle = spawn_test_worker_with_options(
WorkerSpawnOptions::new(source, "https://port-transfer.test/worker.js".to_owned())
.with_global_kind(kind),
);
let message = timeout(Duration::from_secs(30), handle.recv())
.await
.unwrap()
.unwrap();
let WorkerToParentMessage::Console(console) = message else {
panic!("expected MessagePort checks, got {message:?}");
};
let result: serde_json::Value =
serde_json::from_str(console.message.strip_prefix("log: ").unwrap()).unwrap();
assert_eq!(result["total"], 40, "{result}");
assert_eq!(result["failures"], serde_json::json!([]), "{result}");
handle.terminate_and_join();
}
}
@@ -1531,6 +1531,7 @@ fn service_worker_storage_apis_use_explicit_registration_storage_key() {
mod abort_signal_statics;
mod lazy_storage;
mod lifecycle;
mod message_port_transfer;
mod modules;
mod strict_script_mime;