python网络编程——TLS/SSL_python tls

1.TLS简介。

       TLS的前身是 安全套接层 ，是传输层安全协议。TLS能保护的信息包括：

• 与请求URL之间的HTTPS连接以及返回内容。
• 密码或cookies等可能在套接字双向传递的任意认证信息。

       但TLS并不是万能的它也有不能保护的信息：

• 本机和远程连接地址都是可见的，地址信息在每个数据包得IP头信息中以纯文本的形式表示。
• 客户端与服务器的端口号同样在TCP首部信息中可见。
• 客户端进行DNS查询可见。

       但它仍是如今互联网上应用最广泛的传输层加密方法。使用TLS给通信加密时客户端需要的第一样东西就是远程服务器提供的一个二进制文档，称为 证书。书中包含了公钥。公钥是一个整数能对数据加密。只有拥有与公钥对应的私钥，才能解密并理解相应的信息。为了信任发自服务器的证书，TLS会话保存了一个证书机构(CA)列表，该列表包含了我们在对互联网主机进行身份验证是信任的机构。为了证明证书的合法性，CA会为证书加上一个数学标记，该标记叫做 签名 。TLS库使用相应CA证书的公钥验证了证书的签名之后，才会认为该证书是合法的。

2.生成证书。

       使用Linux自带的OpenSSL命令行能快速生成一个供测试代码使用的自签名证书，下面是生成自签名证书的命令：

openssl genrsa  -out server.key 2048    生成私钥
openssl req -new -key server.key -out server.csr  生成CSR（证书签名请求）
openssl x509 -req -days 365 -in server.csr -signkey server.key -out server.crt   生成自签名证书
cat server.crt server.key > localhost.pem    整合文件
1
2
3
4

        注意：在生成CSR文件是会要求填写信息，在填写Common Name 需要填写你要连接的主机名或者IP地址。可以仿照我的信息填写：

Country Name (2 letter code) [AU]:CN
State or Province Name (full name) [Some-State]:beijing
Locality Name (eg, city) []:beijing
Organization Name (eg, company) [Internet Widgits Pty Ltd]:school
Organizational Unit Name (eg, section) []:school
Common Name (e.g. server FQDN or YOUR name) []:localhost
Email Address []:dwj@qq.com
1
2
3
4
5
6
7

3.使用TLS提供的套接字进行安全通信。

代码：

#!/usr/bin/python
#coding:utf-8

import argparse, socket, ssl

def client(host, port, cafile=None):
    purpose = ssl.Purpose.SERVER_AUTH
    #第一个参数表示上下文对象为客户端所用，用于验证其连接的服务器，
    #cafile选项表示脚本验证远程证书时信任的证书机构
    context = ssl.create_default_context(purpose, cafile=cafile)

    raw_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    raw_sock.connect((host, port))
    print('Connected to host {!r} and port {}'.format(host, port))
    #参数调用的是已经使用connect()连接的主机名
    ssl_sock = context.wrap_socket(raw_sock, server_hostname=host)

    while True:
        data = ssl_sock.recv(1024)
        if not data:
            break
        print(repr(data))

def server(host, port, certfile, cafile=None):
    purpose = ssl.Purpose.CLIENT_AUTH
    context = ssl.create_default_context(purpose, cafile=cafile)
    context.load_cert_chain(certfile)

    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((host, port))
    listener.listen(1)
    #将参数设置为server_side=True是因为通信双方必须有一方负责服务器功能。即表明这是服务器
    print('Listening at interface {!r} and port {}'.format(host, port))
    raw_sock, address = listener.accept()
    print('Connection from host {!r} and port {}'.format(*address))
    ssl_sock = context.wrap_socket(raw_sock, server_side=True)

    ssl_sock.sendall('Simple is better than complex.'.encode('ascii'))
    ssl_sock.close()

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Safe TLS client and server')
    parser.add_argument('host', help='hostname or IP address')
    parser.add_argument('port', type=int, help='TCP port number')
    # 客户端-a参数指定签名文件
    parser.add_argument('-a', metavar='cafile', default=None,
                        help='authority: path to CA certificate PEM file')
    #参数-s指定服务器证书文件.pem,也指定运行的是服务器代码。
    parser.add_argument('-s', metavar='certfile', default=None,
                        help='run as server: path to server PEM file')
    args = parser.parse_args()
    if args.s:
        server(args.host, args.port, args.s, args.a)
    else:
        client(args.host, args.port, args.a)
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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

       从以上的代码可以看出。为一个套接字提供安全通信只需要3个步骤。 第一步 是创建一个TLS上下文(context)对象。该对象保存了我们对证书与加密算法选择的设置偏好。 第二步是调用上下文对象的wrap_socket()方法，表示让OpenSSL库负责控制我们的TCP连接，然后与通信对方交换必要的握手信息，并建立加密连接。 最后一步是使用wrap_socket()调用返回的ssl_sock对象，进行所有的后续通信。
