python tcp socket中实现SSL/TLS认证

SSL/TLS介绍

官话说SSL是安全套接层(secure sockets layer)，TLS是SSL的继任者，叫传输层安全(transport layer security)。

说白点，就是在明文的上层和TCP层之间加上一层加密，这样就保证上层信息传输的安全。如HTTP协议是明文传输，加上SSL层之后，就有了雅称HTTPS。它存在的唯一目的就是保证上层通讯安全的一套机制。

传统的 TLS 握手过程:

单向认证和双向认证

SSL认证分为单向认证和双向认证，是在安全通信中使用的两种不同的身份验证方式，它们之间的主要区别在于身份验证的方向和安全性。

单向认证（One-Way Authentication）：

- 客户端认证服务器：
  - 无需客户端拥有客户端证书，只需服务端拥有服务器证书。
  - 例如：浏览器通常会内置一组根证书，这些根证书由浏览器厂商或操作系统供应商预先安装在浏览器中。这组内置的根证书是被广泛信任的证书颁发机构（Certificate Authorities，CAs）签发的根证书，用于验证网站和服务的数字证书的可信性。
  - 当您访问一个使用HTTPS协议的网站时，您的浏览器会使用其内置的根证书来验证该网站的数字证书是否由受信任的CA签发。如果验证通过，浏览器将显示一个锁定图标或其他安全标志，表示连接是安全的。如果验证未通过，浏览器通常会发出警告，提醒您可能存在安全风险。
  - 因此，根证书在浏览器中起到了关键的作用，它们构成了安全通信的基础。浏览器定期更新其内置的根证书，以反映新的CA或吊销的证书，以确保持续的安全性和可信性。
- 适用场景：单向认证适用于大多数Web浏览和服务器通信场景，一般情况下，在打开页面的时候没有提示数据加密插件的，属于单向认证，即浏览器只持有公钥。

双向认证（Two-Way Authentication）：
- 服务器验证客户端和客户端验证服务器：双向认证中，服务器验证客户端的身份，同时客户端也验证服务器的身份。这意味着客户端和服务器都需要提供有效的证书以进行相互验证。
- 更高的安全性：双向认证提供更高的安全性，因为它确保了通信的两端都是合法的，并且双方都可以互相验证。
- 适用场景：双向认证通常在需要更高级别的安全性的场景中使用，例如金融交易、医疗保健、政府通信等，其中双方都需要互相验证以确保身份。类似于支付宝、银行的U顿支付之类的，会要求用户安装插件或驱动，属于双向验证。

总之，单向认证用于服务器验证客户端的情况，而双向认证要求双方都进行身份验证，提供更高级别的安全性。选择哪种认证方式取决于您的应用程序的特定需求和安全性要求。

生成自签名证书文件

数字证书一般由数字证书认证机构签发，证书包含了：

公钥。
证书拥有者身份信息。
数字证书认证机构（发行者）信息。
发行者对这份文件的数字签名及使用的算法。
有效期。

生成 CA 根证书、服务器证书和客户端证书的步骤如下：

步骤 1：创建 CA 根密钥和证书

首先，您需要生成一个 CA 根密钥（私钥）和一个 CA 根证书。以下是一些基本的步骤：

创建 CA 根密钥 (ca-key.pem)。这是用于签署服务器和客户端证书请求的私钥。请确保保护好这个私钥文件，因为它是证书链的根。

openssl genpkey -algorithm RSA -out ca-key.pem

使用 CA 根密钥创建自签名的 CA 根证书 (ca-cert.pem)。

openssl req -new -x509 -key ca-key.pem -out ca-cert.pem -days 3650

在此过程中，您需要提供一些 CA 根证书的信息，如组织、单位、常用名等。

步骤 2：创建服务器证书请求和证书

接下来，您需要为服务器创建证书请求并签署服务器证书。

创建服务器密钥 (server-key.pem)。

openssl genpkey -algorithm RSA -out server-key.pem

使用服务器密钥创建证书请求 (server-csr.pem)。在这一步中，您需要提供服务器的信息，如主机名（通常是服务器的域名）。

openssl req -new -key server-key.pem -out server-csr.pem

