Das deutsche Python-Forum

Code: Alles auswählen

from __future__ import absolute_import, division, print_function

import base64
import binascii
import os
import struct
import time

import six

from cryptography import utils
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.hmac import HMAC

_MAX_CLOCK_SKEW = 20

class InvalidToken(Exception):
    pass

class _FernetEncryptor(object):
    def __init__(self, signing_key, encryption_key, backend):
        self.initialized = False
        self.iv = os.urandom(16)
        self.padder = padding.PKCS7(algorithms.AES.block_size).padder()
        self.encryptor = Cipher(
            algorithms.AES(encryption_key), modes.CBC(self.iv), backend
        ).encryptor()
        self.hmac = HMAC(signing_key, hashes.SHA256(), backend=backend)

    def update(self, data):
        utils._check_bytes("data", data)
        padded_data = self.padder.update(data)
        ciphertext = self.encryptor.update(padded_data)
        if not self.initialized:
            self.initialized = True
            current_time = int(time.time())
            ciphertext = (
                b"\x80" + struct.pack(">Q", current_time) +
                self.iv + ciphertext
            )
        self.hmac.update(ciphertext)
        return ciphertext

    def finalize(self):
        padded_data = self.padder.finalize()
        ciphertext = self.encryptor.update(padded_data)
        ciphertext += self.encryptor.finalize()
        self.hmac.update(ciphertext)
        hmac = self.hmac.finalize()
        return ciphertext + hmac


class _FernetDecryptor(object):
    def __init__(self, ttl, signing_key, encryption_key, backend):
        self.ttl = ttl
        self._encryption_key = encryption_key
        self._backend = backend
        self.initalized = False
        self.unprocessed_data = b""
        self.hmac = HMAC(signing_key, hashes.SHA256(), backend=backend)
        self.unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
        self.decryptor = None
        self.timestamp = None

    def _initialize(self):
        version, timestamp = struct.unpack_from(">BQ", self.unprocessed_data, 0)
        if version != 0x80:
            raise InvalidToken
        if self.ttl is not None:
            current_time = int(time.time())
            if timestamp + ttl < current_time:
                raise InvalidToken
            if current_time + _MAX_CLOCK_SKEW < timestamp:
                raise InvalidToken
        self.timestamp = timestamp
        self.initalized = True
        iv = self.unprocessed_data[9:25]
        self.decryptor = Cipher(
            algorithms.AES(self._encryption_key), modes.CBC(iv), self._backend
        ).decryptor()
        self.hmac.update(self.unprocessed_data[:25])
        self.unprocessed_data = self.unprocessed_data[25:]

    def update(self, data):
        utils._check_bytes("data", data)
        self.unprocessed_data += data
        if not self.initalized:
            if len(self.unprocessed_data) < 25:
                return b""
            self._initialize()
        ciphertext = self.unprocessed_data[:-32]
        self.unprocessed_data = self.unprocessed_data[-32:]
        self.hmac.update(ciphertext)
        plaintext_padded = self.decryptor.update(ciphertext)
        return self.unpadder.update(plaintext_padded)

    def finalize(self):
        try:
            self.hmac.verify(self.unprocessed_data)
        except InvalidSignature:
            raise InvalidToken
        try:
            plaintext_padded = self.decryptor.finalize()
            unpadded = self.unpadder.update(plaintext_padded)
            unpadded += self.unpadder.finalize()
        except ValueError:
            raise InvalidToken
        return unpadded

class FernetStream(object):
    def __init__(self, key, backend=None):
        if backend is None:
            backend = default_backend()
        key = base64.urlsafe_b64decode(key)
        if len(key) != 32:
            raise ValueError(
                "Fernet key must be 32 url-safe base64-encoded bytes."
            )
        self._signing_key = key[:16]
        self._encryption_key = key[16:]
        self._backend = backend

    @classmethod
    def generate_key(cls):
        return base64.urlsafe_b64encode(os.urandom(32))

    def encryptor(self):
        return _FernetEncryptor(self._signing_key,
            self._encryption_key, self._backend)

    def decryptor(self, ttl=None):
        return _FernetDecryptor(ttl, self._signing_key,
            self._encryption_key, self._backend)

Code: Alles auswählen

def verschlüsseln(passwort, dateipfad):
    with open(dateipfad, "r+b") as datei:
        with open(dateipfad + ".verschluesselt", "ab") as datei_verschlüsselt:
            schlüssel, salt = generiere_Schlüssel(passwort)
            f = Fernet.FernetStream(schlüssel)
            encryptor = f.encryptor()
            datei_verschlüsselt.write(salt)
            for i in range(os.path.getsize(dateipfad)):
                byte = datei.read(1)
                datei_verschlüsselt.write(encryptor.update(byte))