| 26 | |
| 27 | |
| 28 | class Fernet: |
| 29 | def __init__( |
| 30 | self, |
| 31 | key: bytes | str, |
| 32 | backend: typing.Any = None, |
| 33 | ) -> None: |
| 34 | try: |
| 35 | key = base64.urlsafe_b64decode(key) |
| 36 | except binascii.Error as exc: |
| 37 | raise ValueError( |
| 38 | "Fernet key must be 32 url-safe base64-encoded bytes." |
| 39 | ) from exc |
| 40 | if len(key) != 32: |
| 41 | raise ValueError( |
| 42 | "Fernet key must be 32 url-safe base64-encoded bytes." |
| 43 | ) |
| 44 | |
| 45 | self._signing_key = key[:16] |
| 46 | self._encryption_key = key[16:] |
| 47 | |
| 48 | @classmethod |
| 49 | def generate_key(cls) -> bytes: |
| 50 | return base64.urlsafe_b64encode(os.urandom(32)) |
| 51 | |
| 52 | def encrypt(self, data: bytes) -> bytes: |
| 53 | return self.encrypt_at_time(data, int(time.time())) |
| 54 | |
| 55 | def encrypt_at_time(self, data: bytes, current_time: int) -> bytes: |
| 56 | iv = os.urandom(16) |
| 57 | return self._encrypt_from_parts(data, current_time, iv) |
| 58 | |
| 59 | def _encrypt_from_parts( |
| 60 | self, data: bytes, current_time: int, iv: bytes |
| 61 | ) -> bytes: |
| 62 | utils._check_bytes("data", data) |
| 63 | |
| 64 | padder = padding.PKCS7(algorithms.AES.block_size).padder() |
| 65 | padded_data = padder.update(data) + padder.finalize() |
| 66 | encryptor = Cipher( |
| 67 | algorithms.AES(self._encryption_key), |
| 68 | modes.CBC(iv), |
| 69 | ).encryptor() |
| 70 | ciphertext = encryptor.update(padded_data) + encryptor.finalize() |
| 71 | |
| 72 | basic_parts = ( |
| 73 | b"\x80" |
| 74 | + current_time.to_bytes(length=8, byteorder="big") |
| 75 | + iv |
| 76 | + ciphertext |
| 77 | ) |
| 78 | |
| 79 | h = HMAC(self._signing_key, hashes.SHA256()) |
| 80 | h.update(basic_parts) |
| 81 | hmac = h.finalize() |
| 82 | return base64.urlsafe_b64encode(basic_parts + hmac) |
| 83 | |
| 84 | def decrypt(self, token: bytes | str, ttl: int | None = None) -> bytes: |
| 85 | timestamp, data = Fernet._get_unverified_token_data(token) |
no outgoing calls