Class MultiFernet

src/cryptography/fernet.py:171–224

class MultiFernet:
    def __init__(self, fernets: Iterable[Fernet]):
        fernets = list(fernets)
        if not fernets:
            raise ValueError(
                "MultiFernet requires at least one Fernet instance"
            )
        self._fernets = fernets

    def encrypt(self, msg: bytes) -> bytes:
        return self.encrypt_at_time(msg, int(time.time()))

    def encrypt_at_time(self, msg: bytes, current_time: int) -> bytes:
        return self._fernets[0].encrypt_at_time(msg, current_time)

    def rotate(self, msg: bytes | str) -> bytes:
        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: bytes | str, ttl: int | None = None) -> bytes:
        for f in self._fernets:
            try:
                return f.decrypt(msg, ttl)
            except InvalidToken:
                pass
        raise InvalidToken

    def decrypt_at_time(
        self, msg: bytes | str, ttl: int, current_time: int
    ) -> bytes:
        for f in self._fernets:
            try:
                return f.decrypt_at_time(msg, ttl, current_time)
            except InvalidToken:
                pass
        raise InvalidToken

    def extract_timestamp(self, msg: bytes | str) -> int:
        for f in self._fernets:
            try:
                return f.extract_timestamp(msg)
            except InvalidToken:
                pass
        raise InvalidToken