dryoc/

dryocstream.rs

//! # Encrypted streams
//!
//! [`DryocStream`] provides libsodium-compatible authenticated encryption for
//! an ordered sequence of messages, also known as a _secret stream_. It uses a
//! shared secret key. Each message is encrypted and authenticated separately,
//! while shared stream state links the messages and requires ordered
//! processing. A tag can mark ordinary messages, rekeying points, or the
//! expected end of the stream.
//!
//! Use [`DryocStream`] to encrypt messages written in order to a file or
//! network connection. A modified, reordered, or duplicated message is
//! rejected when it is pulled; a removed message is detected when the next
//! message is pulled. Removing the end of a stream cannot be detected
//! automatically: the sender must use [`Tag::Final`], and the application must
//! reject a stream that ends without that tag.
//!
//! The shared key can be generated directly or derived with
//! [`Kdf`](crate::kdf), [`Session`](crate::kx), or a password-hashing function
//! such as [`crypto_pwhash`](crate::classic::crypto_pwhash).
//!
//! [`DryocStream::init_push`] generates a public header for each stream. Send
//! the header to the receiving side before the ciphertexts. Never reuse the
//! same key and header for another stream.
//!
//! # Rustaceous API example
//!
//! ```
//! # #[cfg(feature = "alloc")]
//! # {
//! use dryoc::dryocstream::*;
//! use dryoc::types::*;
//! let message1 = b"Arbitrary data to encrypt";
//! let message2 = b"split into";
//! let message3 = b"three messages";
//!
//! // Generate a random secret key for this stream
//! let key = Key::generate();
//!
//! // Initialize the push side, type annotations required on return type
//! let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
//!
//! // Encrypt a series of messages
//! let c1 = push_stream
//!     .push_to_vec(message1, None, Tag::Message)
//!     .expect("Encrypt failed");
//! let c2 = push_stream
//!     .push_to_vec(message2, None, Tag::Message)
//!     .expect("Encrypt failed");
//! let c3 = push_stream
//!     .push_to_vec(message3, None, Tag::Final)
//!     .expect("Encrypt failed");
//!
//! // Initialize the pull side using header generated by the push side
//! let mut pull_stream = DryocStream::init_pull(&key, &header);
//!
//! // Decrypt the encrypted messages, type annotations required
//! let (m1, tag1) = pull_stream.pull_to_vec(&c1, None).expect("Decrypt failed");
//! let (m2, tag2) = pull_stream.pull_to_vec(&c2, None).expect("Decrypt failed");
//! let (m3, tag3) = pull_stream.pull_to_vec(&c3, None).expect("Decrypt failed");
//!
//! assert_eq!(message1, m1.as_slice());
//! assert_eq!(message2, m2.as_slice());
//! assert_eq!(message3, m3.as_slice());
//!
//! assert_eq!(tag1, Tag::Message);
//! assert_eq!(tag2, Tag::Message);
//! assert_eq!(tag3, Tag::Final);
//! # }
//! ```
//!
//! ## Additional resources
//!
//! * See the [libsodium documentation](https://doc.libsodium.org/secret-key_cryptography/secretstream)
//!   for more about secret streams
//! * For public-key encryption, see [`DryocBox`](crate::dryocbox)
//! * For individual messages encrypted with a shared key, see
//!   [`DryocSecretBox`](crate::dryocsecretbox)
//! * See the [`protected`] module for an example that stores keys in protected
//!   memory

#[cfg(feature = "alloc")]
use alloc::vec::Vec;

use zeroize::Zeroize;

use crate::classic::crypto_secretstream_xchacha20poly1305::{
    State, ciphertext_len_from_message_len, crypto_secretstream_xchacha20poly1305_init_pull,
    crypto_secretstream_xchacha20poly1305_init_push, crypto_secretstream_xchacha20poly1305_pull,
    crypto_secretstream_xchacha20poly1305_push, crypto_secretstream_xchacha20poly1305_rekey,
    message_len_from_ciphertext_len,
};
use crate::constants::{
    CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES,
    CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_KEYBYTES,
};
use crate::error::Error;
use crate::types::*;

mod tag;
pub use tag::Tag;

mod sealed {
    pub trait Sealed {}
}

/// Stream mode marker trait: [`Push`] or [`Pull`].
///
/// This trait is sealed and cannot be implemented outside dryoc.
pub trait Mode: sealed::Sealed {}

/// Indicates a push stream
pub struct Push;
/// Indicates a pull stream
pub struct Pull;

