[Python-checkins] cpython (3.5): asyncio: Only allow Unix Stream sockets for loop.create_unix_server/connection
yury.selivanov
python-checkins at python.org
Fri Oct 7 12:40:50 EDT 2016
https://hg.python.org/cpython/rev/1c5459c597ba
changeset: 104343:1c5459c597ba
branch: 3.5
parent: 104338:b252c8079342
user: Yury Selivanov <yury at magic.io>
date: Fri Oct 07 12:39:57 2016 -0400
summary:
asyncio: Only allow Unix Stream sockets for loop.create_unix_server/connection
files:
Lib/asyncio/unix_events.py | 11 +++++++++--
Lib/test/test_asyncio/test_unix_events.py | 11 ++++++++++-
2 files changed, 19 insertions(+), 3 deletions(-)
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -234,6 +234,11 @@
else:
if sock is None:
raise ValueError('no path and sock were specified')
+ if (sock.family != socket.AF_UNIX or
+ sock.type != socket.SOCK_STREAM):
+ raise ValueError(
+ 'A UNIX Domain Stream Socket was expected, got {!r}'
+ .format(sock))
sock.setblocking(False)
transport, protocol = yield from self._create_connection_transport(
@@ -272,9 +277,11 @@
raise ValueError(
'path was not specified, and no sock specified')
- if sock.family != socket.AF_UNIX:
+ if (sock.family != socket.AF_UNIX or
+ sock.type != socket.SOCK_STREAM):
raise ValueError(
- 'A UNIX Domain Socket was expected, got {!r}'.format(sock))
+ 'A UNIX Domain Stream Socket was expected, got {!r}'
+ .format(sock))
server = base_events.Server(self, [sock])
sock.listen(backlog)
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -273,7 +273,16 @@
coro = self.loop.create_unix_server(lambda: None, path=None,
sock=sock)
with self.assertRaisesRegex(ValueError,
- 'A UNIX Domain Socket was expected'):
+ 'A UNIX Domain Stream.*was expected'):
+ self.loop.run_until_complete(coro)
+
+ def test_create_unix_connection_path_inetsock(self):
+ sock = socket.socket()
+ with sock:
+ coro = self.loop.create_unix_connection(lambda: None, path=None,
+ sock=sock)
+ with self.assertRaisesRegex(ValueError,
+ 'A UNIX Domain Stream.*was expected'):
self.loop.run_until_complete(coro)
@mock.patch('asyncio.unix_events.socket')
--
Repository URL: https://hg.python.org/cpython
More information about the Python-checkins
mailing list