diff --git a/toolkit/components/extensions/ConduitsParent.sys.mjs b/toolkit/components/extensions/ConduitsParent.sys.mjs
index 3ea59d6affe..aa42ed15070 100644
--- a/toolkit/components/extensions/ConduitsParent.sys.mjs
+++ b/toolkit/components/extensions/ConduitsParent.sys.mjs
@@ -36,6 +36,7 @@
  * @property {number} [portId]
  * @property {boolean} [native]
  * @property {boolean} [source]
+ * @property {string} [reportOnOpened]
  * @property {string} [reportOnClosed]
  *
  * Lists of recvX, sendX, queryX and castX methods this subject will use.
@@ -66,8 +67,6 @@ import { BaseConduit } from "resource://gre/modules/ConduitsChild.sys.mjs";
 import { ExtensionUtils } from "resource://gre/modules/ExtensionUtils.sys.mjs";
 import { WebNavigationFrames } from "resource://gre/modules/WebNavigationFrames.sys.mjs";
 
-const { DefaultWeakMap, ExtensionError } = ExtensionUtils;
-
 const BATCH_TIMEOUT_MS = 250;
 const ADDON_ENV = new Set(["addon_child", "devtools_child"]);
 
@@ -85,7 +84,10 @@ const Hub = {
   byMethod: new Map(),
 
   /** @type {WeakMap<ConduitsParent, Set<ConduitAddress>>} Conduits by actor. */
-  byActor: new DefaultWeakMap(() => new Set()),
+  byActor: new ExtensionUtils.DefaultWeakMap(() => new Set()),
+
+  /** @type {Map<string, BroadcastConduit>} */
+  reportOnOpened: new Map(),
 
   /** @type {Map<string, BroadcastConduit>} */
   reportOnClosed: new Map(),
@@ -218,6 +220,19 @@ const Hub = {
    */
   recvConduitOpened(address, actor) {
     this.fillInAddress(address, actor);
+
+    for (let [key, conduit] of this.reportOnOpened.entries()) {
+      if (address[key] != null) {
+        // A subject may veto by throwing, leaving the conduit unregistered.
+        try {
+          conduit.subject.recvConduitOpened(address);
+        } catch (e) {
+          Cu.reportError(e);
+          return;
+        }
+      }
+    }
+
     this.remotes.set(address.id, address);
     this.byActor.get(actor).add(address);
   },
@@ -274,6 +289,12 @@ export class BroadcastConduit extends BaseConduit {
       this[`cast${name}`] = this._cast.bind(this, name);
     }
 
+    // Wants to authorize conduits with a specific attribute as they open.
+    // `subject.recvConduitOpened(address)` throws to reject.
+    if (address.reportOnOpened) {
+      Hub.reportOnOpened.set(address.reportOnOpened, this);
+    }
+
     // Wants to know when conduits with a specific attribute are closed.
     // `subject.recvConduitClosed(address)` method will be called.
     if (address.reportOnClosed) {
@@ -312,14 +333,13 @@ export class BroadcastConduit extends BaseConduit {
 
   /**
    * Broadcasts a method call to all conduits of kind that satisfy filtering by
-   * kind-specific properties from arg. If arg.query is true, these broadcasts
-   * are all queries and this returns an array of response promises. Otherwise,
-   * they are not, and undefined is returned.
+   * kind-specific properties from arg. Returns the targeted conduits, along
+   * with the response promises for a query.
    *
    * @param {string} method
    * @param {BroadcastKind} kind
    * @param {object} arg
-   * @returns {undefined | Promise<any[]> | Promise<Response>}
+   * @returns {{ targets: ConduitAddress[], promises: Promise<any>[] }}
    */
   _cast(method, kind, arg) {
     let filters = {
@@ -359,50 +379,7 @@ export class BroadcastConduit extends BaseConduit {
 
     let targets = Array.from(Hub.remotes.values()).filter(filters[kind]);
     let promises = targets.map(c => this._send(method, !!arg.query, c.id, arg));
-    if (arg.query) {
-      return arg.firstResponse
-        ? this._raceResponses(promises)
-        : Promise.allSettled(promises);
-    }
-    return undefined;
-  }
-
-  /**
-   * Custom Promise.race() function that ignores certain resolutions and errors.
-   *
-   * @typedef {{response?: any, received?: boolean}} Response
-   *
-   * @param {Promise<Response>[]} promises
-   * @returns {Promise<Response?>}
-   */
-  _raceResponses(promises) {
-    return new Promise((resolve, reject) => {
-      let result;
-      promises.map(p =>
-        p
-          .then(value => {
-            if (value.response) {
-              // We have an explicit response, resolve immediately.
-              resolve(value);
-            } else if (value.received) {
-              // Message was received, but no response.
-              // Resolve with this only if there is no other explicit response.
-              result = value;
-            }
-          })
-          .catch(err => {
-            // Forward errors that are exposed to extension, but ignore
-            // internal errors such as actor destruction and DataCloneError.
-            if (err instanceof ExtensionError || err?.mozWebExtLocation) {
-              reject(err);
-            } else {
-              Cu.reportError(err);
-            }
-          })
-      );
-      // Ensure resolving when there are no responses.
-      Promise.allSettled(promises).then(() => resolve(result));
-    });
+    return { targets, promises };
   }
 
   async close() {
diff --git a/toolkit/components/extensions/ExtensionParent.sys.mjs b/toolkit/components/extensions/ExtensionParent.sys.mjs
index 72759484121..a9569ad37f3 100644
--- a/toolkit/components/extensions/ExtensionParent.sys.mjs
+++ b/toolkit/components/extensions/ExtensionParent.sys.mjs
@@ -245,15 +245,33 @@ const ProxyMessenger = {
   /** @type {Map<number, Promise>} */
   portPromises: new Map(),
 
+  // Known endpoints of each open port.
+  /** @type {Map<number, object>} portId -> source actor. */
+  portSources: new Map(),
+  /** @type {Map<number, Set<object>>} portId -> known receiver actors. */
+  portReceivers: new Map(),
+
   init() {
     this.conduit = new lazy.BroadcastConduit(ProxyMessenger, {
       id: "ProxyMessenger",
+      reportOnOpened: "portId",
       reportOnClosed: "portId",
       recv: ["PortConnect", "PortMessage", "NativeMessage", "RuntimeMessage"],
       cast: ["PortConnect", "PortMessage", "PortDisconnect", "RuntimeMessage"],
     });
   },
 
+  recvConduitOpened({ portId, source, actor, extensionId }) {
+    if (source) {
+      if (this.portSources.has(portId)) {
+        throw new Error(`Duplicate port source for ${extensionId}`);
+      }
+      this.portSources.set(portId, actor);
+    } else if (!this.portReceivers.get(portId)?.has(actor)) {
+      throw new Error(`Unknown port receiver for ${extensionId}`);
+    }
+  },
+
   openNative(nativeApp, sender) {
     let context = ParentAPIManager.getContextById(sender.childId);
     if (context.extension.hasPermission("geckoViewAddons")) {
@@ -362,10 +380,10 @@ const ProxyMessenger = {
   },
 
   async recvRuntimeMessage(arg, { sender }) {
-    arg.firstResponse = true;
     let kind = await this.normalizeArgs(arg, sender);
     arg.query = true;
-    let result = await this.conduit.castRuntimeMessage(kind, arg);
+    let { promises } = this.conduit.castRuntimeMessage(kind, arg);
+    let result = await raceResponses(promises);
     if (!result) {
       // "throw new ExtensionError" cannot be used because then the stack of the
       // sendMessage call would not be added to the error object generated by
@@ -376,6 +394,12 @@ const ProxyMessenger = {
   },
 
   async recvPortConnect(arg, { sender }) {
+    // openConduit(source=true) precedes queryPortConnect and sets portSources.
+    if (this.portSources.get(arg.portId) !== sender.actor) {
+      Cu.reportError(`Unexpected port source for ${sender.extensionId}`);
+      throw new ExtensionError(ERROR_NO_RECEIVERS);
+    }
+
     if (arg.native) {
       /** @type {ParentPort} */
       let port = this.openNative(arg.name, sender).onConnect(arg.portId, this);
@@ -393,7 +417,12 @@ const ProxyMessenger = {
     try {
       let kind = await this.normalizeArgs(arg, sender);
       arg.query = true;
-      let all = await this.conduit.castPortConnect(kind, arg);
+      let { targets, promises } = this.conduit.castPortConnect(kind, arg);
+      // Holds actors from which we accept recvConduitOpened for this portId,
+      // before castPortConnect settles. No await between cast and assignment,
+      // or a target's recvConduitOpened could arrive first and be rejected.
+      this.portReceivers.set(arg.portId, new Set(targets.map(t => t.actor)));
+      let all = await Promise.allSettled(promises);
       resolve();
 
       // If there are no active onConnect listeners.
@@ -406,6 +435,7 @@ const ProxyMessenger = {
       throw err;
     } finally {
       this.portPromises.delete(arg.portId);
+      this.portReceivers.delete(arg.portId);
     }
   },
 
@@ -422,6 +452,10 @@ const ProxyMessenger = {
   },
 
   async recvConduitClosed(sender) {
+    if (sender.source) {
+      this.portSources.delete(sender.portId);
+    }
+
     let app = this.ports.get(sender.portId);
     if (this.ports.delete(sender.portId) && sender.native) {
       this.untrackNativeAppPort(app);
@@ -478,6 +512,46 @@ const ProxyMessenger = {
 };
 ProxyMessenger.init();
 
+/**
+ * Custom Promise.race() function that ignores certain resolutions and errors,
+ * used to resolve a runtime.sendMessage query with its first explicit response
+ * across all onMessage listeners.
+ *
+ * @typedef {{response?: any, received?: boolean}} Response
+ *
+ * @param {Promise<Response>[]} promises
+ * @returns {Promise<Response?>}
+ */
+function raceResponses(promises) {
+  return new Promise((resolve, reject) => {
+    let result;
+    promises.map(p =>
+      p
+        .then(value => {
+          if (value.response) {
+            // We have an explicit response, resolve immediately.
+            resolve(value);
+          } else if (value.received) {
+            // Message was received, but no response.
+            // Resolve with this only if there is no other explicit response.
+            result = value;
+          }
+        })
+        .catch(err => {
+          // Forward errors that are exposed to extension, but ignore
+          // internal errors such as actor destruction and DataCloneError.
+          if (err instanceof ExtensionError || err?.mozWebExtLocation) {
+            reject(err);
+          } else {
+            Cu.reportError(err);
+          }
+        })
+    );
+    // Ensure resolving when there are no responses.
+    Promise.allSettled(promises).then(() => resolve(result));
+  });
+}
+
 // Responsible for loading extension APIs into the right globals.
 GlobalManager = {
   // Map[extension ID -> Extension]. Determines which extension is
diff --git a/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js b/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js
new file mode 100644
index 00000000000..54c78d6352f
--- /dev/null
+++ b/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js
@@ -0,0 +1,110 @@
+"use strict";
+
+// A compromised child could forge a raw IPC message reusing another portId
+// to join a channel it shouldn't have access to.
+
+// The forged outbound PortMessages below are rejected in the parent,
+// which surfaces as uncaught rejections; they are expected.
+PromiseTestUtils.allowMatchingRejectionsGlobally(
+  /Unknown sender or wrong actor for recvPortMessage/
+);
+
+add_task(async function test_fake_conduits_receive_nothing() {
+  let ext = ExtensionTestUtils.loadExtension({
+    background() {
+      browser.runtime.onConnect.addListener(port => {
+        port.onMessage.addListener(msg => {
+          if (msg === "SECRET") {
+            port.postMessage("ACK");
+          } else {
+            browser.test.fail(`Receiver got a forged message: ${msg}`);
+          }
+        });
+      });
+    },
+    files: {
+      "page.html": `<!DOCTYPE html><meta charset="utf8"><script src="page.js"></script>`,
+      "page.js"() {
+        let port = browser.runtime.connect();
+        port.onMessage.addListener(msg => {
+          if (msg === "ACK") {
+            browser.test.sendMessage("got-ack");
+          }
+        });
+        browser.test.onMessage.addListener(msg => {
+          if (msg === "post-secret") {
+            port.postMessage("SECRET");
+          }
+        });
+        browser.test.sendMessage("page-ready");
+      },
+    },
+  });
+
+  await ext.startup();
+  let url = `moz-extension://${ext.uuid}/page.html`;
+  let page = await ExtensionTestUtils.loadContentPage(url, { extension: ext });
+  await ext.awaitMessage("page-ready");
+
+  // Open fake conduits for both source and receiver using a live portId.
+  // They share the page's actor, so IPC ordering would put any leaked
+  // message before the round-trip ACK. Neither should receive anything.
+  let { messages } = await promiseConsoleOutput(async () => {
+    await page.spawn([ext.id], extensionId => {
+      let actor = content.windowGlobalChild.getActor("Conduits");
+      let port = [...actor.conduits.values()].find(c => c.address?.portId);
+
+      // One fake subject for both port ends, a leak on either fails the test.
+      let fakeSubj = {
+        recvPortMessage(arg) {
+          Assert.ok(
+            false,
+            `Fake conduit received a message: ${JSON.stringify(arg)}`
+          );
+        },
+        recvPortDisconnect() {
+          Assert.ok(false, "Fake conduit received a disconnect");
+        },
+      };
+
+      for (let source of [true, false]) {
+        // Forges openConduit call from ExtensionChild.Port constructor.
+        let conduit = actor.openConduit(fakeSubj, {
+          id: source ? "fake@src" : "fake@rcv",
+          portId: port.address.portId,
+          source,
+          extensionId,
+          envType: "addon_child",
+          recv: ["PortMessage", "PortDisconnect"],
+          send: ["PortMessage"],
+        });
+
+        // Forge a message; parent drops it because the conduit was rejected.
+        conduit.sendPortMessage({ message: "INJECT" });
+      }
+    });
+
+    // Round-trip so the veto logs (and any leak) land before capture ends.
+    ext.sendMessage("post-secret");
+    await ext.awaitMessage("got-ack");
+  });
+
+  Assert.ok(
+    messages.some(m => m.message?.includes("Duplicate port source")),
+    "Parent logged the duplicate source"
+  );
+  Assert.ok(
+    messages.some(m => m.message?.includes("Unknown port receiver")),
+    "Parent logged the unknown receiver"
+  );
+  Assert.equal(
+    messages.filter(m =>
+      m.message?.includes("Unknown sender or wrong actor for recvPortMessage")
+    ).length,
+    2,
+    "Parent rejected both forged outbound messages"
+  );
+
+  await page.close();
+  await ext.unload();
+});
diff --git a/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml b/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
index 374268ea6c2..d3ddec0e4e6 100644
--- a/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
+++ b/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
... (file diff truncated)

diff --git a/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js b/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js
new file mode 100644
index 00000000000..54c78d6352f
--- /dev/null
+++ b/toolkit/components/extensions/test/xpcshell/test_ext_runtime_ports_membership.js
@@ -0,0 +1,110 @@
+"use strict";
+
+// A compromised child could forge a raw IPC message reusing another portId
+// to join a channel it shouldn't have access to.
+
+// The forged outbound PortMessages below are rejected in the parent,
+// which surfaces as uncaught rejections; they are expected.
+PromiseTestUtils.allowMatchingRejectionsGlobally(
+  /Unknown sender or wrong actor for recvPortMessage/
+);
+
+add_task(async function test_fake_conduits_receive_nothing() {
+  let ext = ExtensionTestUtils.loadExtension({
+    background() {
+      browser.runtime.onConnect.addListener(port => {
+        port.onMessage.addListener(msg => {
+          if (msg === "SECRET") {
+            port.postMessage("ACK");
+          } else {
+            browser.test.fail(`Receiver got a forged message: ${msg}`);
+          }
+        });
+      });
+    },
+    files: {
+      "page.html": `<!DOCTYPE html><meta charset="utf8"><script src="page.js"></script>`,
+      "page.js"() {
+        let port = browser.runtime.connect();
+        port.onMessage.addListener(msg => {
+          if (msg === "ACK") {
+            browser.test.sendMessage("got-ack");
+          }
+        });
+        browser.test.onMessage.addListener(msg => {
+          if (msg === "post-secret") {
+            port.postMessage("SECRET");
+          }
+        });
+        browser.test.sendMessage("page-ready");
+      },
+    },
+  });
+
+  await ext.startup();
+  let url = `moz-extension://${ext.uuid}/page.html`;
+  let page = await ExtensionTestUtils.loadContentPage(url, { extension: ext });
+  await ext.awaitMessage("page-ready");
+
+  // Open fake conduits for both source and receiver using a live portId.
+  // They share the page's actor, so IPC ordering would put any leaked
+  // message before the round-trip ACK. Neither should receive anything.
+  let { messages } = await promiseConsoleOutput(async () => {
+    await page.spawn([ext.id], extensionId => {
+      let actor = content.windowGlobalChild.getActor("Conduits");
+      let port = [...actor.conduits.values()].find(c => c.address?.portId);
+
+      // One fake subject for both port ends, a leak on either fails the test.
+      let fakeSubj = {
+        recvPortMessage(arg) {
+          Assert.ok(
+            false,
+            `Fake conduit received a message: ${JSON.stringify(arg)}`
+          );
+        },
+        recvPortDisconnect() {
+          Assert.ok(false, "Fake conduit received a disconnect");
+        },
+      };
+
+      for (let source of [true, false]) {
+        // Forges openConduit call from ExtensionChild.Port constructor.
+        let conduit = actor.openConduit(fakeSubj, {
+          id: source ? "fake@src" : "fake@rcv",
+          portId: port.address.portId,
+          source,
+          extensionId,
+          envType: "addon_child",
+          recv: ["PortMessage", "PortDisconnect"],
+          send: ["PortMessage"],
+        });
+
+        // Forge a message; parent drops it because the conduit was rejected.
+        conduit.sendPortMessage({ message: "INJECT" });
+      }
+    });
+
+    // Round-trip so the veto logs (and any leak) land before capture ends.
+    ext.sendMessage("post-secret");
+    await ext.awaitMessage("got-ack");
+  });
+
+  Assert.ok(
+    messages.some(m => m.message?.includes("Duplicate port source")),
+    "Parent logged the duplicate source"
+  );
+  Assert.ok(
+    messages.some(m => m.message?.includes("Unknown port receiver")),
+    "Parent logged the unknown receiver"
+  );
+  Assert.equal(
+    messages.filter(m =>
+      m.message?.includes("Unknown sender or wrong actor for recvPortMessage")
+    ).length,
+    2,
+    "Parent rejected both forged outbound messages"
+  );
+
+  await page.close();
+  await ext.unload();
+});
diff --git a/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml b/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
index 374268ea6c2..d3ddec0e4e6 100644
--- a/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
+++ b/toolkit/components/extensions/test/xpcshell/xpcshell-common.toml
@@ -614,6 +614,8 @@ skip-if = [
 
 ["test_ext_runtime_ports_gc.js"]
 
+["test_ext_runtime_ports_membership.js"]
+
 ["test_ext_runtime_sendMessage.js"]
 
 ["test_ext_runtime_sendMessage_errors.js"]