ssl 双向认证
client side
1 # coding:utf-8
2
3 import socket, ssl, pprint
4 import os,time
5
6 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7
8 ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
9 ssl_ctx.load_cert_chain("keys/client.pem","keys/client.key")
10 ssl_ctx.load_verify_locations("ca.crt")
11 #ssl_ctx.verify_mode = ssl.CERT_REQUIRED
12
13 ssl_sock = ssl_ctx.wrap_socket(s)
14
15 ssl_sock.connect(('localhost', 6030))
16
17 print repr(ssl_sock.getpeername())
18 print ssl_sock.cipher()
19
20 print pprint.pformat(ssl_sock.getpeercert())
21
22 ssl_sock.write("boo!")
23 time.sleep(2)
24 ssl_sock.write("bye!")
server side
1
2 from __future__ import absolute_import, division, print_function, with_statement
3 import socket, ssl
4 import os
5 import tornado.tcpserver
6 from tornado.ioloop import IOLoop
7 import tornado.gen
8
9 ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
10
11 ssl_ctx.load_cert_chain("./keys/client.pem",
12 "./keys/client.key")
13
14 ssl_ctx.load_verify_locations("keys/ca.crt")
15
16 ssl_ctx.verify_mode = ssl.CERT_REQUIRED
17
18
19 class A(tornado.tcpserver.TCPServer):
20 def handle_stream(self, stream, address):
21 print('recv connection from ' ,address)
22 self.run(stream,address)
23
24 @tornado.gen.coroutine
25 def run(self, stream, address):
26 while True:
27 try:
28 body = yield stream.read_bytes(1024, partial=True)
29 except:
30 print('close connection from ' ,address)
31 return
32 print(body)
33
34 def main():
35 server = A(ssl_options=ssl_ctx)
36 server.listen(6030, '')
37
38
39 io_loop = IOLoop.current()
40
41 io_loop.add_callback(main)
42 io_loop.start()