[pypy-svn] pypy default: Backed out changeset f7e470d7823c
fijal
commits-noreply at bitbucket.org
Tue Feb 1 13:26:29 CET 2011
Author: Maciej Fijalkowski <fijall at gmail.com>
Branch:
Changeset: r41518:0872e654e4d7
Date: 2011-02-01 14:25 +0200
http://bitbucket.org/pypy/pypy/changeset/0872e654e4d7/
Log: Backed out changeset f7e470d7823c
diff --git a/pypy/interpreter/gateway.py b/pypy/interpreter/gateway.py
--- a/pypy/interpreter/gateway.py
+++ b/pypy/interpreter/gateway.py
@@ -344,9 +344,8 @@
self.nextarg()))
def visit__ObjSpace(self, el):
- if self.finger not in [0, 1]:
+ if self.finger != 0:
raise FastFuncNotSupported
- self.args.append('space')
self.unwrap.append("space")
def visit__W_Root(self, el):
@@ -392,9 +391,7 @@
unwrap_info = UnwrapSpec_FastFunc_Unwrap()
unwrap_info.apply_over(unwrap_spec)
narg = unwrap_info.n
- args = unwrap_info.args
- if 'space' not in args:
- args = ['space'] + args
+ args = ['space'] + unwrap_info.args
if args == unwrap_info.unwrap:
fastfunc = func
else:
@@ -531,12 +528,8 @@
self.__class__ = BuiltinCodePassThroughArguments1
self.func__args__ = func
else:
- 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)
+ self.__class__ = globals()['BuiltinCode%d' % arity]
+ setattr(self, 'fastfunc_%d' % arity, fastfunc)
def descr__reduce__(self, space):
from pypy.interpreter.mixedmodule import MixedModule
@@ -724,78 +717,6 @@
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(object):
+class TestBuiltinCode:
def test_signature(self):
def c(space, w_x, w_y, hello_w):
pass
@@ -545,42 +545,7 @@
""")
assert space.is_true(w_res)
- 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]
+ 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
deleted file mode 100644
--- a/lib-python/modified-2.7.0/ssl.py
+++ /dev/null
@@ -1,437 +0,0 @@
-# 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
deleted file mode 100644
--- a/lib-python/modified-2.7.0/test/test_ssl.py
+++ /dev/null
@@ -1,1318 +0,0 @@
-# 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,16 +1,19 @@
try:
- class C:
- def __call__(self):
- pass
+ import pypyjit
+ pypyjit.set_param(threshold=3, inlining=True)
- def f():
- i = 0
- c = C()
- while i < 10:
- c()
- i += 1
- f()
+ 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)
+
except Exception, e:
print "Exception: ", type(e)
print e
More information about the Pypy-commit
mailing list