Шифрование файлов в Python: разбор на живом проекте

Скрипт бэкапов читал базу целиком в память, шифровал через Fernet и складывал в S3. Работало год — пока дамп не дорос до 3,4 ГБ. Тогда воркер с 2 ГБ RAM начал падать с MemoryError, а половина ночных бэкапов оказалась пустыми файлами. Разберём, почему наивное шифрование файлов в Python ломается под реальными данными, и что с этим делать. Всё на библиотеке cryptography версии 42.x, Python 3.11.

Что было в исходном коде

Первоначальный вариант выглядел так — и именно он живёт в половине пет-проектов и внутренних утилит.

# cryptography 42.x, Python 3.11
from cryptography.fernet import Fernet

key = Fernet.generate_key()
f = Fernet(key)

# читаем весь файл в память
with open("dump.sql", "rb") as src:
    data = src.read()

token = f.encrypt(data)

with open("dump.sql.enc", "wb") as dst:
    dst.write(token)

Здесь три проблемы, и все выстреливают именно на больших и боевых данных.

• Память. src.read() держит весь файл в RAM, а encrypt() создаёт ещё одну копию — плюс токен. Пиковое потребление — около трёх размеров файла.
• Fernet не рассчитан на большие файлы. В документации Fernet прямо сказано: он для сообщений, которые целиком помещаются в память. Потоковый режим не предусмотрен.
• Целостность записи. Если процесс убьют посреди dst.write(), на диске останется обрезанный файл, который расшифровать нельзя — а старого уже нет.

Почему потоковое шифрование — не «просто читать по кускам»

Первая мысль — читать файл блоками и шифровать по частям. С Fernet так нельзя: каждый вызов encrypt() порождает самостоятельный токен со своим IV и MAC. Склеить их обратно в один файл и корректно расшифровать штатными средствами не выйдет.

Для потока нужен низкоуровневый примитив с явным режимом сцепления. Берём AES в режиме AES-GCM¹ — он даёт и шифрование, и проверку целостности, но у GCM есть жёсткое ограничение: один и тот же nonce² с одним ключом использовать нельзя никогда. А ещё GCM в реализации AESGCM из cryptography — не потоковый: он тоже требует данные целиком.

Для честного стриминга подходит интерфейс Cipher с режимом CTR плюс отдельный HMAC для целостности, либо готовый пакет с потоковым AEAD. Ниже — вариант на чистой cryptography: encrypt-then-MAC, читаем и пишем блоками.

Потоковый вариант: encrypt-then-MAC

# cryptography 42.x, Python 3.11
import os
import hmac as hmac_mod
import hashlib
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

CHUNK = 64 * 1024  # 64 КБ за раз

def encrypt_file(src_path, dst_path, enc_key, mac_key):
    # enc_key и mac_key — по 32 байта, независимые
    nonce = os.urandom(16)  # для CTR это начальное значение счётчика
    cipher = Cipher(algorithms.AES(enc_key), modes.CTR(nonce))
    encryptor = cipher.encryptor()
    mac = hmac_mod.new(mac_key, digestmod=hashlib.sha256)

    tmp_path = dst_path + ".part"  # пишем во временный файл
    with open(src_path, "rb") as src, open(tmp_path, "wb") as dst:
        dst.write(nonce)          # nonce в начало файла
        mac.update(nonce)
        while True:
            block = src.read(CHUNK)
            if not block:
                break
            ct = encryptor.update(block)
            dst.write(ct)
            mac.update(ct)         # MAC считаем по шифртексту
        dst.write(encryptor.finalize())
        dst.write(mac.digest())    # тег целостности в хвост
        dst.flush()
        os.fsync(dst.fileno())     # гарантируем запись на диск

    # атомарная замена: либо старый файл, либо полный новый
    os.replace(tmp_path, dst_path)

Память здесь фиксированная — 64 КБ независимо от размера дампа. Запись идёт во временный .part, а os.replace() подменяет цель атомарно: обрыв процесса оставит старый бэкап нетронутым.

Расшифровка с проверкой тега

