def encrypt_data(key, plaintext):
    assert isinstance(key, type(b"")), type(key)
    assert isinstance(plaintext, type(b"")), type(plaintext)
    assert len(key) == SecretBox.KEY_SIZE, len(key)
    box = SecretBox(key)
    nonce = utils.random(SecretBox.NONCE_SIZE)
    return box.encrypt(plaintext, nonce)

 def __init__(self, skt, send_key, receive_key):
     self.skt = skt
     self.send_box = SecretBox(send_key)
     self.send_nonce = 0
     self.receive_buf = ReceiveBuffer(self.skt)
     self.receive_box = SecretBox(receive_key)
     self.next_receive_nonce = 0

 def _encrypt_data(self, key, data):
     assert isinstance(key, type(b"")), type(key)
     assert isinstance(data, type(b"")), type(data)
     if len(key) != SecretBox.KEY_SIZE: raise UsageError
     box = SecretBox(key)
     nonce = utils.random(SecretBox.NONCE_SIZE)
     return box.encrypt(data, nonce)

 def _decrypt_data(self, key, encrypted):
     assert isinstance(key, type(b"")), type(key)
     assert isinstance(encrypted, type(b"")), type(encrypted)
     assert len(key) == SecretBox.KEY_SIZE, len(key)
     box = SecretBox(key)
     data = box.decrypt(encrypted)
     return data

class EncryptedSerializer(object):
    """Encrypt session state using PyNaCl.

    :type secret: bytes
    :param secret: a 32-byte random secret for encrypting/decrypting the
                   pickled session state.
    :param serializer:
        An object with two methods: ``loads`` and ``dumps``. The ``loads``
        method should accept bytes and return a Python object. The ``dumps``
        method should accept a Python object and return bytes. A ``ValueError``
        should be raised for malformed inputs. Default: ``None``, which will
        use :class:`pyramid.session.PickleSerializer`.
    """
    def __init__(self, secret, serializer=None):
        if len(secret) != SecretBox.KEY_SIZE:
            raise ValueError(
                "Secret should be a random bytes string of length %d" %
                SecretBox.KEY_SIZE)
        self.box = SecretBox(secret)

        if serializer is None:
            serializer = PickleSerializer()

        self.serializer = serializer

    def loads(self, bstruct):
        """Decrypt session state.

        :type encrypted_state: bytes
        :param encrypted_state: the encrypted session state.

        :rtype: :class:`dict` / picklable mapping
        :returns: the decrypted, unpickled session state, as passed as
                  ``session_state`` to :meth:`dumps`.
        """
        try:
            b64padding = b'=' * (-len(bstruct) % 4)
            fstruct = urlsafe_b64decode(bstruct + b64padding)
        except (binascii.Error, TypeError) as e:
            raise ValueError('Badly formed base64 data: %s' % e)

        try:
            payload = self.box.decrypt(fstruct)
        except CryptoError as e:
            raise ValueError('Possible tampering: %s' % e)
        return self.serializer.loads(payload)

    def dumps(self, session_state):
        """Encrypt session state.

        :type session_state: :class:`dict` / picklable mapping
        :param session_state: the session state to be encrypted.

        :rtype: bytes
        :returns: the encrypted session state
        """
        cstruct = self.serializer.dumps(session_state)
        nonce = random(SecretBox.NONCE_SIZE)
        fstruct = self.box.encrypt(cstruct, nonce)
        return urlsafe_b64encode(fstruct).rstrip(b'=')