impl sealed::Sealed for Push {}
impl sealed::Sealed for Pull {}
impl Mode for Push {}
impl Mode for Pull {}

/// Stack-allocated secret for authenticated secret streams.
pub type Key = StackByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_KEYBYTES>;
/// Stack-allocated header data for authenticated secret streams.
pub type Header = StackByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES>;

#[cfg(any(
    all(feature = "protected", any(unix, windows)),
    all(doc, not(doctest), feature = "std")
))]
#[cfg_attr(all(feature = "nightly", doc), doc(cfg(feature = "protected")))]
pub mod protected {
    //! # Protected memory type aliases for [`DryocStream`]
    //!
    //! Type aliases for using [`DryocStream`] with protected memory.
    //!
    //! ## Example
    //! ```
    //! use dryoc::dryocstream::protected::*;
    //! use dryoc::dryocstream::{DryocStream, Tag};
    //!
    //! // Load some message into locked readonly memory.
    //! let message1 = HeapBytes::from_slice_into_readonly_locked(b"Arbitrary data to encrypt")
    //!     .expect("from slice failed");
    //! let message2 =
    //!     HeapBytes::from_slice_into_readonly_locked(b"split into").expect("from slice failed");
    //! let message3 =
    //!     HeapBytes::from_slice_into_readonly_locked(b"three messages").expect("from slice failed");
    //!
    //! // Generate a random key into locked readonly memory.
    //! let key = Key::generate_readonly_locked().expect("key failed");
    //!
    //! // Initialize the push stream, place the header into locked memory
    //! let (mut push_stream, header): (_, Locked<Header>) = DryocStream::init_push(&key);
    //!
    //! // Encrypt the set of messages, placing everything into locked memory.
    //! let c1: LockedBytes = push_stream
    //!     .push(&message1, None, Tag::Message)
    //!     .expect("Encrypt failed");
    //! let c2: LockedBytes = push_stream
    //!     .push(&message2, None, Tag::Message)
    //!     .expect("Encrypt failed");
    //! let c3: LockedBytes = push_stream
    //!     .push(&message3, None, Tag::Final)
    //!     .expect("Encrypt failed");
    //!
    //! // Initialize the pull stream
    //! let mut pull_stream = DryocStream::init_pull(&key, &header);
    //!
    //! // Decrypt the set of messages, putting everything into locked memory
    //! let (m1, tag1): (LockedBytes, Tag) = pull_stream.pull(&c1, None).expect("Decrypt failed");
    //! let (m2, tag2): (LockedBytes, Tag) = pull_stream.pull(&c2, None).expect("Decrypt failed");
    //! let (m3, tag3): (LockedBytes, Tag) = pull_stream.pull(&c3, None).expect("Decrypt failed");
    //!
    //! assert_eq!(message1.as_slice(), m1.as_slice());
    //! assert_eq!(message2.as_slice(), m2.as_slice());
    //! assert_eq!(message3.as_slice(), m3.as_slice());
    //!
    //! assert_eq!(tag1, Tag::Message);
    //! assert_eq!(tag2, Tag::Message);
    //! assert_eq!(tag3, Tag::Final);
    //! ```
    use super::*;
    pub use crate::protected::*;

    /// Heap-allocated, page-aligned secret key for authenticated secret
    /// streams, for use with protected memory.
    pub type Key = HeapByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_KEYBYTES>;
    /// Heap-allocated, page-aligned header for authenticated secret
    /// streams, for use with protected memory.
    pub type Header = HeapByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES>;
}

/// Secret-key authenticated encrypted streams
#[derive(PartialEq, Eq, Clone, Zeroize)]
pub struct DryocStream<M> {
    state: State,
    phantom: core::marker::PhantomData<M>,
}

impl<M> Drop for DryocStream<M> {
    fn drop(&mut self) {
        self.state.zeroize()
    }
}

impl<M> DryocStream<M> {
    /// Rekeys the stream immediately. The sender and receiver must rekey at the
    /// same position.
    ///
    /// Manual rekeying is unnecessary when the sender uses [`Tag::Rekey`] or
    /// [`Tag::Final`], because those tags rekey after the message. The stream
    /// also rekeys if its internal counter wraps. See the
    /// [libsodium documentation](https://doc.libsodium.org/secret-key_cryptography/secretstream#rekeying)
    /// for details.
    pub fn rekey(&mut self) {
        crypto_secretstream_xchacha20poly1305_rekey(&mut self.state)
    }
}

