使用codecs自定义编/解码方案

发表于 | 分类于 | 阅读次数:

相信很多使用Python的同学熟悉编解码，比如：

1
2
3
4
5

In : print u'\U0001F3F4'.encode('utf-8')  
🏴  
In : '哈哈'.decode('utf8')  
Out: u'\u54c8\u54c8'

这样可以在字符串和unicode之前转换，不过细心的同学可能发现了，我使用了「utf-8」和「utf8」，这2个词长得很像。事实上都能正常使用是由于他们都是「utf_8」的别名，这些别名的对应关系可以用如下方法找到：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

In : import encodings  
  
In : encodings.aliases.aliases['utf8']  
Out: 'utf_8'  
  
In : '哈哈'.decode('u8')  
Out: u'\u54c8\u54c8'  
  
In : '哈哈'.decode('utf')  
Out: u'\u54c8\u54c8'  
  
In : '哈哈'.decode('utf8_ucs2')  
Out: u'\u54c8\u54c8'  
  
In : '哈哈'.decode('utf8_ucs4')  
Out: u'\u54c8\u54c8'

encodings是标准库中自带的编码库，其中包含了上百个标准的编码转换方案。但是通常并不需要使用encodings，而是使用codecs。
codecs包含了编码解码器的注册和其他基本的类，开发者还可以通过codecs提供的接口自定义编/解码方案，也就是可以创造一个新的编解码转换方案，使用encode(‘XX’)和decode(‘XX’)的方式使用。今天我给大家演示直接进行Fernet对称加密的例子。
互联网安全的重要性不必在复述了，大家都应该接触过一些加密技术，可能听过M2Crypto、PyCrypto、Cryptography之类的库。在这里歪个楼，现在的主流是使用Cryptography，它的出现就是为了替代之前的那些库，具体的可以看官方文档。我们使用Cryptography提供的Fernet类来实现，首先实现一个Codec类：

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

import codecs  
  
from cryptography.fernet import Fernet  
from cryptography.fernet import InvalidToken  
  
KEY = Fernet.generate_key()  
f = Fernet(KEY)  
  
  
class FernetCodec(codecs.Codec):  
    def encode(self, input, errors='fernet.strict'):  
        return f.encrypt(input), len(input)  
  
    def decode(self, input, errors='fernet.strict'):  
        try:  
            return f.decrypt(input), len(input)  
        except InvalidToken:  
            error = codecs.lookup_error(errors)  
            return error(UnicodeDecodeError('fernet', input, 0, len(input) + 1,  
                                            'Invalid Token'))

当然也不必在类中实现encode和decode方法，单独的2个函数也可以。我这里是为了给之后的演示到的类复用。
如果你看过内置的字符串encode的方法，它的文档说还接收第二个参数，让你告诉它当出现出错的时候如何去处理，默认是strict，直接就会抛出来错误。
其余可选的还有ignore、replace、xmlcharrefreplace：

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

In [35]: '\x80abc'.decode('utf-8', 'strict')  
---------------------------------------------------------------------------  
UnicodeDecodeError                        Traceback (most recent call last)  
<ipython-input-35-974ebc908d50> in <module>()  
----> 1 '\x80abc'.decode('utf-8', 'strict')  
  
/Users/dongweiming/dae/venv/lib/python2.7/encodings/utf_8.pyc in decode(input, errors)  
 
     15 def decode(input, errors='strict'):  
---> 16     return codecs.utf_8_decode(input, errors, True)  
 
     18 class IncrementalEncoder(codecs.IncrementalEncoder):  
  
UnicodeDecodeError: 'utf8' codec can't decode byte 0x80 in position 0: invalid start byte  
  
In [36]: '\x80abc'.decode('utf-8', 'replace')  
Out[36]: u'\uffabc'  
  
In [37]: '\x80abc'.decode('utf-8', 'ignore')  
Out[37]: u'abc'

事实上Python还内置了其他的选项，如backslashreplace、namereplace、surrogatepass、surrogateescape等，有兴趣的可以看源码实现.
我也会定义2种错误函数，因为在解密（执行decrypt）的时候可能会报InvalidToken错误，但是InvalidToken不包含任何参数，而对错误处理的时候需要知道起始和结束的位置，所以我就直接抛一个UnicodeDecodeError错误了。
接着我们定义递增式和流式的编码类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

class IncrementalEncoder(codecs.IncrementalEncoder, FernetCodec):  
    pass  
  
  
class IncrementalDecoder(codecs.IncrementalDecoder, FernetCodec):  
    pass  
  
  
class StreamReader(FernetCodec, codecs.StreamReader):  
    pass  
  
  
class StreamWriter(FernetCodec, codecs.StreamWriter):  
    pass

由于这个要加密的字符串不会很长，没有必要实现这些类，我就简单的继承然后pass了。接着开开放入口：

1
2
3
4
5
6
7
8
9
10
11

def getregentry():  
    return codecs.CodecInfo(  
        name='fernet',  
        encode=FernetCodec().encode,  
        decode=FernetCodec().decode,  
        incrementalencoder=IncrementalEncoder,  
        incrementaldecoder=IncrementalDecoder,  
        streamwriter=StreamWriter,  
        streamreader=StreamReader,  
    )

其实incrementalencoder、streamwriter这些参数不传递也是可以的，默认是None，今天只是为了让大家知道是有这部分接口的。然后是注册这个入口，为了提供更好的性能，我创建一个函数加上缓存功能：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

_cache = {}  
_unknown = '--unknown--'  
  
