Forbid creating of stream objects outside of asyncio (#13101)

1 files changed, 22 insertions, 10 deletions

diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
index 90fc00de83..fa58e1e858 100644
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -1,6 +1,7 @@
 __all__ = 'create_subprocess_exec', 'create_subprocess_shell'
 
 import subprocess
+import warnings
 
 from . import events
 from . import protocols
@@ -18,8 +19,8 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
                                protocols.SubprocessProtocol):
     """Like StreamReaderProtocol, but for a subprocess."""
 
-    def __init__(self, limit, loop):
-        super().__init__(loop=loop)
+    def __init__(self, limit, loop, *, _asyncio_internal=False):
+        super().__init__(loop=loop, _asyncio_internal=_asyncio_internal)
         self._limit = limit
         self.stdin = self.stdout = self.stderr = None
         self._transport = None
@@ -42,14 +43,16 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
         stdout_transport = transport.get_pipe_transport(1)
         if stdout_transport is not None:
             self.stdout = streams.StreamReader(limit=self._limit,
-                                               loop=self._loop)
+                                               loop=self._loop,
+                                               _asyncio_internal=True)
             self.stdout.set_transport(stdout_transport)
             self._pipe_fds.append(1)
 
         stderr_transport = transport.get_pipe_transport(2)
         if stderr_transport is not None:
             self.stderr = streams.StreamReader(limit=self._limit,
-                                               loop=self._loop)
+                                               loop=self._loop,
+                                               _asyncio_internal=True)
             self.stderr.set_transport(stderr_transport)
             self._pipe_fds.append(2)
 
@@ -58,7 +61,8 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
             self.stdin = streams.StreamWriter(stdin_transport,
                                               protocol=self,
                                               reader=None,
-                                              loop=self._loop)
+                                              loop=self._loop,
+                                              _asyncio_internal=True)
 
     def pipe_data_received(self, fd, data):
         if fd == 1:
@@ -104,7 +108,13 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
 
 
 class Process:
-    def __init__(self, transport, protocol, loop):
+    def __init__(self, transport, protocol, loop, *, _asyncio_internal=False):
+        if not _asyncio_internal:
+            warnings.warn(f"{self.__class__} should be instaniated "
+                          "by asyncio internals only, "
+                          "please avoid its creation from user code",
+                          DeprecationWarning)
+
         self._transport = transport
         self._protocol = protocol
         self._loop = loop
@@ -195,12 +205,13 @@ async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
     if loop is None:
         loop = events.get_event_loop()
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
-                                                        loop=loop)
+                                                        loop=loop,
+                                                        _asyncio_internal=True)
     transport, protocol = await loop.subprocess_shell(
         protocol_factory,
         cmd, stdin=stdin, stdout=stdout,
         stderr=stderr, **kwds)
-    return Process(transport, protocol, loop)
+    return Process(transport, protocol, loop, _asyncio_internal=True)
 
 
 async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
@@ -209,10 +220,11 @@ async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
     if loop is None:
         loop = events.get_event_loop()
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
-                                                        loop=loop)
+                                                        loop=loop,
+                                                        _asyncio_internal=True)
     transport, protocol = await loop.subprocess_exec(
         protocol_factory,
         program, *args,
         stdin=stdin, stdout=stdout,
         stderr=stderr, **kwds)
-    return Process(transport, protocol, loop)
+    return Process(transport, protocol, loop, _asyncio_internal=True)