impl DryocStream<Push> {
    /// Returns a new push stream, initialized from `key`.
    #[must_use]
    pub fn init_push<
        Header: NewByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES>,
        Key: ByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_KEYBYTES>,
    >(
        key: &Key,
    ) -> (Self, Header) {
        let mut state = State::new();
        let mut header = Header::new_byte_array();
        crypto_secretstream_xchacha20poly1305_init_push(
            &mut state,
            header.as_mut_array(),
            key.as_array(),
        );
        (
            Self {
                state,
                phantom: core::marker::PhantomData,
            },
            header,
        )
    }

    /// Encrypts `message` for this stream with `associated_data` and `tag`,
    /// returning the ciphertext.
    ///
    /// # Errors
    ///
    /// Returns an error if the message exceeds the stream's maximum message
    /// length, or the output storage does not resize to exactly the required
    /// ciphertext length.
    pub fn push<Output: NewBytes + ResizableBytes, Input: Bytes + ?Sized>(
        &mut self,
        message: &Input,
        associated_data: Option<&[u8]>,
        tag: Tag,
    ) -> Result<Output, Error> {
        let mut ciphertext = Output::new_bytes();
        ciphertext.resize(
            ciphertext_len_from_message_len(message.as_slice().len())?,
            0,
        );
        crypto_secretstream_xchacha20poly1305_push(
            &mut self.state,
            ciphertext.as_mut_slice(),
            message.as_slice(),
            associated_data,
            tag.bits(),
        )?;
        Ok(ciphertext)
    }

    /// Encrypts `message` for this stream with `associated_data` and `tag`,
    /// returning the ciphertext.
    ///
    /// # Errors
    ///
    /// Returns an error if the message exceeds the stream's maximum message
    /// length.
    #[cfg(feature = "alloc")]
    pub fn push_to_vec<Input: Bytes + ?Sized>(
        &mut self,
        message: &Input,
        associated_data: Option<&[u8]>,
        tag: Tag,
    ) -> Result<Vec<u8>, Error> {
        self.push(message, associated_data, tag)
    }
}

impl DryocStream<Pull> {
    /// Returns a new pull stream, initialized from `key` and `header`.
    #[must_use]
    pub fn init_pull<
        Key: ByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_KEYBYTES>,
        Header: ByteArray<CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES>,
    >(
        key: &Key,
        header: &Header,
    ) -> Self {
        let mut state = State::new();
        crypto_secretstream_xchacha20poly1305_init_pull(
            &mut state,
            header.as_array(),
            key.as_array(),
        );
        Self {
            state,
            phantom: core::marker::PhantomData,
        }
    }

    /// Decrypts `ciphertext` for this stream with `associated_data`, returning
    /// the decrypted message and tag.
    ///
    /// # Errors
    ///
    /// Returns an error if the ciphertext is too short or too long, the output
    /// storage cannot hold the plaintext, or authentication fails.
    /// Authentication fails for a wrong key or header, mismatched associated
    /// data, modified ciphertext, or messages processed out of order.
    /// An authenticated tag byte that is not one of the four [`Tag`] values is
    /// also rejected without advancing the stream.
    pub fn pull<Output: NewBytes + ResizableBytes, Input: Bytes + ?Sized>(
        &mut self,
        ciphertext: &Input,
        associated_data: Option<&[u8]>,
    ) -> Result<(Output, Tag), Error> {
        let message_len = message_len_from_ciphertext_len(ciphertext.as_slice().len())?;

        let mut message = Output::new_bytes();
        message.resize(message_len, 0);
        let mut tag = 0u8;
        let mut next_state = self.state.clone();
        crypto_secretstream_xchacha20poly1305_pull(
            &mut next_state,
            message.as_mut_slice(),
            &mut tag,
            ciphertext.as_slice(),
            associated_data,
        )?;

        let tag = match Tag::try_from(tag) {
            Ok(tag) => tag,
            Err(error) => {
                message.as_mut_slice().zeroize();
                return Err(error);
            }
        };
        self.state = next_state;

        Ok((message, tag))
    }

    /// Decrypts `ciphertext` for this stream with `associated_data`, returning
    /// the decrypted message and tag into a [`Vec`].
    ///
    /// # Errors
    ///
    /// Returns an error if the ciphertext is too short or too long, or
    /// authentication fails because the key, header, associated data, stream
    /// position, or ciphertext does not match. An authenticated tag byte that
    /// is not one of the four [`Tag`] values is also rejected without
    /// advancing the stream.
    #[cfg(feature = "alloc")]
    pub fn pull_to_vec<Input: Bytes + ?Sized>(
        &mut self,
        ciphertext: &Input,
        associated_data: Option<&[u8]>,
    ) -> Result<(Vec<u8>, Tag), Error> {
        self.pull(ciphertext, associated_data)
    }
}

