普通的 socket 通信

我们用 Python 来实现一个简单的 TCP 服务器,它实现 echo 功能。

服务端代码:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socketserver

class MyTcpHandler(socketserver.BaseRequestHandler):

    def handle(self) -> None:
        self.data = self.request.recv(1024).strip()
        self.request.sendall(self.data)


HOST, PORT = "localhost", 9999

with socketserver.TCPServer((HOST, PORT), MyTcpHandler) as server:
    server.serve_forever()

客户端代码:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socket

HOST, PORT = "localhost", 9999


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.connect((HOST, PORT))
    sock.sendall(bytes("Hello, I am Client", "utf-8"))
    received = str(sock.recv(1024), "utf-8")
    print(received)

增加 TLS 支持

单向认证

证书我们直接通过 openssl 生成

使用 openssl 生成证书

测试不需要真的让 CA 签发证书,所以我们用 openssl 生成自签名的根证书

  1. 生成 CA 私钥

    1
    
    openssl genrsa -des3 -out ca.key 2048
    

    -des3 使用 des3 算法,强度为 2048, 输出到 ca.key ,生成过程中输入密码 123456

  2. 生成自签名的 CA 证书

    1
    
    openssl req -new -x509 -days 36500 -key ca.key -out ca.crt
    

    openssl 会要求填写下面的信息

     1
     2
     3
     4
     5
     6
     7
     8
     9
    10
    11
    12
    
    Country Name (2 letter code) [AU]:CN // 国家码
    State or Province Name (full name) [Some-State]:GuangDong // 省份
    Locality Name (eg, city) []:GuangZhou // 城市
    Organization Name (eg, company) [Internet Widgits Pty Ltd]: Personal // 组织机构或公司名
    Organizational Unit Name (eg, section) []: Personal // 机构部门
    Common Name (e.g. server FQDN or YOUR name) []:域名 // *.abc.com
    Email Address []:[email protected]  // 邮件地址
    
    Please enter the following 'extra' attributes
    to be sent with your certificate request
    A challenge password []:123456 // 证书密码
    An optional company name []:Personal // 公司名
    

    这里要注意一点:上面的步骤好像全程没有公钥参与,实际上公钥就在证书中。下面的命令查看证书中的公钥:

    1
    
    openssl x509 -in .\ca.crt -pubkey -noout
    
  3. 生成服务端的私钥

    1
    
    openssl genrsa -out server.key 2048
    

    输出到 server.key ,不用密码保护

  4. 生成服务端的证书签名请求 CSR 文件

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

    同样要填相关信息

  5. 使用上面生成的 CA 给服务端签证书

    1
    
    openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 36500
    

    x509: 证书格式 -days 36500: 过期时间 -in: 指定请求文件

服务端支持 tls

python 直接用 ssl 库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import socketserver
import ssl

class MyTcpHandler(socketserver.BaseRequestHandler):

    def handle(self) -> None:
        self.data = self.request.recv(1024).strip()
        self.request.sendall(self.data)

context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile="server.crt", keyfile="server.key")

class MyTCPServer(socketserver.TCPServer):
    def get_request(self):
        (sock, addr) = super().get_request()
        return (context.wrap_socket(sock=sock, server_side=True), addr)

HOST, PORT = "127.0.0.1", 9999

with MyTCPServer((HOST, PORT), MyTcpHandler) as server:
    server.serve_forever()

可以看到,服务端 SSL 就需要刚刚生成的 server.crt 证书,=server.key= 私钥

客户端访问 SSL 服务端

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socket
import ssl

HOST, PORT = "127.0.0.1", 9999

context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.maximum_version = ssl.TLSVersion.TLSv1_2
context.load_default_certs()
context.load_verify_locations("ca.crt")

with socket.create_connection((HOST, PORT)) as sock:
    with context.wrap_socket(sock) as ssock:
        ssock.sendall(bytes("Hello, I am Client", "utf-8"))
        received = str(ssock.recv(1024), "utf-8")
        print(received)

我们这里不检测 hostname 。因为我们的证书是自签名的,所以要将验证位置为 “ca.crt” 才能验证成功

双向认证

双向认证服务器也需要认证客户端的身份,同样我们先生成客户端的私钥和证书。

服务端代码,区别只在需要验证客户端证书:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socketserver
import ssl

class MyTcpHandler(socketserver.BaseRequestHandler):

    def handle(self) -> None:
        self.data = self.request.recv(1024).strip()
        self.request.sendall(self.data)

context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.verify_mode = ssl.VerifyMode.CERT_REQUIRED # 需要验证客户端证书

context.load_cert_chain(certfile="server.crt", keyfile="server.key")
context.load_verify_locations("ca.crt")

class MyTCPServer(socketserver.TCPServer):
    def get_request(self):
        (sock, addr) = super().get_request()
        return (context.wrap_socket(sock=sock, server_side=True), addr)

HOST, PORT = "127.0.0.1", 9999

with MyTCPServer((HOST, PORT), MyTcpHandler) as server:
    server.serve_forever()

客户端代码,客户端需要设置证书与密钥:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socket
import ssl

HOST, PORT = "127.0.0.1", 9999

context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.maximum_version = ssl.TLSVersion.TLSv1_2
context.load_cert_chain(certfile="client.crt", keyfile="client.key", password="123456")
context.load_verify_locations("ca.crt")

with socket.create_connection((HOST, PORT)) as sock:
    with context.wrap_socket(sock) as ssock:
        ssock.sendall(bytes("Hello, I am Client", "utf-8"))
        received = str(ssock.recv(1024), "utf-8")
        print(received)