Add Multifernet.rotate method

AUTHORS.rst

@@ -37,3 +37,4 @@ PGP key fingerprints are enclosed in parentheses.
 * 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>

CHANGELOG.rst

@@ -9,6 +9,8 @@ Changelog
 .. 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:
 

docs/fernet.rst

@@ -86,7 +86,8 @@ has support for implementing key rotation via :class:`MultiFernet`.
     .. 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 @@ has support for implementing key rotation via :class:`MultiFernet`.
     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
 

docs/spelling_wordlist.txt

@@ -21,6 +21,7 @@ cryptographic
 cryptographically
 Debian
 decrypt
+decrypts
 Decrypts
 decrypted
 decrypting

src/cryptography/fernet.py

@@ -71,11 +71,14 @@ def _encrypt_from_parts(self, data, current_time, iv):
         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 @@ def decrypt(self, token, ttl=None):
             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 __init__(self, fernets):
     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:

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_no_fernets(self, backend):
     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"))