#[cfg(all(test, feature = "alloc"))]
mod validation_tests {
    use super::*;
    use crate::constants::CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_ABYTES;

    /// Key and header of the [`TAG_KAT`] stream.
    pub(super) const TAG_KAT_KEY: [u8; 32] = *b"dryoc secretstream tag kat key!!";
    pub(super) const TAG_KAT_HEADER: [u8; 24] = *b"fixed secretstream hdr!!";
    /// `(message, associated data, tag, hex ciphertext)`.
    pub(super) type TagKat = (&'static [u8], Option<&'static [u8]>, Tag, &'static str);
    /// One stream carrying every tag, as [`TagKat`]s, pushed in order by
    /// libsodium 1.0.22 from [`TAG_KAT_KEY`] and [`TAG_KAT_HEADER`]. The
    /// message after `Rekey` checks that the tag rekeyed the stream.
    pub(super) const TAG_KAT: [TagKat; 5] = [
        (
            b"message",
            None,
            Tag::Message,
            "a59b40f34c0389665e80425058a5d5700805c1050cf44c1e",
        ),
        (
            b"push",
            Some(b"tag kat"),
            Tag::Push,
            "11f5df40c058fcf62f4caec9785bb132550a9ea0f7",
        ),
        (
            b"rekey",
            None,
            Tag::Rekey,
            "682581e6eb47fee588aa927a183e85660ed0f186de83",
        ),
        (
            b"after rekey",
            Some(b"tag kat"),
            Tag::Message,
            "54547a88d24cdb2d009376b265e49973dde328ac249762bfbc9cd83d",
        ),
        (
            b"final",
            None,
            Tag::Final,
            "84a19391176cc87a0e58212ec9cfe35d1d348223e901",
        ),
    ];

    #[test]
    fn every_tag_matches_libsodium_known_answers() {
        // `init_push` always draws a random header, so build the push side
        // for the fixed header directly: after initialization a push state
        // equals the pull state for the same key and header.
        let mut state = State::new();
        crypto_secretstream_xchacha20poly1305_init_pull(&mut state, &TAG_KAT_HEADER, &TAG_KAT_KEY);
        let mut push_stream = DryocStream::<Push> {
            state,
            phantom: core::marker::PhantomData,
        };
        let mut pull_stream =
            DryocStream::init_pull(&Key::from(TAG_KAT_KEY), &Header::from(TAG_KAT_HEADER));

        for (message, associated_data, tag, expected) in TAG_KAT {
            let expected = hex::decode(expected).expect("hex");
            let ciphertext = push_stream
                .push_to_vec(message, associated_data, tag)
                .expect("push failed");
            assert_eq!(ciphertext, expected, "push {tag:?}");

            let (decrypted, pulled_tag) = pull_stream
                .pull_to_vec(&expected, associated_data)
                .expect("pull failed");
            assert_eq!((decrypted.as_slice(), pulled_tag), (message, tag));
        }
    }

    #[test]
    fn rustaceous_pull_rejects_every_unknown_tag_without_advancing_state() {
        let key = Key::generate();
        let mut push_state = State::new();
        let mut raw_header =
            [0u8; crate::constants::CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_HEADERBYTES];
        crypto_secretstream_xchacha20poly1305_init_push(
            &mut push_state,
            &mut raw_header,
            key.as_array(),
        );
        let header = Header::try_from(raw_header.as_slice()).expect("header conversion failed");
        let mut pull_stream = DryocStream::init_pull(&key, &header);
        let original_pull_state = pull_stream.state.clone();
        let message = b"authenticated unknown tag";

        // Classic push authenticates any tag byte; the Rustaceous pull must
        // reject every byte other than the four `Tag` values.
        for bits in 4..=u8::MAX {
            let mut invalid_ciphertext =
                vec![0u8; message.len() + CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_ABYTES];
            crypto_secretstream_xchacha20poly1305_push(
                &mut push_state.clone(),
                &mut invalid_ciphertext,
                message,
                None,
                bits,
            )
            .expect("classic push failed");

            let error = pull_stream
                .pull_to_vec(&invalid_ciphertext, None)
                .expect_err("unknown tag must be rejected");
            assert!(
                matches!(
                    error,
                    Error::InvalidValue {
                        context: crate::ErrorContext::Tag,
                        actual,
                        ..
                    } if actual == u64::from(bits)
                ),
                "tag byte {bits:#04x}: {error:?}"
            );
            assert!(pull_stream.state == original_pull_state);
        }

        let mut valid_ciphertext =
            vec![0u8; message.len() + CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_ABYTES];
        crypto_secretstream_xchacha20poly1305_push(
            &mut push_state,
            &mut valid_ciphertext,
            message,
            None,
            Tag::Message.bits(),
        )
        .expect("classic push failed");
        let (decrypted, tag) = pull_stream
            .pull_to_vec(&valid_ciphertext, None)
            .expect("state must remain usable after rejection");
        assert_eq!(decrypted, message);
        assert_eq!(tag, Tag::Message);
    }

    #[test]
    fn rustaceous_pull_rejects_short_ciphertext_without_advancing_state() {
        let key = Key::generate();
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
        let c1 = push_stream
            .push_to_vec(b"first", None, Tag::Message)
            .expect("push failed");

        let mut pull_stream = DryocStream::init_pull(&key, &header);
        let original_state = pull_stream.state.clone();
        for len in [0, 1, CRYPTO_SECRETSTREAM_XCHACHA20POLY1305_ABYTES - 1] {
            let short: &[u8] = &c1[..len];
            assert!(matches!(
                pull_stream.pull_to_vec(&short, None),
                Err(Error::InvalidLength {
                    context: crate::ErrorContext::Ciphertext,
                    actual,
                    ..
                }) if actual == len
            ));
            assert!(pull_stream.state == original_state);
        }

        let (message, tag) = pull_stream.pull_to_vec(&c1, None).expect("pull failed");
        assert_eq!(message, b"first");
        assert_eq!(tag, Tag::Message);
    }

    #[test]
    fn rustaceous_pull_out_of_order_fails_and_state_stays_usable() {
        let key = Key::generate();
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
        let aad: &[u8] = b"stream aad";
        let c1 = push_stream
            .push_to_vec(b"first", Some(aad), Tag::Message)
            .expect("push failed");
        let c2 = push_stream
            .push_to_vec(b"second", Some(aad), Tag::Push)
            .expect("push failed");
        let c3 = push_stream
            .push_to_vec(b"third", Some(aad), Tag::Final)
            .expect("push failed");

        let mut pull_stream = DryocStream::init_pull(&key, &header);
        let initial_state = pull_stream.state.clone();

        // Skipping ahead is rejected and does not consume the position.
        assert!(matches!(
            pull_stream.pull_to_vec(&c2, Some(aad)),
            Err(Error::AuthenticationFailed)
        ));
        assert!(matches!(
            pull_stream.pull_to_vec(&c3, Some(aad)),
            Err(Error::AuthenticationFailed)
        ));
        // Mismatched associated data is rejected the same way.
        assert!(matches!(
            pull_stream.pull_to_vec(&c1, None),
            Err(Error::AuthenticationFailed)
        ));
        assert!(pull_stream.state == initial_state);

        let (m1, t1) = pull_stream.pull_to_vec(&c1, Some(aad)).expect("pull c1");
        assert_eq!((m1.as_slice(), t1), (&b"first"[..], Tag::Message));

        // Replaying the consumed message and skipping c2 both fail.
        let after_c1 = pull_stream.state.clone();
        assert!(pull_stream.pull_to_vec(&c1, Some(aad)).is_err());
        assert!(pull_stream.pull_to_vec(&c3, Some(aad)).is_err());
        assert!(pull_stream.state == after_c1);

        let (m2, t2) = pull_stream.pull_to_vec(&c2, Some(aad)).expect("pull c2");
        assert_eq!((m2.as_slice(), t2), (&b"second"[..], Tag::Push));
        let (m3, t3) = pull_stream.pull_to_vec(&c3, Some(aad)).expect("pull c3");
        assert_eq!((m3.as_slice(), t3), (&b"third"[..], Tag::Final));
    }

    #[test]
    fn manual_rekey_must_be_mirrored_by_the_pull_side() {
        let key = Key::generate();
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
        let c1 = push_stream
            .push_to_vec(b"before rekey", None, Tag::Message)
            .expect("push failed");
        push_stream.rekey();
        let c2 = push_stream
            .push_to_vec(b"after rekey", None, Tag::Message)
            .expect("push failed");
        let c3 = push_stream
            .push_to_vec(b"final", None, Tag::Final)
            .expect("push failed");

        let mut pull_stream = DryocStream::init_pull(&key, &header);
        let (m1, _) = pull_stream.pull_to_vec(&c1, None).expect("pull c1");
        assert_eq!(m1, b"before rekey");

        // Without the matching rekey the stream is desynchronized...
        let before_rekey = pull_stream.state.clone();
        assert!(matches!(
            pull_stream.pull_to_vec(&c2, None),
            Err(Error::AuthenticationFailed)
        ));
        assert!(pull_stream.state == before_rekey);

        // ...and the same rekey resynchronizes it.
        pull_stream.rekey();
        let (m2, t2) = pull_stream.pull_to_vec(&c2, None).expect("pull c2");
        assert_eq!((m2.as_slice(), t2), (&b"after rekey"[..], Tag::Message));
        let (m3, t3) = pull_stream.pull_to_vec(&c3, None).expect("pull c3");
        assert_eq!((m3.as_slice(), t3), (&b"final"[..], Tag::Final));
    }
}

#[cfg(all(test, dryoc_native_tests, feature = "alloc"))]
mod tests {
    use super::*;
    use crate::native_test_util::{
        SECRETSTREAM_TAG_FINAL, SECRETSTREAM_TAG_MESSAGE, SECRETSTREAM_TAG_PUSH,
        SECRETSTREAM_TAG_REKEY, SecretStream as SOStream,
    };