测试结果：

       传输的数据的内容都将会被加密算法加密，使用抓包工具将无法识别数据内容。这种模式就是在连接套接字前先包装套接字。如果是客户端套接字的话，就先包装套接字，然后运行connect()；如果是服务器套接字的话，就先包装套接字，然后运行accep()。

4.指定加密算法的套接字通信。

代码：

#!/usr/bin/python
#coding:utf-8

import argparse,socket,ssl,sys,textwrap
import ctypes
#使打印的消息更加规范
from pprint import pprint

#设置TLS连接
def open_tls(context,address,server=False):
    raw_sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    if server:
        raw_sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
        raw_sock.bind(address)
        raw_sock.listen(1)
        say('Interface where we are listening',address)
        raw_client_sock,address = raw_sock.accept()
        say('Client has connected from address',address)
        return context.wrap_socket(raw_client_sock,server_side=True)
    else:
        say('Address we want to talk to',address)
        raw_sock.connect(address)
        return context.wrap_socket(raw_sock)

#获取主机名并验证签名信息
def describe(ssl_sock, hostname, server=False, debug=False):
    cert = ssl_sock.getpeercert()
    if cert is None:
        say('Peer certificate', 'none')
    else:
        say('Peer certificate', 'provided')
        subject = cert.get('subject', [])
        names = [name for names in subject for (key, name) in names
                 if key == 'commonName']
        if 'subjectAltName' in cert:
            names.extend(name for (key, name) in cert['subjectAltName']
                         if key == 'DNS')

        say('Name(s) on peer certificate', *names or ['none'])
        if (not server) and names:
            try:
                ssl.match_hostname(cert, hostname)
            except ssl.CertificateError as e:
                message = str(e)
            else:
                message = 'Yes'
            say('Whether name(s) match the hostname', message)
        for category, count in sorted(context.cert_store_stats().items()):
            say('Certificates loaded of type {}'.format(category), count)

    try:
        protocol_version = SSL_get_version(ssl_sock)
    except Exception:
        if debug:
            raise
    else:
        say('Protocol version negotiated', protocol_version)

    cipher, version, bits = ssl_sock.cipher()
    compression = ssl_sock.compression()

    say('Cipher chosen for this connection', cipher)
    say('Cipher defined in TLS version', version)
    say('Cipher key has this many bits', bits)
    say('Compression algorithm in use', compression or 'none')

    return cert

class PySSLSocket(ctypes.Structure):
    """The first few fields of a PySSLSocket (see Python's Modules/_ssl.c)."""

    _fields_ = [('ob_refcnt', ctypes.c_ulong), ('ob_type', ctypes.c_void_p),
                ('Socket', ctypes.c_void_p), ('ssl', ctypes.c_void_p)]

#获取ssl的版本
def SSL_get_version(ssl_sock):
    """Reach behind the scenes for a socket's TLS protocol version."""

    lib = ctypes.CDLL(ssl._ssl.__file__)
    lib.SSL_get_version.restype = ctypes.c_char_p
    address = id(ssl_sock._sslobj)
    struct = ctypes.cast(address, ctypes.POINTER(PySSLSocket)).contents
    version_bytestring = lib.SSL_get_version(struct.ssl)
    return version_bytestring.decode('ascii')