(py3.8) root@localhost:/opt/lianhaifeng/test2# openssl req -new -key server-key.pem -out server-csr.pem
You are about to be asked to enter information that will be incorporated
into your certificate request.
What you are about to enter is what is called a Distinguished Name or a DN.
There are quite a few fields but you can leave some blank
For some fields there will be a default value,
If you enter '.', the field will be left blank.
-----
Country Name (2 letter code) [AU]:CN
State or Province Name (full name) [Some-State]:fujian
Locality Name (eg, city) []:xiamen
Organization Name (eg, company) [Internet Widgits Pty Ltd]:
Organizational Unit Name (eg, section) []:
Common Name (e.g. server FQDN or YOUR name) []:
Email Address []:

Please enter the following 'extra' attributes
to be sent with your certificate request
A challenge password []:
An optional company name []:

使用 CA 根证书和私钥签署服务器证书请求，生成服务器证书 (server-cert.pem)。

openssl x509 -req -in server-csr.pem -CA ca-cert.pem -CAkey ca-key.pem -out server-cert.pem -CAcreateserial -days 3650

(py3.8) root@localhost:/opt/lianhaifeng/test2# openssl x509 -req -in server-csr.pem -CA ca-cert.pem -CAkey ca-key.pem -out server-cert.pem -CAcreateserial -days 3650
Signature ok
subject=C = CN, ST = fujian, L = xiamen, O = Internet Widgits Pty Ltd
Getting CA Private Key

验证生成的服务器证书，如果验证成功，它将不会输出任何错误消息。

openssl verify -CAfile ca-cert.pem client-cert.pem  # 验证证书链完整性
openssl x509 -text -noout -in server-cert.pem  # 验证证书属性

(py3.8) root@localhost:/opt/lianhaifeng/test2# openssl verify -CAfile ca-cert.pem server-cert.pem
server-cert.pem: OK

步骤 3：创建客户端证书请求和证书（可选）

如果您需要客户端证书以进行双向认证，则可以执行以下步骤：

创建客户端密钥 (client-key.pem)。

openssl genpkey -algorithm RSA -out client-key.pem

(py3.8) root@localhost:/opt/lianhaifeng/test2# openssl genpkey -algorithm RSA -out client-key.pem
....................................................................................................................................................................+++++
........+++++

使用客户端密钥创建证书请求 (client-csr.pem)。

openssl req -new -key client-key.pem -out client-csr.pem

(py3.8) root@localhost:/opt/lianhaifeng/test2# openssl req -new -key client-key.pem -out client-csr.pem
You are about to be asked to enter information that will be incorporated
into your certificate request.
What you are about to enter is what is called a Distinguished Name or a DN.
There are quite a few fields but you can leave some blank
For some fields there will be a default value,
If you enter '.', the field will be left blank.
-----
Country Name (2 letter code) [AU]:CN
State or Province Name (full name) [Some-State]:fujian
Locality Name (eg, city) []:qz
Organization Name (eg, company) [Internet Widgits Pty Ltd]:
Organizational Unit Name (eg, section) []:
Common Name (e.g. server FQDN or YOUR name) []:
Email Address []:

Please enter the following 'extra' attributes
to be sent with your certificate request
A challenge password []:
An optional company name []:

使用 CA 根证书和私钥签署客户端证书请求，生成客户端证书 (client-cert.pem)。

openssl x509 -req -in client-csr.pem -CA ca-cert.pem -CAkey ca-key.pem -out client-cert.pem -CAcreateserial -days 3650

注意： 在实际生产环境中，需要更多的安全性和细化控制。此示例仅提供了基本步骤和示例命令，具体的配置可能因您的环境和需求而有所不同。同时，请确保适当地保护和存储您的密钥和证书文件。

验证生成的服务器证书，如果验证成功，它将不会输出任何错误消息。

openssl verify -CAfile ca-cert.pem client-cert.pem  # 验证证书链完整性
openssl x509 -text -noout -in server-cert.pem  # 验证证书属性

python实现SSL单向认证

通过上面的介绍，我们已经了解了单向认证的一般概念及流程。

下面是一个简单的Python示例代码，演示了如何在客户端认证服务端证书：

ssl_server.py

import socket
import ssl
import threading

