1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
# DP: Fix issue #9729, Unconnected SSLSocket.{send,recv} raises TypeError.
--- a/Lib/ssl.py
+++ b/Lib/ssl.py
@@ -184,14 +184,14 @@
else:
return v
else:
- return socket.send(self, data, flags)
+ return self._sock.send(data, flags)
def sendto(self, data, addr, flags=0):
if self._sslobj:
raise ValueError("sendto not allowed on instances of %s" %
self.__class__)
else:
- return socket.sendto(self, data, addr, flags)
+ return self._sock.sendto(data, flags, addr)
def sendall(self, data, flags=0):
if self._sslobj:
@@ -216,7 +216,7 @@
self.__class__)
return self.read(buflen)
else:
- return socket.recv(self, buflen, flags)
+ return self._sock.recv(buflen, flags)
def recv_into(self, buffer, nbytes=None, flags=0):
if buffer and (nbytes is None):
@@ -233,21 +233,21 @@
buffer[:v] = tmp_buffer
return v
else:
- return socket.recv_into(self, buffer, nbytes, flags)
+ return self._sock.recv_into(buffer, nbytes, flags)
def recvfrom(self, addr, buflen=1024, flags=0):
if self._sslobj:
raise ValueError("recvfrom not allowed on instances of %s" %
self.__class__)
else:
- return socket.recvfrom(self, addr, buflen, flags)
+ return self._sock.recvfrom(buflen, flags)
def recvfrom_into(self, buffer, nbytes=None, flags=0):
if self._sslobj:
raise ValueError("recvfrom_into not allowed on instances of %s" %
self.__class__)
else:
- return socket.recvfrom_into(self, buffer, nbytes, flags)
+ return self._sock.recvfrom_into(buffer, nbytes, flags)
def pending(self):
if self._sslobj:
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -151,6 +151,19 @@
del ss
self.assertEqual(wr(), None)
+ def test_wrapped_unconnected(self):
+ # The _delegate_methods in socket.py are correctly delegated to by an
+ # unconnected SSLSocket, so they will raise a socket.error rather than
+ # something unexpected like TypeError.
+ s = socket.socket(socket.AF_INET)
+ ss = ssl.wrap_socket(s)
+ self.assertRaises(socket.error, ss.recv, 1)
+ self.assertRaises(socket.error, ss.recv_into, bytearray('x'))
+ self.assertRaises(socket.error, ss.recvfrom, None)
+ self.assertRaises(socket.error, ss.recvfrom_into, bytearray(''), 0)
+ self.assertRaises(socket.error, ss.send, 'x')
+ self.assertRaises(socket.error, ss.sendto, 'x', ('0.0.0.0', 0))
+
class NetworkedTests(unittest.TestCase):
|