    /// The known answers in [`validation_tests::TAG_KAT`] are libsodium's,
    /// and every [`Tag`] byte is libsodium's constant.
    #[test]
    fn tag_known_answers_are_libsodiums() {
        use super::validation_tests::{TAG_KAT, TAG_KAT_HEADER, TAG_KAT_KEY};

        for (tag, so_tag) in [
            (Tag::Message, SECRETSTREAM_TAG_MESSAGE),
            (Tag::Push, SECRETSTREAM_TAG_PUSH),
            (Tag::Rekey, SECRETSTREAM_TAG_REKEY),
            (Tag::Final, SECRETSTREAM_TAG_FINAL),
        ] {
            assert_eq!(tag.bits(), so_tag, "{tag:?}");
        }

        // libsodium's push state for a fixed header is its pull state.
        let mut so_push = SOStream::init_pull(&TAG_KAT_HEADER, &TAG_KAT_KEY).expect("init pull");
        for (message, associated_data, tag, expected) in TAG_KAT {
            assert_eq!(
                so_push
                    .push(message, associated_data, tag.bits())
                    .expect("push"),
                hex::decode(expected).expect("hex"),
                "{tag:?}"
            );
        }
    }

    /// Shared stream key; each stream still derives a random header.
    fn fixed_key() -> Key {
        Key::from(*b"dryoc secretstream rekey key 32b")
    }