def lookup(prefix, name):
    if not name.startswith(prefix):
        name = prefix + name
    try:
        return getattr(ssl, name)
    except AttributeError:
        matching_names = (s for s in dir(ssl) if s.startswith(prefix))
        message = 'Error: {!r} is not one of the available names:\n {}'.format(
            name, ' '.join(sorted(matching_names)))
        print(fill(message), file=sys.stderr)
        sys.exit(2)

def say(title,*words):
    print(fill(title.ljust(36,'.')+ ' ' + ' '.join(str(w) for w in words)))

def fill(text):
    return textwrap.fill(text,subsequent_indent='   ',
                         break_long_words=False,break_on_hyphens=False)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Protect a socket with TLS')
    parser.add_argument('host', help='hostname or IP address')
    parser.add_argument('port', type=int, help='TCP port number')
    #客户端-a参数指定签名文件
    parser.add_argument('-a', metavar='cafile', default=None,
                        help='authority: path to CA certificate PEM file')
    #-c参数指定服务器证书文件.pem
    parser.add_argument('-c', metavar='certfile', default=None,
                        help='path to PEM file with client certificate')
    #-C参数指定加密方式
    parser.add_argument('-C', metavar='ciphers', default='ALL',
                        help='list of ciphers, formatted per OpenSSL')
    #-p参数指定ssl的版本
    parser.add_argument('-p', metavar='PROTOCOL', default='SSLv23',
                        help='protocol version (default: "SSLv23")')
    #-s指定运行服务器
    parser.add_argument('-s', metavar='certfile', default=None,
                        help='run as server: path to certificate PEM file')
    #-d参数启动调试模式
    parser.add_argument('-d', action='store_true', default=False,
                        help='debug mode: do not hide "ctypes" exceptions')
    #-v参数打印证书信息
    parser.add_argument('-v', action='store_true', default=False,
                        help='verbose: print out remote certificate')
    args = parser.parse_args()

    address = (args.host, args.port)
    protocol = lookup('PROTOCOL_', args.p)

    context = ssl.SSLContext(protocol)
    context.set_ciphers(args.C)
    context.check_hostname = False
    if (args.s is not None) and (args.c is not None):
        parser.error('you cannot specify both -c and -s')
    elif args.s is not None:
        context.verify_mode = ssl.CERT_OPTIONAL
        purpose = ssl.Purpose.CLIENT_AUTH
        context.load_cert_chain(args.s)
    else:
        context.verify_mode = ssl.CERT_REQUIRED
        purpose = ssl.Purpose.SERVER_AUTH
        if args.c is not None:
            context.load_cert_chain(args.c)
    if args.a is None:
        context.load_default_certs(purpose)
    else:
        context.load_verify_locations(args.a)

    print()
    ssl_sock = open_tls(context, address, args.s)
    cert = describe(ssl_sock, args.host, args.s, args.d)
    print()
    if args.v:
        pprint(cert)
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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

       自己选择加密协议能实现 前向安全，即就算某种加密算法过时，你也能指定更完美的加密算法，用来保证通信的安全性。还有就是连接的安全程度其实是由服务器负责决定的，客户端的作者通常只是希望在不完全暴露数据的情况下让操作尽可能的按照他的想法进行。
测试结果：

附上查看您的ssl模块并报告您的Python版本中提供的选项代码：

try:
    import ssl
except ImportError:
    ssl = None

def main():
    if ssl is None:
        print('This Python is not compiled with SSL support')
        return
    names = dir(ssl)
    print()
    display(names, ' protocol ', lambda s: s.startswith('PROTOCOL_'))
    display(names, ' verify_mode ', lambda s: s.startswith('CERT_'))
    display(names, ' verify_flags ', lambda s: s.startswith('VERIFY_'))
    display(names, ' options ', lambda s: s.startswith('OP_'))
    display(names, ' feature availability ', lambda s: s.startswith('HAS_'))

def display(names, title, test):
    items = [(fix(getattr(ssl, name)), name) for name in names if test(name)]
    print(title.center(72, '-'))
    for value, name in sorted(items):
        print('{:27} {:10}  {:>32}'.format(name, value, bin(value)[2:]))
    print()

def fix(value):
    """Turn negative 32-bit numbers into positive numbers."""
    return (value + 2 ** 32) if (value < 0) else value

if __name__ == '__main__':
    main()
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
28
29
30