[pypy-svn] pypy default: Fix ssl.py and test_ssl.py (by chance, one more test passing on nightly
fijal
commits-noreply at bitbucket.org
Tue Feb 1 13:21:41 CET 2011
Author: Maciej Fijalkowski <fijall at gmail.com>
Branch:
Changeset: r41517:f7e470d7823c
Date: 2011-02-01 14:21 +0200
http://bitbucket.org/pypy/pypy/changeset/f7e470d7823c/
Log: Fix ssl.py and test_ssl.py (by chance, one more test passing on
nightly run as well)
diff --git a/pypy/interpreter/gateway.py b/pypy/interpreter/gateway.py
--- a/pypy/interpreter/gateway.py
+++ b/pypy/interpreter/gateway.py
@@ -344,8 +344,9 @@
self.nextarg()))
def visit__ObjSpace(self, el):
- if self.finger != 0:
+ if self.finger not in [0, 1]:
raise FastFuncNotSupported
+ self.args.append('space')
self.unwrap.append("space")
def visit__W_Root(self, el):
@@ -391,7 +392,9 @@
unwrap_info = UnwrapSpec_FastFunc_Unwrap()
unwrap_info.apply_over(unwrap_spec)
narg = unwrap_info.n
- args = ['space'] + unwrap_info.args
+ args = unwrap_info.args
+ if 'space' not in args:
+ args = ['space'] + args
if args == unwrap_info.unwrap:
fastfunc = func
else:
@@ -528,8 +531,12 @@
self.__class__ = BuiltinCodePassThroughArguments1
self.func__args__ = func
else:
- self.__class__ = globals()['BuiltinCode%d' % arity]
- setattr(self, 'fastfunc_%d' % arity, fastfunc)
+ if len(unwrap_spec) >= 2 and unwrap_spec[1] is ObjSpace:
+ self.__class__ = globals()['BuiltinCodeSpace%d' % arity]
+ setattr(self, 'fastfunc_space_%d' % arity, fastfunc)
+ else:
+ self.__class__ = globals()['BuiltinCode%d' % arity]
+ setattr(self, 'fastfunc_%d' % arity, fastfunc)
def descr__reduce__(self, space):
from pypy.interpreter.mixedmodule import MixedModule
@@ -717,6 +724,78 @@
w_result = space.w_None
return w_result
+class BuiltinCodeSpace1(BuiltinCode1):
+ _immutable_ = True
+ fast_natural_arity = 1
+
+ def fastcall_1(self, space, w_func, w1):
+ try:
+ w_result = self.fastfunc_space_1(w1, space)
+ except DescrMismatch:
+ return w1.descr_call_mismatch(space,
+ self.descrmismatch_op,
+ self.descr_reqcls,
+ Arguments(space, [w1]))
+ except Exception, e:
+ raise self.handle_exception(space, e)
+ if w_result is None:
+ w_result = space.w_None
+ return w_result
+
+class BuiltinCodeSpace2(BuiltinCode2):
+ _immutable_ = True
+ fast_natural_arity = 2
+
+ def fastcall_2(self, space, w_func, w1, w2):
+ try:
+ w_result = self.fastfunc_space_2(w1, space, w2)
+ except DescrMismatch:
+ return w1.descr_call_mismatch(space,
+ self.descrmismatch_op,
+ self.descr_reqcls,
+ Arguments(space, [w1, w2]))
+ except Exception, e:
+ raise self.handle_exception(space, e)
+ if w_result is None:
+ w_result = space.w_None
+ return w_result
+
+class BuiltinCodeSpace3(BuiltinCode3):
+ _immutable_ = True
+ fast_natural_arity = 3
+
+ def fastcall_3(self, space, func, w1, w2, w3):
+ try:
+ w_result = self.fastfunc_space_3(w1, space, w2, w3)
+ except DescrMismatch:
+ return w1.descr_call_mismatch(space,
+ self.descrmismatch_op,
+ self.descr_reqcls,
+ Arguments(space, [w1, w2, w3]))
+ except Exception, e:
+ raise self.handle_exception(space, e)
+ if w_result is None:
+ w_result = space.w_None
+ return w_result
+
+class BuiltinCodeSpace4(BuiltinCode4):
+ _immutable_ = True
+ fast_natural_arity = 4
+
+ def fastcall_4(self, space, func, w1, w2, w3, w4):
+ try:
+ w_result = self.fastfunc_space_4(w1, space, w2, w3, w4)
+ except DescrMismatch:
+ return w1.descr_call_mismatch(space,
+ self.descrmismatch_op,
+ self.descr_reqcls,
+ Arguments(space,
+ [w1, w2, w3, w4]))
+ except Exception, e:
+ raise self.handle_exception(space, e)
+ if w_result is None:
+ w_result = space.w_None
+ return w_result
class interp2app(Wrappable):
"""Build a gateway that calls 'f' at interp-level."""
diff --git a/pypy/interpreter/test/test_gateway.py b/pypy/interpreter/test/test_gateway.py
--- a/pypy/interpreter/test/test_gateway.py
+++ b/pypy/interpreter/test/test_gateway.py
@@ -14,7 +14,7 @@
self.name = name
self.defs_w = []
-class TestBuiltinCode:
+class TestBuiltinCode(object):
def test_signature(self):
def c(space, w_x, w_y, hello_w):
pass
@@ -545,7 +545,42 @@
""")
assert space.is_true(w_res)
- assert called == [w_app_f, w_app_f]
+ assert called == [w_app_f, w_app_f]
+
+ def test_interp2app_fastcall_method_space(self):
+ space = self.space
+ w = space.wrap
+ w_3 = w(3)
+
+ def f(w_self, space, w_x):
+ return w_x
+ app_f = gateway.interp2app_temp(f, unwrap_spec=[gateway.W_Root,
+ gateway.ObjSpace,
+ gateway.W_Root])
+ w_app_f = w(app_f)
+
+ # sanity
+ assert isinstance(w_app_f.code, gateway.BuiltinCode2)
+
+ called = []
+ fastcall_2 = w_app_f.code.fastcall_2
+ def witness_fastcall_2(space, w_func, w_a, w_b):
+ called.append(w_func)
+ return fastcall_2(space, w_func, w_a, w_b)
+
+ w_app_f.code.fastcall_2 = witness_fastcall_2
+
+ w_res = space.appexec([w_app_f, w_3], """(f, x):
+ class A(object):
+ m = f # not a builtin function, so works as method
+ y = A().m(x)
+ b = A().m
+ z = b(x)
+ return y is x and z is x
+ """)
+
+ assert space.is_true(w_res)
+ assert called == [w_app_f, w_app_f]
def test_plain(self):
space = self.space
diff --git a/lib-python/modified-2.7.0/ssl.py b/lib-python/modified-2.7.0/ssl.py
new file mode 100644
--- /dev/null
+++ b/lib-python/modified-2.7.0/ssl.py
@@ -0,0 +1,437 @@
+# Wrapper module for _ssl, providing some additional facilities
+# implemented in Python. Written by Bill Janssen.
+
+"""\
+This module provides some more Pythonic support for SSL.
+
+Object types:
+
+ SSLSocket -- subtype of socket.socket which does SSL over the socket
+
+Exceptions:
+
+ SSLError -- exception raised for I/O errors
+
+Functions:
+
+ cert_time_to_seconds -- convert time string used for certificate
+ notBefore and notAfter functions to integer
+ seconds past the Epoch (the time values
+ returned from time.time())
+
+ fetch_server_certificate (HOST, PORT) -- fetch the certificate provided
+ by the server running on HOST at port PORT. No
+ validation of the certificate is performed.
+
+Integer constants:
+
+SSL_ERROR_ZERO_RETURN
+SSL_ERROR_WANT_READ
+SSL_ERROR_WANT_WRITE
+SSL_ERROR_WANT_X509_LOOKUP
+SSL_ERROR_SYSCALL
+SSL_ERROR_SSL
+SSL_ERROR_WANT_CONNECT
+
+SSL_ERROR_EOF
+SSL_ERROR_INVALID_ERROR_CODE
+
+The following group define certificate requirements that one side is
+allowing/requiring from the other side:
+
+CERT_NONE - no certificates from the other side are required (or will
+ be looked at if provided)
+CERT_OPTIONAL - certificates are not required, but if provided will be
+ validated, and if validation fails, the connection will
+ also fail
+CERT_REQUIRED - certificates are required, and will be validated, and
+ if validation fails, the connection will also fail
+
+The following constants identify various SSL protocol variants:
+
+PROTOCOL_SSLv2
+PROTOCOL_SSLv3
+PROTOCOL_SSLv23
+PROTOCOL_TLSv1
+"""
+
+import textwrap
+
+import _ssl # if we can't import it, let the error propagate
+
+from _ssl import OPENSSL_VERSION_NUMBER, OPENSSL_VERSION_INFO, OPENSSL_VERSION
+from _ssl import SSLError
+from _ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
+from _ssl import PROTOCOL_SSLv2, PROTOCOL_SSLv3, PROTOCOL_SSLv23, PROTOCOL_TLSv1
+from _ssl import RAND_status, RAND_egd, RAND_add
+from _ssl import \
+ SSL_ERROR_ZERO_RETURN, \
+ SSL_ERROR_WANT_READ, \
+ SSL_ERROR_WANT_WRITE, \
+ SSL_ERROR_WANT_X509_LOOKUP, \
+ SSL_ERROR_SYSCALL, \
+ SSL_ERROR_SSL, \
+ SSL_ERROR_WANT_CONNECT, \
+ SSL_ERROR_EOF, \
+ SSL_ERROR_INVALID_ERROR_CODE
+
+from socket import socket, _fileobject, error as socket_error
+from socket import getnameinfo as _getnameinfo
+import base64 # for DER-to-PEM translation
+import errno
+
+class SSLSocket(socket):
+
+ """This class implements a subtype of socket.socket that wraps
+ the underlying OS socket in an SSL context when necessary, and
+ provides read and write methods over that channel."""
+
+ def __init__(self, sock, keyfile=None, certfile=None,
+ server_side=False, cert_reqs=CERT_NONE,
+ ssl_version=PROTOCOL_SSLv23, ca_certs=None,
+ do_handshake_on_connect=True,
+ suppress_ragged_eofs=True, ciphers=None):
+ socket.__init__(self, _sock=sock._sock)
+
+ if certfile and not keyfile:
+ keyfile = certfile
+ # see if it's connected
+ try:
+ socket.getpeername(self)
+ except socket_error, e:
+ if e.errno != errno.ENOTCONN:
+ raise
+ # no, no connection yet
+ self._sslobj = None
+ else:
+ # yes, create the SSL object
+ self._sslobj = _ssl.sslwrap(self._sock, server_side,
+ keyfile, certfile,
+ cert_reqs, ssl_version, ca_certs,
+ ciphers)
+ if do_handshake_on_connect:
+ self.do_handshake()
+ self.keyfile = keyfile
+ self.certfile = certfile
+ self.cert_reqs = cert_reqs
+ self.ssl_version = ssl_version
+ self.ca_certs = ca_certs
+ self.ciphers = ciphers
+ self.do_handshake_on_connect = do_handshake_on_connect
+ self.suppress_ragged_eofs = suppress_ragged_eofs
+ self._makefile_refs = 0
+
+ def read(self, len=1024):
+
+ """Read up to LEN bytes and return them.
+ Return zero-length string on EOF."""
+
+ try:
+ return self._sslobj.read(len)
+ except SSLError, x:
+ if x.args[0] == SSL_ERROR_EOF and self.suppress_ragged_eofs:
+ return ''
+ else:
+ raise
+
+ def write(self, data):
+
+ """Write DATA to the underlying SSL channel. Returns
+ number of bytes of DATA actually transmitted."""
+
+ return self._sslobj.write(data)
+
+ def getpeercert(self, binary_form=False):
+
+ """Returns a formatted version of the data in the
+ certificate provided by the other end of the SSL channel.
+ Return None if no certificate was provided, {} if a
+ certificate was provided, but not validated."""
+
+ return self._sslobj.peer_certificate(binary_form)
+
+ def cipher(self):
+
+ if not self._sslobj:
+ return None
+ else:
+ return self._sslobj.cipher()
+
+ def send(self, data, flags=0):
+ if self._sslobj:
+ if flags != 0:
+ raise ValueError(
+ "non-zero flags not allowed in calls to send() on %s" %
+ self.__class__)
+ while True:
+ try:
+ v = self._sslobj.write(data)
+ except SSLError, x:
+ if x.args[0] == SSL_ERROR_WANT_READ:
+ return 0
+ elif x.args[0] == SSL_ERROR_WANT_WRITE:
+ return 0
+ else:
+ raise
+ else:
+ return v
+ else:
+ return self._sock.send(data, flags)
+
+ def sendto(self, data, flags_or_addr, addr=None):
+ if self._sslobj:
+ raise ValueError("sendto not allowed on instances of %s" %
+ self.__class__)
+ elif addr is None:
+ return self._sock.sendto(data, flags_or_addr)
+ else:
+ return self._sock.sendto(data, flags_or_addr, addr)
+
+ def sendall(self, data, flags=0):
+ if self._sslobj:
+ if flags != 0:
+ raise ValueError(
+ "non-zero flags not allowed in calls to sendall() on %s" %
+ self.__class__)
+ amount = len(data)
+ count = 0
+ while (count < amount):
+ v = self.send(data[count:])
+ count += v
+ return amount
+ else:
+ return socket.sendall(self, data, flags)
+
+ def recv(self, buflen=1024, flags=0):
+ if self._sslobj:
+ if flags != 0:
+ raise ValueError(
+ "non-zero flags not allowed in calls to recv() on %s" %
+ self.__class__)
+ return self.read(buflen)
+ else:
+ return self._sock.recv(buflen, flags)
+
+ def recv_into(self, buffer, nbytes=None, flags=0):
+ if buffer and (nbytes is None):
+ nbytes = len(buffer)
+ elif nbytes is None:
+ nbytes = 1024
+ if self._sslobj:
+ if flags != 0:
+ raise ValueError(
+ "non-zero flags not allowed in calls to recv_into() on %s" %
+ self.__class__)
+ tmp_buffer = self.read(nbytes)
+ v = len(tmp_buffer)
+ buffer[:v] = tmp_buffer
+ return v
+ else:
+ return self._sock.recv_into(buffer, nbytes, flags)
+
+ def recvfrom(self, buflen=1024, flags=0):
+ if self._sslobj:
+ raise ValueError("recvfrom not allowed on instances of %s" %
+ self.__class__)
+ else:
+ 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 self._sock.recvfrom_into(buffer, nbytes, flags)
+
+ def pending(self):
+ if self._sslobj:
+ return self._sslobj.pending()
+ else:
+ return 0
+
+ def unwrap(self):
+ if self._sslobj:
+ s = self._sslobj.shutdown()
+ self._sslobj = None
+ return s
+ else:
+ raise ValueError("No SSL wrapper around " + str(self))
+
+ def shutdown(self, how):
+ self._sslobj = None
+ socket.shutdown(self, how)
+
+ def close(self):
+ if self._makefile_refs < 1:
+ self._sslobj = None
+ socket.close(self)
+ else:
+ self._makefile_refs -= 1
+
+ def do_handshake(self):
+
+ """Perform a TLS/SSL handshake."""
+
+ self._sslobj.do_handshake()
+
+ def connect(self, addr):
+
+ """Connects to remote ADDR, and then wraps the connection in
+ an SSL channel."""
+
+ # Here we assume that the socket is client-side, and not
+ # connected at the time of the call. We connect it, then wrap it.
+ if self._sslobj:
+ raise ValueError("attempt to connect already-connected SSLSocket!")
+ socket.connect(self, addr)
+ self._sslobj = _ssl.sslwrap(self._sock, False, self.keyfile, self.certfile,
+ self.cert_reqs, self.ssl_version,
+ self.ca_certs, self.ciphers)
+ if self.do_handshake_on_connect:
+ self.do_handshake()
+
+ def accept(self):
+
+ """Accepts a new connection from a remote client, and returns
+ a tuple containing that new connection wrapped with a server-side
+ SSL channel, and the address of the remote client."""
+
+ newsock, addr = socket.accept(self)
+ return (SSLSocket(newsock,
+ keyfile=self.keyfile,
+ certfile=self.certfile,
+ server_side=True,
+ cert_reqs=self.cert_reqs,
+ ssl_version=self.ssl_version,
+ ca_certs=self.ca_certs,
+ ciphers=self.ciphers,
+ do_handshake_on_connect=self.do_handshake_on_connect,
+ suppress_ragged_eofs=self.suppress_ragged_eofs),
+ addr)
+
+ def makefile(self, mode='r', bufsize=-1):
+
+ """Make and return a file-like object that
+ works with the SSL connection. Just use the code
+ from the socket module."""
+
+ self._makefile_refs += 1
+ # close=True so as to decrement the reference count when done with
+ # the file-like object.
+ return _fileobject(self, mode, bufsize, close=True)
+
+
+
+def wrap_socket(sock, keyfile=None, certfile=None,
+ server_side=False, cert_reqs=CERT_NONE,
+ ssl_version=PROTOCOL_SSLv23, ca_certs=None,
+ do_handshake_on_connect=True,
+ suppress_ragged_eofs=True, ciphers=None):
+
+ return SSLSocket(sock, keyfile=keyfile, certfile=certfile,
+ server_side=server_side, cert_reqs=cert_reqs,
+ ssl_version=ssl_version, ca_certs=ca_certs,
+ do_handshake_on_connect=do_handshake_on_connect,
+ suppress_ragged_eofs=suppress_ragged_eofs,
+ ciphers=ciphers)
+
+
+# some utility functions
+
+def cert_time_to_seconds(cert_time):
+
+ """Takes a date-time string in standard ASN1_print form
+ ("MON DAY 24HOUR:MINUTE:SEC YEAR TIMEZONE") and return
+ a Python time value in seconds past the epoch."""
+
+ import time
+ return time.mktime(time.strptime(cert_time, "%b %d %H:%M:%S %Y GMT"))
+
+PEM_HEADER = "-----BEGIN CERTIFICATE-----"
+PEM_FOOTER = "-----END CERTIFICATE-----"
+
+def DER_cert_to_PEM_cert(der_cert_bytes):
+
+ """Takes a certificate in binary DER format and returns the
+ PEM version of it as a string."""
+
+ if hasattr(base64, 'standard_b64encode'):
+ # preferred because older API gets line-length wrong
+ f = base64.standard_b64encode(der_cert_bytes)
+ return (PEM_HEADER + '\n' +
+ textwrap.fill(f, 64) + '\n' +
+ PEM_FOOTER + '\n')
+ else:
+ return (PEM_HEADER + '\n' +
+ base64.encodestring(der_cert_bytes) +
+ PEM_FOOTER + '\n')
+
+def PEM_cert_to_DER_cert(pem_cert_string):
+
+ """Takes a certificate in ASCII PEM format and returns the
+ DER-encoded version of it as a byte sequence"""
+
+ if not pem_cert_string.startswith(PEM_HEADER):
+ raise ValueError("Invalid PEM encoding; must start with %s"
+ % PEM_HEADER)
+ if not pem_cert_string.strip().endswith(PEM_FOOTER):
+ raise ValueError("Invalid PEM encoding; must end with %s"
+ % PEM_FOOTER)
+ d = pem_cert_string.strip()[len(PEM_HEADER):-len(PEM_FOOTER)]
+ return base64.decodestring(d)
+
+def get_server_certificate(addr, ssl_version=PROTOCOL_SSLv3, ca_certs=None):
+
+ """Retrieve the certificate from the server at the specified address,
+ and return it as a PEM-encoded string.
+ If 'ca_certs' is specified, validate the server cert against it.
+ If 'ssl_version' is specified, use it in the connection attempt."""
+
+ host, port = addr
+ if (ca_certs is not None):
+ cert_reqs = CERT_REQUIRED
+ else:
+ cert_reqs = CERT_NONE
+ s = wrap_socket(socket(), ssl_version=ssl_version,
+ cert_reqs=cert_reqs, ca_certs=ca_certs)
+ s.connect(addr)
+ dercert = s.getpeercert(True)
+ s.close()
+ return DER_cert_to_PEM_cert(dercert)
+
+def get_protocol_name(protocol_code):
+ if protocol_code == PROTOCOL_TLSv1:
+ return "TLSv1"
+ elif protocol_code == PROTOCOL_SSLv23:
+ return "SSLv23"
+ elif protocol_code == PROTOCOL_SSLv2:
+ return "SSLv2"
+ elif protocol_code == PROTOCOL_SSLv3:
+ return "SSLv3"
+ else:
+ return "<unknown>"
+
+
+# a replacement for the old socket.ssl function
+
+def sslwrap_simple(sock, keyfile=None, certfile=None):
+
+ """A replacement for the old socket.ssl function. Designed
+ for compability with Python 2.5 and earlier. Will disappear in
+ Python 3.0."""
+
+ if hasattr(sock, "_sock"):
+ sock = sock._sock
+
+ ssl_sock = _ssl.sslwrap(sock, 0, keyfile, certfile, CERT_NONE,
+ PROTOCOL_SSLv23, None)
+ try:
+ sock.getpeername()
+ except socket_error:
+ # no, no connection yet
+ pass
+ else:
+ # yes, do the handshake
+ ssl_sock.do_handshake()
+
+ return ssl_sock
diff --git a/lib-python/modified-2.7.0/test/test_ssl.py b/lib-python/modified-2.7.0/test/test_ssl.py
new file mode 100644
--- /dev/null
+++ b/lib-python/modified-2.7.0/test/test_ssl.py
@@ -0,0 +1,1318 @@
+# Test the support for SSL and sockets
+
+import sys
+import unittest
+from test import test_support
+import asyncore
+import socket
+import select
+import time
+import gc
+import os
+import errno
+import pprint
+import urllib, urlparse
+import traceback
+import weakref
+import functools
+import platform
+
+from BaseHTTPServer import HTTPServer
+from SimpleHTTPServer import SimpleHTTPRequestHandler
+
+# Optionally test SSL support, if we have it in the tested platform
+skip_expected = False
+try:
+ import ssl
+except ImportError:
+ skip_expected = True
+
+HOST = test_support.HOST
+CERTFILE = None
+SVN_PYTHON_ORG_ROOT_CERT = None
+
+def handle_error(prefix):
+ exc_format = ' '.join(traceback.format_exception(*sys.exc_info()))
+ if test_support.verbose:
+ sys.stdout.write(prefix + exc_format)
+
+
+class BasicTests(unittest.TestCase):
+
+ def test_sslwrap_simple(self):
+ # A crude test for the legacy API
+ try:
+ ssl.sslwrap_simple(socket.socket(socket.AF_INET))
+ except IOError, e:
+ if e.errno == 32: # broken pipe when ssl_sock.do_handshake(), this test doesn't care about that
+ pass
+ else:
+ raise
+ try:
+ ssl.sslwrap_simple(socket.socket(socket.AF_INET)._sock)
+ except IOError, e:
+ if e.errno == 32: # broken pipe when ssl_sock.do_handshake(), this test doesn't care about that
+ pass
+ else:
+ raise
+
+# Issue #9415: Ubuntu hijacks their OpenSSL and forcefully disables SSLv2
+def skip_if_broken_ubuntu_ssl(func):
+ # We need to access the lower-level wrapper in order to create an
+ # implicit SSL context without trying to connect or listen.
+ try:
+ import _ssl
+ except ImportError:
+ # The returned function won't get executed, just ignore the error
+ pass
+ @functools.wraps(func)
+ def f(*args, **kwargs):
+ try:
+ s = socket.socket(socket.AF_INET)
+ _ssl.sslwrap(s._sock, 0, None, None,
+ ssl.CERT_NONE, ssl.PROTOCOL_SSLv2, None, None)
+ except ssl.SSLError as e:
+ if (ssl.OPENSSL_VERSION_INFO == (0, 9, 8, 15, 15) and
+ platform.linux_distribution() == ('debian', 'squeeze/sid', '')
+ and 'Invalid SSL protocol variant specified' in str(e)):
+ raise unittest.SkipTest("Patched Ubuntu OpenSSL breaks behaviour")
+ return func(*args, **kwargs)
+ return f
+
+
+class BasicSocketTests(unittest.TestCase):
+
+ def test_constants(self):
+ ssl.PROTOCOL_SSLv2
+ ssl.PROTOCOL_SSLv23
+ ssl.PROTOCOL_SSLv3
+ ssl.PROTOCOL_TLSv1
+ ssl.CERT_NONE
+ ssl.CERT_OPTIONAL
+ ssl.CERT_REQUIRED
+
+ def test_random(self):
+ v = ssl.RAND_status()
+ if test_support.verbose:
+ sys.stdout.write("\n RAND_status is %d (%s)\n"
+ % (v, (v and "sufficient randomness") or
+ "insufficient randomness"))
+ try:
+ ssl.RAND_egd(1)
+ except TypeError:
+ pass
+ else:
+ print "didn't raise TypeError"
+ ssl.RAND_add("this is a random string", 75.0)
+
+ @test_support.impl_detail("obscure test")
+ def test_parse_cert(self):
+ # note that this uses an 'unofficial' function in _ssl.c,
+ # provided solely for this test, to exercise the certificate
+ # parsing code
+ p = ssl._ssl._test_decode_cert(CERTFILE, False)
+ if test_support.verbose:
+ sys.stdout.write("\n" + pprint.pformat(p) + "\n")
+
+ def test_DER_to_PEM(self):
+ with open(SVN_PYTHON_ORG_ROOT_CERT, 'r') as f:
+ pem = f.read()
+ d1 = ssl.PEM_cert_to_DER_cert(pem)
+ p2 = ssl.DER_cert_to_PEM_cert(d1)
+ d2 = ssl.PEM_cert_to_DER_cert(p2)
+ self.assertEqual(d1, d2)
+ if not p2.startswith(ssl.PEM_HEADER + '\n'):
+ self.fail("DER-to-PEM didn't include correct header:\n%r\n" % p2)
+ if not p2.endswith('\n' + ssl.PEM_FOOTER + '\n'):
+ self.fail("DER-to-PEM didn't include correct footer:\n%r\n" % p2)
+
+ def test_openssl_version(self):
+ n = ssl.OPENSSL_VERSION_NUMBER
+ t = ssl.OPENSSL_VERSION_INFO
+ s = ssl.OPENSSL_VERSION
+ self.assertIsInstance(n, (int, long))
+ self.assertIsInstance(t, tuple)
+ self.assertIsInstance(s, str)
+ # Some sanity checks follow
+ # >= 0.9
+ self.assertGreaterEqual(n, 0x900000)
+ # < 2.0
+ self.assertLess(n, 0x20000000)
+ major, minor, fix, patch, status = t
+ self.assertGreaterEqual(major, 0)
+ self.assertLess(major, 2)
+ self.assertGreaterEqual(minor, 0)
+ self.assertLess(minor, 256)
+ self.assertGreaterEqual(fix, 0)
+ self.assertLess(fix, 256)
+ self.assertGreaterEqual(patch, 0)
+ self.assertLessEqual(patch, 26)
+ self.assertGreaterEqual(status, 0)
+ self.assertLessEqual(status, 15)
+ # Version string as returned by OpenSSL, the format might change
+ self.assertTrue(s.startswith("OpenSSL {:d}.{:d}.{:d}".format(major, minor, fix)),
+ (s, t))
+
+ def test_ciphers(self):
+ if not test_support.is_resource_enabled('network'):
+ return
+ remote = ("svn.python.org", 443)
+ with test_support.transient_internet(remote[0]):
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_NONE, ciphers="ALL")
+ s.connect(remote)
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_NONE, ciphers="DEFAULT")
+ s.connect(remote)
+ # Error checking occurs when connecting, because the SSL context
+ # isn't created before.
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_NONE, ciphers="^$:,;?*'dorothyx")
+ with self.assertRaisesRegexp(ssl.SSLError, "No cipher can be selected"):
+ s.connect(remote)
+
+ @test_support.cpython_only
+ def test_refcycle(self):
+ # Issue #7943: an SSL object doesn't create reference cycles with
+ # itself.
+ s = socket.socket(socket.AF_INET)
+ ss = ssl.wrap_socket(s)
+ wr = weakref.ref(ss)
+ 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(b'x'))
+ self.assertRaises(socket.error, ss.recvfrom, 1)
+ self.assertRaises(socket.error, ss.recvfrom_into, bytearray(b'x'), 1)
+ self.assertRaises(socket.error, ss.send, b'x')
+ self.assertRaises(socket.error, ss.sendto, b'x', ('0.0.0.0', 0))
+
+
+class NetworkedTests(unittest.TestCase):
+
+ def test_connect(self):
+ with test_support.transient_internet("svn.python.org"):
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_NONE)
+ s.connect(("svn.python.org", 443))
+ c = s.getpeercert()
+ if c:
+ self.fail("Peer cert %s shouldn't be here!")
+ s.close()
+
+ # this should fail because we have no verification certs
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_REQUIRED)
+ try:
+ s.connect(("svn.python.org", 443))
+ except ssl.SSLError:
+ pass
+ finally:
+ s.close()
+
+ # this should succeed because we specify the root cert
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_REQUIRED,
+ ca_certs=SVN_PYTHON_ORG_ROOT_CERT)
+ try:
+ s.connect(("svn.python.org", 443))
+ finally:
+ s.close()
+
+ @unittest.skipIf(os.name == "nt", "Can't use a socket as a file under Windows")
+ def test_makefile_close(self):
+ # Issue #5238: creating a file-like object with makefile() shouldn't
+ # delay closing the underlying "real socket" (here tested with its
+ # file descriptor, hence skipping the test under Windows).
+ with test_support.transient_internet("svn.python.org"):
+ ss = ssl.wrap_socket(socket.socket(socket.AF_INET))
+ ss.connect(("svn.python.org", 443))
+ fd = ss.fileno()
+ f = ss.makefile()
+ f.close()
+ # The fd is still open
+ os.read(fd, 0)
+ # Closing the SSL socket should close the fd too
+ ss.close()
+ gc.collect()
+ with self.assertRaises(OSError) as e:
+ os.read(fd, 0)
+ self.assertEqual(e.exception.errno, errno.EBADF)
+
+ def test_non_blocking_handshake(self):
+ with test_support.transient_internet("svn.python.org"):
+ s = socket.socket(socket.AF_INET)
+ s.connect(("svn.python.org", 443))
+ s.setblocking(False)
+ s = ssl.wrap_socket(s,
+ cert_reqs=ssl.CERT_NONE,
+ do_handshake_on_connect=False)
+ count = 0
+ while True:
+ try:
+ count += 1
+ s.do_handshake()
+ break
+ except ssl.SSLError, err:
+ if err.args[0] == ssl.SSL_ERROR_WANT_READ:
+ select.select([s], [], [])
+ elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
+ select.select([], [s], [])
+ else:
+ raise
+ s.close()
+ if test_support.verbose:
+ sys.stdout.write("\nNeeded %d calls to do_handshake() to establish session.\n" % count)
+
+ def test_get_server_certificate(self):
+ with test_support.transient_internet("svn.python.org"):
+ pem = ssl.get_server_certificate(("svn.python.org", 443))
+ if not pem:
+ self.fail("No server certificate on svn.python.org:443!")
+
+ try:
+ pem = ssl.get_server_certificate(("svn.python.org", 443), ca_certs=CERTFILE)
+ except ssl.SSLError:
+ #should fail
+ pass
+ else:
+ self.fail("Got server certificate %s for svn.python.org!" % pem)
+
+ pem = ssl.get_server_certificate(("svn.python.org", 443), ca_certs=SVN_PYTHON_ORG_ROOT_CERT)
+ if not pem:
+ self.fail("No server certificate on svn.python.org:443!")
+ if test_support.verbose:
+ sys.stdout.write("\nVerified certificate for svn.python.org:443 is\n%s\n" % pem)
+
+ def test_algorithms(self):
+ # Issue #8484: all algorithms should be available when verifying a
+ # certificate.
+ # SHA256 was added in OpenSSL 0.9.8
+ if ssl.OPENSSL_VERSION_INFO < (0, 9, 8, 0, 15):
+ self.skipTest("SHA256 not available on %r" % ssl.OPENSSL_VERSION)
+ # NOTE: https://sha256.tbs-internet.com is another possible test host
+ remote = ("sha2.hboeck.de", 443)
+ sha256_cert = os.path.join(os.path.dirname(__file__), "sha256.pem")
+ with test_support.transient_internet("sha2.hboeck.de"):
+ s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+ cert_reqs=ssl.CERT_REQUIRED,
+ ca_certs=sha256_cert,)
+ try:
+ s.connect(remote)
+ if test_support.verbose:
+ sys.stdout.write("\nCipher with %r is %r\n" %
+ (remote, s.cipher()))
+ sys.stdout.write("Certificate is:\n%s\n" %
+ pprint.pformat(s.getpeercert()))
+ finally:
+ s.close()
+
+
+try:
+ import threading
+except ImportError:
+ _have_threads = False
+else:
+ _have_threads = True
+
+ class ThreadedEchoServer(threading.Thread):
+
+ class ConnectionHandler(threading.Thread):
+
+ """A mildly complicated class, because we want it to work both
+ with and without the SSL wrapper around the socket connection, so
+ that we can test the STARTTLS functionality."""
+
+ def __init__(self, server, connsock):
+ self.server = server
+ self.running = False
+ self.sock = connsock
+ self.sock.setblocking(1)
+ self.sslconn = None
+ threading.Thread.__init__(self)
+ self.daemon = True
+
+ def show_conn_details(self):
+ if self.server.certreqs == ssl.CERT_REQUIRED:
+ cert = self.sslconn.getpeercert()
+ if test_support.verbose and self.server.chatty:
+ sys.stdout.write(" client cert is " + pprint.pformat(cert) + "\n")
+ cert_binary = self.sslconn.getpeercert(True)
+ if test_support.verbose and self.server.chatty:
+ sys.stdout.write(" cert binary is " + str(len(cert_binary)) + " bytes\n")
+ cipher = self.sslconn.cipher()
+ if test_support.verbose and self.server.chatty:
+ sys.stdout.write(" server: connection cipher is now " + str(cipher) + "\n")
+
+ def wrap_conn(self):
+ try:
+ self.sslconn = ssl.wrap_socket(self.sock, server_side=True,
+ certfile=self.server.certificate,
+ ssl_version=self.server.protocol,
+ ca_certs=self.server.cacerts,
+ cert_reqs=self.server.certreqs,
+ ciphers=self.server.ciphers)
+ except ssl.SSLError:
+ # XXX Various errors can have happened here, for example
+ # a mismatching protocol version, an invalid certificate,
+ # or a low-level bug. This should be made more discriminating.
+ if self.server.chatty:
+ handle_error("\n server: bad connection attempt from " +
+ str(self.sock.getpeername()) + ":\n")
+ self.close()
+ self.running = False
+ self.server.stop()
+ return False
+ else:
+ return True
+
+ def read(self):
+ if self.sslconn:
+ return self.sslconn.read()
+ else:
+ return self.sock.recv(1024)
+
+ def write(self, bytes):
+ if self.sslconn:
+ return self.sslconn.write(bytes)
+ else:
+ return self.sock.send(bytes)
+
+ def close(self):
+ if self.sslconn:
+ self.sslconn.close()
+ else:
+ self.sock._sock.close()
+
+ def run(self):
+ self.running = True
+ if not self.server.starttls_server:
+ if isinstance(self.sock, ssl.SSLSocket):
+ self.sslconn = self.sock
+ elif not self.wrap_conn():
+ return
+ self.show_conn_details()
+ while self.running:
+ try:
+ msg = self.read()
+ if not msg:
+ # eof, so quit this handler
+ self.running = False
+ self.close()
+ elif msg.strip() == 'over':
+ if test_support.verbose and self.server.connectionchatty:
+ sys.stdout.write(" server: client closed connection\n")
+ self.close()
+ return
+ elif self.server.starttls_server and msg.strip() == 'STARTTLS':
+ if test_support.verbose and self.server.connectionchatty:
+ sys.stdout.write(" server: read STARTTLS from client, sending OK...\n")
+ self.write("OK\n")
+ if not self.wrap_conn():
+ return
+ elif self.server.starttls_server and self.sslconn and msg.strip() == 'ENDTLS':
+ if test_support.verbose and self.server.connectionchatty:
+ sys.stdout.write(" server: read ENDTLS from client, sending OK...\n")
+ self.write("OK\n")
+ self.sslconn.unwrap()
+ self.sslconn = None
+ if test_support.verbose and self.server.connectionchatty:
+ sys.stdout.write(" server: connection is now unencrypted...\n")
+ else:
+ if (test_support.verbose and
+ self.server.connectionchatty):
+ ctype = (self.sslconn and "encrypted") or "unencrypted"
+ sys.stdout.write(" server: read %s (%s), sending back %s (%s)...\n"
+ % (repr(msg), ctype, repr(msg.lower()), ctype))
+ self.write(msg.lower())
+ except ssl.SSLError:
+ if self.server.chatty:
+ handle_error("Test server failure:\n")
+ self.close()
+ self.running = False
+ # normally, we'd just stop here, but for the test
+ # harness, we want to stop the server
+ self.server.stop()
+
+ def __init__(self, certificate, ssl_version=None,
+ certreqs=None, cacerts=None,
+ chatty=True, connectionchatty=False, starttls_server=False,
+ wrap_accepting_socket=False, ciphers=None):
+
+ if ssl_version is None:
+ ssl_version = ssl.PROTOCOL_TLSv1
+ if certreqs is None:
+ certreqs = ssl.CERT_NONE
+ self.certificate = certificate
+ self.protocol = ssl_version
+ self.certreqs = certreqs
+ self.cacerts = cacerts
+ self.ciphers = ciphers
+ self.chatty = chatty
+ self.connectionchatty = connectionchatty
+ self.starttls_server = starttls_server
+ self.sock = socket.socket()
+ self.flag = None
+ if wrap_accepting_socket:
+ self.sock = ssl.wrap_socket(self.sock, server_side=True,
+ certfile=self.certificate,
+ cert_reqs = self.certreqs,
+ ca_certs = self.cacerts,
+ ssl_version = self.protocol,
+ ciphers = self.ciphers)
+ if test_support.verbose and self.chatty:
+ sys.stdout.write(' server: wrapped server socket as %s\n' % str(self.sock))
+ self.port = test_support.bind_port(self.sock)
+ self.active = False
+ threading.Thread.__init__(self)
+ self.daemon = True
+
+ def start(self, flag=None):
+ self.flag = flag
+ threading.Thread.start(self)
+
+ def run(self):
+ self.sock.settimeout(0.05)
+ self.sock.listen(5)
+ self.active = True
+ if self.flag:
+ # signal an event
+ self.flag.set()
+ while self.active:
+ try:
+ newconn, connaddr = self.sock.accept()
+ if test_support.verbose and self.chatty:
+ sys.stdout.write(' server: new connection from '
+ + str(connaddr) + '\n')
+ handler = self.ConnectionHandler(self, newconn)
+ handler.start()
+ except socket.timeout:
+ pass
+ except KeyboardInterrupt:
+ self.stop()
+ self.sock.close()
+
+ def stop(self):
+ self.active = False
+
+ class AsyncoreEchoServer(threading.Thread):
+
+ class EchoServer(asyncore.dispatcher):
+
+ class ConnectionHandler(asyncore.dispatcher_with_send):
+
+ def __init__(self, conn, certfile):
+ asyncore.dispatcher_with_send.__init__(self, conn)
+ self.socket = ssl.wrap_socket(conn, server_side=True,
+ certfile=certfile,
+ do_handshake_on_connect=False)
+ self._ssl_accepting = True
+
+ def readable(self):
+ if isinstance(self.socket, ssl.SSLSocket):
+ while self.socket.pending() > 0:
+ self.handle_read_event()
+ return True
+
+ def _do_ssl_handshake(self):
+ try:
+ self.socket.do_handshake()
+ except ssl.SSLError, err:
+ if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
+ ssl.SSL_ERROR_WANT_WRITE):
+ return
+ elif err.args[0] == ssl.SSL_ERROR_EOF:
+ return self.handle_close()
+ raise
+ except socket.error, err:
+ if err.args[0] == errno.ECONNABORTED:
+ return self.handle_close()
+ else:
+ self._ssl_accepting = False
+
+ def handle_read(self):
+ if self._ssl_accepting:
+ self._do_ssl_handshake()
+ else:
+ data = self.recv(1024)
+ if data and data.strip() != 'over':
+ self.send(data.lower())
+
+ def handle_close(self):
+ self.close()
+ if test_support.verbose:
+ sys.stdout.write(" server: closed connection %s\n" % self.socket)
+
+ def handle_error(self):
+ raise
+
+ def __init__(self, certfile):
+ self.certfile = certfile
+ asyncore.dispatcher.__init__(self)
+ self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.port = test_support.bind_port(self.socket)
+ self.listen(5)
+
+ def handle_accept(self):
+ sock_obj, addr = self.accept()
+ if test_support.verbose:
+ sys.stdout.write(" server: new connection from %s:%s\n" %addr)
+ self.ConnectionHandler(sock_obj, self.certfile)
+
+ def handle_error(self):
+ raise
+
+ def __init__(self, certfile):
+ self.flag = None
+ self.active = False
+ self.server = self.EchoServer(certfile)
+ self.port = self.server.port
+ threading.Thread.__init__(self)
+ self.daemon = True
+
+ def __str__(self):
+ return "<%s %s>" % (self.__class__.__name__, self.server)
+
+ def start(self, flag=None):
+ self.flag = flag
+ threading.Thread.start(self)
+
+ def run(self):
+ self.active = True
+ if self.flag:
+ self.flag.set()
+ while self.active:
+ asyncore.loop(0.05)
+
+ def stop(self):
+ self.active = False
+ self.server.close()
+
+ class SocketServerHTTPSServer(threading.Thread):
+
+ class HTTPSServer(HTTPServer):
+
+ def __init__(self, server_address, RequestHandlerClass, certfile):
+ HTTPServer.__init__(self, server_address, RequestHandlerClass)
+ # we assume the certfile contains both private key and certificate
+ self.certfile = certfile
+ self.allow_reuse_address = True
+
+ def __str__(self):
+ return ('<%s %s:%s>' %
+ (self.__class__.__name__,
+ self.server_name,
+ self.server_port))
+
+ def get_request(self):
+ # override this to wrap socket with SSL
+ sock, addr = self.socket.accept()
+ sslconn = ssl.wrap_socket(sock, server_side=True,
+ certfile=self.certfile)
+ return sslconn, addr
+
+ class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
+ # need to override translate_path to get a known root,
+ # instead of using os.curdir, since the test could be
+ # run from anywhere
+
+ server_version = "TestHTTPS/1.0"
+
+ root = None
+
+ def translate_path(self, path):
+ """Translate a /-separated PATH to the local filename syntax.
+
+ Components that mean special things to the local file system
+ (e.g. drive or directory names) are ignored. (XXX They should
+ probably be diagnosed.)
+
+ """
+ # abandon query parameters
+ path = urlparse.urlparse(path)[2]
+ path = os.path.normpath(urllib.unquote(path))
+ words = path.split('/')
+ words = filter(None, words)
+ path = self.root
+ for word in words:
+ drive, word = os.path.splitdrive(word)
+ head, word = os.path.split(word)
+ if word in self.root: continue
+ path = os.path.join(path, word)
+ return path
+
+ def log_message(self, format, *args):
+
+ # we override this to suppress logging unless "verbose"
+
+ if test_support.verbose:
+ sys.stdout.write(" server (%s:%d %s):\n [%s] %s\n" %
+ (self.server.server_address,
+ self.server.server_port,
+ self.request.cipher(),
+ self.log_date_time_string(),
+ format%args))
+
+
+ def __init__(self, certfile):
+ self.flag = None
+ self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0]
+ self.server = self.HTTPSServer(
+ (HOST, 0), self.RootedHTTPRequestHandler, certfile)
+ self.port = self.server.server_port
+ threading.Thread.__init__(self)
+ self.daemon = True
+
+ def __str__(self):
+ return "<%s %s>" % (self.__class__.__name__, self.server)
+
+ def start(self, flag=None):
+ self.flag = flag
+ threading.Thread.start(self)
+
+ def run(self):
+ if self.flag:
+ self.flag.set()
+ self.server.serve_forever(0.05)
+
+ def stop(self):
+ self.server.shutdown()
+
+
+ def bad_cert_test(certfile):
+ """
+ Launch a server with CERT_REQUIRED, and check that trying to
+ connect to it with the given client certificate fails.
+ """
+ server = ThreadedEchoServer(CERTFILE,
+ certreqs=ssl.CERT_REQUIRED,
+ cacerts=CERTFILE, chatty=False)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ try:
+ try:
+ s = ssl.wrap_socket(socket.socket(),
+ certfile=certfile,
+ ssl_version=ssl.PROTOCOL_TLSv1)
+ s.connect((HOST, server.port))
+ except ssl.SSLError, x:
+ if test_support.verbose:
+ sys.stdout.write("\nSSLError is %s\n" % x[1])
+ except socket.error, x:
+ if test_support.verbose:
+ sys.stdout.write("\nsocket.error is %s\n" % x[1])
+ else:
+ raise AssertionError("Use of invalid cert should have failed!")
+ finally:
+ server.stop()
+ server.join()
+
+ def server_params_test(certfile, protocol, certreqs, cacertsfile,
+ client_certfile, client_protocol=None, indata="FOO\n",
+ ciphers=None, chatty=True, connectionchatty=False,
+ wrap_accepting_socket=False):
+ """
+ Launch a server, connect a client to it and try various reads
+ and writes.
+ """
+ server = ThreadedEchoServer(certfile,
+ certreqs=certreqs,
+ ssl_version=protocol,
+ cacerts=cacertsfile,
+ ciphers=ciphers,
+ chatty=chatty,
+ connectionchatty=connectionchatty,
+ wrap_accepting_socket=wrap_accepting_socket)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ if client_protocol is None:
+ client_protocol = protocol
+ try:
+ s = ssl.wrap_socket(socket.socket(),
+ certfile=client_certfile,
+ ca_certs=cacertsfile,
+ ciphers=ciphers,
+ cert_reqs=certreqs,
+ ssl_version=client_protocol)
+ s.connect((HOST, server.port))
+ for arg in [indata, bytearray(indata), memoryview(indata)]:
+ if connectionchatty:
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: sending %s...\n" % (repr(arg)))
+ s.write(arg)
+ outdata = s.read()
+ if connectionchatty:
+ if test_support.verbose:
+ sys.stdout.write(" client: read %s\n" % repr(outdata))
+ if outdata != indata.lower():
+ raise AssertionError(
+ "bad data <<%s>> (%d) received; expected <<%s>> (%d)\n"
+ % (outdata[:min(len(outdata),20)], len(outdata),
+ indata[:min(len(indata),20)].lower(), len(indata)))
+ s.write("over\n")
+ if connectionchatty:
+ if test_support.verbose:
+ sys.stdout.write(" client: closing connection.\n")
+ s.close()
+ finally:
+ server.stop()
+ server.join()
+
+ def try_protocol_combo(server_protocol,
+ client_protocol,
+ expect_success,
+ certsreqs=None):
+ if certsreqs is None:
+ certsreqs = ssl.CERT_NONE
+ certtype = {
+ ssl.CERT_NONE: "CERT_NONE",
+ ssl.CERT_OPTIONAL: "CERT_OPTIONAL",
+ ssl.CERT_REQUIRED: "CERT_REQUIRED",
+ }[certsreqs]
+ if test_support.verbose:
+ formatstr = (expect_success and " %s->%s %s\n") or " {%s->%s} %s\n"
+ sys.stdout.write(formatstr %
+ (ssl.get_protocol_name(client_protocol),
+ ssl.get_protocol_name(server_protocol),
+ certtype))
+ try:
+ # NOTE: we must enable "ALL" ciphers, otherwise an SSLv23 client
+ # will send an SSLv3 hello (rather than SSLv2) starting from
+ # OpenSSL 1.0.0 (see issue #8322).
+ server_params_test(CERTFILE, server_protocol, certsreqs,
+ CERTFILE, CERTFILE, client_protocol,
+ ciphers="ALL", chatty=False)
+ # Protocol mismatch can result in either an SSLError, or a
+ # "Connection reset by peer" error.
+ except ssl.SSLError:
+ if expect_success:
+ raise
+ except socket.error as e:
+ if expect_success or e.errno != errno.ECONNRESET:
+ raise
+ else:
+ if not expect_success:
+ raise AssertionError(
+ "Client protocol %s succeeded with server protocol %s!"
+ % (ssl.get_protocol_name(client_protocol),
+ ssl.get_protocol_name(server_protocol)))
+
+
+ class ThreadedTests(unittest.TestCase):
+
+ def test_rude_shutdown(self):
+ """A brutal shutdown of an SSL server should raise an IOError
+ in the client when attempting handshake.
+ """
+ listener_ready = threading.Event()
+ listener_gone = threading.Event()
+
+ s = socket.socket()
+ port = test_support.bind_port(s, HOST)
+
+ # `listener` runs in a thread. It sits in an accept() until
+ # the main thread connects. Then it rudely closes the socket,
+ # and sets Event `listener_gone` to let the main thread know
+ # the socket is gone.
+ def listener():
+ s.listen(5)
+ listener_ready.set()
+ s.accept()
+ s.close()
+ listener_gone.set()
+
+ def connector():
+ listener_ready.wait()
+ c = socket.socket()
+ c.connect((HOST, port))
+ listener_gone.wait()
+ try:
+ ssl_sock = ssl.wrap_socket(c)
+ except IOError:
+ pass
+ else:
+ self.fail('connecting to closed SSL socket should have failed')
+
+ t = threading.Thread(target=listener)
+ t.start()
+ try:
+ connector()
+ finally:
+ t.join()
+
+ @skip_if_broken_ubuntu_ssl
+ def test_echo(self):
+ """Basic test of an SSL client connecting to a server"""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ server_params_test(CERTFILE, ssl.PROTOCOL_TLSv1, ssl.CERT_NONE,
+ CERTFILE, CERTFILE, ssl.PROTOCOL_TLSv1,
+ chatty=True, connectionchatty=True)
+
+ def test_getpeercert(self):
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ s2 = socket.socket()
+ server = ThreadedEchoServer(CERTFILE,
+ certreqs=ssl.CERT_NONE,
+ ssl_version=ssl.PROTOCOL_SSLv23,
+ cacerts=CERTFILE,
+ chatty=False)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ try:
+ s = ssl.wrap_socket(socket.socket(),
+ certfile=CERTFILE,
+ ca_certs=CERTFILE,
+ cert_reqs=ssl.CERT_REQUIRED,
+ ssl_version=ssl.PROTOCOL_SSLv23)
+ s.connect((HOST, server.port))
+ cert = s.getpeercert()
+ self.assertTrue(cert, "Can't get peer certificate.")
+ cipher = s.cipher()
+ if test_support.verbose:
+ sys.stdout.write(pprint.pformat(cert) + '\n')
+ sys.stdout.write("Connection cipher is " + str(cipher) + '.\n')
+ if 'subject' not in cert:
+ self.fail("No subject field in certificate: %s." %
+ pprint.pformat(cert))
+ if ((('organizationName', 'Python Software Foundation'),)
+ not in cert['subject']):
+ self.fail(
+ "Missing or invalid 'organizationName' field in certificate subject; "
+ "should be 'Python Software Foundation'.")
+ s.close()
+ finally:
+ server.stop()
+ server.join()
+
+ def test_empty_cert(self):
+ """Connecting with an empty cert file"""
+ bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
+ "nullcert.pem"))
+ def test_malformed_cert(self):
+ """Connecting with a badly formatted certificate (syntax error)"""
+ bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
+ "badcert.pem"))
+ def test_nonexisting_cert(self):
+ """Connecting with a non-existing cert file"""
+ bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
+ "wrongcert.pem"))
+ def test_malformed_key(self):
+ """Connecting with a badly formatted key (syntax error)"""
+ bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,
+ "badkey.pem"))
+
+ @skip_if_broken_ubuntu_ssl
+ def test_protocol_sslv2(self):
+ """Connecting to an SSLv2 server with various client options"""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True)
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_OPTIONAL)
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_REQUIRED)
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv23, True)
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv3, False)
+ try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLSv1, False)
+
+ @skip_if_broken_ubuntu_ssl
+ def test_protocol_sslv23(self):
+ """Connecting to an SSLv23 server with various client options"""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ try:
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv2, True)
+ except (ssl.SSLError, socket.error), x:
+ # this fails on some older versions of OpenSSL (0.9.7l, for instance)
+ if test_support.verbose:
+ sys.stdout.write(
+ " SSL2 client to SSL23 server test unexpectedly failed:\n %s\n"
+ % str(x))
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True)
+
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_OPTIONAL)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
+
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23, True, ssl.CERT_REQUIRED)
+ try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
+
+ @skip_if_broken_ubuntu_ssl
+ def test_protocol_sslv3(self):
+ """Connecting to an SSLv3 server with various client options"""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True)
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_OPTIONAL)
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, True, ssl.CERT_REQUIRED)
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv2, False)
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv23, False)
+ try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLSv1, False)
+
+ @skip_if_broken_ubuntu_ssl
+ def test_protocol_tlsv1(self):
+ """Connecting to a TLSv1 server with various client options"""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True)
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_OPTIONAL)
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, True, ssl.CERT_REQUIRED)
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv2, False)
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False)
+ try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv23, False)
+
+ def test_starttls(self):
+ """Switching from clear text to encrypted and back again."""
+ msgs = ("msg 1", "MSG 2", "STARTTLS", "MSG 3", "msg 4", "ENDTLS", "msg 5", "msg 6")
+
+ server = ThreadedEchoServer(CERTFILE,
+ ssl_version=ssl.PROTOCOL_TLSv1,
+ starttls_server=True,
+ chatty=True,
+ connectionchatty=True)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ wrapped = False
+ try:
+ s = socket.socket()
+ s.setblocking(1)
+ s.connect((HOST, server.port))
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ for indata in msgs:
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: sending %s...\n" % repr(indata))
+ if wrapped:
+ conn.write(indata)
+ outdata = conn.read()
+ else:
+ s.send(indata)
+ outdata = s.recv(1024)
+ if (indata == "STARTTLS" and
+ outdata.strip().lower().startswith("ok")):
+ # STARTTLS ok, switch to secure mode
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: read %s from server, starting TLS...\n"
+ % repr(outdata))
+ conn = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_TLSv1)
+ wrapped = True
+ elif (indata == "ENDTLS" and
+ outdata.strip().lower().startswith("ok")):
+ # ENDTLS ok, switch back to clear text
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: read %s from server, ending TLS...\n"
+ % repr(outdata))
+ s = conn.unwrap()
+ wrapped = False
+ else:
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: read %s from server\n" % repr(outdata))
+ if test_support.verbose:
+ sys.stdout.write(" client: closing connection.\n")
+ if wrapped:
+ conn.write("over\n")
+ else:
+ s.send("over\n")
+ s.close()
+ finally:
+ server.stop()
+ server.join()
+
+ def test_socketserver(self):
+ """Using a SocketServer to create and manage SSL connections."""
+ server = SocketServerHTTPSServer(CERTFILE)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ try:
+ if test_support.verbose:
+ sys.stdout.write('\n')
+ with open(CERTFILE, 'rb') as f:
+ d1 = f.read()
+ d2 = ''
+ # now fetch the same data from the HTTPS server
+ url = 'https://127.0.0.1:%d/%s' % (
+ server.port, os.path.split(CERTFILE)[1])
+ with test_support.check_py3k_warnings():
+ f = urllib.urlopen(url)
+ dlen = f.info().getheader("content-length")
+ if dlen and (int(dlen) > 0):
+ d2 = f.read(int(dlen))
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: read %d bytes from remote server '%s'\n"
+ % (len(d2), server))
+ f.close()
+ self.assertEqual(d1, d2)
+ finally:
+ server.stop()
+ server.join()
+
+ def test_wrapped_accept(self):
+ """Check the accept() method on SSL sockets."""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ server_params_test(CERTFILE, ssl.PROTOCOL_SSLv23, ssl.CERT_REQUIRED,
+ CERTFILE, CERTFILE, ssl.PROTOCOL_SSLv23,
+ chatty=True, connectionchatty=True,
+ wrap_accepting_socket=True)
+
+ def test_asyncore_server(self):
+ """Check the example asyncore integration."""
+ indata = "TEST MESSAGE of mixed case\n"
+
+ if test_support.verbose:
+ sys.stdout.write("\n")
+ server = AsyncoreEchoServer(CERTFILE)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ try:
+ s = ssl.wrap_socket(socket.socket())
+ s.connect(('127.0.0.1', server.port))
+ if test_support.verbose:
+ sys.stdout.write(
+ " client: sending %s...\n" % (repr(indata)))
+ s.write(indata)
+ outdata = s.read()
+ if test_support.verbose:
+ sys.stdout.write(" client: read %s\n" % repr(outdata))
+ if outdata != indata.lower():
+ self.fail(
+ "bad data <<%s>> (%d) received; expected <<%s>> (%d)\n"
+ % (outdata[:min(len(outdata),20)], len(outdata),
+ indata[:min(len(indata),20)].lower(), len(indata)))
+ s.write("over\n")
+ if test_support.verbose:
+ sys.stdout.write(" client: closing connection.\n")
+ s.close()
+ finally:
+ server.stop()
+ # wait for server thread to end
+ server.join()
+
+ def test_recv_send(self):
+ """Test recv(), send() and friends."""
+ if test_support.verbose:
+ sys.stdout.write("\n")
+
+ server = ThreadedEchoServer(CERTFILE,
+ certreqs=ssl.CERT_NONE,
+ ssl_version=ssl.PROTOCOL_TLSv1,
+ cacerts=CERTFILE,
+ chatty=True,
+ connectionchatty=False)
+ flag = threading.Event()
+ server.start(flag)
+ # wait for it to start
+ flag.wait()
+ # try to connect
+ s = ssl.wrap_socket(socket.socket(),
+ server_side=False,
+ certfile=CERTFILE,
+ ca_certs=CERTFILE,
+ cert_reqs=ssl.CERT_NONE,
+ ssl_version=ssl.PROTOCOL_TLSv1)
+ s.connect((HOST, server.port))
+ try:
+ # helper methods for standardising recv* method signatures
+ def _recv_into():
+ b = bytearray("\0"*100)
+ count = s.recv_into(b)
+ return b[:count]
+
+ def _recvfrom_into():
+ b = bytearray("\0"*100)
+ count, addr = s.recvfrom_into(b)
+ return b[:count]
+
+ # (name, method, whether to expect success, *args)
+ send_methods = [
+ ('send', s.send, True, []),
+ ('sendto', s.sendto, False, ["some.address"]),
+ ('sendall', s.sendall, True, []),
+ ]
+ recv_methods = [
+ ('recv', s.recv, True, []),
+ ('recvfrom', s.recvfrom, False, ["some.address"]),
+ ('recv_into', _recv_into, True, []),
+ ('recvfrom_into', _recvfrom_into, False, []),
+ ]
+ data_prefix = u"PREFIX_"
+
+ for meth_name, send_meth, expect_success, args in send_methods:
+ indata = data_prefix + meth_name
+ try:
+ send_meth(indata.encode('ASCII', 'strict'), *args)
+ outdata = s.read()
+ outdata = outdata.decode('ASCII', 'strict')
+ if outdata != indata.lower():
+ self.fail(
+ "While sending with <<%s>> bad data "
+ "<<%r>> (%d) received; "
+ "expected <<%r>> (%d)\n" % (
+ meth_name, outdata[:20], len(outdata),
+ indata[:20], len(indata)
+ )
+ )
+ except ValueError as e:
+ if expect_success:
+ self.fail(
+ "Failed to send with method <<%s>>; "
+ "expected to succeed.\n" % (meth_name,)
+ )
+ if not str(e).startswith(meth_name):
+ self.fail(
+ "Method <<%s>> failed with unexpected "
+ "exception message: %s\n" % (
+ meth_name, e
+ )
+ )
+
+ for meth_name, recv_meth, expect_success, args in recv_methods:
+ indata = data_prefix + meth_name
+ try:
+ s.send(indata.encode('ASCII', 'strict'))
+ outdata = recv_meth(*args)
+ outdata = outdata.decode('ASCII', 'strict')
+ if outdata != indata.lower():
+ self.fail(
+ "While receiving with <<%s>> bad data "
+ "<<%r>> (%d) received; "
+ "expected <<%r>> (%d)\n" % (
+ meth_name, outdata[:20], len(outdata),
+ indata[:20], len(indata)
+ )
+ )
+ except ValueError as e:
+ if expect_success:
+ self.fail(
+ "Failed to receive with method <<%s>>; "
+ "expected to succeed.\n" % (meth_name,)
+ )
+ if not str(e).startswith(meth_name):
+ self.fail(
+ "Method <<%s>> failed with unexpected "
+ "exception message: %s\n" % (
+ meth_name, e
+ )
+ )
+ # consume data
+ s.read()
+
+ s.write("over\n".encode("ASCII", "strict"))
+ s.close()
+ finally:
+ server.stop()
+ server.join()
+
+ def test_handshake_timeout(self):
+ # Issue #5103: SSL handshake must respect the socket timeout
+ server = socket.socket(socket.AF_INET)
+ host = "127.0.0.1"
+ port = test_support.bind_port(server)
+ started = threading.Event()
+ finish = False
+
+ def serve():
+ server.listen(5)
+ started.set()
+ conns = []
+ while not finish:
+ r, w, e = select.select([server], [], [], 0.1)
+ if server in r:
+ # Let the socket hang around rather than having
+ # it closed by garbage collection.
+ conns.append(server.accept()[0])
+
+ t = threading.Thread(target=serve)
+ t.start()
+ started.wait()
+
+ try:
+ try:
+ c = socket.socket(socket.AF_INET)
+ c.settimeout(0.2)
+ c.connect((host, port))
+ # Will attempt handshake and time out
+ self.assertRaisesRegexp(ssl.SSLError, "timed out",
+ ssl.wrap_socket, c)
+ finally:
+ c.close()
+ try:
+ c = socket.socket(socket.AF_INET)
+ c.settimeout(0.2)
+ c = ssl.wrap_socket(c)
+ # Will attempt handshake and time out
+ self.assertRaisesRegexp(ssl.SSLError, "timed out",
+ c.connect, (host, port))
+ finally:
+ c.close()
+ finally:
+ finish = True
+ t.join()
+ server.close()
+
+
+def test_main(verbose=False):
+ if skip_expected:
+ raise unittest.SkipTest("No SSL support")
+
+ global CERTFILE, SVN_PYTHON_ORG_ROOT_CERT
+ CERTFILE = test_support.findfile("keycert.pem")
+ SVN_PYTHON_ORG_ROOT_CERT = test_support.findfile(
+ "https_svn_python_org_root.pem")
+
+ if (not os.path.exists(CERTFILE) or
+ not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT)):
+ raise test_support.TestFailed("Can't read certificate files!")
+
+ tests = [BasicTests, BasicSocketTests]
+
+ if test_support.is_resource_enabled('network'):
+ tests.append(NetworkedTests)
+
+ if _have_threads:
+ thread_info = test_support.threading_setup()
+ if thread_info and test_support.is_resource_enabled('network'):
+ tests.append(ThreadedTests)
+
+ try:
+ test_support.run_unittest(*tests)
+ finally:
+ if _have_threads:
+ test_support.threading_cleanup(*thread_info)
+
+if __name__ == "__main__":
+ test_main()
diff --git a/pypy/jit/tl/pypyjit_demo.py b/pypy/jit/tl/pypyjit_demo.py
--- a/pypy/jit/tl/pypyjit_demo.py
+++ b/pypy/jit/tl/pypyjit_demo.py
@@ -1,19 +1,16 @@
try:
- import pypyjit
- pypyjit.set_param(threshold=3, inlining=True)
+ class C:
+ def __call__(self):
+ pass
- def sqrt(y, n=10000):
- x = y / 2
- while n > 0:
- #assert y > 0 and x > 0
- if y > 0 and x > 0: pass
- n -= 1
- x = (x + y/x) / 2
- return x
-
- print sqrt(1234, 4)
-
+ def f():
+ i = 0
+ c = C()
+ while i < 10:
+ c()
+ i += 1
+ f()
except Exception, e:
print "Exception: ", type(e)
print e
More information about the Pypy-commit
mailing list