Django自定义Fernet加密字段无法解密问题排查

问题场景 #

在Django模型中使用自定义SecureString字段，迁移操作成功且能向数据表插入加密值，但查询数据时解密失败，抛出InvalidToken错误。该加密解密逻辑在独立Python脚本中可正常运行，使用的数据库为PostgreSQL 12.4。

自定义字段代码：

class SecureString(CharField):
    """Custom Encrypted Field"""

    #kdf = X963KDF(algorithm=hashes.SHA256(),
                    # length=32,
                     #sharedinfo=None,
                    # backend=default_backend())

    key = bytes(settings.FERNET_KEY,'utf-8')
    f = Fernet(key)

    def from_db_value(self, value, expression, connection):
        return self.f.decrypt(str.encode(value))

    def get_prep_value(self, value):
        return self.f.encrypt(bytes(value, 'utf-8'))

错误栈信息 #

查询时触发的错误：

File /usr/local/lib/python3.8/site-packages/IPython/core/formatters.py:706, in PlainTextFormatter.__call__(self, obj)
    699 stream = StringIO()
    700 printer = pretty.RepresentationPrinter(stream, self.verbose,
    701     self.max_width, self.newline,
    702     max_seq_length=self.max_seq_length,
    703     singleton_pprinters=self.singleton_printers,
    704     type_pprinters=self.type_printers,
    705     deferred_pprinters=self.deferred_printers)
--> 706 printer.pretty(obj)
    707 printer.flush()
    708 return stream.getvalue()

File /usr/local/lib/python3.8/site-packages/IPython/lib/pretty.py:410, in RepresentationPrinter.pretty(self, obj)
    407                         return meth(obj, self, cycle)
    408                 if cls is not object \
    409                         and callable(cls.__dict__.get('__repr__')):
--> 410                     return _repr_pprint(obj, self, cycle)
    412     return _default_pprint(obj, self, cycle)
    413 finally:

File /usr/local/lib/python3.8/site-packages/IPython/lib/pretty.py:778, in _repr_pprint(obj, p, cycle)
    776 """A pprint that just redirects to the normal repr function."""
    777 # Find newlines and replace them with p.break_()
--> 778 output = repr(obj)
    779 lines = output.splitlines()
    780 with p.group():

File /usr/local/lib/python3.8/site-packages/django/db/models/query.py:370, in QuerySet.__repr__(self)
    369 def __repr__(self):
--> 370     data = list(self[: REPR_OUTPUT_SIZE + 1])
    371     if len(data) > REPR_OUTPUT_SIZE:
    372         data[-1] = "...(remaining elements truncated)..."

File /usr/local/lib/python3.8/site-packages/django/db/models/query.py:376, in QuerySet.__len__(self)
    375 def __len__(self):
--> 376     self._fetch_all()
    377     return len(self._result_cache)

File /usr/local/lib/python3.8/site-packages/django/db/models/query.py:1867, in QuerySet._fetch_all(self)
   1865 def _fetch_all(self):
   1866     if self._result_cache is None:
-> 1867         self._result_cache = list(self._iterable_class(self))
   1868     if self._prefetch_related_lookups and not self._prefetch_done:
   1869         self._prefetch_related_objects()

File /usr/local/lib/python3.8/site-packages/django/db/models/query.py:204, in ValuesIterable.__iter__(self)
    198 names = [
    199     *query.extra_select,
    200     *query.values_select,
    201     *query.annotation_select,
    202 ]
    203 indexes = range(len(names))
--> 204 for row in compiler.results_iter(
    205     chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size
    206 ):
    207     yield {names[i]: row[i] for i in indexes}

File /usr/local/lib/python3.8/site-packages/django/db/models/sql/compiler.py:1336, in SQLCompiler.apply_converters(self, rows, converters)
   1334     value = row[pos]
   1335     for converter in convs:
-> 1336         value = converter(value, expression, connection)
   1337     row[pos] = value
   1338 yield row

File /code/server/identity/model_mixins.py:61, in SecureString.from_db_value(self, value, expression, connection)
     59 key = bytes(settings.FERNET_KEY, 'utf-8')
     60 f = Fernet(key)
--> 61 return f.decrypt(str.encode(value))