def search_function(encoding):  
    import encodings  
    encoding = encodings.normalize_encoding(encoding)  
    entry = _cache.get(encoding, _unknown)  
    if entry is not _unknown:  
        return entry  
  
    if encoding == 'fernet':  
        entry = getregentry()  
        _cache[encoding] = entry  
        return entry  
                         
                         
codecs.register(search_function)

这里简单介绍下normalize_encoding的意义，之前我说到了别名，我们再感受下：

1
2
3
4
5
6

In : u'哈哈'.encode('utf-8')  
Out: '\xe5\x93\x88\xe5\x93\x88'  
  
In : u'哈哈'.encode('utf_8')  
Out: '\xe5\x93\x88\xe5\x93\x88'

这种中划线和下划线最后无差别对待，就是通过这个函数标准化的。
codecs模块底层维护了一个搜索函数的列表，通过调用codecs.register方法就把上述函数append进去了。
最后我们注册2个错误处理的函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

def strict_errors(exc):  
    if isinstance(exc, UnicodeDecodeError):  
        raise TypeError('Invalid Token')  
  
  
def fallback_errors(exc):  
    s = []  
    for c in exc.object[exc.start:exc.end]:  # 只是为了演示提供的接口，实施上就是返回原来的input  
        s.append(c)  
    return ''.join(s), exc.end  
  
codecs.register_error('fernet.strict', strict_errors)  
codecs.register_error('fernet.fallback', fallback_errors)

默认使用的是fernet.strict这种处理方案，也可以使用fallback模式。ok，我们现在感受一下：

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

In [1]: import fernet  
  
In [2]: input = 'hah'  
  
In [3]: output = input.encode('fernet')  
  
In [4]: output  # 已经是加密后的结果了  
Out[4]: 'gAAAAABZCFa6Znp2a_e9O0VqP6qToO6T3xRbF7O-adtpFC4QYO7jvVc6Yrcwbo6YGQfL8g5HCXcsaan_THWNhjZAorPTwlQQTA=='  
  
In [5]: output.decode('fernet')  # 解密后还原成原来的字符串  
Out[5]: 'hah'  
  
In [6]: input.decode('fernet')  
'fernet' codec can't decode bytes in position 0-3: Invalid Token  
---------------------------------------------------------------------------  
TypeError                                 Traceback (most recent call last)  
<ipython-input-6-1c20dc246c52> in <module>()  
----> 1 input.decode('fernet')  
  
/Users/dongweiming/test/tmp/fernet.pyc in decode(self, input, errors)  
     20             error = codecs.lookup_error(errors)  
     21             return error(UnicodeDecodeError('fernet', input, 0, len(input) + 1,  
---> 22                                             'Invalid Token'))  
 
 
  
/Users/dongweiming/test/tmp/fernet.pyc in strict_errors(exc)  
     68     print exc  
     69     if isinstance(exc, UnicodeDecodeError):  
---> 70         raise TypeError('Invalid Token')  
 
 
  
TypeError: Invalid Token  # 抛了个自定义错误  
  
In [7]: input.decode('fernet', 'fernet.fallback')  
Out[7]: 'hah'  # 解密不成功，返回原来的字符串

好了，自定义的加解密方案完成了，整理全部的代码如下：

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

import codecs  
  
from cryptography.fernet import Fernet  
from cryptography.fernet import InvalidToken  
  
KEY = Fernet.generate_key()  
f = Fernet(KEY)  
_cache = {}  
_unknown = '--unknown--'  
  
  
class FernetCodec(codecs.Codec):  
    def encode(self, input, errors='fernet.strict'):  
        return f.encrypt(input), len(input)  
  
    def decode(self, input, errors='fernet.strict'):  
        try:  
            return f.decrypt(input), len(input)  
        except InvalidToken:  
            error = codecs.lookup_error(errors)  
            return error(UnicodeDecodeError('fernet', input, 0, len(input) + 1,  
                                            'Invalid Token'))  
  
  
class IncrementalEncoder(codecs.IncrementalEncoder, FernetCodec):  
    pass  
  
  
class IncrementalDecoder(codecs.IncrementalDecoder, FernetCodec):  
    pass  
  
  
class StreamReader(FernetCodec, codecs.StreamReader):  
    pass  
  
  
class StreamWriter(FernetCodec, codecs.StreamWriter):  
    pass  
  
  
  
def getregentry():  
    return codecs.CodecInfo(  
        name='fernet',  
        encode=FernetCodec().encode,  
        decode=FernetCodec().decode,  
        incrementalencoder=IncrementalEncoder,  
        incrementaldecoder=IncrementalDecoder,  
        streamwriter=StreamWriter,  
        streamreader=StreamReader,  
    )  
  
  
def search_function(encoding):  
    import encodings  
    encoding = encodings.normalize_encoding(encoding)  
    entry = _cache.get(encoding, _unknown)  
    if entry is not _unknown:  
        return entry  
  
    if encoding == 'fernet':  
        entry = getregentry()  
        _cache[encoding] = entry  
        return entry  
  
  
def strict_errors(exc):  
    if isinstance(exc, UnicodeDecodeError):  
        raise TypeError('Invalid Token')  
  
  
def fallback_errors(exc):  
    s = []  
    for c in exc.object[exc.start:exc.end]:  
        s.append(c)  
    return ''.join(s), exc.end  
  
  
codecs.register(search_function)  
codecs.register_error('fernet.strict', strict_errors)  
codecs.register_error('fernet.fallback', fallback_errors)

版权声明：本文由 董伟明 原创，未经作者授权禁止任何微信公众号和向掘金(juejin.im)转载，技术博客转载采用 保留署名-非商业性使用-禁止演绎 4.0-国际许可协议
python