def get_ssl_context(self, protocol=None):
"""
This method returns a SSL context based on the file that was specified
when this object was created.
Arguments:
- An optional protocol. SSLv2 is used by default.
Returns:
- The SSL context
"""
# Validate the arguments
if protocol is None:
protocol = ssl.PROTOCOL_SSLv2
# Create an SSL context from the stored file and password.
ssl_context = ssl.SSLContext(protocol)
ssl_context.load_cert_chain(self._cert_filename,
password=self._password)
# Return the context
return ssl_context
python类PROTOCOL_SSLv2()的实例源码
def _connect(self, host, port, timeout, isssl = False):
conn = None
try:
if isssl and not _SUPPORT_SSL:
raise 'Not SUPPORT SSL'
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
if isssl:
try:
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv23)
except ssl.SSLError as _:
conn.close()
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv2)
conn.settimeout(timeout)
except Exception as e:
raise CurlError(Curl.CURLE_COULDNT_CONNECT)
return conn
def _connect(self, host, port, timeout, isssl = False):
conn = None
try:
if isssl and not _SUPPORT_SSL:
raise 'Not SUPPORT SSL'
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
if isssl:
try:
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv23)
except ssl.SSLError as _:
conn.close()
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv2)
conn.settimeout(timeout)
except Exception as e:
raise CurlError(Curl.CURLE_COULDNT_CONNECT)
return conn
def _connect(self, host, port, timeout, isssl = False):
conn = None
try:
if isssl and not _SUPPORT_SSL:
raise 'Not SUPPORT SSL'
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
if isssl:
try:
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv23)
except ssl.SSLError as _:
conn.close()
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.connect((host, port))
conn = ssl.wrap_socket(conn, ssl_version=ssl.PROTOCOL_SSLv2)
conn.settimeout(timeout)
except Exception as e:
raise CurlError(Curl.CURLE_COULDNT_CONNECT)
return conn
def ssl_wrap_socket(self):
# Allow sending of keep-alive messages - seems to prevent some servers
# from closing SSL, leading to deadlocks.
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
try:
import ssl
if self.ca_certs is not None:
cert_reqs = ssl.CERT_REQUIRED
else:
cert_reqs = ssl.CERT_NONE
if self.ssl_version == "tls1":
ssl_version = ssl.PROTOCOL_TLSv1
elif self.ssl_version == "ssl2":
ssl_version = ssl.PROTOCOL_SSLv2
elif self.ssl_version == "ssl3":
ssl_version = ssl.PROTOCOL_SSLv3
elif self.ssl_version == "ssl23" or self.ssl_version is None:
ssl_version = ssl.PROTOCOL_SSLv23
else:
raise socket.sslerror("Invalid SSL version requested: %s", self.ssl_version)
self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ca_certs=self.ca_certs, cert_reqs=cert_reqs, ssl_version=ssl_version)
ssl_exc = ssl.SSLError
self.read_fd = self.sock.fileno()
except ImportError:
# No ssl module, and socket.ssl has no fileno(), and does not allow certificate verification
raise socket.sslerror("imaplib2 SSL mode does not work without ssl module")
if self.cert_verify_cb is not None:
cert_err = self.cert_verify_cb(self.sock.getpeercert(), self.host)
if cert_err:
raise ssl_exc(cert_err)
def test_ssl2_disabled(self):
self.assertRaises(socket.error,
self._connect_socket,
ssl_version=ssl.PROTOCOL_SSLv2)
def open_ssl_socket(version=ssl.PROTOCOL_SSLv2):
pass
def ssl_wrap_socket(self):
# Allow sending of keep-alive messages - seems to prevent some servers
# from closing SSL, leading to deadlocks.
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
try:
import ssl
if self.ca_certs is not None:
cert_reqs = ssl.CERT_REQUIRED
else:
cert_reqs = ssl.CERT_NONE
if self.ssl_version == "tls1":
ssl_version = ssl.PROTOCOL_TLSv1
elif self.ssl_version == "ssl2":
ssl_version = ssl.PROTOCOL_SSLv2
elif self.ssl_version == "ssl3":
ssl_version = ssl.PROTOCOL_SSLv3
elif self.ssl_version == "ssl23" or self.ssl_version is None:
ssl_version = ssl.PROTOCOL_SSLv23
else:
raise socket.sslerror("Invalid SSL version requested: %s", self.ssl_version)
self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ca_certs=self.ca_certs, cert_reqs=cert_reqs, ssl_version=ssl_version)
ssl_exc = ssl.SSLError
self.read_fd = self.sock.fileno()
except ImportError:
# No ssl module, and socket.ssl has no fileno(), and does not allow certificate verification
raise socket.sslerror("imaplib2 SSL mode does not work without ssl module")
if self.cert_verify_cb is not None:
cert_err = self.cert_verify_cb(self.sock.getpeercert(), self.host)
if cert_err:
raise ssl_exc(cert_err)
def try_protocol_combo(self, server_protocol, client_protocol):
with self.server.lock:
self.server.handler.ssl_version = server_protocol
self.client.ssl_version = client_protocol
self.client.close()
self.client.connect(self.server.host, self.server.port)
try:
self.client.login()
except (ssl.SSLError, socket.error):
self.client.close()
else:
self.client.quit()
# def test_ssl_version(self):
# protos = [ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv23,
# ssl.PROTOCOL_TLSv1]
# if hasattr(ssl, "PROTOCOL_SSLv2"):
# protos.append(ssl.PROTOCOL_SSLv2)
# for proto in protos:
# self.try_protocol_combo(ssl.PROTOCOL_SSLv2, proto)
# for proto in protos:
# self.try_protocol_combo(ssl.PROTOCOL_SSLv3, proto)
# for proto in protos:
# self.try_protocol_combo(ssl.PROTOCOL_SSLv23, proto)
# for proto in protos:
# self.try_protocol_combo(ssl.PROTOCOL_TLSv1, proto)
def test_sslv2(self):
self.client.ssl_version = ssl.PROTOCOL_SSLv2
self.client.close()
with self.server.lock:
self.client.connect(self.server.host, self.server.port)
self.assertRaises(socket.error, self.client.login)
self.client.ssl_version = ssl.PROTOCOL_SSLv2