File /usr/local/lib/python3.8/site-packages/cryptography/fernet.py:86, in Fernet.decrypt(self, token, ttl)
     83 def decrypt(
     84     self, token: typing.Union[bytes, str], ttl: typing.Optional[int] = None
     85 ) -> bytes:
--> 86     timestamp, data = Fernet._get_unverified_token_data(token)
     87     if ttl is None:
     88         time_info = None

File /usr/local/lib/python3.8/site-packages/cryptography/fernet.py:119, in Fernet._get_unverified_token_data(token)
    117     data = base64.urlsafe_b64decode(token)
    118 except (TypeError, binascii.Error):
--> 119     raise InvalidToken
    121 if not data or data[0] != 0x80:
    122     raise InvalidToken

InvalidToken: 

关键差异点 #

数据库中存储的值为十六进制格式：\x674141414141426b384d34596e535f6d6a6b3173556b5344455f57585a58326851374b35576c56536938676d3358556165515a767a677a754f6c4b7a646c613276397644455874675276795971724b794e616744596f6d516d78694a5167335334654d526b784d495156566d7344575242776e3639487143515044334f676e65334e524268624a5742786f544258336f435930675f6c646c6c524a44714b56466d4f7836575f5f6c5772556e4a6d4f3372325372636b33536a6a346d7a536c56415635386f474c44596d646962774c42384e4e4d645657464a3743567a55597854673d3d，而独立脚本中使用同一密钥加密后的值为标准Base64字符串：gAAAAABk8M1BhMeOF10wcEZ7U6zb_vQpaQ8zHxlDuGLyFCA6JQu0NSYfshulqorntWQS4OF7PAyyQ7BCJZ2r0QNt7e8FBZbjTQ==。

问题原因 #

1. 字段类型不匹配：SecureString继承自CharField，但Fernet加密结果是字节串，PostgreSQL会将字节串以十六进制格式存储（\x开头），而非预期的Base64字符串。解密时直接对该十六进制字符串编码后解密，无法解析为有效的Fernet token。
2. 类属性初始化风险：key和f作为类属性，会在类加载时初始化一次。若settings.FERNET_KEY在运行时发生变化，会导致加密和解密使用的Fernet实例不一致，不过这并非当前问题的核心原因。

解决方案 #

方案1：改用BinaryField存储二进制数据 #

Fernet加密结果是字节串，BinaryField适合直接存储二进制数据，无需格式转换：

from django.db import models
from cryptography.fernet import Fernet
from django.conf import settings

class SecureString(models.BinaryField):
    """Custom Encrypted Field"""

    def from_db_value(self, value, expression, connection):
        if value is None:
            return value
        key = bytes(settings.FERNET_KEY, 'utf-8')
        f = Fernet(key)
        return f.decrypt(value).decode('utf-8')

    def get_prep_value(self, value):
        if value is None:
            return value
        key = bytes(settings.FERNET_KEY, 'utf-8')
        f = Fernet(key)
        return f.encrypt(bytes(value, 'utf-8'))

方案2：保持CharField，转存Base64字符串 #

若必须使用CharField，需将加密后的字节串转为Base64字符串存储，解密时先解码Base64：

from django.db import models
from cryptography.fernet import Fernet
from django.conf import settings
import base64

class SecureString(models.CharField):
    """Custom Encrypted Field"""

    def from_db_value(self, value, expression, connection):
        if value is None:
            return value
        key = bytes(settings.FERNET_KEY, 'utf-8')
        f = Fernet(key)
        # 将存储的Base64字符串解码为字节串
        token_bytes = base64.urlsafe_b64decode(value)
        return f.decrypt(token_bytes).decode('utf-8')

    def get_prep_value(self, value):
        if value is None:
            return value
        key = bytes(settings.FERNET_KEY, 'utf-8')
        f = Fernet(key)
        # 将加密后的字节串转为Base64字符串存储
        token_bytes = f.encrypt(bytes(value, 'utf-8'))
        return base64.urlsafe_b64encode(token_bytes).decode('utf-8')

额外优化：避免类属性初始化问题 #

不要将Fernet实例作为类属性，而是在每个方法内部初始化，或使用实例属性，避免类加载时初始化导致的配置不一致问题。

内容的提问来源于stack exchange，提问作者Tyler