在Python中使用ssl context.set_servername_callback

Question

我的目标是允许ssl客户端从服务器中的多个有效证书对中进行选择。客户端有一个CA证书，它将用来验证来自服务器的证书。

所以要尝试完成此操作，我在服务器上使用ssl.SSLContext.set_servername_callback()并结合使用ssl.SSLSocket.wrap_socket's parameter: server_hostname`来尝试允许客户端指定要使用的密钥对。下面的代码是什么样子：

Server代码：

import sys 
import pickle 
import ssl 
import socket 
import select 

request = {'msgtype': 0, 'value': 'Ping', 'test': [chr(i) for i in range(256)]} 
response = {'msgtype': 1, 'value': 'Pong'} 

def handle_client(c, a): 
    print("Connection from {}:{}".format(*a)) 
    req_raw = c.recv(10000) 
    req = pickle.loads(req_raw) 
    print("Received message: {}".format(req)) 
    res = pickle.dumps(response) 
    print("Sending message: {}".format(response)) 
    c.send(res) 

def run_server(hostname, port): 
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
    s.bind((hostname, port)) 
    s.listen(8) 
    print("Serving on {}:{}".format(hostname, port)) 

    try: 
     while True: 
      (c, a) = s.accept() 

      def servername_callback(sock, req_hostname, cb_context, as_callback=True): 
       print('Loading certs for {}'.format(req_hostname)) 
       server_cert = "ssl/{}/server".format(req_hostname) # NOTE: This use of socket input is INSECURE 
       cb_context.load_cert_chain(certfile="{}.crt".format(server_cert), keyfile="{}.key".format(server_cert)) 

       # Seems like this is designed usage: https://github.com/python/cpython/blob/3.4/Modules/_ssl.c#L1469 
       sock.context = cb_context 
       return None 

      context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH) 
      context.set_servername_callback(servername_callback) 
      default_cert = "ssl/3.1/server" 
      context.load_cert_chain(certfile="{}.crt".format(default_cert), keyfile="{}.key".format(default_cert)) 
      ssl_sock = context.wrap_socket(c, server_side=True) 

      try: 
       handle_client(ssl_sock, a) 
      finally: 
       c.close() 

    except KeyboardInterrupt: 
     s.close() 

if __name__ == '__main__': 
    hostname = '' 
    port = 6789 
    run_server(hostname, port) 

客户端代码：

import sys 
import pickle 
import socket 
import ssl 

request = {'msgtype': 0, 'value': 'Ping', 'test': [chr(i) for i in range(256)]} 
response = {'msgtype': 1, 'value': 'Pong'} 


def client(hostname, port): 
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
    print("Connecting to {}:{}".format(hostname, port)) 
    s.connect((hostname, port)) 

    ssl_sock = ssl.SSLSocket(sock=s, ca_certs="server_old.crt", cert_reqs=ssl.CERT_REQUIRED, server_hostname='3.2') 

    print("Sending message: {}".format(request)) 
    req = pickle.dumps(request) 
    ssl_sock.send(req) 

    resp_raw = ssl_sock.recv(10000) 
    resp = pickle.loads(resp_raw) 
    print("Received message: {}".format(resp)) 

    ssl_sock.close() 

if __name__ == '__main__': 
    hostname = 'localhost' 
    port = 6789 
    client(hostname, port) 

但它不工作。什么似乎发生的是servername_callback被调用，正在获取指定的“主机名”，并且在回调中对context.load_cert_chain的调用不失败（尽管如果给定的路径不存在，它会失败）。但是，服务器始终返回在调用context.wrap_socket(c, server_side=True)之前加载的证书对。所以我的问题是：在servername_callback中有没有办法修改ssl上下文使用的密钥对，并获得该密钥对的证书用于连接？

我还应该注意到，我检查了流量，并且在servername_callback函数返回之后才会发送服务器证书（并且如果未成功完成或返回“失败”值，将永远不会发送）。

Answer 1

在您的回调中，cb_context与调用wrap_socket()的上下文相同，与socket.context相同，因此socket.context = cb_context将上下文设置为与之前相同。

更改上下文的证书链不会影响用于当前wrap_socket()操作的证书。对此的解释是OpenSSL的是如何创建它的基本对象，在这种情况下，已经创建基础SSL结构和使用copies of the chains：

NOTES

与SSL_CTX结构链关联被复制到任何调用SSL_new（）时的SSL结构。 SSL结构不会受随后在父SSL_CTX中更改的链影响。

设置新的上下文时，会更新SSL结构，但在new context is equal to the old one时不执行更新。

您需要将sock.context设置为不同的上下文才能使其工作。您目前在每个新的传入连接上实例化一个新的上下文，这是不需要的。相反，您应该只实例化一次标准上下文并重用它。这同样适用于动态加载的背景，你可以在启动时创建所有这些，把它们放在一个字典，所以你可以做一个查找，e.g：

... 

contexts = {} 

for hostname in os.listdir("ssl"): 
    print('Loading certs for {}'.format(hostname)) 
    server_cert = "ssl/{}/server".format(hostname) 
    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH) 
    context.load_cert_chain(certfile="{}.crt".format(server_cert), 
          keyfile="{}.key".format(server_cert)) 
    contexts[hostname] = context 

def servername_callback(sock, req_hostname, cb_context, as_callback=True): 
    context = contexts.get(req_hostname) 
    if context is not None: 
     sock.context = context 
    else: 
     pass # handle unknown hostname case 

def run_server(hostname, port): 
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
    s.bind((hostname, port)) 
    s.listen(8) 
    print("Serving on {}:{}".format(hostname, port)) 

    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH) 
    context.set_servername_callback(servername_callback) 
    default_cert = "ssl/3.1/server" 
    context.load_cert_chain(certfile="{}.crt".format(default_cert), 
          keyfile="{}.key".format(default_cert)) 

    try: 
     while True: 
      (c, a) = s.accept() 
      ssl_sock = context.wrap_socket(c, server_side=True) 
      try: 
       handle_client(ssl_sock, a) 
      finally: 
       c.close() 

    except KeyboardInterrupt: 
     s.close()