Python如何为socket连接实现类似HTTPS的TLS加密安全隧道

你需要的加密效果本质是为普通TCP Socket套上TLS协议层，和HTTPS底层加密逻辑完全一致，Python内置的ssl模块可直接实现，自带证书校验能力可防范MITM攻击，原有业务逻辑几乎不需要修改。

1. 生成测试用自签名证书 #

执行以下openssl命令生成服务端密钥与证书：

openssl req -x509 -newkey rsa:4096 -keyout server.key -out server.crt -days 365 -nodes

执行时按提示填写信息，Common Name字段请填写服务端的实际IP或域名，否则后续客户端证书校验会失败。测试阶段需将生成的server.crt文件复制到客户端运行目录。

2. 完整加密版服务端代码 #

import socket 
import threading
import ssl

HEADER = 64
PORT = 5050
SERVER = "*Server IP*" # 替换为实际服务端IP
ADDR = (SERVER, PORT)
FORMAT = 'utf-8'
DISCONNECT_MESSAGE = "!DISCONNECT"

# 原始Socket基础上套TLS层
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server = ssl.wrap_socket(server, keyfile="server.key", certfile="server.crt", server_side=True)
server.bind(ADDR)

def handle_client(conn, addr):
    print(f"[NEW CONNECTION] {addr} connected.")

    connected = True
    while connected:
        msg_length = conn.recv(HEADER).decode(FORMAT)
        if msg_length:
            msg_length = int(msg_length)
            msg = conn.recv(msg_length).decode(FORMAT)
            if msg == DISCONNECT_MESSAGE:
                connected = False

            print(f"[{addr}] {msg}")
            conn.send("Msg received".encode(FORMAT))

    conn.close()


def start():
    server.listen()
    print(f"[LISTENING] Server is listening on {SERVER}")
    while True:
        conn, addr = server.accept()
        thread = threading.Thread(target=handle_client, args=(conn, addr))
        thread.start()
        print(f"[ACTIVE CONNECTIONS] {threading.activeCount() - 1}")


print("[STARTING] server is starting...")
start()

3. 完整加密版客户端代码 #

import socket
import ssl

HEADER = 64
PORT = 5050
FORMAT = 'utf-8'
DISCONNECT_MESSAGE = "!DISCONNECT"
SERVER = "*Server IP*" # 替换为实际服务端IP，和证书Common Name一致
ADDR = (SERVER, PORT)

# 原始Socket基础上套TLS层，开启服务端证书校验防MITM
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client = ssl.wrap_socket(client, ca_certs="server.crt", cert_reqs=ssl.CERT_REQUIRED, server_hostname=SERVER)
client.connect(ADDR)

def send(msg):
    message = msg.encode(FORMAT)
    msg_length = len(message)
    send_length = str(msg_length).encode(FORMAT)
    send_length += b' ' * (HEADER - len(send_length))
    client.send(send_length)
    client.send(message)
    print(client.recv(2048).decode(FORMAT))

user_input = None
while user_input != "exit":
    user_input = (input())
    send(user_input)


send(DISCONNECT_MESSAGE)

注意事项 #

• 生产环境请使用由可信CA机构签发的正式证书，不要使用自签名证书
• 请勿删除客户端的cert_reqs=ssl.CERT_REQUIRED配置，否则TLS无法防范中间人攻击
• 原有业务逻辑（消息长度处理、多线程处理、断开逻辑等）无需做任何修改，TLS层会自动处理数据的加密解密

内容的提问来源于stack exchange，提问作者TheAngrySocalist