Add Multifernet.rotate method (#3979)
* add rotate method
* add some more tests for the failure modes
* start adding some documentation for the rotate method
* operate on a single token at a time, leave lists to the caller
* add versionadded
add versionadded, drop rotate from class doctest
* give rotate a doctest
* single level, not aligned
* add changelog for mf.rotate
* show that, once rotated, the old fernet instance can no longer decrypt the token
* add the instead of just the how
* update docs to reflect removal of ttl from rotate
* update tests
* refactor internal methods so that we can extract the timestamp
* implement rotate
* update wordlist (case sensitive?)
* lints
* consistent naming
* get_token_data/get_unverified_token_data -> better name
* doc changes
* use the static method, do not treat as imethod
* move up to MultiFernet docs
* add to authors
* alter wording
* monkeypatch time to make it less possible for the test to pass simply due to calls occuring in less than one second
* set the time after encryption to make sure that the time is preserved as part of re-encryption
diff --git a/AUTHORS.rst b/AUTHORS.rst
index 4444bf5..60f5de1 100644
--- a/AUTHORS.rst
+++ b/AUTHORS.rst
@@ -37,3 +37,4 @@
* Ofek Lev <ofekmeister@gmail.com> (FFB6 B92B 30B1 7848 546E 9912 972F E913 DAD5 A46E)
* Erik Daguerre <fallenwolf@wolfthefallen.com>
* Aviv Palivoda <palaviv@gmail.com>
+* Chris Wolfe <chriswwolfe@gmail.com>
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index f25bc28..9e1d0eb 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -9,6 +9,8 @@
.. note:: This version is not yet released and is under active development.
* **BACKWARDS INCOMPATIBLE:** Support for Python 2.6 has been dropped.
+* Added token rotation support to :doc:`Fernet </fernet>` with
+ :meth:`~cryptography.fernet.MultiFernet.rotate`.
.. _v2-1-1:
diff --git a/docs/fernet.rst b/docs/fernet.rst
index 82d94fa..a0ffe64 100644
--- a/docs/fernet.rst
+++ b/docs/fernet.rst
@@ -86,7 +86,8 @@
.. versionadded:: 0.7
This class implements key rotation for Fernet. It takes a ``list`` of
- :class:`Fernet` instances, and implements the same API:
+ :class:`Fernet` instances and implements the same API with the exception
+ of one additional method: :meth:`MultiFernet.rotate`:
.. doctest::
@@ -109,6 +110,50 @@
the front of the list to start encrypting new messages, and remove old keys
as they are no longer needed.
+ Token rotation as offered by :meth:`MultiFernet.rotate` is a best practice
+ and manner of cryptographic hygiene designed to limit damage in the event of
+ an undetected event and to increase the difficulty of attacks. For example,
+ if an employee who had access to your company's fernet keys leaves, you'll
+ want to generate new fernet key, rotate all of the tokens currently deployed
+ using that new key, and then retire the old fernet key(s) to which the
+ employee had access.
+
+ .. method:: rotate(msg)
+
+ .. versionadded:: 2.2
+
+ Rotates a token by re-encrypting it under the :class:`MultiFernet`
+ instance's primary key. This preserves the timestamp that was originally
+ saved with the token. If a token has successfully been rotated then the
+ rotated token will be returned. If rotation fails this will raise an
+ exception.
+
+ .. doctest::
+
+ >>> from cryptography.fernet import Fernet, MultiFernet
+ >>> key1 = Fernet(Fernet.generate_key())
+ >>> key2 = Fernet(Fernet.generate_key())
+ >>> f = MultiFernet([key1, key2])
+ >>> token = f.encrypt(b"Secret message!")
+ >>> token
+ '...'
+ >>> f.decrypt(token)
+ 'Secret message!'
+ >>> key3 = Fernet(Fernet.generate_key())
+ >>> f2 = MultiFernet([key3, key1, key2])
+ >>> rotated = f2.rotate(token)
+ >>> f2.decrypt(rotated)
+ 'Secret message!'
+
+ :param bytes msg: The token to re-encrypt.
+ :returns bytes: A secure message that cannot be read or altered without
+ the key. This is URL-safe base64-encoded. This is referred to as a
+ "Fernet token".
+ :raises cryptography.fernet.InvalidToken: If a ``token`` is in any
+ way invalid this exception is raised.
+ :raises TypeError: This exception is raised if the ``msg`` is not
+ ``bytes``.
+
.. class:: InvalidToken
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index c53cc80..4cf31f5 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -21,6 +21,7 @@
cryptographically
Debian
decrypt
+decrypts
Decrypts
decrypted
decrypting
diff --git a/src/cryptography/fernet.py b/src/cryptography/fernet.py
index 99eb10e..1f33a12 100644
--- a/src/cryptography/fernet.py
+++ b/src/cryptography/fernet.py
@@ -71,11 +71,14 @@
return base64.urlsafe_b64encode(basic_parts + hmac)
def decrypt(self, token, ttl=None):
+ timestamp, data = Fernet._get_unverified_token_data(token)
+ return self._decrypt_data(data, timestamp, ttl)
+
+ @staticmethod
+ def _get_unverified_token_data(token):
if not isinstance(token, bytes):
raise TypeError("token must be bytes.")
- current_time = int(time.time())
-
try:
data = base64.urlsafe_b64decode(token)
except (TypeError, binascii.Error):
@@ -88,6 +91,10 @@
timestamp, = struct.unpack(">Q", data[1:9])
except struct.error:
raise InvalidToken
+ return timestamp, data
+
+ def _decrypt_data(self, data, timestamp, ttl):
+ current_time = int(time.time())
if ttl is not None:
if timestamp + ttl < current_time:
raise InvalidToken
@@ -134,6 +141,20 @@
def encrypt(self, msg):
return self._fernets[0].encrypt(msg)
+ def rotate(self, msg):
+ timestamp, data = Fernet._get_unverified_token_data(msg)
+ for f in self._fernets:
+ try:
+ p = f._decrypt_data(data, timestamp, None)
+ break
+ except InvalidToken:
+ pass
+ else:
+ raise InvalidToken
+
+ iv = os.urandom(16)
+ return self._fernets[0]._encrypt_from_parts(p, timestamp, iv)
+
def decrypt(self, msg, ttl=None):
for f in self._fernets:
try:
diff --git a/tests/test_fernet.py b/tests/test_fernet.py
index dbce44f..6558d11 100644
--- a/tests/test_fernet.py
+++ b/tests/test_fernet.py
@@ -6,6 +6,7 @@
import base64
import calendar
+import datetime
import json
import os
import time
@@ -156,3 +157,55 @@
def test_non_iterable_argument(self, backend):
with pytest.raises(TypeError):
MultiFernet(None)
+
+ def test_rotate(self, backend):
+ f1 = Fernet(base64.urlsafe_b64encode(b"\x00" * 32), backend=backend)
+ f2 = Fernet(base64.urlsafe_b64encode(b"\x01" * 32), backend=backend)
+
+ mf1 = MultiFernet([f1])
+ mf2 = MultiFernet([f2, f1])
+
+ plaintext = b"abc"
+ mf1_ciphertext = mf1.encrypt(plaintext)
+
+ assert mf2.decrypt(mf1_ciphertext) == plaintext
+
+ rotated = mf2.rotate(mf1_ciphertext)
+
+ assert rotated != mf1_ciphertext
+ assert mf2.decrypt(rotated) == plaintext
+
+ with pytest.raises(InvalidToken):
+ mf1.decrypt(rotated)
+
+ def test_rotate_preserves_timestamp(self, backend, monkeypatch):
+ f1 = Fernet(base64.urlsafe_b64encode(b"\x00" * 32), backend=backend)
+ f2 = Fernet(base64.urlsafe_b64encode(b"\x01" * 32), backend=backend)
+
+ mf1 = MultiFernet([f1])
+ mf2 = MultiFernet([f2, f1])
+
+ plaintext = b"abc"
+ mf1_ciphertext = mf1.encrypt(plaintext)
+
+ later = datetime.datetime.now() + datetime.timedelta(minutes=5)
+ later_time = time.mktime(later.timetuple())
+ monkeypatch.setattr(time, "time", lambda: later_time)
+
+ original_time, _ = Fernet._get_unverified_token_data(mf1_ciphertext)
+ rotated_time, _ = Fernet._get_unverified_token_data(
+ mf2.rotate(mf1_ciphertext)
+ )
+
+ assert later_time != rotated_time
+ assert original_time == rotated_time
+
+ def test_rotate_decrypt_no_shared_keys(self, backend):
+ f1 = Fernet(base64.urlsafe_b64encode(b"\x00" * 32), backend=backend)
+ f2 = Fernet(base64.urlsafe_b64encode(b"\x01" * 32), backend=backend)
+
+ mf1 = MultiFernet([f1])
+ mf2 = MultiFernet([f2])
+
+ with pytest.raises(InvalidToken):
+ mf2.rotate(mf1.encrypt(b"abc"))