gh-143376: validate usages of _interpchannels.ChannelID in _channelid_shared (#146574)

Author: picnixz

Lib/test/test_interpreters/test_channels.py

@@ -8,7 +8,7 @@
 # Raise SkipTest if subinterpreters not supported.
 _channels = import_helper.import_module('_interpchannels')
 from concurrent import interpreters
-from test.support import channels
+from test.support import channels, force_not_colorized
 from .utils import _run_output, TestBase
 
 
@@ -34,6 +34,39 @@ def test_highlevel_reloaded(self):
             importlib.reload(channels)
             """));
 
+    @force_not_colorized
+    def test_channel_id_types(self):
+        # See https://github.com/python/cpython/issues/143376.
+        code = """\
+        import _interpchannels
+        import re
+        from concurrent import interpreters
+
+        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
+
+        _interpchannels._register_end_types(Send, Recv)
+        chan = _interpchannels._channel_id(_interpchannels.create(), _resolve=True)
+        interp = interpreters.create()
+        try:
+            interp.prepare_main(chan=Send(chan))
+        finally:
+            interp.close()
+        """
+
+        out, err = self.assert_python_failure("-c", code)
+        self.assertEqual(out, '')
+        errmsg = "TypeError: '_id' attribute must be an _interpchannels.ChannelID object, not object"
+        self.assertIn(errmsg, err)
+
 
 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

@@ -2634,6 +2634,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