    #[test]
    fn rekey_on_push_side_is_mirrored_by_libsodium_pull() {
        let key = fixed_key();
        let aad: &[u8] = b"associated";
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
        let c1 = push_stream
            .push_to_vec(b"before rekey", Some(aad), Tag::Message)
            .expect("push failed");
        push_stream.rekey();
        let c2 = push_stream
            .push_to_vec(b"after rekey", Some(aad), Tag::Push)
            .expect("push failed");
        let c3 = push_stream
            .push_to_vec(b"final", Some(aad), Tag::Final)
            .expect("push failed");

        let so_header = header.as_slice();
        let so_key = key.as_slice();

        // Without the rekey, sodium cannot continue past c1.
        let mut unsynced = SOStream::init_pull(so_header, so_key).expect("init pull");
        assert_eq!(
            unsynced.pull(&c1, Some(aad)).expect("pull c1"),
            (b"before rekey".to_vec(), SECRETSTREAM_TAG_MESSAGE)
        );
        assert!(unsynced.pull(&c2, Some(aad)).is_err());

        let mut so_pull = SOStream::init_pull(so_header, so_key).expect("init pull");
        assert_eq!(
            so_pull.pull(&c1, Some(aad)).expect("pull c1"),
            (b"before rekey".to_vec(), SECRETSTREAM_TAG_MESSAGE)
        );
        so_pull.rekey().expect("rekey");
        assert_eq!(
            so_pull.pull(&c2, Some(aad)).expect("pull c2"),
            (b"after rekey".to_vec(), SECRETSTREAM_TAG_PUSH)
        );
        assert_eq!(
            so_pull.pull(&c3, Some(aad)).expect("pull c3"),
            (b"final".to_vec(), SECRETSTREAM_TAG_FINAL)
        );
        assert!(so_pull.is_finalized());
    }

