rs_matter/transport/

session.rs

/*
 *
 *    Copyright (c) 2022-2026 Project CHIP Authors
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

use core::fmt;
use core::num::NonZeroU8;
use embassy_time::Instant;

use cfg_if::cfg_if;

use rand_core::RngCore;

use crate::crypto::{canon, CanonAeadKey, CanonAeadKeyRef, Crypto, CryptoSensitive, Kdf};
use crate::dm::clusters::basic_info::BasicInfoConfig;
use crate::error::{Error, ErrorCode};
use crate::fabric::Fabrics;
use crate::group_keys::KeySet;
use crate::sc::SessionParameters;
use crate::transport::exchange::ExchangeId;
use crate::transport::mrp::{self, ReliableMessage};
use crate::transport::TransportRunner;
use crate::utils::init::{init, Init, IntoFallibleInit};
use crate::utils::storage::{ParseBuf, Vec, WriteBuf};
use crate::{Matter, MatterState};

use super::dedup::{GroupCtrStore, RxCtrState};
use super::exchange::{ExchangeState, MessageMeta, Role};
use super::mrp::RetransEntry;
use super::network::Address;
use super::packet::PacketHdr;
use super::plain_hdr::PlainHdr;
use super::proto_hdr::ProtoHdr;
use super::Packet;

pub const MAX_CAT_IDS_PER_NOC: usize = 3;
pub type NocCatIds = [u32; MAX_CAT_IDS_PER_NOC];

pub const ATT_CHALLENGE_LEN: usize = 16;

canon!(
    ATT_CHALLENGE_LEN,
    ATT_CHALLENGE_ZEROED,
    AttChallenge,
    AttChallengeRef
);

#[derive(Debug, PartialEq, Eq, Clone, Default)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum SessionMode {
    // The Case session will capture the local fabric index
    // and the local fabric index
    Case {
        fab_idx: NonZeroU8,
        cat_ids: NocCatIds,
    },
    // The Pase session always starts with a fabric index of 0
    // (i.e. no fabric) but will be upgraded to the actual fabric index
    // once AddNOC or UpdateNOC is received
    Pase {
        fab_idx: u8,
    },
    // A group session used for group (multicast) messaging.
    Group {
        fab_idx: NonZeroU8,
        group_id: u16,
    },
    #[default]
    PlainText,
}

impl SessionMode {
    pub fn fab_idx(&self) -> u8 {
        match self {
            SessionMode::Case { fab_idx, .. } => fab_idx.get(),
            SessionMode::Pase { fab_idx, .. } => *fab_idx,
            SessionMode::Group { fab_idx, .. } => fab_idx.get(),
            SessionMode::PlainText => 0,
        }
    }
}

pub struct Session {
    // Internal ID which is guaranteeed to be unique accross all sessions and not change when sessions are added/removed
    pub(crate) id: u32,
    peer_addr: Address,
    local_nodeid: u64,
    peer_nodeid: Option<u64>,
    // I find the session initiator/responder role getting confused with exchange initiator/responder
    // So, we might keep this as enc_key and dec_key for now
    dec_key: CanonAeadKey,
    enc_key: CanonAeadKey,
    att_challenge: AttChallenge,
    local_sess_id: u16,
    peer_sess_id: u16,
    msg_ctr: u32,
    rx_ctr_state: RxCtrState,
    mode: SessionMode,
    pub(crate) exchanges: Vec<Option<ExchangeState>, MAX_EXCHANGES>,
    last_use: Instant,
    /// Peer's effective `MRP_SESSION_ACTIVE_INTERVAL` (ms) — drives our
    /// MRP retransmission base interval when transmitting to this peer.
    peer_active_interval_ms: u32,
    /// Peer's effective `MRP_SESSION_IDLE_INTERVAL` (ms) — see
    /// `peer_active_interval_ms`. Currently informational on the responder
    /// side until idle-vs-active classification lands in the MRP code.
    peer_idle_interval_ms: u32,
    /// Peer's effective `MRP_SESSION_ACTIVE_THRESHOLD` (ms).
    peer_active_threshold_ms: u16,
    /// If `true` then the session is considered "expired". Session expiration happens
    /// for the session on behalf of which a fabric is removed.
    ///
    /// Expired sessions can still process their ongoing exchanges, but do not accept any new ones.
    /// Furthermore, expired sessions are the prime candidates for eviction.
    expired: bool,
    reserved: bool,
}

impl Session {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        id: u32,
        msg_ctr: u32,
        reserved: bool,
        peer_addr: Address,
        peer_nodeid: Option<u64>,
        peer_active_interval_ms: u32,
        peer_idle_interval_ms: u32,
        peer_active_threshold_ms: u16,
    ) -> Self {
        Self {
            id,
            reserved,
            peer_addr,
            local_nodeid: 0,
            peer_nodeid,
            dec_key: CanonAeadKey::new(),
            enc_key: CanonAeadKey::new(),
            att_challenge: AttChallenge::new(),
            peer_sess_id: 0,
            local_sess_id: 0,
            msg_ctr: msg_ctr & MATTER_MSG_CTR_RANGE,
            rx_ctr_state: RxCtrState::new(0),
            mode: SessionMode::PlainText,
            exchanges: Vec::new(),
            last_use: Instant::now(),
            peer_active_interval_ms,
            peer_idle_interval_ms,
            peer_active_threshold_ms,
            expired: false,
        }
    }

    #[allow(clippy::too_many_arguments)]
    pub fn init(
        id: u32,
        msg_ctr: u32,
        reserved: bool,
        peer_addr: Address,
        peer_nodeid: Option<u64>,
        peer_active_interval_ms: u32,
        peer_idle_interval_ms: u32,
        peer_active_threshold_ms: u16,
    ) -> impl Init<Self> {
        init!(Self {
            id,
            reserved,
            peer_addr,
            local_nodeid: 0,
            peer_nodeid,
            dec_key <- CanonAeadKey::init(),
            enc_key <- CanonAeadKey::init(),
            att_challenge <- AttChallenge::init(),
            peer_sess_id: 0,
            local_sess_id: 0,
            msg_ctr: msg_ctr & MATTER_MSG_CTR_RANGE,
            rx_ctr_state: RxCtrState::new(0),
            mode: SessionMode::PlainText,
            exchanges <- Vec::init(),
            last_use: Instant::now(),
            peer_active_interval_ms,
            peer_idle_interval_ms,
            peer_active_threshold_ms,
            expired: false,
        })
    }

    /// Get the internal ID of the session
    /// This ID is guaranteed to be unique across all sessions
    pub const fn id(&self) -> u32 {
        self.id
    }

    pub fn get_local_sess_id(&self) -> u16 {
        self.local_sess_id
    }

    #[cfg(test)]
    pub fn set_local_sess_id(&mut self, sess_id: u16) {
        self.local_sess_id = sess_id;
    }

    pub(crate) fn set_local_nodeid(&mut self, nodeid: u64) {
        self.local_nodeid = nodeid;
    }

    pub fn get_peer_sess_id(&self) -> u16 {
        self.peer_sess_id
    }

    pub fn get_peer_addr(&self) -> Address {
        self.peer_addr
    }

    pub fn is_encrypted(&self) -> bool {
        match self.mode {
            SessionMode::Case { .. } | SessionMode::Pase { .. } | SessionMode::Group { .. } => true,
            SessionMode::PlainText => false,
        }
    }

    pub fn get_peer_node_id(&self) -> Option<u64> {
        self.peer_nodeid
    }

    pub fn get_local_fabric_idx(&self) -> u8 {
        self.mode.fab_idx()
    }

    pub fn get_session_mode(&self) -> &SessionMode {
        &self.mode
    }

    pub(crate) fn set_session_mode(&mut self, mode: SessionMode) {
        self.mode = mode;
    }

    pub fn get_peer_active_interval_ms(&self) -> u32 {
        self.peer_active_interval_ms
    }

    pub fn get_peer_idle_interval_ms(&self) -> u32 {
        self.peer_idle_interval_ms
    }

    pub fn get_peer_active_threshold_ms(&self) -> u16 {
        self.peer_active_threshold_ms
    }

    /// Record the peer's `session_parameters` (any combination of `sai` /
    /// `sii` / `sat`) for use by MRP retransmission timing on later sends
    /// to this peer. Only the fields the peer actually advertised get
    /// overwritten; absent fields leave the seeded default in place so
    /// repeated handshakes don't clobber a stronger earlier hint with a
    /// later-but-emptier one.
    ///
    /// `Some(0)` is dropped (with a warning) — this method runs on
    /// Sigma1 / PBKDFParamRequest TLV from an unauthenticated peer, and
    /// an "interval" of zero would either collapse the MRP backoff to
    /// a tight retransmit loop or, in the SAT case, mark the session
    /// active for zero milliseconds. Neither is a legitimate value, so
    /// rejecting them protects against a trivial pre-auth DoS.
    pub(crate) fn set_peer_session_params(&mut self, params: &SessionParameters) {
        if let Some(sai) = params.sai {
            if sai > 0 {
                self.peer_active_interval_ms = sai;
            } else {
                warn!("Peer advertised session_parameters.sai=0; ignoring");
            }
        }

        if let Some(sii) = params.sii {
            if sii > 0 {
                self.peer_idle_interval_ms = sii;
            } else {
                warn!("Peer advertised session_parameters.sii=0; ignoring");
            }
        }

        if let Some(sat) = params.sat {
            if sat > 0 {
                self.peer_active_threshold_ms = sat;
            } else {
                warn!("Peer advertised session_parameters.sat=0; ignoring");
            }
        }
    }

    fn get_msg_ctr(&mut self) -> u32 {
        let ctr = self.msg_ctr;
        self.msg_ctr += 1;
        ctr
    }

    pub fn get_dec_key(&self) -> Option<CanonAeadKeyRef<'_>> {
        match self.mode {
            SessionMode::Case { .. } | SessionMode::Pase { .. } | SessionMode::Group { .. } => {
                Some(self.dec_key.reference())
            }
            SessionMode::PlainText => None,
        }
    }

    pub fn get_enc_key(&self) -> Option<CanonAeadKeyRef<'_>> {
        match self.mode {
            SessionMode::Case { .. } | SessionMode::Pase { .. } | SessionMode::Group { .. } => {
                Some(self.enc_key.reference())
            }
            SessionMode::PlainText => None,
        }
    }

    pub fn get_att_challenge(&self) -> Option<AttChallengeRef<'_>> {
        match self.mode {
            SessionMode::Case { .. } | SessionMode::Pase { .. } => {
                Some(self.att_challenge.reference())
            }
            SessionMode::PlainText | SessionMode::Group { .. } => None,
        }
    }

    /// Whether this is a CASE session to the given peer node ID and fabric index.
    pub(crate) fn is_for_node(&self, fabric_idx: NonZeroU8, peer_node_id: u64) -> bool {
        self.get_local_fabric_idx() == fabric_idx.get()
            && self.peer_nodeid == Some(peer_node_id)
            && self.is_encrypted()
            && !self.reserved
    }

    /// Whether this is a PASE session to the given peer address.
    ///
    /// PASE sessions are all keyed at fabric 0 / node 0 (no operational
    /// identity yet), so the peer *address* is what distinguishes one PASE
    /// session from another - which matters on a commissioner that may have
    /// several PASE sessions (to different devices) in flight at once.
    pub(crate) fn is_pase_for_addr(&self, peer_addr: &Address) -> bool {
        matches!(self.mode, SessionMode::Pase { .. })
            && self.peer_addr.canonical() == peer_addr.canonical()
            && !self.reserved
    }

    pub(crate) fn is_for_rx(&self, rx_peer: &Address, rx_plain: &PlainHdr) -> bool {
        let nodeid_matches = self.peer_nodeid.is_none()
            || rx_plain.get_src_nodeid().is_none()
            || self.peer_nodeid == rx_plain.get_src_nodeid();

        // For unsecured sessions, also match by destination node ID (the echoed
        // ephemeral initiator node ID) to disambiguate multiple unsecured sessions
        // for the same peer (spec).
        let dest_nodeid_matches = self.is_encrypted()
            || self.local_nodeid == 0
            || rx_plain.get_dst_unicast_nodeid().is_none()
            || rx_plain.get_dst_unicast_nodeid() == Some(self.local_nodeid);

        nodeid_matches
            && dest_nodeid_matches
            && self.local_sess_id == rx_plain.sess_id
            // Compare canonically: a dual-stack socket may report a peer as
            // `::ffff:a.b.c.d` on receive while the session stored the plain
            // `V4` address it was created with (or vice versa). Canonicalizing
            // only the *comparison* (not the stored address) lets the two match
            // without disturbing the address used for reply routing. See
            // `Address::canonical`.
            && self.peer_addr.canonical() == rx_peer.canonical()
            && self.is_encrypted() == rx_plain.is_encrypted()
            && !self.reserved
    }

    pub(crate) fn is_for_tx(&self, session_id: u32) -> bool {
        self.id == session_id
    }

    /// Return `true` if the session is expired.
    pub(crate) fn is_expired(&self) -> bool {
        self.expired
    }

    pub fn upgrade_fabric_idx(&mut self, fabric_idx: NonZeroU8) -> Result<(), Error> {
        if let SessionMode::Pase { fab_idx } = &mut self.mode {
            if *fab_idx == 0 {
                *fab_idx = fabric_idx.get();
            } else {
                // Upgrading a PASE session can happen only once
                Err(ErrorCode::Invalid)?;
            }
        } else {
            // CASE sessions are not upgradeable, as per spec
            // And for plain text sessions - we shoudn't even get here in the first place
            Err(ErrorCode::Invalid)?;
        }

        Ok(())
    }

    /// Update the session state with the data in the received packet headers.
    ///
    /// Return `true` if a new exchange was created, and `false` otherwise.
    pub(crate) fn post_recv(&mut self, rx_header: &PacketHdr) -> Result<bool, Error> {
        if !self
            .rx_ctr_state
            .post_recv(rx_header.plain.ctr, self.is_encrypted(), false)
        {
            Err(ErrorCode::Duplicate)?;
        }

        let exch_index = self.get_exch_for_rx(&rx_header.proto);
        if let Some(exch_index) = exch_index {
            let exch = unwrap!(self.exchanges[exch_index].as_mut());

            exch.post_recv(&rx_header.plain, &rx_header.proto)?;

            Ok(false)
        } else {
            if !rx_header.proto.is_initiator()
                || !MessageMeta::from(&rx_header.proto).is_new_exchange()
            {
                // Do not create a new exchange if the peer is not an initiator, or if
                // the packet is NOT a candidate for a new exchange
                // (i.e. it is a standalone ACK or a SC status response)
                Err(ErrorCode::NoExchange)?;
            }

            if self.expired {
                // Per Matter Core spec, an expired session must not
                // accept new inbound messages. Skipping expired sessions here lets the
                // caller surface a `SessionNotFound` to the peer rather than running
                // the request through ACL checks against a removed fabric.
                Err(ErrorCode::NoSession)?;
            }

            if let Some(exch_index) =
                self.add_exch(rx_header.proto.exch_id, Role::Responder(Default::default()))
            {
                // unwrap is safe as we just created the exchange
                let exch = unwrap!(self.exchanges[exch_index].as_mut());

                exch.post_recv(&rx_header.plain, &rx_header.proto)?;

                Ok(true)
            } else {
                Err(ErrorCode::NoSpaceExchanges)?
            }
        }
    }

    pub(crate) fn pre_send(
        &mut self,
        exch_index: Option<usize>,
        tx_header: &mut PacketHdr,
        session_active_interval_ms: Option<u32>,
        session_idle_interval_ms: Option<u32>,
    ) -> Result<(Address, bool), Error> {
        let ctr = if let Some(exchange_index) = exch_index {
            let exchange = unwrap!(self.exchanges[exchange_index].as_mut());
            exchange.mrp.retrans.as_ref().map(RetransEntry::get_msg_ctr)
        } else {
            None
        };

        let retransmission = ctr.is_some();

        tx_header.plain.sess_id = self.get_peer_sess_id();
        tx_header.plain.ctr = ctr.unwrap_or_else(|| self.get_msg_ctr());
        // For unsecured initiator sessions, set Source Node ID to our ephemeral
        // initiator node ID (spec: "enclosed by initiator as Source Node ID").
        // Encrypted sessions and responder unsecured sessions (local_nodeid=0) send no Source.
        tx_header.plain.set_src_nodeid(
            (!self.is_encrypted() && self.local_nodeid != 0).then_some(self.local_nodeid),
        );
        tx_header.plain.set_dst_unicast_nodeid(
            (self.mode == SessionMode::PlainText)
                .then_some(self.peer_nodeid)
                .flatten(),
        );

        tx_header.proto.adjust_reliability(false, &self.peer_addr);

        if let Some(exchange_index) = exch_index {
            let exchange = unwrap!(self.exchanges[exchange_index].as_mut());

            exchange.pre_send(
                &tx_header.plain,
                &mut tx_header.proto,
                session_active_interval_ms,
                session_idle_interval_ms,
            )?;
        }

        Ok((self.peer_addr, retransmission))
    }

    /// Decode the remaining part of the packet after the plain header and then consume the `ParseBuf`
    /// instance as it no longer would be necessary.
    ///
    /// Returns the range of the decoded packet payload
    pub(crate) fn decode_remaining<C: Crypto>(
        &self,
        crypto: C,
        rx_header: &mut PacketHdr,
        mut pb: ParseBuf,
    ) -> Result<(usize, usize), Error> {
        rx_header.decode_remaining(
            crypto,
            self.get_dec_key(),
            self.peer_nodeid.unwrap_or_default(),
            &mut pb,
        )?;

        rx_header.proto.adjust_reliability(true, &self.peer_addr);

        Ok(pb.slice_range())
    }

    pub(crate) fn encode<C: Crypto>(
        &self,
        crypto: C,
        tx: &PacketHdr,
        wb: &mut WriteBuf,
    ) -> Result<(), Error> {
        tx.encode(crypto, self.get_enc_key(), self.local_nodeid, wb)
    }

    fn update_last_used(&mut self) {
        self.last_use = Instant::now();
    }

    pub(crate) fn get_exch_for_rx(&self, rx_proto: &ProtoHdr) -> Option<usize> {
        self.exchanges
            .iter()
            .enumerate()
            .filter(|(_, exch)| {
                exch.as_ref()
                    .map(|exch| exch.is_for_rx(rx_proto))
                    .unwrap_or(false)
            })
            .map(|(index, _)| index)
            .next()
    }

    pub(crate) fn add_exch(&mut self, exch_id: u16, role: Role) -> Option<usize> {
        let exch_state = Some(ExchangeState {
            exch_id,
            role,
            mrp: ReliableMessage::new(),
        });

        let exch_index = if self.exchanges.len() < MAX_EXCHANGES {
            let _ = self.exchanges.push(exch_state);

            self.exchanges.len() - 1
        } else {
            let index = self.exchanges.iter().position(Option::is_none);

            if let Some(index) = index {
                self.exchanges[index] = exch_state;

                index
            } else {
                error!(
                    "Too many exchanges for session {} [SID:{:x},RSID:{:x}]; exchange creation failed",
                    self.id,
                    self.get_local_sess_id(),
                    self.get_peer_sess_id()
                );

                return None;
            }
        };

        let exch_id = ExchangeId::new(self.id, exch_index);

        debug!("New exchange: {} :: {:?}", exch_id.display(self), role);

        Some(exch_index)
    }

    pub(crate) fn remove_exch(&mut self, index: usize) -> bool {
        let exchange = unwrap!(self.exchanges[index].as_mut());
        let exchange_id = ExchangeId::new(self.id, index);

        if exchange.mrp.is_retrans_pending() {
            exchange.role.set_dropped_state();
            error!("Exchange {}: A packet is still (re)transmitted! Marking as dropped, but session will be closed", exchange_id.display(self));

            false
        } else if exchange.mrp.is_ack_pending() {
            exchange.role.set_dropped_state();
            warn!(
                "Exchange {}: Pending ACK. Marking as dropped",
                exchange_id.display(self)
            );

            false
        } else {
            trace!("Exchange {}: Dropped cleanly", exchange_id.display(self));
            self.exchanges[index] = None;

            true
        }
    }
}

impl fmt::Display for Session {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "peer: {:?}, peer_nodeid: {:?}, local: {}, remote: {}, msg_ctr: {}, mode: {:?}, ts: {:?}, expired: {}",
            self.peer_addr,
            self.peer_nodeid,
            self.local_sess_id,
            self.peer_sess_id,
            self.msg_ctr,
            self.mode,
            self.last_use,
            self.expired,
        )
    }
}

/// A helper struct for reserving a session slot in the session table when we don't have all the necessary information to create a full session yet.
///
/// Public for testing purposes, but should not be used outside of the transport module.
pub struct ReservedSession<'a> {
    id: u32,
    matter: &'a Matter<'a>,
    complete: bool,
}

impl<'a> ReservedSession<'a> {
    pub fn reserve_now<C: Crypto>(matter: &'a Matter<'a>, crypto: C) -> Result<Self, Error> {
        let dev_det = matter.dev_det();
        matter.with_state(|state| {
            let mut rand = crypto.weak_rand()?;

            let id = state
                .sessions
                .add(rand.next_u32(), true, Address::new(), None, dev_det)?
                .id;

            Ok(Self {
                id,
                matter,
                complete: false,
            })
        })
    }

    pub async fn reserve<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
    ) -> Result<ReservedSession<'a>, Error> {
        let session = Self::reserve_now(matter, &crypto);

        if let Ok(session) = session {
            Ok(session)
        } else {
            TransportRunner::new(matter, &crypto)
                .evict_some_session()
                .await?;

            Self::reserve_now(matter, &crypto)
        }
    }

    #[allow(clippy::too_many_arguments)]
    pub fn update(
        &mut self,
        local_nodeid: u64,
        peer_nodeid: u64,
        peer_sessid: u16,
        local_sessid: u16,
        peer_addr: Address,
        mode: SessionMode,
        dec_key: Option<CanonAeadKeyRef<'_>>,
        enc_key: Option<CanonAeadKeyRef<'_>>,
        att_challenge: Option<AttChallengeRef<'_>>,
    ) -> Result<(), Error> {
        self.matter.with_state(|state| {
            self.update_with_state(
                state,
                local_nodeid,
                peer_nodeid,
                peer_sessid,
                local_sessid,
                peer_addr,
                mode,
                dec_key,
                enc_key,
                att_challenge,
            )
        })
    }

    #[allow(clippy::too_many_arguments)]
    pub fn update_with_state(
        &mut self,
        state: &mut MatterState,
        local_nodeid: u64,
        peer_nodeid: u64,
        peer_sessid: u16,
        local_sessid: u16,
        peer_addr: Address,
        mode: SessionMode,
        dec_key: Option<CanonAeadKeyRef<'_>>,
        enc_key: Option<CanonAeadKeyRef<'_>>,
        att_challenge: Option<AttChallengeRef<'_>>,
    ) -> Result<(), Error> {
        let session = state.sessions.get(self.id).ok_or(ErrorCode::NoSession)?;

        session.local_nodeid = local_nodeid;
        session.peer_nodeid = Some(peer_nodeid);
        session.peer_sess_id = peer_sessid;
        session.local_sess_id = local_sessid;
        session.peer_addr = peer_addr;
        session.mode = mode;

        if let Some(dec_key) = dec_key {
            session.dec_key.load(dec_key);
        }

        if let Some(enc_key) = enc_key {
            session.enc_key.load(enc_key);
        }

        if let Some(att_challenge) = att_challenge {
            session.att_challenge.load(att_challenge);
        }

        Ok(())
    }

    /// Record the peer's MRP `session_parameters` (Sigma1 / Sigma2 /
    /// PBKDFParamRequest / PBKDFParamResponse) on the reserved session so
    /// they are in place by the time it transitions to a full Session.
    /// Per Matter Core spec, MRP retransmission backoff to this
    /// peer should derive from the peer-advertised `sai` (and idle
    /// detection later from `sii`/`sat`).
    pub(crate) fn set_peer_session_params(
        &mut self,
        params: &SessionParameters,
    ) -> Result<(), Error> {
        self.matter.with_state(|state| {
            let session = state.sessions.get(self.id).ok_or(ErrorCode::NoSession)?;
            session.set_peer_session_params(params);
            Ok(())
        })
    }

    pub fn complete(mut self) {
        self.complete = true;
    }
}

impl Drop for ReservedSession<'_> {
    fn drop(&mut self) {
        self.matter.with_state(|state| {
            if self.complete {
                let session = unwrap!(state.sessions.get(self.id));
                session.reserved = false;
            } else {
                state.sessions.remove(self.id);
            }
        })
    }
}

cfg_if! {
    if #[cfg(feature = "max-sessions-64")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 64;
    } else if #[cfg(feature = "max-sessions-32")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 32;
    } else if #[cfg(feature = "max-sessions-16")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 16;
    } else if #[cfg(feature = "max-sessions-8")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 8;
    } else if #[cfg(feature = "max-sessions-7")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 7;
    } else if #[cfg(feature = "max-sessions-6")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 6;
    } else if #[cfg(feature = "max-sessions-5")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 5;
    } else if #[cfg(feature = "max-sessions-4")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 4;
    } else if #[cfg(feature = "max-sessions-3")] {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 3;
    } else {
        /// Max number of supported sessions
        pub const MAX_SESSIONS: usize = 16;
    }
}

cfg_if! {
    if #[cfg(feature = "max-exchanges-per-session-16")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 16;
    } else if #[cfg(feature = "max-exchanges-per-session-8")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 8;
    } else if #[cfg(feature = "max-exchanges-per-session-7")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 7;
    } else if #[cfg(feature = "max-exchanges-per-session-6")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 6;
    } else if #[cfg(feature = "max-exchanges-per-session-5")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 5;
    } else if #[cfg(feature = "max-exchanges-per-session-4")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 4;
    } else if #[cfg(feature = "max-exchanges-per-session-3")] {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 3;
    } else {
        /// Max number of supported exchanges per session
        pub const MAX_EXCHANGES: usize = 5;
    }
}

const MATTER_MSG_CTR_RANGE: u32 = 0x0fffffff;

/// All sessions
pub struct Sessions {
    next_sess_unique_id: u32,
    next_sess_id: u16,
    next_exch_id: u16,
    sessions: Vec<Session, MAX_SESSIONS>,
    group_ctr_store: GroupCtrStore,
}

impl Sessions {
    /// Create a new Sessions instance.
    #[inline(always)]
    pub const fn new() -> Self {
        Self {
            sessions: Vec::new(),
            group_ctr_store: GroupCtrStore::new(),
            next_sess_unique_id: 0,
            next_sess_id: 1,
            next_exch_id: 1,
        }
    }

    /// Create an in-place initializer for a new Sessions instance.
    pub fn init() -> impl Init<Self> {
        init!(Self {
            sessions <- Vec::init(),
            group_ctr_store: GroupCtrStore::new(),
            next_sess_unique_id: 0,
            next_sess_id: 1,
            next_exch_id: 1,
        })
    }

    pub fn reset(&mut self) {
        self.sessions.clear();
        self.group_ctr_store = GroupCtrStore::new();
        self.next_sess_id = 1;
        self.next_exch_id = 1;
    }

    /// Attempt to decrypt and accept a group (multicast) message.
    ///
    /// Derives group operational keys on-the-fly from `FabricMgr`, matching
    /// the packet's `(session_id, group_id)`, tries to decrypt with each,
    /// validates the group message counter, and creates an ephemeral group
    /// session on success.
    ///
    /// Returns the created session and payload range, mirroring how unicast
    /// uses `get_for_rx()` + `decode_remaining()`.
    pub(crate) fn get_or_create_for_group_rx<const N: usize, C: Crypto>(
        &mut self,
        crypto: C,
        fabrics: &Fabrics,
        packet: &mut Packet<N>,
        dev_det: &BasicInfoConfig<'_>,
    ) -> Result<(&mut Session, (usize, usize)), Error> {
        let src_nodeid = packet
            .header
            .plain
            .get_src_nodeid()
            .ok_or(ErrorCode::InvalidData)?;
        let group_id = packet
            .header
            .plain
            .get_dst_groupcast_nodeid()
            .ok_or(ErrorCode::InvalidData)?;
        let expected_sess_id = packet.header.plain.sess_id;
        let msg_ctr = packet.header.plain.ctr;

        debug!(
            "Group: Attempting decrypt for PEER={:?} SID=0x{:04x}, GRP=0x{:04x}, SRC=0x{:016x}, CTR={}",
            packet.peer, expected_sess_id, group_id, src_nodeid, msg_ctr
        );

        // Parse the plain header to determine encrypted portion offset
        let mut pb = ParseBuf::new(&mut packet.buf[packet.payload_start..]);
        packet.header.plain.decode(&mut pb)?;

        // Save the encrypted payload so we can restore it between decryption attempts
        let encrypted_offset = pb.read_off();
        let encrypted_len = pb.as_slice().len();
        let mut saved_encrypted = [0u8; 1280];
        if encrypted_len > saved_encrypted.len() {
            return Err(ErrorCode::BufferTooSmall.into());
        }
        saved_encrypted[..encrypted_len].copy_from_slice(pb.as_slice());

        // Derive keys on-the-fly and try each one
        let mut group_key_found: Option<(NonZeroU8, (usize, usize))> = None;

        'outer: for fabric in fabrics.iter() {
            let fab_idx = fabric.fab_idx();
            let compressed_fabric_id = fabric.compressed_fabric_id();

            for map_entry in fabric.groups().key_map_iter() {
                if map_entry.group_id != group_id {
                    continue;
                }

                let Some(key_set_entry) = fabric.groups().key_set_get(map_entry.group_key_set_id)
                else {
                    continue;
                };

                for epoch_key_entry in key_set_entry.epoch_keys.iter() {
                    let mut temp_key_set = KeySet::new();
                    if temp_key_set
                        .update(
                            &crypto,
                            epoch_key_entry.epoch_key.reference(),
                            &compressed_fabric_id,
                        )
                        .is_err()
                    {
                        continue;
                    }

                    let op_key_ref = temp_key_set.op_key();
                    let Ok(session_id) = derive_group_session_id(&crypto, op_key_ref) else {
                        continue;
                    };

                    if session_id != expected_sess_id {
                        continue;
                    }

                    if let Some(payload_range) = Self::try_group_decrypt(
                        &crypto,
                        packet,
                        &saved_encrypted[..encrypted_len],
                        encrypted_offset,
                        op_key_ref,
                        src_nodeid,
                    ) {
                        group_key_found = Some((fab_idx, payload_range));
                        break 'outer;
                    }
                }
            }
        }

        if group_key_found.is_none() {
            debug!(
                "Group: No key could decrypt the message (SID=0x{:04x}, GRP=0x{:04x})",
                expected_sess_id, group_id
            );
        }

        let (fab_idx, payload_range) = group_key_found.ok_or(ErrorCode::NoSession)?;

        // Validate group message counter before creating the session
        if !self
            .group_ctr_store
            .post_recv(fab_idx.get(), src_nodeid, msg_ctr)
        {
            debug!(
                "Group: Duplicate message counter {} from node 0x{:016x} fab_idx={}",
                msg_ctr, src_nodeid, fab_idx
            );
            return Err(ErrorCode::Duplicate.into());
        }

        // Create ephemeral group session
        let peer = packet.peer;
        let mut rand = crypto.weak_rand()?;
        let session = match self.add(rand.next_u32(), false, peer, Some(src_nodeid), dev_det) {
            Ok(session) => session,
            Err(_) => {
                // Session table is full; evict the least-recently-used session
                if let Some(lru_id) = self.get_session_for_eviction().map(|sess| sess.id) {
                    debug!("Group: Evicting session {} to make room", lru_id);
                    self.remove(lru_id);
                    self.add(rand.next_u32(), false, peer, Some(src_nodeid), dev_det)?
                } else {
                    return Err(ErrorCode::NoSpaceSessions.into());
                }
            }
        };
        session.set_session_mode(SessionMode::Group { fab_idx, group_id });
        session.local_sess_id = expected_sess_id;

        debug!(
            "Group: Created group session for fab_idx={}, group_id=0x{:04x}, src_nodeid=0x{:016x}",
            fab_idx, group_id, src_nodeid
        );

        // Re-borrow the current created session for returning
        let session = unwrap!(self.sessions.last_mut());
        session.update_last_used();

        Ok((session, payload_range))
    }

    /// Try to decrypt a group message with a candidate key.
    /// Restores the ciphertext before attempting.
    /// On success, returns the payload range; the packet buffer contains decrypted data.
    fn try_group_decrypt<const N: usize, C: Crypto>(
        crypto: C,
        packet: &mut Packet<N>,
        saved_encrypted: &[u8],
        encrypted_offset: usize,
        op_key: CanonAeadKeyRef<'_>,
        src_nodeid: u64,
    ) -> Option<(usize, usize)> {
        // Restore ciphertext
        let start = packet.payload_start + encrypted_offset;
        let encrypted_len = saved_encrypted.len();
        packet.buf[start..start + encrypted_len].copy_from_slice(saved_encrypted);

        // Re-create ParseBuf and re-parse plain header
        let mut pb = ParseBuf::new(&mut packet.buf[packet.payload_start..]);
        if packet.header.plain.decode(&mut pb).is_err() {
            error!("Plain header parse error");
            return None;
        }

        if packet
            .header
            .decode_remaining(crypto, Some(op_key), src_nodeid, &mut pb)
            .is_ok()
        {
            packet.header.proto.adjust_reliability(true, &packet.peer);
            Some(pb.slice_range())
        } else {
            None
        }
    }

    pub fn get_next_sess_id(&mut self) -> u16 {
        let mut next_sess_id: u16;
        loop {
            next_sess_id = self.next_sess_id;

            // Increment next sess id
            self.next_sess_id = self.next_sess_id.overflowing_add(1).0;
            if self.next_sess_id == 0 {
                self.next_sess_id = 1;
            }

            // Ensure the currently selected id doesn't match any existing session
            if self
                .sessions
                .iter()
                .all(|sess| sess.get_local_sess_id() != next_sess_id)
            {
                break;
            }
        }
        next_sess_id
    }

    pub fn get_next_exch_id(&mut self) -> u16 {
        let mut next_exch_id: u16;
        loop {
            next_exch_id = self.next_exch_id;

            // Increment next exch id
            self.next_exch_id = self.next_exch_id.overflowing_add(1).0;
            if self.next_exch_id == 0 {
                self.next_exch_id = 1;
            }

            // Ensure the currently selected id doesn't match any existing exchange
            if self
                .sessions
                .iter()
                .flat_map(|sess| sess.exchanges.iter())
                .filter_map(|exch| exch.as_ref())
                .all(|exch| {
                    !matches!(exch.role, Role::Responder(_)) || exch.exch_id != next_exch_id
                })
            {
                break;
            }
        }
        next_exch_id
    }

    pub fn get_session_for_eviction(&mut self) -> Option<&mut Session> {
        let mut lru_index = None;
        let mut lru_ts = Instant::now();
        for (i, s) in self.sessions.iter().enumerate() {
            if (s.expired || s.last_use < lru_ts)
                && !s.reserved
                && s.exchanges.iter().all(Option::is_none)
            {
                lru_ts = s.last_use;
                lru_index = Some(i);

                if s.expired {
                    // Expired sessons are the prime candidates for eviction,
                    // so we can break early
                    break;
                }
            }
        }

        lru_index.map(|index| &mut self.sessions[index])
    }

    pub fn add(
        &mut self,
        msg_ctr: u32,
        reserved: bool,
        peer_addr: Address,
        peer_nodeid: Option<u64>,
        dev_det: &BasicInfoConfig<'_>,
    ) -> Result<&mut Session, Error> {
        let session_id = self.next_sess_unique_id;

        self.next_sess_unique_id += 1;
        if self.next_sess_unique_id > 0x0fff_ffff {
            // Reserve the upper 4 bits for the exchange index
            self.next_sess_unique_id = 0;
        }

        // Seed the peer's MRP intervals from our own configured defaults;
        // they'll be overwritten by Sigma1 / PBKDFParamRequest (or the
        // initiator's Sigma2 / PBKDFParamResponse) once the peer
        // advertises its own `session_parameters`.
        let (peer_active_interval_ms, peer_idle_interval_ms, peer_active_threshold_ms) =
            mrp::default_peer_mrp_params(dev_det);

        let session = Session::init(
            session_id,
            msg_ctr,
            reserved,
            peer_addr,
            peer_nodeid,
            peer_active_interval_ms,
            peer_idle_interval_ms,
            peer_active_threshold_ms,
        );

        self.sessions
            .push_init(session.into_fallible::<Error>(), || {
                ErrorCode::NoSpaceSessions.into()
            })?;

        Ok(unwrap!(self.sessions.last_mut()))
    }

    /// This assumes that the higher layer has taken care of doing anything required
    /// as per the spec before the session is removed
    pub fn remove(&mut self, id: u32) -> Option<Session> {
        if let Some(index) = self.sessions.iter().position(|sess| sess.id == id) {
            Some(self.sessions.swap_remove(index))
        } else {
            None
        }
    }

    /// This assumes that the higher layer has taken care of doing anything required
    /// as per the spec before the sessions are removed or expired
    pub fn remove_for_fabric(&mut self, fabric_idx: NonZeroU8, expire_sess_id: Option<u32>) {
        while let Some(index) = self.sessions.iter().position(|sess| {
            sess.get_local_fabric_idx() == fabric_idx.get() && Some(sess.id) != expire_sess_id
        }) {
            info!(
                "Dropping session with ID {} for fabric index {} immediately",
                self.sessions[index].id, fabric_idx
            );
            self.sessions.swap_remove(index);
        }

        if let Some(expire_sess_id) = expire_sess_id {
            let expire_sess = self
                .sessions
                .iter_mut()
                .find(|sess| sess.id == expire_sess_id);
            if let Some(expire_sess) = expire_sess {
                expire_sess.expired = true;
                info!(
                    "Marking session with ID {} as expired for fabric index {}",
                    expire_sess_id,
                    fabric_idx.get()
                );
            } else {
                warn!(
                    "No session with ID {} found for fabric index {} to mark as expired",
                    expire_sess_id,
                    fabric_idx.get()
                );
            }
        }
    }

    pub fn get(&mut self, id: u32) -> Option<&mut Session> {
        let mut session = self.sessions.iter_mut().find(|sess| sess.id == id);

        if let Some(session) = session.as_mut() {
            session.update_last_used();
        }

        session
    }

    /// Find the operational (CASE) session for a `(fabric, node)` pair.
    ///
    /// Operational sessions are by definition encrypted and on a real fabric,
    /// so the lookup always matches an encrypted session - there is no
    /// "unsecured by node" lookup (unsecured/PASE sessions carry no operational
    /// identity; see [`Sessions::get_pase_for_addr`]).
    pub(crate) fn get_for_node(
        &mut self,
        fabric_idx: NonZeroU8,
        peer_node_id: u64,
    ) -> Option<&mut Session> {
        // Prefer a TCP-backed session (larger payloads, no MRP fragmentation
        // limits) over UDP when both are available for the same peer. This
        // is required e.g. for the WebRTC Transport Provider's outbound
        // `Answer(sdp)` invoke whose payload can easily exceed a UDP MTU.
        let idx = self
            .sessions
            .iter()
            .enumerate()
            .filter(|(_, s)| !s.expired && s.is_for_node(fabric_idx, peer_node_id))
            .max_by_key(|(_, s)| i32::from(s.peer_addr.is_tcp()))
            .map(|(i, _)| i)?;

        let session = &mut self.sessions[idx];

        session.update_last_used();

        Some(session)
    }

    /// Find an in-flight PASE session to the given peer address.
    ///
    /// Used by [`Exchange::initiate_pase`](crate::transport::exchange::Exchange::initiate_pase)
    /// to reuse a PASE session per peer (rather than assuming a single global
    /// one), so a commissioner can drive several concurrent commissionings.
    pub(crate) fn get_pase_for_addr(&mut self, peer_addr: &Address) -> Option<&mut Session> {
        let mut session = self
            .sessions
            .iter_mut()
            .find(|s| !s.expired && s.is_pase_for_addr(peer_addr));

        if let Some(session) = session.as_mut() {
            session.update_last_used();
        }

        session
    }

    pub(crate) fn get_for_rx(
        &mut self,
        rx_peer: &Address,
        rx_plain: &PlainHdr,
    ) -> Option<&mut Session> {
        let mut session = self
            .sessions
            .iter_mut()
            .find(|sess| sess.is_for_rx(rx_peer, rx_plain));

        if let Some(session) = session.as_mut() {
            session.update_last_used();
        }

        session
    }

    pub(crate) fn get_for_tx(&mut self, session_id: u32) -> Option<&mut Session> {
        let mut session = self
            .sessions
            .iter_mut()
            .find(|sess| sess.is_for_tx(session_id));

        if let Some(session) = session.as_mut() {
            session.update_last_used();
        }

        session
    }

    pub(crate) fn get_exch<F>(&mut self, f: F) -> Option<(&mut Session, usize)>
    where
        F: Fn(&Session, &ExchangeState) -> bool,
    {
        let exch = self
            .sessions
            .iter()
            .flat_map(|sess| {
                sess.exchanges
                    .iter()
                    .enumerate()
                    .filter_map(move |(exch_index, exch)| {
                        exch.as_ref().map(|exch| (sess, exch, exch_index))
                    })
            })
            .filter(|(sess, exch, _)| f(sess, exch))
            .map(|(sess, _, exch_index)| (sess.id, exch_index))
            .next();

        if let Some((id, exch_index)) = exch {
            let session = unwrap!(self.get(id));
            session.update_last_used();

            Some((session, exch_index))
        } else {
            None
        }
    }

    /// Iterate over the sessions
    pub fn iter(&self) -> impl Iterator<Item = &Session> {
        self.sessions.iter()
    }

    /// Drop every PASE session, whether unpromoted (still
    /// `SessionMode::Pase { fab_idx: 0 }`) or already promoted to a
    /// fabric. Used by:
    ///
    /// * `RevokeCommissioning` and a fail-safe expiry over a PASE
    ///   session (Matter Core spec): when the
    ///   commissioning window is torn down, any in-flight PASE sessions
    ///   associated with it must be terminated. A PASE that was
    ///   promoted via `AddNOC` is rolled back by the same fail-safe
    ///   expiry, so its session must go too.
    /// * `CommissioningComplete` (Matter Core spec): once the device
    ///   transitions to operational state,
    ///   all PASE sessions SHALL be terminated. Without this each
    ///   commissioning round leaks the promoted PASE it ran on, and
    ///   the session table eventually exhausts — visible as `BUSY` on
    ///   the next round's `PBKDFParamRequest`.
    ///
    /// `expire_sess_id` is the optional ID of a session that should NOT
    /// be removed immediately — typically the session that issued the
    /// triggering command, so its response can still be sent. That
    /// session is marked as `expired` instead, so it stops accepting
    /// new exchanges but the in-flight one can complete; the transport
    /// reclaims the slot via the usual LRU eviction path.
    pub fn remove_pase(&mut self, expire_sess_id: Option<u32>) {
        while let Some(index) = self.sessions.iter().position(|sess| {
            matches!(sess.get_session_mode(), SessionMode::Pase { .. })
                && Some(sess.id) != expire_sess_id
        }) {
            info!("Dropping PASE session with ID {}", self.sessions[index].id);
            self.sessions.swap_remove(index);
        }

        if let Some(expire_sess_id) = expire_sess_id {
            if let Some(sess) = self.sessions.iter_mut().find(|sess| {
                sess.id == expire_sess_id
                    && matches!(sess.get_session_mode(), SessionMode::Pase { .. })
            }) {
                sess.expired = true;
                info!("Marking PASE session with ID {} as expired", expire_sess_id);
            }
        }
    }
}

impl Default for Sessions {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Display for Sessions {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        writeln!(f, "{{[")?;
        for s in &self.sessions {
            writeln!(f, "{{ {}, }},", s)?;
        }
        write!(f, "], next_sess_id: {}", self.next_sess_id)?;
        write!(f, "}}")
    }
}

/// Derive the Group Session ID from an operational group key.
///
/// Per Matter Spec:
/// ```text
/// GroupKeyHash = Crypto_KDF(
///     InputKey = OperationalGroupKey,
///     Salt     = [],
///     Info     = "GroupKeyHash",
///     Length   = 16 bits
/// )
/// GroupSessionId = (GroupKeyHash[0] << 8) | GroupKeyHash[1]
/// ```
pub fn derive_group_session_id<C: Crypto>(
    crypto: C,
    op_key: CanonAeadKeyRef<'_>,
) -> Result<u16, Error> {
    const GRP_KEY_HASH_INFO: &[u8] = b"GroupKeyHash";

    let mut hash = CryptoSensitive::<2>::new();

    crypto
        .kdf()?
        .expand(&[], op_key, GRP_KEY_HASH_INFO, &mut hash)
        .map_err(|_| ErrorCode::InvalidData)?;

    let bytes = hash.access();
    Ok(((bytes[0] as u16) << 8) | (bytes[1] as u16))
}

#[cfg(test)]
mod tests {
    use crate::crypto::{test_only_crypto, AEAD_KEY_ZEROED};
    use crate::dm::clusters::basic_info::BasicInfoConfig;
    use crate::transport::network::Address;

    use super::*;

    /// Stand-in `BasicInfoConfig` for tests that don't care about the
    /// peer-MRP defaults — `Sessions::add` only reads `sai`/`sii` from it.
    const TEST_DEV_DET: BasicInfoConfig<'static> = BasicInfoConfig::new();

    #[test]
    fn test_next_sess_id_doesnt_reuse() {
        let mut sm = Sessions::new();
        let sess = unwrap!(sm.add(0, false, Address::default(), None, &TEST_DEV_DET));
        sess.set_local_sess_id(1);
        assert_eq!(sm.get_next_sess_id(), 2);
        assert_eq!(sm.get_next_sess_id(), 3);
        let sess = unwrap!(sm.add(0, false, Address::default(), None, &TEST_DEV_DET));
        sess.set_local_sess_id(4);
        assert_eq!(sm.get_next_sess_id(), 5);
    }

    #[test]
    fn test_next_sess_id_overflows() {
        let mut sm = Sessions::new();
        let sess = unwrap!(sm.add(0, false, Address::default(), None, &TEST_DEV_DET));
        sess.set_local_sess_id(1);
        assert_eq!(sm.get_next_sess_id(), 2);
        sm.next_sess_id = 65534;
        assert_eq!(sm.get_next_sess_id(), 65534);
        assert_eq!(sm.get_next_sess_id(), 65535);
        assert_eq!(sm.get_next_sess_id(), 2);
    }

    #[test]
    fn test_derive_group_session_id() {
        // Spec test vector:
        // Operational Group Key: a6:f5:30:6b:af:6d:05:0a:f2:3b:a4:bd:6b:9d:d9:60
        // Expected GroupSessionId: 0xB9F7 (47607)
        let op_key_bytes: [u8; 16] = [
            0xa6, 0xf5, 0x30, 0x6b, 0xaf, 0x6d, 0x05, 0x0a, 0xf2, 0x3b, 0xa4, 0xbd, 0x6b, 0x9d,
            0xd9, 0x60,
        ];

        let mut op_key = AEAD_KEY_ZEROED;
        op_key.try_load_from_slice(&op_key_bytes).unwrap();

        let crypto = test_only_crypto();
        let session_id = derive_group_session_id(&crypto, op_key.reference()).unwrap();

        assert_eq!(
            session_id, 0xB9F7,
            "Group Session ID mismatch: got 0x{:04X}, expected 0xB9F7",
            session_id
        );
    }
}