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

1
2
3
4
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
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
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
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)
14
15 def decode(input, errors='strict'):
---> 16 return codecs.utf_8_decode(input, errors, True)
17
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
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
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
_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
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
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
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'))
23
24
/Users/dongweiming/test/tmp/fernet.pyc in strict_errors(exc)
68 print exc
69 if isinstance(exc, UnicodeDecodeError):
---> 70 raise TypeError('Invalid Token')
71
72
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
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)