def test_secret_box_decryption_combined(key, nonce, plaintext, ciphertext):
    box = SecretBox(key, encoder=HexEncoder)

    combined = binascii.hexlify(binascii.unhexlify(nonce) + binascii.unhexlify(ciphertext))
    decrypted = binascii.hexlify(box.decrypt(combined, encoder=HexEncoder))

    assert decrypted == plaintext

 def _decrypt_data(self, key, encrypted):
     assert isinstance(key, type(b"")), type(key)
     assert isinstance(encrypted, type(b"")), type(encrypted)
     if len(key) != SecretBox.KEY_SIZE: raise UsageError
     box = SecretBox(key)
     data = box.decrypt(encrypted)
     return data

 def add_third_party_caveat(self,
                            macaroon,
                            location,
                            key,
                            key_id,
                            nonce=None,
                            **kwargs):
     derived_key = truncate_or_pad(
         generate_derived_key(convert_to_bytes(key))
     )
     old_key = truncate_or_pad(binascii.unhexlify(macaroon.signature_bytes))
     box = SecretBox(key=old_key)
     nonce = nonce or nacl.utils.random(box.NONCE_SIZE)
     verification_key_id = box.encrypt(
         derived_key, nonce=nonce
     )
     caveat = Caveat(
         caveat_id=key_id,
         location=location,
         verification_key_id=verification_key_id
     )
     macaroon.caveats.append(caveat)
     encode_key = binascii.unhexlify(macaroon.signature_bytes)
     macaroon.signature = sign_third_party_caveat(
         encode_key,
         caveat._verification_key_id,
         caveat._caveat_id
     )
     return macaroon

def encrypt(message, keyPath):
    """
    Encrypts a message given a path to a local file containing a key.

    :param message: The message to be encrypted.
    :param keyPath: A path to a file containing a 256-bit key (and nothing else).
    :type message: str
    :type keyPath: str
    :rtype: str

    A constant overhead is added to every encrypted message (for the nonce and MAC).
    >>> import tempfile
    >>> k = tempfile.mktemp()
    >>> with open(k, 'w') as f:
    ...     f.write(nacl.utils.random(SecretBox.KEY_SIZE))
    >>> message = 'test'
    >>> len(encrypt(message, k)) == encryptionOverhead + len(message)
    True
    """
    with open(keyPath) as f:
        key = f.read()
    if len(key) != SecretBox.KEY_SIZE:
        raise ValueError("Key is %d bytes, but must be exactly %d bytes" % (len(key),
                                                                            SecretBox.KEY_SIZE))
    sb = SecretBox(key)
    # We generate the nonce using secure random bits. For long enough
    # nonce size, the chance of a random nonce collision becomes
    # *much* smaller than the chance of a subtle coding error causing
    # a nonce reuse. Currently the nonce size is 192 bits--the chance
    # of a collision is astronomically low. (This approach is
    # recommended in the libsodium documentation.)
    nonce = nacl.utils.random(SecretBox.NONCE_SIZE)
    assert len(nonce) == SecretBox.NONCE_SIZE
    return str(sb.encrypt(message, nonce))