def decrypt_file(src_path, dst_path, enc_key, mac_key):
    size = os.path.getsize(src_path)
    tag_len = 32           # sha256
    nonce_len = 16
    body_len = size - nonce_len - tag_len
    if body_len < 0:
        raise ValueError("файл повреждён или слишком короткий")

    with open(src_path, "rb") as src:
        nonce = src.read(nonce_len)
        mac = hmac_mod.new(mac_key, digestmod=hashlib.sha256)
        mac.update(nonce)
        cipher = Cipher(algorithms.AES(enc_key), modes.CTR(nonce))
        decryptor = cipher.decryptor()

        tmp_path = dst_path + ".part"
        remaining = body_len
        with open(tmp_path, "wb") as dst:
            while remaining > 0:
                block = src.read(min(CHUNK, remaining))
                remaining -= len(block)
                mac.update(block)
                dst.write(decryptor.update(block))
            dst.write(decryptor.finalize())
        stored_tag = src.read(tag_len)

    # сверяем тег ДО того, как отдать файл наружу
    if not hmac_mod.compare_digest(mac.digest(), stored_tag):
        os.remove(tmp_path)
        raise ValueError("тег не совпал — файл повреждён или подменён")
    os.replace(tmp_path, dst_path)

Тонкий момент: тег проверяется compare_digest — сравнение за постоянное время, чтобы не давать подсказок по таймингу. И проверяем мы его после того, как весь файл прочитан, но до os.replace — расшифрованный мусор наружу не попадёт.

Encrypt-then-MAC — не стилистика, а порядок: сначала считаем MAC по шифртексту, при чтении сначала сверяем тег, и только потом доверяем данным. Проверять целостность после расшифровки — классическая дыра.

Откуда брать ключи

Пароль пользователя нельзя использовать как ключ AES напрямую. Прогоняем его через KDF. В боевом коде для этого проекта взяли Scrypt из cryptography — он тяжёл для перебора и даёт два независимых ключа из одного пароля через разные соли.

from cryptography.hazmat.primitives.kdf.scrypt import Scrypt

def derive_keys(password: bytes, salt: bytes):
    # 64 байта = 32 на шифрование + 32 на MAC
    kdf = Scrypt(salt=salt, length=64, n=2**15, r=8, p=1)
    material = kdf.derive(password)
    return material[:32], material[32:]

Соль (os.urandom(16)) генерируется при шифровании и пишется в файл рядом с nonce — она не секретна. Секретен только пароль. Параметры n, r, p подбирайте по актуальным рекомендациям в документации KDF: n=2**15 — компромисс, на слабом сервере он может занять заметное время.

Что сломалось на бою и как чинили

Ротация ключа

Через полгода понадобилось сменить пароль. Оказалось, что старые файлы никак не помечены версией ключа — непонятно, каким паролем что шифровалось. Лечится заголовком: первый байт файла — номер версии схемы, дальше соль и nonce. При смене ключа увеличиваете версию, а при расшифровке по байту понимаете, какой набор параметров применять.

Замер: память и время

Сравнение на файле 3,4 ГБ, сервер 2 vCPU / 2 ГБ RAM, диск SSD. Замеряйте у себя — цифры зависят от железа и параметров Scrypt; ниже порядок величин, а не гарантия.

Размер чанка от 32 КБ до 1 МБ на общее время влияет слабо — упирается в диск и в саму AES. Ставить 1 байт или весь файл — обе крайности плохи.

Где этот подход ломается

• Повтор nonce. Если os.urandom подменить на детерминированный источник и зашифровать два файла одним ключом с одинаковым nonce, CTR раскрывает XOR открытых текстов. Nonce обязан быть случайным на каждое шифрование.
• Разные ключи для шифрования и MAC. Использовать один ключ и там, и там — ошибка. Мы получаем оба из KDF независимо, это принципиально.
• Симметричное шифрование. Пароль нужен и для шифровки, и для расшифровки. Если файлы шифрует один сервис, а читает другой без общего секрета — нужна асимметрия (обмен ключами), это отдельная тема, здесь её не трогаем.
• Метаданные. Имя файла, размер и время доступа не шифруются. Если утечёт сам факт «есть бэкап на 3 ГБ от вторника» — это уже утечка.

Что делать дальше

Если пишете это не для одного скрипта, а всерьёз — посмотрите на готовые потоковые AEAD-решения вместо ручного encrypt-then-MAC: собственная сборка криптопримитивов легко даёт незаметную дыру. Мерьте пиковую память под реальным размером данных, а не на тестовом килобайте, и обязательно проверьте сценарий обрыва процесса посреди записи — убейте воркер во время шифрования и убедитесь, что старый бэкап цел. В проде смотрите на время работы KDF: если оно растёт, ротация паролей превратится в узкое место.