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

1 files changed, 51 insertions, 16 deletions

diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py
index 33fc303..c9b1f32 100644
--- a/Lib/asyncio/streams.py
+++ b/Lib/asyncio/streams.py
@@ -4,6 +4,7 @@ __all__ = (
 
 import socket
 import sys
+import warnings
 import weakref
 
 if hasattr(socket, 'AF_UNIX'):
@@ -42,11 +43,14 @@ async def open_connection(host=None, port=None, *,
     """
     if loop is None:
         loop = events.get_event_loop()
-    reader = StreamReader(limit=limit, loop=loop)
-    protocol = StreamReaderProtocol(reader, loop=loop)
+    reader = StreamReader(limit=limit, loop=loop,
+                          _asyncio_internal=True)
+    protocol = StreamReaderProtocol(reader, loop=loop,
+                                    _asyncio_internal=True)
     transport, _ = await loop.create_connection(
         lambda: protocol, host, port, **kwds)
-    writer = StreamWriter(transport, protocol, reader, loop)
+    writer = StreamWriter(transport, protocol, reader, loop,
+                          _asyncio_internal=True)
     return reader, writer
 
 
@@ -77,9 +81,11 @@ async def start_server(client_connected_cb, host=None, port=None, *,
         loop = events.get_event_loop()
 
     def factory():
-        reader = StreamReader(limit=limit, loop=loop)
+        reader = StreamReader(limit=limit, loop=loop,
+                              _asyncio_internal=True)
         protocol = StreamReaderProtocol(reader, client_connected_cb,
-                                        loop=loop)
+                                        loop=loop,
+                                        _asyncio_internal=True)
         return protocol
 
     return await loop.create_server(factory, host, port, **kwds)
@@ -93,11 +99,14 @@ if hasattr(socket, 'AF_UNIX'):
         """Similar to `open_connection` but works with UNIX Domain Sockets."""
         if loop is None:
             loop = events.get_event_loop()
-        reader = StreamReader(limit=limit, loop=loop)
-        protocol = StreamReaderProtocol(reader, loop=loop)
+        reader = StreamReader(limit=limit, loop=loop,
+                              _asyncio_internal=True)
+        protocol = StreamReaderProtocol(reader, loop=loop,
+                                        _asyncio_internal=True)
         transport, _ = await loop.create_unix_connection(
             lambda: protocol, path, **kwds)
-        writer = StreamWriter(transport, protocol, reader, loop)
+        writer = StreamWriter(transport, protocol, reader, loop,
+                              _asyncio_internal=True)
         return reader, writer
 
     async def start_unix_server(client_connected_cb, path=None, *,
@@ -107,9 +116,11 @@ if hasattr(socket, 'AF_UNIX'):
             loop = events.get_event_loop()
 
         def factory():
-            reader = StreamReader(limit=limit, loop=loop)
+            reader = StreamReader(limit=limit, loop=loop,
+                                  _asyncio_internal=True)
             protocol = StreamReaderProtocol(reader, client_connected_cb,
-                                            loop=loop)
+                                            loop=loop,
+                                            _asyncio_internal=True)
             return protocol
 
         return await loop.create_unix_server(factory, path, **kwds)
@@ -125,11 +136,20 @@ class FlowControlMixin(protocols.Protocol):
     StreamWriter.drain() must wait for _drain_helper() coroutine.
     """
 
-    def __init__(self, loop=None):
+    def __init__(self, loop=None, *, _asyncio_internal=False):
         if loop is None:
             self._loop = events.get_event_loop()
         else:
             self._loop = loop
+        if not _asyncio_internal:
+            # NOTE:
+            # Avoid inheritance from FlowControlMixin
+            # Copy-paste the code to your project
+            # if you need flow control helpers
+            warnings.warn(f"{self.__class__} should be instaniated "
+                          "by asyncio internals only, "
+                          "please avoid its creation from user code",
+                          DeprecationWarning)
         self._paused = False
         self._drain_waiter = None
         self._connection_lost = False
@@ -191,8 +211,9 @@ class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
 
     _source_traceback = None
 
-    def __init__(self, stream_reader, client_connected_cb=None, loop=None):
-        super().__init__(loop=loop)
+    def __init__(self, stream_reader, client_connected_cb=None, loop=None,
+                 *, _asyncio_internal=False):
+        super().__init__(loop=loop, _asyncio_internal=_asyncio_internal)
         if stream_reader is not None:
             self._stream_reader_wr = weakref.ref(stream_reader,
                                                  self._on_reader_gc)
@@ -253,7 +274,8 @@ class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
         if self._client_connected_cb is not None:
             self._stream_writer = StreamWriter(transport, self,
                                                reader,
-                                               self._loop)
+                                               self._loop,
+                                               _asyncio_internal=True)
             res = self._client_connected_cb(reader,
                                             self._stream_writer)
             if coroutines.iscoroutine(res):
@@ -311,7 +333,13 @@ class StreamWriter:
     directly.
     """
 
-    def __init__(self, transport, protocol, reader, loop):
+    def __init__(self, transport, protocol, reader, 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
         # drain() expects that the reader has an exception() method
@@ -388,7 +416,14 @@ class StreamReader:
 
     _source_traceback = None
 
-    def __init__(self, limit=_DEFAULT_LIMIT, loop=None):
+    def __init__(self, limit=_DEFAULT_LIMIT, loop=None,
+                 *, _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)
+
         # The line length limit is  a security feature;
         # it also doubles as half the buffer limit.