    #[test]
    fn rekey_on_libsodium_push_side_is_mirrored_by_rustaceous_pull() {
        let key = fixed_key();
        let (mut so_push, so_header) = SOStream::init_push(key.as_slice());
        let c1 = so_push
            .push(b"before rekey", None, SECRETSTREAM_TAG_MESSAGE)
            .expect("push failed");
        so_push.rekey().expect("rekey");
        let c2 = so_push
            .push(b"after rekey", None, SECRETSTREAM_TAG_REKEY)
            .expect("push failed");
        let c3 = so_push
            .push(b"after tag rekey", None, SECRETSTREAM_TAG_MESSAGE)
            .expect("push failed");
        let c4 = so_push
            .push(b"final", None, SECRETSTREAM_TAG_FINAL)
            .expect("push failed");

        let header = Header::try_from(so_header.as_slice()).expect("header");
        let mut pull_stream = DryocStream::init_pull(&key, &header);
        let (m1, t1) = pull_stream.pull_to_vec(&c1, None).expect("pull c1");
        assert_eq!((m1.as_slice(), t1), (&b"before rekey"[..], Tag::Message));

        let before_rekey = pull_stream.state.clone();
        assert!(matches!(
            pull_stream.pull_to_vec(&c2, None),
            Err(Error::AuthenticationFailed)
        ));
        assert!(pull_stream.state == before_rekey);

        pull_stream.rekey();
        let (m2, t2) = pull_stream.pull_to_vec(&c2, None).expect("pull c2");
        assert_eq!((m2.as_slice(), t2), (&b"after rekey"[..], Tag::Rekey));
        // The authenticated REKEY tag rekeys automatically; no manual call.
        let (m3, t3) = pull_stream.pull_to_vec(&c3, None).expect("pull c3");
        assert_eq!((m3.as_slice(), t3), (&b"after tag rekey"[..], Tag::Message));
        let (m4, t4) = pull_stream.pull_to_vec(&c4, None).expect("pull c4");
        assert_eq!((m4.as_slice(), t4), (&b"final"[..], Tag::Final));
    }

    #[test]
    fn libsodium_messages_pulled_out_of_order_fail_and_state_stays_usable() {
        let key = fixed_key();
        let (mut so_push, so_header) = SOStream::init_push(key.as_slice());
        let c1 = so_push
            .push(b"one", None, SECRETSTREAM_TAG_MESSAGE)
            .expect("push");
        let c2 = so_push
            .push(b"two", None, SECRETSTREAM_TAG_MESSAGE)
            .expect("push");
        let c3 = so_push
            .push(b"three", None, SECRETSTREAM_TAG_FINAL)
            .expect("push");

        let header = Header::try_from(so_header.as_slice()).expect("header");
        let mut pull_stream = DryocStream::init_pull(&key, &header);

        assert!(matches!(
            pull_stream.pull_to_vec(&c2, None),
            Err(Error::AuthenticationFailed)
        ));
        let (m1, _) = pull_stream.pull_to_vec(&c1, None).expect("pull c1");
        assert_eq!(m1, b"one");
        assert!(pull_stream.pull_to_vec(&c3, None).is_err());
        let (m2, _) = pull_stream.pull_to_vec(&c2, None).expect("pull c2");
        assert_eq!(m2, b"two");
        let (m3, t3) = pull_stream.pull_to_vec(&c3, None).expect("pull c3");
        assert_eq!((m3.as_slice(), t3), (&b"three"[..], Tag::Final));
    }

    #[test]
    fn test_stream_push() {
        let message1 = b"Arbitrary data to encrypt";
        let message2 = b"split into";
        let message3 = b"three messages";

        // Generate a random secret key for this stream
        let key = Key::generate();

        // Initialize the push side, type annotations required on return type
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);
        // Encrypt a series of messages
        let c1: Vec<u8> = push_stream
            .push(message1, None, Tag::Message)
            .expect("Encrypt failed");
        let c2: Vec<u8> = push_stream
            .push(message2, None, Tag::Message)
            .expect("Encrypt failed");
        let c3: Vec<u8> = push_stream
            .push(message3, None, Tag::Final)
            .expect("Encrypt failed");

        // Initialize the pull side using header generated by the push side
        let mut so_stream_pull =
            SOStream::init_pull(header.as_slice(), key.as_slice()).expect("pull init failed");

        let (m1, tag1) = so_stream_pull.pull(&c1, None).expect("decrypt failed");
        let (m2, tag2) = so_stream_pull.pull(&c2, None).expect("decrypt failed");
        let (m3, tag3) = so_stream_pull.pull(&c3, None).expect("decrypt failed");

