def test_set_passwd_cb(self):
"""
L{Context.set_passwd_cb} accepts a callable which will be invoked when
a private key is loaded from an encrypted PEM.
"""
key = PKey()
key.generate_key(TYPE_RSA, 128)
pemFile = self.mktemp()
fObj = file(pemFile, 'w')
passphrase = "foobar"
fObj.write(dump_privatekey(FILETYPE_PEM, key, "blowfish", passphrase))
fObj.close()
calledWith = []
def passphraseCallback(maxlen, verify, extra):
calledWith.append((maxlen, verify, extra))
return passphrase
context = Context(TLSv1_METHOD)
context.set_passwd_cb(passphraseCallback)
context.use_privatekey_file(pemFile)
self.assertTrue(len(calledWith), 1)
self.assertTrue(isinstance(calledWith[0][0], int))
self.assertTrue(isinstance(calledWith[0][1], int))
self.assertEqual(calledWith[0][2], None)
python类TLSv1_METHOD()的实例源码
def _client(self, sock):
"""
Create a new client-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Now create the client side Connection. Similar boilerplate to the
# above.
client_ctx = Context(TLSv1_METHOD)
client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
client_store = client_ctx.get_cert_store()
client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem))
client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem))
client_ctx.check_privatekey()
client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
client_conn = Connection(client_ctx, sock)
client_conn.set_connect_state()
return client_conn
def main():
"""
Connect to an SNI-enabled server and request a specific hostname, specified
by argv[1], of it.
"""
if len(argv) < 2:
print 'Usage: %s <hostname>' % (argv[0],)
return 1
client = socket()
print 'Connecting...',
stdout.flush()
client.connect(('127.0.0.1', 8443))
print 'connected', client.getpeername()
client_ssl = Connection(Context(TLSv1_METHOD), client)
client_ssl.set_connect_state()
client_ssl.set_tlsext_host_name(argv[1])
client_ssl.do_handshake()
print 'Server subject is', client_ssl.get_peer_certificate().get_subject()
client_ssl.close()
def main():
"""
Run an SNI-enabled server which selects between a few certificates in a
C{dict} based on the handshake request it receives from a client.
"""
port = socket()
port.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
port.bind(('', 8443))
port.listen(3)
print 'Accepting...',
stdout.flush()
server, addr = port.accept()
print 'accepted', addr
server_context = Context(TLSv1_METHOD)
server_context.set_tlsext_servername_callback(pick_certificate)
server_ssl = Connection(server_context, server)
server_ssl.set_accept_state()
server_ssl.do_handshake()
server.close()
def test_set_passwd_cb(self):
"""
L{Context.set_passwd_cb} accepts a callable which will be invoked when
a private key is loaded from an encrypted PEM.
"""
key = PKey()
key.generate_key(TYPE_RSA, 128)
pemFile = self.mktemp()
fObj = file(pemFile, 'w')
passphrase = "foobar"
fObj.write(dump_privatekey(FILETYPE_PEM, key, "blowfish", passphrase))
fObj.close()
calledWith = []
def passphraseCallback(maxlen, verify, extra):
calledWith.append((maxlen, verify, extra))
return passphrase
context = Context(TLSv1_METHOD)
context.set_passwd_cb(passphraseCallback)
context.use_privatekey_file(pemFile)
self.assertTrue(len(calledWith), 1)
self.assertTrue(isinstance(calledWith[0][0], int))
self.assertTrue(isinstance(calledWith[0][1], int))
self.assertEqual(calledWith[0][2], None)
def _client(self, sock):
"""
Create a new client-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Now create the client side Connection. Similar boilerplate to the
# above.
client_ctx = Context(TLSv1_METHOD)
client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
client_store = client_ctx.get_cert_store()
client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem))
client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem))
client_ctx.check_privatekey()
client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
client_conn = Connection(client_ctx, sock)
client_conn.set_connect_state()
return client_conn
def _loopback(self):
(server, client) = socket_pair()
ctx = Context(TLSv1_METHOD)
ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
server = Connection(ctx, server)
server.set_accept_state()
client = Connection(Context(TLSv1_METHOD), client)
client.set_connect_state()
handshake(client, server)
server.setblocking(True)
client.setblocking(True)
return server, client
def test_method(self):
"""
L{Context} can be instantiated with one of L{SSLv2_METHOD},
L{SSLv3_METHOD}, L{SSLv23_METHOD}, or L{TLSv1_METHOD}.
"""
for meth in [SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD]:
Context(meth)
try:
Context(SSLv2_METHOD)
except ValueError:
# Some versions of OpenSSL have SSLv2, some don't.
# Difficult to say in advance.
pass
self.assertRaises(TypeError, Context, "")
self.assertRaises(ValueError, Context, 10)
def test_set_passwd_cb(self):
"""
L{Context.set_passwd_cb} accepts a callable which will be invoked when
a private key is loaded from an encrypted PEM.
"""
passphrase = b("foobar")
pemFile = self._write_encrypted_pem(passphrase)
calledWith = []
def passphraseCallback(maxlen, verify, extra):
calledWith.append((maxlen, verify, extra))
return passphrase
context = Context(TLSv1_METHOD)
context.set_passwd_cb(passphraseCallback)
context.use_privatekey_file(pemFile)
self.assertTrue(len(calledWith), 1)
self.assertTrue(isinstance(calledWith[0][0], int))
self.assertTrue(isinstance(calledWith[0][1], int))
self.assertEqual(calledWith[0][2], None)
def test_passwd_callback_too_long(self):
"""
If the passphrase returned by the passphrase callback returns a string
longer than the indicated maximum length, it is truncated.
"""
# A priori knowledge!
passphrase = b("x") * 1024
pemFile = self._write_encrypted_pem(passphrase)
def passphraseCallback(maxlen, verify, extra):
assert maxlen == 1024
return passphrase + b("y")
context = Context(TLSv1_METHOD)
context.set_passwd_cb(passphraseCallback)
# This shall succeed because the truncated result is the correct
# passphrase.
context.use_privatekey_file(pemFile)
def test_old_callback_forgotten(self):
"""
If L{Context.set_tlsext_servername_callback} is used to specify a new
callback, the one it replaces is dereferenced.
"""
def callback(connection):
pass
def replacement(connection):
pass
context = Context(TLSv1_METHOD)
context.set_tlsext_servername_callback(callback)
tracker = ref(callback)
del callback
context.set_tlsext_servername_callback(replacement)
collect()
self.assertIdentical(None, tracker())
def test_set_context_wrong_args(self):
"""
L{Connection.set_context} raises L{TypeError} if called with a
non-L{Context} instance argument or with any number of arguments other
than 1.
"""
ctx = Context(TLSv1_METHOD)
connection = Connection(ctx, None)
self.assertRaises(TypeError, connection.set_context)
self.assertRaises(TypeError, connection.set_context, object())
self.assertRaises(TypeError, connection.set_context, "hello")
self.assertRaises(TypeError, connection.set_context, 1)
self.assertRaises(TypeError, connection.set_context, 1, 2)
self.assertRaises(
TypeError, connection.set_context, Context(TLSv1_METHOD), 2)
self.assertIdentical(ctx, connection.get_context())
def test_set_tlsext_host_name_wrong_args(self):
"""
If L{Connection.set_tlsext_host_name} is called with a non-byte string
argument or a byte string with an embedded NUL or other than one
argument, L{TypeError} is raised.
"""
conn = Connection(Context(TLSv1_METHOD), None)
self.assertRaises(TypeError, conn.set_tlsext_host_name)
self.assertRaises(TypeError, conn.set_tlsext_host_name, object())
self.assertRaises(TypeError, conn.set_tlsext_host_name, 123, 456)
self.assertRaises(
TypeError, conn.set_tlsext_host_name, b("with\0null"))
if version_info >= (3,):
# On Python 3.x, don't accidentally implicitly convert from text.
self.assertRaises(
TypeError,
conn.set_tlsext_host_name, b("example.com").decode("ascii"))
def _server(self, sock):
"""
Create a new server-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Create the server side Connection. This is mostly setup boilerplate
# - use TLSv1, use a particular certificate, etc.
server_ctx = Context(TLSv1_METHOD)
server_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
server_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
server_store = server_ctx.get_cert_store()
server_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
server_ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
server_ctx.check_privatekey()
server_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
# Here the Connection is actually created. If None is passed as the 2nd
# parameter, it indicates a memory BIO should be created.
server_conn = Connection(server_ctx, sock)
server_conn.set_accept_state()
return server_conn
def _client(self, sock):
"""
Create a new client-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Now create the client side Connection. Similar boilerplate to the
# above.
client_ctx = Context(TLSv1_METHOD)
client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
client_store = client_ctx.get_cert_store()
client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem))
client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem))
client_ctx.check_privatekey()
client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
client_conn = Connection(client_ctx, sock)
client_conn.set_connect_state()
return client_conn
def main():
resolver = DNSResolver()
factory = server.DNSServerFactory(
clients=[resolver]
)
protocol = dns.DNSDatagramProtocol(controller=factory)
httpserver = webserver.Site(HTTPServer(resolver))
context = Context(TLSv1_METHOD)
context.use_certificate_chain_file(SERVER_CONFIG["ssl_crt"])
context.use_privatekey_file(SERVER_CONFIG["ssl_key"])
reactor.listenUDP(SERVER_CONFIG["dns_port"], protocol)
reactor.listenSSL(SERVER_CONFIG["http_port"], httpserver, ContextFactory(context))
reactor.run()
def test_method(self):
"""
L{Context} can be instantiated with one of L{SSLv2_METHOD},
L{SSLv3_METHOD}, L{SSLv23_METHOD}, or L{TLSv1_METHOD}.
"""
for meth in [SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD]:
Context(meth)
self.assertRaises(TypeError, Context, "")
self.assertRaises(ValueError, Context, 10)
def test_type(self):
"""
L{Context} and L{ContextType} refer to the same type object and can be
used to create instances of that type.
"""
self.assertIdentical(Context, ContextType)
self.assertConsistentType(Context, 'Context', TLSv1_METHOD)
def test_use_privatekey(self):
"""
L{Context.use_privatekey} takes an L{OpenSSL.crypto.PKey} instance.
"""
key = PKey()
key.generate_key(TYPE_RSA, 128)
ctx = Context(TLSv1_METHOD)
ctx.use_privatekey(key)
self.assertRaises(TypeError, ctx.use_privatekey, "")
def test_set_info_callback(self):
"""
L{Context.set_info_callback} accepts a callable which will be invoked
when certain information about an SSL connection is available.
"""
(server, client) = socket_pair()
clientSSL = Connection(Context(TLSv1_METHOD), client)
clientSSL.set_connect_state()
called = []
def info(conn, where, ret):
called.append((conn, where, ret))
context = Context(TLSv1_METHOD)
context.set_info_callback(info)
context.use_certificate(
load_certificate(FILETYPE_PEM, cleartextCertificatePEM))
context.use_privatekey(
load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM))
serverSSL = Connection(context, server)
serverSSL.set_accept_state()
while not called:
for ssl in clientSSL, serverSSL:
try:
ssl.do_handshake()
except WantReadError:
pass
# Kind of lame. Just make sure it got called somehow.
self.assertTrue(called)
def test_load_verify_invalid_file(self):
"""
L{Context.load_verify_locations} raises L{Error} when passed a
non-existent cafile.
"""
clientContext = Context(TLSv1_METHOD)
self.assertRaises(
Error, clientContext.load_verify_locations, self.mktemp())
def test_set_default_verify_paths_signature(self):
"""
L{Context.set_default_verify_paths} takes no arguments and raises
L{TypeError} if given any.
"""
context = Context(TLSv1_METHOD)
self.assertRaises(TypeError, context.set_default_verify_paths, None)
self.assertRaises(TypeError, context.set_default_verify_paths, 1)
self.assertRaises(TypeError, context.set_default_verify_paths, "")
def test_add_extra_chain_cert_invalid_cert(self):
"""
L{Context.add_extra_chain_cert} raises L{TypeError} if called with
other than one argument or if called with an object which is not an
instance of L{X509}.
"""
context = Context(TLSv1_METHOD)
self.assertRaises(TypeError, context.add_extra_chain_cert)
self.assertRaises(TypeError, context.add_extra_chain_cert, object())
self.assertRaises(TypeError, context.add_extra_chain_cert, object(), object())
def test_add_extra_chain_cert(self):
"""
L{Context.add_extra_chain_cert} accepts an L{X509} instance to add to
the certificate chain.
"""
context = Context(TLSv1_METHOD)
context.add_extra_chain_cert(load_certificate(FILETYPE_PEM, cleartextCertificatePEM))
# XXX Oh no, actually asserting something about its behavior would be really hard.
# See #477521.
def test_type(self):
"""
L{Connection} and L{ConnectionType} refer to the same type object and
can be used to create instances of that type.
"""
self.assertIdentical(Connection, ConnectionType)
ctx = Context(TLSv1_METHOD)
self.assertConsistentType(Connection, 'Connection', ctx, None)
def test_set_client_ca_list_errors(self):
"""
L{Context.set_client_ca_list} raises a L{TypeError} if called with a
non-list or a list that contains objects other than X509Names.
"""
ctx = Context(TLSv1_METHOD)
self.assertRaises(TypeError, ctx.set_client_ca_list, "spam")
self.assertRaises(TypeError, ctx.set_client_ca_list, ["spam"])
self.assertIdentical(ctx.set_client_ca_list([]), None)
def test_add_client_ca_errors(self):
"""
L{Context.add_client_ca} raises L{TypeError} if called with a non-X509
object or with a number of arguments other than one.
"""
ctx = Context(TLSv1_METHOD)
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
self.assertRaises(TypeError, ctx.add_client_ca)
self.assertRaises(TypeError, ctx.add_client_ca, "spam")
self.assertRaises(TypeError, ctx.add_client_ca, cacert, cacert)
def __init__(self, *args, **kw):
kw['sslmethod'] = SSL.TLSv1_METHOD
ssl.DefaultOpenSSLContextFactory.__init__(self, *args, **kw)
def getContext(self):
# FIXME -- we should use sslv23 to allow for tlsv1.2
# and, if possible, explicitely disable sslv3 clientside.
# Servers should avoid sslv3
self.method = SSL.TLSv1_METHOD # SSLv23_METHOD
ctx = ssl.ClientContextFactory.getContext(self)
ctx.use_certificate_file(self.cert)
ctx.use_privatekey_file(self.key)
return ctx
def pick_certificate(connection):
try:
key, cert = certificates[connection.get_servername()]
except KeyError:
pass
else:
new_context = Context(TLSv1_METHOD)
new_context.use_privatekey(key)
new_context.use_certificate(cert)
connection.set_context(new_context)