gh-143376: validate usages of _interpchannels.ChannelID in _channelid_shared

Author: picnixz

Lib/test/test_interpreters/test_channels.py

@@ -30,6 +30,30 @@ def test_highlevel_reloaded(self):
         # See gh-115490 (https://github.com/python/cpython/issues/115490).
         importlib.reload(channels)
 
+    def test_channel_id_types(self):
+        # See https://github.com/python/cpython/issues/143376.
+        class End:
+            __slots__ = ()
+            def __new__(cls, chan=None):
+                return object.__new__(cls)
+            @property
+            def _id(self):
+                return object()
+
+        class Send(End): pass
+        class Recv(End): pass
+
+        _ = importlib.reload(_channels)
+        _._register_end_types(Send, Recv)
+        chan = _._channel_id(_.create(), _resolve=True)
+        interp = interpreters.create()
+        # The TypeError exception arises from our internal check,
+        # but subinterpreters use a generic exception message instead
+        # while retaining the original exception type.
+        err = "does not support cross-interpreter data"
+        with self.assertRaisesRegex(TypeError, err):
+            interp.prepare_main(chan=Send(chan))
+
 
 class TestChannels(TestBase):
 

Misc/NEWS.d/next/Library/2026-03-28-14-34-18.gh-issue-143376.FvWTwi.rst

@@ -0,0 +1,2 @@
+:mod:`concurrent.interpreters`: fix a crash when internal cross-interpreter
+channel ends have incorrect channel IDs. Patch by Bénédikt Tran.

Modules/_interpchannelsmodule.c

@@ -2633,6 +2633,17 @@ _channelid_from_xid(_PyXIData_t *data)
 static int
 _channelid_shared(PyThreadState *tstate, PyObject *obj, _PyXIData_t *data)
 {
+    module_state *st = _get_current_module_state();
+    if (st == NULL) {
+        return -1;
+    }
+    if (!PyObject_IsInstance(obj, (PyObject *)st->ChannelIDType)) {
+        PyErr_Format(PyExc_TypeError,
+                     "'_id' attribute must be an %N object, not %T",
+                     st->ChannelIDType, obj);
+        return -1;
+    }
+
     if (_PyXIData_InitWithSize(
             data, tstate->interp, sizeof(struct _channelid_xid), obj,
             _channelid_from_xid