        assert_eq!(message1, m1.as_slice());
        assert_eq!(message2, m2.as_slice());
        assert_eq!(message3, m3.as_slice());

        assert_eq!(tag1, SECRETSTREAM_TAG_MESSAGE);
        assert_eq!(tag2, SECRETSTREAM_TAG_MESSAGE);
        assert_eq!(tag3, SECRETSTREAM_TAG_FINAL);
    }

    #[test]
    fn test_stream_pull() {
        use core::convert::TryFrom;

        let message1 = b"Arbitrary data to encrypt";
        let message2 = b"split into";
        let message3 = b"three messages";

        // Generate a random secret key for this stream
        let key = Key::generate();

        // Initialize the push side, type annotations required on return type
        let (mut so_push_stream, so_header) = SOStream::init_push(key.as_slice());
        // Encrypt a series of messages
        let c1: Vec<u8> = so_push_stream
            .push(message1, None, SECRETSTREAM_TAG_MESSAGE)
            .expect("Encrypt failed");
        let c2: Vec<u8> = so_push_stream
            .push(message2, None, SECRETSTREAM_TAG_MESSAGE)
            .expect("Encrypt failed");
        let c3: Vec<u8> = so_push_stream
            .push(message3, None, SECRETSTREAM_TAG_FINAL)
            .expect("Encrypt failed");

        // Initialize the pull side using header generated by the push side
        let mut pull_stream = DryocStream::init_pull(
            &key,
            &Header::try_from(so_header.as_slice()).expect("header"),
        );

        // Decrypt the encrypted messages, type annotations required
        let (m1, tag1): (Vec<u8>, Tag) = pull_stream.pull(&c1, None).expect("Decrypt failed");
        let (m2, tag2): (Vec<u8>, Tag) = pull_stream.pull(&c2, None).expect("Decrypt failed");
        let (m3, tag3): (Vec<u8>, Tag) = pull_stream.pull(&c3, None).expect("Decrypt failed");

        assert_eq!(message1, m1.as_slice());
        assert_eq!(message2, m2.as_slice());
        assert_eq!(message3, m3.as_slice());

        assert_eq!(tag1, Tag::Message);
        assert_eq!(tag2, Tag::Message);
        assert_eq!(tag3, Tag::Final);
    }

    #[cfg(all(feature = "protected", any(unix, windows)))]
    #[test]
    fn test_protected_memory() {
        use crate::protected::*;

        let message1 = b"Arbitrary data to encrypt";
        let message2 = b"split into";
        let message3 = b"three messages";

        // Generate a random secret key for this stream
        let key = protected::Key::generate_locked().expect("generate locked");

        // Initialize the push side, type annotations required on return type
        let (mut push_stream, header): (_, Header) = DryocStream::init_push(&key);

        // Set secret key memory to no-access, but it must be unlocked first
        let key = key
            .munlock()
            .expect("munlock")
            .mprotect_noaccess()
            .expect("mprotect");

        // Encrypt a series of messages
        let c1: Locked<HeapBytes> = push_stream
            .push(message1, None, Tag::Message)
            .expect("Encrypt failed");
        let c2: Vec<u8> = push_stream
            .push(message2, None, Tag::Message)
            .expect("Encrypt failed");
        let c3: Vec<u8> = push_stream
            .push(message3, None, Tag::Final)
            .expect("Encrypt failed");

        // allow access again
        let key = key.mprotect_readonly().expect("mprotect");

        // Initialize the pull side using header generated by the push side
        let mut pull_stream = DryocStream::init_pull(&key, &header);

        // Set secret key memory to no-access
        let _key = key.mprotect_noaccess().expect("mprotect");

        // Decrypt the encrypted messages, type annotations required
        let (m1, tag1): (Locked<HeapBytes>, Tag) =
            pull_stream.pull(&c1, None).expect("Decrypt failed");
        let (m2, tag2): (Locked<HeapBytes>, Tag) =
            pull_stream.pull(&c2, None).expect("Decrypt failed");
        let (m3, tag3): (Locked<HeapBytes>, Tag) =
            pull_stream.pull(&c3, None).expect("Decrypt failed");

        assert_eq!(message1, m1.as_slice());
        assert_eq!(message2, m2.as_slice());
        assert_eq!(message3, m3.as_slice());

        assert_eq!(tag1, Tag::Message);
        assert_eq!(tag2, Tag::Message);
        assert_eq!(tag3, Tag::Final);
    }
}