使用codecs自定义编/解码方案
/ / / 阅读数:4048相信很多使用 Python 的同学熟悉编解码,比如:
In : print u'\U0001F3F4'.encode('utf-8') 🏴 In : '哈哈'.decode('utf8') Out: u'\u54c8\u54c8' |
这样可以在字符串和 unicode 之前转换,不过细心的同学可能发现了,我使用了「utf-8」和「utf8」,这 2 个词长得很像。事实上都能正常使用是由于他们都是「utf_8」的别名,这些别名的对应关系可以用如下方法找到:
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 类:
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:
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 错误了。
接着我们定义递增式和流式的编码类:
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 了。接着开开放入口:
def getregentry(): return codecs.CodecInfo( name='fernet', encode=FernetCodec().encode, decode=FernetCodec().decode, incrementalencoder=IncrementalEncoder, incrementaldecoder=IncrementalDecoder, streamwriter=StreamWriter, streamreader=StreamReader, ) |
其实 incrementalencoder、streamwriter 这些参数不传递也是可以的,默认是 None,今天只是为了让大家知道是有这部分接口的。然后是注册这个入口,为了提供更好的性能,我创建一个函数加上缓存功能:
_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 的意义,之前我说到了别名,我们再感受下:
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 个错误处理的函数:
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,我们现在感受一下:
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' # 解密不成功,返回原来的字符串
好了,自定义的加解密方案完成了,整理全部的代码如下:
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) |
明白了