class server_ssl:
    def build_listen(self):
        # CA_FILE = "ca-cert.pem"
        KEY_FILE = "server-key.pem"
        CERT_FILE = "server-cert.pem"
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(certfile=CERT_FILE, keyfile=KEY_FILE)  # 加载服务端证书和私钥
        # context.load_verify_locations(CA_FILE)  # 加载根证书
        context.verify_mode = ssl.CERT_NONE  # 不需要客户端提供证书

        # 监听端口
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
            # 将socket打包成SSL socket
            with context.wrap_socket(sock, server_side=True) as ssock:
                ssock.bind(('127.0.0.1', 10036))
                ssock.listen(5)
                print("Server is listening for connections...")

                while True:
                    # 接收客户端连接
                    client_socket, addr = ssock.accept()
                    print(f"Accepted connection from {addr}")

                    # 创建新线程来处理客户端请求
                    client_thread = threading.Thread(target=self.handle_client, args=(client_socket, addr))
                    client_thread.start()

    def handle_client(self, client_socket, addr):
        try:
            while True:
                # 接收客户端信息
                msg = client_socket.recv(1024).decode("utf-8")
                if not msg:
                    break  # 客户端断开连接

                print(f"Received message from client {addr}: {msg}")

                # 向客户端发送信息
                response = f"Received: {msg}".encode("utf-8")
                client_socket.send(response)
        except Exception as e:
            print(f"Error: {str(e)}")
        finally:
            client_socket.close()
            print("Connection closed")

if __name__ == "__main__":
    server = server_ssl()
    server.build_listen()

运行server端代码：

(py3.8) root@localhost:/opt/lianhaifeng# python ssl_server.py
Server is listening for connections...
Accepted connection from ('127.0.0.1', 49814)
Received message from client ('127.0.0.1', 49814): hello
Received message from client ('127.0.0.1', 49814): world

ssl_client.py

import socket
import ssl

class client_ssl:
    def send_hello(self):
        CA_FILE = "ca-cert.pem"
        # CLIENT_KEY_FILE = "client-key.pem"
        # CLIENT_CERT_FILE = "client-cert.pem"

        # 创建SSL上下文对象
        context = ssl.SSLContext(ssl.PROTOCOL_TLS)
        context.check_hostname = False
        # context.load_cert_chain(certfile=CLIENT_CERT_FILE, keyfile=CLIENT_KEY_FILE)  # 服务器不需要认证客户端证书，故不需要
        context.load_verify_locations(CA_FILE)   # 使用根证书认证服务端证书
        context.verify_mode = ssl.CERT_REQUIRED

        # 与服务端建立socket连接
        with socket.socket() as sock:
            # 将socket打包成SSL socket
            with context.wrap_socket(sock, server_side=False) as ssock:
                ssock.connect(('127.0.0.1', 10036))

                while True:
                    # 输入要发送的消息
                    msg = input("Enter a message to send (or 'quit' to exit): ")
                    if msg.lower() == 'quit':
                        break

                    # 向服务端发送消息
                    ssock.send(msg.encode("utf-8"))

                    # 接收并打印服务端返回的消息
                    response = ssock.recv(1024).decode("utf-8")
                    print(f"Received message from the server: {response}")

if __name__ == "__main__":
    client = client_ssl()
    client.send_hello()

运行客户端代码：

(py3.8) root@localhost:/opt/lianhaifeng# python ssl_client.py
Enter a message to send (or 'quit' to exit): hello
Received message from the server: Received: hello
Enter a message to send (or 'quit' to exit): world
Received message from the server: Received: world
Enter a message to send (or 'quit' to exit):

python实现SSL双向认证

下面是一个简单的Python示例代码，演示了如何在服务器和客户端之间进行双向认证：

服务器端代码 server.py：

import socket
import ssl
import threading