def decrypt(ciphertext, keyPath):
    """
    Decrypts a given message that was encrypted with the encrypt() method.

    :param ciphertext: The encrypted message (as a string).
    :param keyPath: A path to a file containing a 256-bit key (and nothing else).
    :type keyPath: str
    :rtype: str

    Raises an error if ciphertext was modified
    >>> import tempfile
    >>> k = tempfile.mktemp()
    >>> with open(k, 'w') as f:
    ...     f.write(nacl.utils.random(SecretBox.KEY_SIZE))
    >>> ciphertext = encrypt("testMessage", k)
    >>> ciphertext = chr(ord(ciphertext[0]) ^ 1) + ciphertext[1:]
    >>> decrypt(ciphertext, k)
    Traceback (most recent call last):
    ...
    CryptoError: Decryption failed. Ciphertext failed verification

    Otherwise works correctly
    >>> decrypt(encrypt("testMessage", k), k)
    'testMessage'
    """
    with open(keyPath) as f:
        key = f.read()
    if len(key) != SecretBox.KEY_SIZE:
        raise ValueError("Key is %d bytes, but must be exactly %d bytes" % (len(key),
                                                                            SecretBox.KEY_SIZE))
    sb = SecretBox(key)
    # The nonce is kept with the message.
    return sb.decrypt(ciphertext)

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Re-encrypt.
        """
        # Derive the key from the passphrase.
        derived = util.derive_passphrase(self.passphrase)

        # Generate two random nonces.
        nonce1 = random(SecretBox.NONCE_SIZE)
        nonce2 = random(SecretBox.NONCE_SIZE)

        sign_box = SecretBox(derived)
        enc_box = SecretBox(derived)

        s_p = self.sign.encode()
        e_p = self.encrypt.encode()

        s_e = sign_box.encrypt(s_p, nonce1)
        e_e = enc_box.encrypt(e_p, nonce2)

        # Update `self.key`.
        self.key._private_key_raw = e_e.ciphertext
        self.key._private_signing_key_raw = s_e.ciphertext

        # Bit of a mixed up name.
        self.key._private_nonce = e_e.nonce
        self.key._private_signing_nonce = s_e.nonce

        if exc_type is not None:
            raise exc_type(exc_val)

    def __enter__(self) -> typing.Tuple[PrivateKey, PrivateKey]:
        """
        Provides a pair of private keys.
        """
        # Derive the key from the passphrase.
        derived = util.derive_passphrase(self.passphrase)

        sign_box = SecretBox(derived)
        enc_box = SecretBox(derived)

        # Decrypt, using the two nonces.
        s_d = sign_box.decrypt(self.key._private_signing_seed, self.key._private_signing_nonce)
        e_d = enc_box.decrypt(self.key._private_key_raw, self.key._private_nonce)

        # Generate a SigningKey out of the seed.
        self.sign = SigningKey(s_d)
        self.encrypt = PrivateKey(e_d)

        # Update the key's public keys.
        if self.key._public_key is None:
            self.key._public_key = self.encrypt.public_key

        if self.key._public_signing_key is None:
            self.key._public_signing_key = self.sign.verify_key

        return self.encrypt, self.sign

 def _encrypt_data(self, key, data):
     assert isinstance(key, type(b"")), type(key)
     assert isinstance(data, type(b"")), type(data)
     assert len(key) == SecretBox.KEY_SIZE, len(key)
     box = SecretBox(key)
     nonce = utils.random(SecretBox.NONCE_SIZE)
     return box.encrypt(data, nonce)

 def add_third_party_caveat(self,
                            macaroon,
                            location,
                            key,
                            key_id,
                            **kwargs):
     derived_key = truncate_or_pad(
         generate_derived_key(convert_to_bytes(key))
     )
     old_key = truncate_or_pad(binascii.unhexlify(macaroon.signature_bytes))
     box = SecretBox(key=old_key)
     verification_key_id = box.encrypt(
         derived_key, nonce=kwargs.get('nonce')
     )
     caveat = Caveat(
         caveat_id=key_id,
         location=location,
         verification_key_id=verification_key_id,
         version=macaroon.version
     )
     macaroon.caveats.append(caveat)
     encode_key = binascii.unhexlify(macaroon.signature_bytes)
     macaroon.signature = sign_third_party_caveat(
         encode_key,
         caveat._verification_key_id,
         caveat._caveat_id
     )
     return macaroon

def test_secret_box_encryption_generates_different_nonces(key, nonce, plaintext, ciphertext):
    box = SecretBox(key, encoder=HexEncoder)

    nonce_0 = box.encrypt(binascii.unhexlify(plaintext), encoder=HexEncoder).nonce

    nonce_1 = box.encrypt(binascii.unhexlify(plaintext), encoder=HexEncoder).nonce

    assert nonce_0 != nonce_1

def test_secret_box_optional_nonce(key, nonce, plaintext, ciphertext):
    box = SecretBox(key, encoder=HexEncoder)

    encrypted = box.encrypt(binascii.unhexlify(plaintext), encoder=HexEncoder)

    decrypted = binascii.hexlify(box.decrypt(encrypted, encoder=HexEncoder))

    assert decrypted == plaintext

def handshake_initiate(private_key, redis_client):
    try:
        request = expect_json_request(bottle.request, INITIATE_SCHEMA)

        symmetric_key = redis_get_cookie(
            redis_client, request[INITIATE_COOKIE_FIELD])
        cookie_sbox = SecretBox(symmetric_key)
        cookie = cookie_sbox.decrypt(
            str(request[INITIATE_COOKIE_FIELD]), encoder=Base64Encoder)

        if len(cookie) != 2 * CURVE25519_KEY_BYTES:
            bottle.response.status = HTTP_INTERNAL_SERVER_ERROR
            return {'error': 'An invalid cookie was sent to the client.'}
        client_transient_pkey = PublicKey(cookie[0:CURVE25519_KEY_BYTES])
        transient_skey = PrivateKey(cookie[CURVE25519_KEY_BYTES:])

        if request[INITIATE_CLIENT_TRANSIENT_PKEY_FIELD] != \
           client_transient_pkey.encode(Base64Encoder):
            raise InvalidClientRequest(
                'Initiate: non matching transient public keys.')

        vouch_json = open_box(request[INITIATE_VOUCH_FIELD],
                              transient_skey, client_transient_pkey)
        vouch = parse_and_verify_json(vouch_json, VOUCH_SCHEMA)

        client_pkey = PublicKey(
            str(vouch[VOUCH_CLIENT_PKEY_FIELD]), encoder=Base64Encoder)
        vouch_for_transient_pkey = open_box(
            vouch[VOUCH_TRANSIENT_KEY_BOX_FIELD], private_key, client_pkey)
        if vouch_for_transient_pkey != client_transient_pkey.encode():
            raise InvalidClientRequest(
                'Initiate: non matching transient public keys.')

        resp = 'I believe you are {} and you want {}'.format(
            client_pkey.encode(Base64Encoder), vouch[VOUCH_MESSAGE_FIELD])
        print(resp)
        response_nonce = nacl.utils.random(Box.NONCE_SIZE)
        response_box = Box(transient_skey, client_transient_pkey)
        response_box_cipher = response_box.encrypt(
            resp, response_nonce, encoder=Base64Encoder)
        return {'response': response_box_cipher}
    except jsonschema.ValidationError as e:
        log.exception(e)
        bottle.response.status = HTTP_BAD_REQUEST
        return {'error': str(e)}
    except InvalidClientRequest as e:
        log.exception(e)
        bottle.response.status = HTTP_BAD_REQUEST
        return {'error': str(e)}
    except MissingCookie as e:
        log.exception(e)
        bottle.response.status = HTTP_BAD_REQUEST
        return {'error': str(e)}
    except CryptoError as e:
        log.exception(e)
        bottle.response.status = HTTP_BAD_REQUEST
        return {'error': 'Bad encryption in handshake.'}
    return {'error': ''}

def test_secret_box_encryption(key, nonce, plaintext, ciphertext):
    box = SecretBox(key, encoder=HexEncoder)
    encrypted = box.encrypt(binascii.unhexlify(plaintext), binascii.unhexlify(nonce), encoder=HexEncoder)

    expected = binascii.hexlify(binascii.unhexlify(nonce) + binascii.unhexlify(ciphertext))

    assert encrypted == expected
    assert encrypted.nonce == nonce
    assert encrypted.ciphertext == ciphertext

def test_secret_box_wrong_lengths():
    with pytest.raises(ValueError):
        SecretBox(b"")

    box = SecretBox(b"ec2bee2d5be613ca82e377c96a0bf2220d823ce980cdff6279473edc52862798", encoder=HexEncoder)
    with pytest.raises(ValueError):
        box.encrypt(b"", b"")
    with pytest.raises(ValueError):
        box.decrypt(b"", b"")

def decrypt_list_entry(boxed, symkey, tmppub):
    sbox = SecretBox(symkey)
    msg = remove_prefix(sbox.decrypt(boxed),
                        "list:", NotListResponseError)
    (got_tmppub, fetch_token, delete_token,
     length) = struct.unpack(">32s32s32sQ", msg)
    if not equal(got_tmppub, tmppub):
        raise WrongPubkeyError
    return fetch_token, delete_token, length