1.创建根CA证书
生成 rootCA.key 以及 rootCA.crt. 用你的域名或者ip地址替换demo.mlopshub.com，比如192.168.10.11
openssl req -x509 \-sha256 -days 356 \-nodes \-newkey rsa:2048 \-subj "/CN=demo.mlopshub.com/C=CN/L=Beijing" \-keyout rootCA.key -out rootCA.crt
或者
openssl req -x509 \-sha256 -days 356 \-nodes \-newkey rsa:2048 \-subj "/CN=192.168.10.11/C=CN/L=Beijing" \-keyout rootCA.key -out rootCA.crt
如果上述命令提示Can’t load /home/username/.rnd into RNG，则需要你手动创建这个文件
2. 生成自签名证书
2.1生成server的私有key
openssl genrsa -out server.key 2048
2.2.创建证书签名请求配置
根据实际情况替换域名以及IP
cat > csr.conf <<EOF
[ req ]
default_bits = 2048
prompt = no
default_md = sha256
req_extensions = req_ext
distinguished_name = dn[ dn ]
C = US
ST = Beijing
L = Beijing
O = MLopsHub
OU = MlopsHub Dev
CN =test.com[ req_ext ]
subjectAltName = @alt_names[ alt_names ]
DNS.1 = test.com
DNS.2 = test2.com
IP.1 = 192.168.10.11
IP.2 = 192.168.1.22EOF
2.3.使用服务器私钥生成证书签名请求(CSR)
openssl req -new -key server.key -out server.csr -config csr.conf
2.4.创建一个外部文件
根据实际情况替换域名以及IP
cat > cert.conf <<EOFauthorityKeyIdentifier=keyid,issuer
basicConstraints=CA:FALSE
keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
subjectAltName = @alt_names[alt_names]
DNS.1 = test.com
IP.1 = 192.168.10.62
EOF
2.5.使用自签名CA生成SSL证书
openssl x509 -req \-in server.csr \-CA rootCA.crt -CAkey rootCA.key \-CAcreateserial -out server.crt \-days 365 \-sha256 -extfile cert.conf
将server.crt server.key放到服务器目录下
将rootCA.crt放到客户端进行验签
3.python验证
3.1客户端单向/双向/不认证
（1）服务器端的代码
import socket
import ssl# 服务器配置
SERVER_HOST = '192.168.10.62'
SERVER_PORT = 12345def ssl_auth_double():# 创建一个普通的 TCP 套接字server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)print("Server socket created")# 绑定到本地地址和端口server_address = (SERVER_HOST, SERVER_PORT)server_socket.bind(server_address)print("Bound to address and port")# 开始监听连接server_socket.listen(1)print("Listening for connections")# 创建 SSL 上下文ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)ssl_context.load_cert_chain(certfile="server.crt", keyfile="server.key")ssl_context.load_verify_locations(cafile="root_cli.crt")ssl_context.verify_mode = ssl.CERT_REQUIREDtry:while True:# 接受客户端连接client_socket, client_address = server_socket.accept()print(f"Accepted connection from {client_address}")# 将客户端 socket 包装成 SSL socketssl_socket = ssl_context.wrap_socket(client_socket, server_side=True)try:while True:# 接收客户端数据data = ssl_socket.recv(1024)if not data:break# 将接收到的数据原样返回ssl_socket.sendall(data)except socket.error as e:print(f"Socket error: {e}")finally:# 关闭 SSL socketssl_socket.close()except KeyboardInterrupt:print("Server stopped by user")finally:# 关闭服务器 socketserver_socket.close()def ssl_auth_single():# 创建一个socket对象server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)print("Socket created")# 绑定到本地地址和端口server_address = (SERVER_HOST, SERVER_PORT)server_socket.bind(server_address)print("Bound to address and port")# 开始监听连接server_socket.listen(1)print("Listening for connections")# 创建SSL上下文ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)ssl_context.load_cert_chain(certfile='server.crt', keyfile='server.key')print('Server is listening on {}:{}'.format(*server_address))while True:# 接受客户端连接client_socket, client_address = server_socket.accept()print('Accepted connection from {}:{}'.format(*client_address))# 将客户端socket包装成SSL socketssl_socket = ssl_context.wrap_socket(client_socket, server_side=True)try:while True:# 接收客户端数据data = ssl_socket.recv(1024)if not data:break# 将接收到的数据原样返回ssl_socket.sendall(data)except socket.error as e:print('Socket error:', e)finally:# 关闭SSL socketssl_socket.close()def ssl_only():# 创建一个普通的 TCP/IP 套接字server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server_socket.bind((SERVER_HOST, SERVER_PORT))server_socket.listen(5)# 创建 SSL 上下文，禁用客户端证书验证context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)context.load_cert_chain(certfile='server.crt', keyfile='server.key')context.verify_mode = ssl.CERT_NONE # 不验证客户端证书# 将普通套接字封装为 SSL 套接字，开启 SSL 加密server_socket_ssl = context.wrap_socket(server_socket, server_side=True)print(f"Server is listening on {SERVER_HOST}:{SERVER_PORT}...")while True:# 等待客户端连接client_socket, addr = server_socket_ssl.accept()print(f"Connection from {addr} established!")try:# 接收数据data = client_socket.recv(1024)if data:print(f"Received: {data.decode('utf-8')}")# 将接收到的数据原样返回client_socket.sendall(data)else:print("No data received")except Exception as e:print(f"Error: {e}")finally:# 关闭客户端连接client_socket.close()
if __name__ == '__main__':ssl_only()（2）客户端代码
import socket
import ssl
import os
import time
# 客户端配置
SERVER_HOST = '192.168.10.62'
SERVER_PORT = 12345
CA_CERT_FILE = 'rootCA.crt' # CA 证书（客户端用于验证服务端）
current_directory = os.getcwd()
print(f"当前目录是: {current_directory}")def ssl_auth_double():client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)print("Client socket created")# 创建 SSL 上下文ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)ssl_context.load_cert_chain(certfile="client.crt", keyfile="client.key")ssl_context.load_verify_locations(cafile="rootCA.crt")ssl_context.verify_mode = ssl.CERT_REQUIRED# 将套接字包装为 SSL 套接字client_socket_ssl = ssl_context.wrap_socket(client_socket, server_hostname=SERVER_HOST)print("Client SSL socket wrapped")try:client_socket_ssl.connect((SERVER_HOST, SERVER_PORT))print(f"Connected to {SERVER_HOST}:{SERVER_PORT} over SSL.")while True:# 发送数据client_socket_ssl.sendall(b"Hello from SSL client!")print("Data sent to server")# 接收响应data = client_socket_ssl.recv(1024)print(f"Received from server: {data.decode('utf-8')}")except Exception as e:print(f"Error: {e}")finally:# 关闭连接client_socket_ssl.close()
def ssl_auth_single():# 创建一个普通的 TCP 套接字client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)print("Client socket created")# 将套接字包装为 SSL 套接字client_socket_ssl = ssl.wrap_socket(client_socket, certfile=None, keyfile=None, server_side=False, ssl_version=ssl.PROTOCOL_SSLv23, ca_certs=CA_CERT_FILE, cert_reqs=ssl.CERT_REQUIRED)print("Client SSL socket wrapped")try:client_socket_ssl.connect((SERVER_HOST, SERVER_PORT))print(f"Connected to {SERVER_HOST}:{SERVER_PORT} over SSL.")while True:# 发送数据client_socket_ssl.sendall(b"Hello from SSL client!")print("Data sent to server")# 接收响应data = client_socket_ssl.recv(1024)print(f"Received from server: {data.decode('utf-8')}")time.sleep(1)except Exception as e:print(f"Error: {e}")finally:# 关闭连接client_socket_ssl.close()print("Connection closed")
def ssl_only():# 创建一个普通的 TCP 套接字client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)print("Client socket created")# 创建 SSL 上下文ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)ssl_context.check_hostname = Falsessl_context.verify_mode = ssl.CERT_NONE# 设置支持的加密套件ssl_context.set_ciphers('ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384:DHE-RSA-AES128-GCM-SHA256:DHE-DSS-AES128-GCM-SHA256:kEDH+AESGCM:ECDHE-RSA-AES128-SHA256:ECDHE-ECDSA-AES128-SHA256:ECDHE-RSA-AES128-SHA:ECDHE-ECDSA-AES128-SHA:ECDHE-RSA-AES256-SHA384:ECDHE-ECDSA-AES256-SHA384:ECDHE-RSA-AES256-SHA:ECDHE-ECDSA-AES256-SHA:DHE-RSA-AES128-SHA256:DHE-RSA-AES128-SHA:DHE-DSS-AES128-SHA256:DHE-RSA-AES256-SHA256:DHE-DSS-AES256-SHA:DHE-RSA-AES256-SHA:AES128-GCM-SHA256:AES256-GCM-SHA384:AES128-SHA256:AES256-SHA256:AES128-SHA:AES256-SHA:AES:CAMELLIA:DES-CBC3-SHA:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!aECDH:!EDH-DSS-DES-CBC3-SHA:!EDH-RSA-DES-CBC3-SHA:!KRB5-DES-CBC3-SHA')# 将套接字包装为 SSL 套接字client_socket_ssl = ssl_context.wrap_socket(client_socket, server_hostname=SERVER_HOST)print("Client SSL socket wrapped")try:client_socket_ssl.connect((SERVER_HOST, SERVER_PORT))print(f"Connected to {SERVER_HOST}:{SERVER_PORT} over SSL.")while True:# 发送数据client_socket_ssl.sendall(b"Hello from SSL client!")print("Data sent to server")# 接收响应data = client_socket_ssl.recv(1024)print(f"Received from server: {data.decode('utf-8')}")time.sleep(1)except Exception as e:print(f"Error: {e}")finally:# 关闭连接client_socket_ssl.close()print("Connection closed")
if __name__ == '__main__':ssl_only()