class server_ssl:
    def build_listen(self):
        CA_FILE = "ca-cert.pem"
        KEY_FILE = "server-key.pem"
        CERT_FILE = "server-cert.pem"
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(certfile=CERT_FILE, keyfile=KEY_FILE)
        context.load_verify_locations(CA_FILE)
        context.verify_mode = ssl.CERT_REQUIRED  # 如果服务器不验证客户端证书：ssl.CERT_NONE 
        context.check_hostname = False

        # 监听端口
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
            # 将socket打包成SSL socket
            with context.wrap_socket(sock, server_side=True) as ssock:
                ssock.bind(('127.0.0.1', 10028))
                ssock.listen(5)
                print("Server is listening for connections...")

                while True:
                    # 接收客户端连接
                    client_socket, addr = ssock.accept()
                    print(f"Accepted connection from {addr}")

                    # 创建新线程来处理客户端请求
                    client_thread = threading.Thread(target=self.handle_client, args=(client_socket, addr))
                    client_thread.start()

    def handle_client(self, client_socket, addr):
        try:
            while True:
                # 接收客户端信息
                msg = client_socket.recv(1024).decode("utf-8")
                if not msg:
                    break  # 客户端断开连接

                print(f"Received message from client {addr}: {msg}")

                # 向客户端发送信息
                response = f"Received: {msg}".encode("utf-8")
                client_socket.send(response)
        except Exception as e:
            print(f"Error: {str(e)}")
        finally:
            client_socket.close()
            print("Connection closed")

if __name__ == "__main__":
    server = server_ssl()
    server.build_listen()

(py3.8) root@localhost:/opt/lianhaifeng/test2# python server.py
Server is listening for connections...
Accepted connection from ('127.0.0.1', 55364)
Received message from client ('127.0.0.1', 55364): hello
Received message from client ('127.0.0.1', 55364): haige1
Accepted connection from ('127.0.0.1', 55366)
Received message from client ('127.0.0.1', 55366): hello
Received message from client ('127.0.0.1', 55366): haige2

客户端代码 client.py：

import socket
import ssl

class client_ssl:
    def send_hello(self):
        CA_FILE = "ca-cert.pem"
        SERVER_CERT_FILE = "server-cert.pem"  # 服务器证书文件路径
        CLIENT_KEY_FILE = "client-key.pem"
        CLIENT_CERT_FILE = "client-cert.pem"

        # 创建SSL上下文对象
        context = ssl.SSLContext(ssl.PROTOCOL_TLS)
        context.check_hostname = False
        context.load_cert_chain(certfile=CLIENT_CERT_FILE, keyfile=CLIENT_KEY_FILE)
        context.load_verify_locations(CA_FILE)
        context.verify_mode = ssl.CERT_REQUIRED
        context.load_verify_locations(cafile=CA_FILE)  # 设置根证书

        # 与服务端建立socket连接
        with socket.socket() as sock:
            # 将socket打包成SSL socket
            with context.wrap_socket(sock, server_side=False) as ssock:
                ssock.connect(('127.0.0.1', 10026))

                while True:
                    # 输入要发送的消息
                    msg = input("Enter a message to send (or 'quit' to exit): ")
                    if msg.lower() == 'quit':
                        break

                    # 向服务端发送消息
                    ssock.send(msg.encode("utf-8"))

                    # 接收并打印服务端返回的消息
                    response = ssock.recv(1024).decode("utf-8")
                    print(f"Received message from the server: {response}")

if __name__ == "__main__":
    client = client_ssl()
    client.send_hello()

客户端1

(py3.8) root@localhost:/opt/lianhaifeng/test2# python client.py
Enter a message to send (or 'quit' to exit): hello
Received message from the server: Received: hello
Enter a message to send (or 'quit' to exit): haige1
Received message from the server: Received: haige1
Enter a message to send (or 'quit' to exit):

客户端2

(py3.8) root@localhost:/opt/lianhaifeng/test2# python client.py
Enter a message to send (or 'quit' to exit): hello
Received message from the server: Received: hello
Enter a message to send (or 'quit' to exit): haige2
Received message from the server: Received: haige2
Enter a message to send (or 'quit' to exit):

若是服务端要求验证客户端证书，但是客户端没有携带证书，则会引发一个常见的网络错误：ConnectionResetError: [Errno 104] Connection reset by peer，此时在服务器端也会有详细信息：ssl.SSLError: [SSL: PEER_DID_NOT_RETURN_A_CERTIFICATE] peer did not return a certificate (_ssl.c:1131)

如果本篇文章对你所有帮助，欢迎转发、点赞、收藏、在看，非常感谢。

参考

https://www.jianshu.com/p/ffcf1b765d76
https://blog.csdn.net/wuliganggang/article/details/78428866
https://blog.csdn.net/zhangtaoym/article/details/55259889
https://cloud.tencent.com/document/product/214/54254