use std::collections::HashSet;
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use fastmcp_core::logging::{debug, targets, warn};
use fastmcp_core::{McpContext, SessionState, Sha256Digest, sha256_bounded};
use fastmcp_protocol::{
ClientCapabilities, ClientInfo, JsonRpcRequest, LogLevel, ResourceUpdatedNotificationParams,
ServerCapabilities, ServerInfo,
};
use crate::NotificationSender;
static NEXT_SESSION_ID: AtomicU64 = AtomicU64::new(1);
const fn checked_session_id_successor(current: u64) -> Option<u64> {
current.checked_add(1)
}
fn next_session_id() -> u64 {
NEXT_SESSION_ID
.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
checked_session_id_successor(current)
})
.unwrap_or_else(|_| {
panic!("process-local MCP session identity space exhausted")
})
}
pub(crate) const MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION: usize = 32;
pub(crate) const MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION: usize = 4 * 1024 * 1024;
const RESOURCE_LOG_HASH_INPUT_LIMIT: usize = 4 * 1024;
const RESOURCE_LOG_DIGEST_PREFIX_BYTES: usize = 8;
#[derive(Clone, Copy)]
struct SafeResourceLogLabel {
byte_len: usize,
hashed_bytes: usize,
digest_prefix: [u8; RESOURCE_LOG_DIGEST_PREFIX_BYTES],
}
impl fmt::Display for SafeResourceLogLabel {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "bytes={},sha256_prefix=", self.byte_len)?;
for byte in self.digest_prefix {
write!(f, "{byte:02x}")?;
}
if self.hashed_bytes < self.byte_len {
write!(f, ",hashed_prefix_bytes={}", self.hashed_bytes)?;
}
Ok(())
}
}
fn safe_resource_log_label(uri: &str) -> SafeResourceLogLabel {
let bytes = uri.as_bytes();
let hashed_bytes = bytes.len().min(RESOURCE_LOG_HASH_INPUT_LIMIT);
let bounded_prefix = &bytes[..hashed_bytes];
let mut digest_prefix = [0_u8; RESOURCE_LOG_DIGEST_PREFIX_BYTES];
if let Ok(digest) = sha256_bounded(bounded_prefix, RESOURCE_LOG_HASH_INPUT_LIMIT) {
digest_prefix.copy_from_slice(&digest.as_bytes()[..RESOURCE_LOG_DIGEST_PREFIX_BYTES]);
}
SafeResourceLogLabel {
byte_len: bytes.len(),
hashed_bytes,
digest_prefix,
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SubscriptionAdmission {
Accepted,
Duplicate,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SubscriptionAdmissionError {
CapacityExceeded,
RequestNotLive,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SubscriptionRemoval {
Removed,
NotSubscribed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SubscriptionRemovalError {
RequestNotLive,
}
#[derive(Clone)]
pub(crate) struct InitializationSnapshot {
initialized: bool,
client_info: Option<ClientInfo>,
client_capabilities: Option<ClientCapabilities>,
protocol_version: Option<String>,
}
#[derive(Clone, Default)]
pub(crate) struct SessionPrincipalBinding {
fingerprint: Arc<Mutex<Option<Sha256Digest>>>,
}
impl fmt::Debug for SessionPrincipalBinding {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("SessionPrincipalBinding")
.field("bound", &self.is_bound_for_debug())
.finish()
}
}
impl SessionPrincipalBinding {
pub(crate) fn bind_or_verify(&self, fingerprint: Sha256Digest) -> bool {
let mut bound = self
.fingerprint
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
match *bound {
Some(existing) => existing == fingerprint,
None => {
*bound = Some(fingerprint);
true
}
}
}
pub(crate) fn verify_existing(&self, fingerprint: Sha256Digest) -> bool {
self.fingerprint
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.is_some_and(|existing| existing == fingerprint)
}
fn is_bound_for_debug(&self) -> bool {
self.fingerprint
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.is_some()
}
}
pub struct Session {
id: u64,
initialized: bool,
client_info: Option<ClientInfo>,
client_capabilities: Option<ClientCapabilities>,
server_info: ServerInfo,
server_capabilities: ServerCapabilities,
protocol_version: Option<String>,
resource_subscriptions: HashSet<String>,
resource_subscription_bytes: usize,
log_level: Option<LogLevel>,
state: SessionState,
principal_binding: SessionPrincipalBinding,
}
impl fmt::Debug for Session {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Session")
.field("initialized", &self.initialized)
.field("has_client_info", &self.client_info.is_some())
.field(
"has_client_capabilities",
&self.client_capabilities.is_some(),
)
.field("has_protocol_version", &self.protocol_version.is_some())
.field(
"resource_subscription_count",
&self.resource_subscriptions.len(),
)
.field(
"resource_subscription_bytes",
&self.resource_subscription_bytes,
)
.field("has_log_level", &self.log_level.is_some())
.field("state_entry_count", &self.state.len())
.field(
"principal_bound",
&self.principal_binding.is_bound_for_debug(),
)
.finish()
}
}
impl Session {
#[must_use]
pub fn new(server_info: ServerInfo, server_capabilities: ServerCapabilities) -> Self {
Self {
id: next_session_id(),
initialized: false,
client_info: None,
client_capabilities: None,
server_info,
server_capabilities,
protocol_version: None,
resource_subscriptions: HashSet::new(),
resource_subscription_bytes: 0,
log_level: None,
state: SessionState::new(),
principal_binding: SessionPrincipalBinding::default(),
}
}
#[must_use]
pub(crate) fn id(&self) -> u64 {
self.id
}
#[must_use]
pub fn state(&self) -> &SessionState {
&self.state
}
pub(crate) fn principal_binding(&self) -> SessionPrincipalBinding {
self.principal_binding.clone()
}
#[must_use]
pub fn is_initialized(&self) -> bool {
self.initialized
}
pub fn initialize(
&mut self,
client_info: ClientInfo,
client_capabilities: ClientCapabilities,
protocol_version: String,
) {
self.client_info = Some(client_info);
self.client_capabilities = Some(client_capabilities);
self.protocol_version = Some(protocol_version);
self.initialized = true;
}
pub(crate) fn initialization_snapshot(&self) -> InitializationSnapshot {
InitializationSnapshot {
initialized: self.initialized,
client_info: self.client_info.clone(),
client_capabilities: self.client_capabilities.clone(),
protocol_version: self.protocol_version.clone(),
}
}
pub(crate) fn restore_initialization(&mut self, snapshot: InitializationSnapshot) {
self.initialized = snapshot.initialized;
self.client_info = snapshot.client_info;
self.client_capabilities = snapshot.client_capabilities;
self.protocol_version = snapshot.protocol_version;
}
#[must_use]
pub fn client_info(&self) -> Option<&ClientInfo> {
self.client_info.as_ref()
}
#[must_use]
pub fn client_capabilities(&self) -> Option<&ClientCapabilities> {
self.client_capabilities.as_ref()
}
#[must_use]
pub fn server_info(&self) -> &ServerInfo {
&self.server_info
}
#[must_use]
pub fn server_capabilities(&self) -> &ServerCapabilities {
&self.server_capabilities
}
#[must_use]
pub fn protocol_version(&self) -> Option<&str> {
self.protocol_version.as_deref()
}
pub(crate) fn subscribe_resource(
&mut self,
ctx: &McpContext,
uri: String,
) -> Result<SubscriptionAdmission, SubscriptionAdmissionError> {
if ctx.ensure_live().is_err() {
return Err(SubscriptionAdmissionError::RequestNotLive);
}
if uri.len() > MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION {
warn!(
target: targets::SESSION,
"resource subscription rejected by session limits; subscription_count={}; subscription_count_limit={}; retained_uri_bytes={}; retained_uri_bytes_limit={}; candidate_uri={}",
self.resource_subscriptions.len(),
MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION,
self.resource_subscription_bytes,
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION,
safe_resource_log_label(&uri),
);
return Err(SubscriptionAdmissionError::CapacityExceeded);
}
if self.resource_subscriptions.contains(&uri) {
return Ok(SubscriptionAdmission::Duplicate);
}
let prospective_bytes = self.resource_subscription_bytes.checked_add(uri.len());
if self.resource_subscriptions.len() >= MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION
|| prospective_bytes
.is_none_or(|bytes| bytes > MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION)
{
warn!(
target: targets::SESSION,
"resource subscription rejected by session limits; subscription_count={}; subscription_count_limit={}; retained_uri_bytes={}; retained_uri_bytes_limit={}; candidate_uri={}",
self.resource_subscriptions.len(),
MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION,
self.resource_subscription_bytes,
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION,
safe_resource_log_label(&uri),
);
return Err(SubscriptionAdmissionError::CapacityExceeded);
}
let Some(prospective_bytes) = prospective_bytes else {
return Err(SubscriptionAdmissionError::CapacityExceeded);
};
if self.resource_subscriptions.try_reserve(1).is_err() {
warn!(
target: targets::SESSION,
"resource subscription rejected because bounded storage could not be reserved; subscription_count={}; subscription_count_limit={}; retained_uri_bytes={}; retained_uri_bytes_limit={}; candidate_uri={}",
self.resource_subscriptions.len(),
MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION,
self.resource_subscription_bytes,
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION,
safe_resource_log_label(&uri),
);
return Err(SubscriptionAdmissionError::CapacityExceeded);
}
if ctx.ensure_live().is_err() {
return Err(SubscriptionAdmissionError::RequestNotLive);
}
if self.resource_subscriptions.insert(uri) {
self.resource_subscription_bytes = prospective_bytes;
Ok(SubscriptionAdmission::Accepted)
} else {
Ok(SubscriptionAdmission::Duplicate)
}
}
pub(crate) fn unsubscribe_resource(
&mut self,
ctx: &McpContext,
uri: &str,
) -> Result<SubscriptionRemoval, SubscriptionRemovalError> {
self.unsubscribe_resource_with_precommit(ctx, uri, || {})
}
fn unsubscribe_resource_with_precommit(
&mut self,
ctx: &McpContext,
uri: &str,
before_commit: impl FnOnce(),
) -> Result<SubscriptionRemoval, SubscriptionRemovalError> {
if ctx.ensure_live().is_err() {
return Err(SubscriptionRemovalError::RequestNotLive);
}
if uri.len() > MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION {
return Ok(SubscriptionRemoval::NotSubscribed);
}
let subscribed = self.resource_subscriptions.contains(uri);
before_commit();
if ctx.ensure_live().is_err() {
return Err(SubscriptionRemovalError::RequestNotLive);
}
if !subscribed {
return Ok(SubscriptionRemoval::NotSubscribed);
}
if let Some(removed) = self.resource_subscriptions.take(uri) {
self.resource_subscription_bytes = self
.resource_subscription_bytes
.saturating_sub(removed.len());
Ok(SubscriptionRemoval::Removed)
} else {
Ok(SubscriptionRemoval::NotSubscribed)
}
}
pub(crate) fn rollback_resource_subscription(&mut self, uri: &str) {
if let Some(removed) = self.resource_subscriptions.take(uri) {
self.resource_subscription_bytes = self
.resource_subscription_bytes
.saturating_sub(removed.len());
}
}
pub(crate) fn restore_resource_subscription(&mut self, uri: String) {
let uri_len = uri.len();
if self.resource_subscriptions.insert(uri) {
self.resource_subscription_bytes = self
.resource_subscription_bytes
.checked_add(uri_len)
.expect("restoring a previously retained subscription cannot overflow");
}
debug_assert!(self.resource_subscriptions.len() <= MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION);
debug_assert!(
self.resource_subscription_bytes <= MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION
);
}
#[must_use]
pub fn is_resource_subscribed(&self, uri: &str) -> bool {
self.resource_subscriptions.contains(uri)
}
#[must_use]
pub fn subscribed_resource_uris(&self) -> impl Iterator<Item = &str> {
self.resource_subscriptions.iter().map(String::as_str)
}
pub fn set_log_level(&mut self, level: LogLevel) {
self.log_level = Some(level);
}
pub(crate) fn restore_log_level(&mut self, level: Option<LogLevel>) {
self.log_level = level;
}
#[must_use]
pub fn log_level(&self) -> Option<LogLevel> {
self.log_level
}
#[must_use]
pub fn supports_sampling(&self) -> bool {
self.client_capabilities
.as_ref()
.is_some_and(|caps| caps.sampling.is_some())
}
#[must_use]
pub fn supports_elicitation(&self) -> bool {
self.client_capabilities
.as_ref()
.is_some_and(|caps| caps.elicitation.is_some())
}
#[must_use]
pub fn supports_roots(&self) -> bool {
self.client_capabilities
.as_ref()
.is_some_and(|caps| caps.roots.is_some())
}
pub fn notify_resource_updated(&self, uri: &str, sender: &NotificationSender) -> bool {
if !self.is_resource_subscribed(uri) {
return false;
}
let safe_uri = safe_resource_log_label(uri);
let params = ResourceUpdatedNotificationParams {
uri: uri.to_string(),
};
let payload = match serde_json::to_value(params) {
Ok(value) => value,
Err(err) => {
warn!(
target: targets::SESSION,
"failed to serialize resource update; uri={}: {}",
safe_uri,
err
);
return false;
}
};
debug!(
target: targets::SESSION,
"sending resource update notification; uri={}",
safe_uri
);
let notification =
JsonRpcRequest::notification("notifications/resources/updated", Some(payload));
if crate::catch_extension_unwind(|| sender(notification)).is_err() {
warn!(
target: targets::SESSION,
"resource update notification sender terminated unexpectedly; uri={}; detail=panic_payload_redacted",
safe_uri
);
return false;
}
true
}
}
#[cfg(test)]
mod tests {
use super::*;
use asupersync::Cx;
use fastmcp_protocol::{ElicitationCapability, RootsCapability, SamplingCapability};
use std::sync::{Arc, Mutex};
fn make_server_info() -> ServerInfo {
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
}
}
fn make_client_info() -> ClientInfo {
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
}
}
fn make_session() -> Session {
Session::new(make_server_info(), ServerCapabilities::default())
}
fn live_context() -> McpContext {
McpContext::new(Cx::for_testing(), 1)
}
fn admit_resource(session: &mut Session, uri: impl Into<String>) -> SubscriptionAdmission {
session
.subscribe_resource(&live_context(), uri.into())
.expect("test subscription should be admitted")
}
fn remove_resource(session: &mut Session, uri: &str) -> SubscriptionRemoval {
session
.unsubscribe_resource(&live_context(), uri)
.expect("live test unsubscription should succeed")
}
#[test]
fn new_session_is_not_initialized() {
let session = make_session();
assert!(!session.is_initialized());
}
#[test]
fn session_id_successor_rejects_wraparound() {
assert_eq!(checked_session_id_successor(1), Some(2));
assert_eq!(checked_session_id_successor(u64::MAX - 1), Some(u64::MAX));
assert_eq!(checked_session_id_successor(u64::MAX), None);
}
#[test]
fn new_session_has_no_client_info() {
let session = make_session();
assert!(session.client_info().is_none());
}
#[test]
fn new_session_has_no_client_capabilities() {
let session = make_session();
assert!(session.client_capabilities().is_none());
}
#[test]
fn new_session_has_no_protocol_version() {
let session = make_session();
assert!(session.protocol_version().is_none());
}
#[test]
fn new_session_has_no_log_level() {
let session = make_session();
assert!(session.log_level().is_none());
}
#[test]
fn new_session_returns_server_info() {
let session = make_session();
assert_eq!(session.server_info().name, "test");
assert_eq!(session.server_info().version, "1.0");
}
#[test]
fn new_session_returns_server_capabilities() {
let caps = ServerCapabilities::default();
let session = Session::new(make_server_info(), caps);
let _ = session.server_capabilities();
}
#[test]
fn initialize_sets_initialized_flag() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
assert!(session.is_initialized());
}
#[test]
fn initialize_stores_client_info() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
let info = session.client_info().expect("client_info set");
assert_eq!(info.name, "test-client");
assert_eq!(info.version, "1.0");
}
#[test]
fn initialize_stores_client_capabilities() {
let mut session = make_session();
let caps = ClientCapabilities {
sampling: Some(SamplingCapability {}),
elicitation: None,
roots: None,
};
session.initialize(make_client_info(), caps, "2024-11-05".to_string());
let stored = session.client_capabilities().expect("caps set");
assert!(stored.sampling.is_some());
}
#[test]
fn initialize_stores_protocol_version() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2025-03-26".to_string(),
);
assert_eq!(session.protocol_version(), Some("2025-03-26"));
}
#[test]
fn subscribe_and_check_resource() {
let mut session = make_session();
assert!(!session.is_resource_subscribed("file:///a.txt"));
assert_eq!(
admit_resource(&mut session, "file:///a.txt"),
SubscriptionAdmission::Accepted
);
assert!(session.is_resource_subscribed("file:///a.txt"));
}
#[test]
fn unsubscribe_resource_removes_it() {
let mut session = make_session();
admit_resource(&mut session, "file:///a.txt");
assert_eq!(
remove_resource(&mut session, "file:///a.txt"),
SubscriptionRemoval::Removed
);
assert!(!session.is_resource_subscribed("file:///a.txt"));
}
#[test]
fn unsubscribe_nonexistent_resource_is_noop() {
let mut session = make_session();
assert_eq!(
remove_resource(&mut session, "file:///does-not-exist"),
SubscriptionRemoval::NotSubscribed
);
assert!(!session.is_resource_subscribed("file:///does-not-exist"));
}
#[test]
fn cancellation_during_unsubscribe_lookup_cannot_commit_removal() {
let mut session = make_session();
let uri = "resource://retained";
admit_resource(&mut session, uri);
let raw_cx = Cx::for_testing();
let ctx = McpContext::new(raw_cx.clone(), 44);
let result = session.unsubscribe_resource_with_precommit(&ctx, uri, || {
raw_cx.set_cancel_requested(true);
});
assert_eq!(result, Err(SubscriptionRemovalError::RequestNotLive));
assert!(session.is_resource_subscribed(uri));
assert_eq!(session.resource_subscription_bytes, uri.len());
}
#[test]
fn multiple_subscriptions_are_independent() {
let mut session = make_session();
admit_resource(&mut session, "a://1");
admit_resource(&mut session, "b://2");
assert!(session.is_resource_subscribed("a://1"));
assert!(session.is_resource_subscribed("b://2"));
remove_resource(&mut session, "a://1");
assert!(!session.is_resource_subscribed("a://1"));
assert!(session.is_resource_subscribed("b://2"));
}
#[test]
fn duplicate_subscribe_is_idempotent() {
let mut session = make_session();
assert_eq!(
admit_resource(&mut session, "r://x"),
SubscriptionAdmission::Accepted
);
let retained_bytes = session.resource_subscription_bytes;
assert_eq!(
admit_resource(&mut session, "r://x"),
SubscriptionAdmission::Duplicate
);
assert!(session.is_resource_subscribed("r://x"));
assert_eq!(session.resource_subscriptions.len(), 1);
assert_eq!(session.resource_subscription_bytes, retained_bytes);
remove_resource(&mut session, "r://x");
assert!(!session.is_resource_subscribed("r://x"));
assert_eq!(session.resource_subscription_bytes, 0);
}
#[test]
fn resource_subscription_count_admits_exact_cap_and_rejects_cap_plus_one() {
let mut session = make_session();
for index in 0..MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION {
assert_eq!(
admit_resource(&mut session, format!("test://subscription/{index}")),
SubscriptionAdmission::Accepted
);
}
assert_eq!(
session.resource_subscriptions.len(),
MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION
);
let retained_bytes = session.resource_subscription_bytes;
let over_limit_uri = "test://subscription/over-limit";
let rejection = session.subscribe_resource(&live_context(), over_limit_uri.to_string());
assert_eq!(rejection, Err(SubscriptionAdmissionError::CapacityExceeded));
assert!(!session.is_resource_subscribed(over_limit_uri));
assert_eq!(session.resource_subscription_bytes, retained_bytes);
remove_resource(&mut session, "test://subscription/0");
assert_eq!(
admit_resource(&mut session, over_limit_uri),
SubscriptionAdmission::Accepted
);
assert!(session.is_resource_subscribed(over_limit_uri));
assert_eq!(
session.resource_subscriptions.len(),
MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION
);
}
#[test]
fn resource_subscription_bytes_admit_exact_cap_reject_cap_plus_one_and_release() {
let mut session = make_session();
let exact_limit_uri = "é".repeat(MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION / 2);
assert_eq!(
exact_limit_uri.len(),
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION
);
assert_eq!(
admit_resource(&mut session, exact_limit_uri.clone()),
SubscriptionAdmission::Accepted
);
assert!(session.is_resource_subscribed(&exact_limit_uri));
assert_eq!(
session.resource_subscription_bytes,
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION
);
assert_eq!(
admit_resource(&mut session, exact_limit_uri.clone()),
SubscriptionAdmission::Duplicate
);
assert_eq!(
session.resource_subscription_bytes,
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION
);
remove_resource(&mut session, &exact_limit_uri);
assert_eq!(session.resource_subscription_bytes, 0);
assert!(session.resource_subscriptions.is_empty());
let over_limit_uri = format!("{exact_limit_uri}x");
assert_eq!(
over_limit_uri.len(),
MAX_RESOURCE_SUBSCRIPTION_BYTES_PER_SESSION + 1
);
let rejection = session.subscribe_resource(&live_context(), over_limit_uri.clone());
assert_eq!(rejection, Err(SubscriptionAdmissionError::CapacityExceeded));
assert!(!session.is_resource_subscribed(&over_limit_uri));
assert_eq!(session.resource_subscription_bytes, 0);
assert_eq!(
remove_resource(&mut session, &over_limit_uri),
SubscriptionRemoval::NotSubscribed
);
}
#[test]
fn non_live_request_cannot_admit_or_duplicate_a_subscription() {
let mut session = make_session();
admit_resource(&mut session, "resource://retained");
let cancelled_cx = Cx::for_testing();
cancelled_cx.set_cancel_requested(true);
let cancelled = McpContext::new(cancelled_cx, 2);
assert_eq!(
session.subscribe_resource(&cancelled, "resource://new".to_string()),
Err(SubscriptionAdmissionError::RequestNotLive)
);
assert_eq!(
session.subscribe_resource(&cancelled, "resource://retained".to_string()),
Err(SubscriptionAdmissionError::RequestNotLive)
);
assert_eq!(
session.unsubscribe_resource(&cancelled, "resource://retained"),
Err(SubscriptionRemovalError::RequestNotLive)
);
assert!(!session.is_resource_subscribed("resource://new"));
assert!(session.is_resource_subscribed("resource://retained"));
let (expired, lease) = McpContext::new(Cx::for_testing(), 3)
.begin_request_scope()
.expect("fresh context should acquire its request lease");
drop(lease);
assert_eq!(
session.subscribe_resource(&expired, "resource://expired".to_string()),
Err(SubscriptionAdmissionError::RequestNotLive)
);
assert_eq!(
session.unsubscribe_resource(&expired, "resource://retained"),
Err(SubscriptionRemovalError::RequestNotLive)
);
assert!(!session.is_resource_subscribed("resource://expired"));
assert!(session.is_resource_subscribed("resource://retained"));
}
#[test]
fn set_log_level_and_read_back() {
let mut session = make_session();
session.set_log_level(LogLevel::Warning);
assert_eq!(session.log_level(), Some(LogLevel::Warning));
}
#[test]
fn set_log_level_overwrites_previous() {
let mut session = make_session();
session.set_log_level(LogLevel::Debug);
session.set_log_level(LogLevel::Error);
assert_eq!(session.log_level(), Some(LogLevel::Error));
}
#[test]
fn state_is_accessible() {
let session = make_session();
let state = session.state();
let val: Option<String> = state.get("key");
assert!(val.is_none());
}
#[test]
fn notify_resource_updated_returns_false_when_not_subscribed() {
let session = make_session();
let sender: NotificationSender = Arc::new(|_| {});
assert!(!session.notify_resource_updated("file:///a.txt", &sender));
}
#[test]
fn notify_resource_updated_sends_when_subscribed() {
let mut session = make_session();
admit_resource(&mut session, "file:///a.txt");
let sent = Arc::new(Mutex::new(Vec::new()));
let sent_clone = Arc::clone(&sent);
let sender: NotificationSender = Arc::new(move |req| {
sent_clone.lock().unwrap().push(req);
});
let result = session.notify_resource_updated("file:///a.txt", &sender);
assert!(result);
let messages = sent.lock().unwrap();
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].method, "notifications/resources/updated");
}
#[test]
fn notify_resource_updated_includes_uri_in_params() {
let mut session = make_session();
admit_resource(&mut session, "test://res");
let sent = Arc::new(Mutex::new(Vec::new()));
let sent_clone = Arc::clone(&sent);
let sender: NotificationSender = Arc::new(move |req| {
sent_clone.lock().unwrap().push(req);
});
session.notify_resource_updated("test://res", &sender);
let messages = sent.lock().unwrap();
let params = messages[0].params.as_ref().expect("params present");
let uri = params
.get("uri")
.and_then(|v| v.as_str())
.expect("uri field");
assert_eq!(uri, "test://res");
}
#[test]
fn notify_resource_updated_does_not_fire_for_other_uri() {
let mut session = make_session();
admit_resource(&mut session, "file:///a.txt");
let sent = Arc::new(Mutex::new(Vec::new()));
let sent_clone = Arc::clone(&sent);
let sender: NotificationSender = Arc::new(move |req| {
sent_clone.lock().unwrap().push(req);
});
let result = session.notify_resource_updated("file:///b.txt", &sender);
assert!(!result);
assert!(sent.lock().unwrap().is_empty());
}
#[test]
fn resource_update_log_label_is_bounded_and_redacts_uri_text() {
const URI_CANARY: &str = "private://customer-pii-canary@example.invalid/secret-token";
let metadata = format!("{}", safe_resource_log_label(URI_CANARY));
assert!(metadata.contains(&format!("bytes={}", URI_CANARY.len())));
assert!(metadata.contains("sha256_prefix="));
assert!(!metadata.contains(URI_CANARY));
let long_uri = "x".repeat(RESOURCE_LOG_HASH_INPUT_LIMIT + 1);
let bounded_metadata = format!("{}", safe_resource_log_label(&long_uri));
assert!(bounded_metadata.contains(&format!(
"hashed_prefix_bytes={RESOURCE_LOG_HASH_INPUT_LIMIT}"
)));
assert!(!bounded_metadata.contains(&long_uri));
}
#[test]
fn notify_resource_updated_catches_sender_panic_and_returns_false() {
let mut session = make_session();
admit_resource(&mut session, "test://subscribed");
let sender: NotificationSender = Arc::new(|_| {
panic!("notification-sender-panic-payload-canary-2d941");
});
assert!(!session.notify_resource_updated("test://subscribed", &sender));
assert!(session.is_resource_subscribed("test://subscribed"));
}
#[test]
fn session_debug_format_includes_fields() {
let session = make_session();
let debug = format!("{:?}", session);
assert!(debug.contains("Session"));
assert!(debug.contains("initialized: false"));
}
#[test]
fn session_debug_reports_only_safe_counts_and_booleans() {
const SERVER_NAME_CANARY: &str = "server-tenant-canary-85fd";
const SERVER_VERSION_CANARY: &str = "server-version-canary-12aa";
const CLIENT_NAME_CANARY: &str = "customer-pii-canary@example.invalid";
const CLIENT_VERSION_CANARY: &str = "client-version-canary-53bc";
const PROTOCOL_CANARY: &str = "protocol-canary-secret-71f2";
const URI_CANARY: &str = "private://subscription-uri-canary-2bd4";
const STATE_KEY_CANARY: &str = "fastmcp.auth.canary-key-59cf";
const STATE_VALUE_CANARY: &str = "Bearer auth-token-canary-b300";
let mut session = Session::new(
ServerInfo {
name: SERVER_NAME_CANARY.to_string(),
version: SERVER_VERSION_CANARY.to_string(),
},
ServerCapabilities::default(),
);
session.initialize(
ClientInfo {
name: CLIENT_NAME_CANARY.to_string(),
version: CLIENT_VERSION_CANARY.to_string(),
},
ClientCapabilities::default(),
PROTOCOL_CANARY.to_string(),
);
admit_resource(&mut session, URI_CANARY);
assert!(session.state().set(STATE_KEY_CANARY, STATE_VALUE_CANARY));
session.set_log_level(LogLevel::Debug);
let debug = format!("{session:?}");
assert!(debug.contains("Session"));
assert!(debug.contains("initialized: true"));
assert!(debug.contains("resource_subscription_count: 1"));
assert!(debug.contains("state_entry_count: 1"));
for canary in [
SERVER_NAME_CANARY,
SERVER_VERSION_CANARY,
CLIENT_NAME_CANARY,
CLIENT_VERSION_CANARY,
PROTOCOL_CANARY,
URI_CANARY,
STATE_KEY_CANARY,
STATE_VALUE_CANARY,
] {
assert!(
!debug.contains(canary),
"Session Debug leaked canary: {canary}"
);
}
}
#[test]
fn test_session_supports_sampling() {
let mut session = Session::new(
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
},
ServerCapabilities::default(),
);
assert!(!session.supports_sampling());
session.initialize(
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
},
ClientCapabilities {
sampling: Some(SamplingCapability {}),
elicitation: None,
roots: None,
},
"2024-11-05".to_string(),
);
assert!(session.supports_sampling());
assert!(!session.supports_elicitation());
assert!(!session.supports_roots());
}
#[test]
fn test_session_supports_elicitation() {
let mut session = Session::new(
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
},
ServerCapabilities::default(),
);
session.initialize(
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
},
ClientCapabilities {
sampling: None,
elicitation: Some(ElicitationCapability::form()),
roots: None,
},
"2024-11-05".to_string(),
);
assert!(!session.supports_sampling());
assert!(session.supports_elicitation());
assert!(!session.supports_roots());
}
#[test]
fn test_session_supports_roots() {
let mut session = Session::new(
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
},
ServerCapabilities::default(),
);
session.initialize(
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
},
ClientCapabilities {
sampling: None,
elicitation: None,
roots: Some(RootsCapability { list_changed: true }),
},
"2024-11-05".to_string(),
);
assert!(!session.supports_sampling());
assert!(!session.supports_elicitation());
assert!(session.supports_roots());
}
#[test]
fn test_session_supports_all_capabilities() {
let mut session = Session::new(
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
},
ServerCapabilities::default(),
);
session.initialize(
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
},
ClientCapabilities {
sampling: Some(SamplingCapability {}),
elicitation: Some(ElicitationCapability::both()),
roots: Some(RootsCapability {
list_changed: false,
}),
},
"2024-11-05".to_string(),
);
assert!(session.supports_sampling());
assert!(session.supports_elicitation());
assert!(session.supports_roots());
}
#[test]
fn test_session_no_capabilities() {
let mut session = Session::new(
ServerInfo {
name: "test".to_string(),
version: "1.0".to_string(),
},
ServerCapabilities::default(),
);
session.initialize(
ClientInfo {
name: "test-client".to_string(),
version: "1.0".to_string(),
},
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
assert!(!session.supports_sampling());
assert!(!session.supports_elicitation());
assert!(!session.supports_roots());
}
#[test]
fn reinitialize_overwrites_client_info() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
session.initialize(
ClientInfo {
name: "new-client".to_string(),
version: "2.0".to_string(),
},
ClientCapabilities {
sampling: Some(SamplingCapability {}),
elicitation: None,
roots: None,
},
"2025-03-26".to_string(),
);
assert!(session.is_initialized());
let info = session.client_info().unwrap();
assert_eq!(info.name, "new-client");
assert_eq!(info.version, "2.0");
assert_eq!(session.protocol_version(), Some("2025-03-26"));
assert!(session.supports_sampling());
}
#[test]
fn state_persists_after_initialization() {
let mut session = make_session();
session.state().set("key", "before_init");
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
let val: Option<String> = session.state().get("key");
assert_eq!(val.as_deref(), Some("before_init"));
}
#[test]
fn notify_resource_updated_after_unsubscribe_returns_false() {
let mut session = make_session();
admit_resource(&mut session, "r://x");
let sent = Arc::new(Mutex::new(Vec::new()));
let sent_clone = Arc::clone(&sent);
let sender: NotificationSender = Arc::new(move |req| {
sent_clone.lock().unwrap().push(req);
});
assert!(session.notify_resource_updated("r://x", &sender));
assert_eq!(sent.lock().unwrap().len(), 1);
remove_resource(&mut session, "r://x");
assert!(!session.notify_resource_updated("r://x", &sender));
assert_eq!(sent.lock().unwrap().len(), 1);
}
#[test]
fn resubscribe_after_unsubscribe_works() {
let mut session = make_session();
admit_resource(&mut session, "r://x");
remove_resource(&mut session, "r://x");
assert!(!session.is_resource_subscribed("r://x"));
admit_resource(&mut session, "r://x");
assert!(session.is_resource_subscribed("r://x"));
}
#[test]
fn session_debug_after_init_shows_initialized_true() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
let debug = format!("{:?}", session);
assert!(debug.contains("initialized: true"));
}
#[test]
fn session_with_custom_server_capabilities() {
use fastmcp_protocol::{LoggingCapability, TasksCapability, ToolsCapability};
let caps = ServerCapabilities {
tools: Some(ToolsCapability { list_changed: true }),
logging: Some(LoggingCapability {}),
tasks: Some(TasksCapability {
list_changed: false,
}),
..ServerCapabilities::default()
};
let session = Session::new(make_server_info(), caps);
assert!(session.server_capabilities().tools.is_some());
assert!(session.server_capabilities().logging.is_some());
assert!(session.server_capabilities().tasks.is_some());
}
#[test]
fn set_log_level_all_variants() {
let mut session = make_session();
for level in [
LogLevel::Debug,
LogLevel::Info,
LogLevel::Warning,
LogLevel::Error,
] {
session.set_log_level(level);
assert_eq!(session.log_level(), Some(level));
}
}
#[test]
fn log_level_persists_across_reinitialization() {
let mut session = make_session();
session.set_log_level(LogLevel::Warning);
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
assert_eq!(session.log_level(), Some(LogLevel::Warning));
session.initialize(
ClientInfo {
name: "other".to_string(),
version: "2.0".to_string(),
},
ClientCapabilities::default(),
"2025-03-26".to_string(),
);
assert_eq!(session.log_level(), Some(LogLevel::Warning));
}
#[test]
fn resource_subscriptions_persist_across_reinitialization() {
let mut session = make_session();
admit_resource(&mut session, "file:///keep.txt");
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
assert!(session.is_resource_subscribed("file:///keep.txt"));
}
#[test]
fn state_set_after_init_persists_through_reinit() {
let mut session = make_session();
session.initialize(
make_client_info(),
ClientCapabilities::default(),
"2024-11-05".to_string(),
);
session.state().set("counter", 42);
session.initialize(
ClientInfo {
name: "new".to_string(),
version: "3.0".to_string(),
},
ClientCapabilities::default(),
"2025-03-26".to_string(),
);
let val: Option<i32> = session.state().get("counter");
assert_eq!(val, Some(42));
}
#[test]
fn notify_resource_updated_fires_independently_per_subscription() {
let mut session = make_session();
admit_resource(&mut session, "a://1");
admit_resource(&mut session, "b://2");
let sent = Arc::new(Mutex::new(Vec::new()));
let sent_clone = Arc::clone(&sent);
let sender: NotificationSender = Arc::new(move |req| {
sent_clone.lock().unwrap().push(req);
});
assert!(session.notify_resource_updated("a://1", &sender));
assert_eq!(sent.lock().unwrap().len(), 1);
let uri = sent.lock().unwrap()[0]
.params
.as_ref()
.unwrap()
.get("uri")
.unwrap()
.as_str()
.unwrap()
.to_string();
assert_eq!(uri, "a://1");
assert!(session.notify_resource_updated("b://2", &sender));
assert_eq!(sent.lock().unwrap().len(), 2);
let uri2 = sent.lock().unwrap()[1]
.params
.as_ref()
.unwrap()
.get("uri")
.unwrap()
.as_str()
.unwrap()
.to_string();
assert_eq!(uri2, "b://2");
}
#[test]
fn supports_elicitation_and_roots_false_before_init() {
let session = make_session();
assert!(!session.supports_elicitation());
assert!(!session.supports_roots());
}
#[test]
fn principal_binding_is_write_once_and_shared_across_clones() {
let session = make_session();
let binding = session.principal_binding();
let clone = binding.clone();
let alice = Sha256Digest::from_bytes([0x11; 32]);
let bob = Sha256Digest::from_bytes([0x22; 32]);
assert!(!binding.is_bound_for_debug());
assert!(!binding.verify_existing(alice));
assert!(binding.bind_or_verify(alice));
assert!(binding.verify_existing(alice));
assert!(!binding.verify_existing(bob));
assert!(clone.bind_or_verify(alice));
assert!(!clone.bind_or_verify(bob));
assert!(binding.bind_or_verify(alice));
}
#[test]
fn session_debug_reports_only_principal_presence() {
let session = make_session();
let canary = Sha256Digest::from_bytes([0xAB; 32]);
assert!(session.principal_binding().bind_or_verify(canary));
let debug = format!("{session:?}");
assert!(debug.contains("principal_bound: true"));
assert!(!debug.contains("abab"));
}
}