fix: address 2025-11-25 conformance audit findings (#951)
* fix: interpret task ttl as milliseconds * fix: use text/plain for default text mime type * fix: include resource param in token refresh * test: align conformance prompt args with runner * ci: run server conformance suite on PRs * ci: build client bin and gate pending scenarios
This commit is contained in:
parent
8e44af499b
commit
bdf0c32e8c
5 changed files with 169 additions and 19 deletions
|
|
@ -297,7 +297,7 @@ impl ContentBlock {
|
|||
ContentBlock::Resource(EmbeddedResource::new(
|
||||
ResourceContents::TextResourceContents {
|
||||
uri: uri.into(),
|
||||
mime_type: Some("text".to_string()),
|
||||
mime_type: Some("text/plain".to_string()),
|
||||
text: content.into(),
|
||||
meta: None,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -193,7 +193,7 @@ impl ResourceContents {
|
|||
pub fn text(text: impl Into<String>, uri: impl Into<String>) -> Self {
|
||||
Self::TextResourceContents {
|
||||
uri: uri.into(),
|
||||
mime_type: Some("text".into()),
|
||||
mime_type: Some("text/plain".into()),
|
||||
text: text.into(),
|
||||
meta: None,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ impl OperationDescriptor {
|
|||
self
|
||||
}
|
||||
|
||||
/// Time-to-live in milliseconds, matching `TaskMetadata.ttl` from the MCP spec.
|
||||
pub fn with_ttl(mut self, ttl: u64) -> Self {
|
||||
self.ttl = Some(ttl);
|
||||
self
|
||||
|
|
@ -75,7 +76,11 @@ pub trait OperationResultTransport: Send + Sync + 'static {
|
|||
}
|
||||
|
||||
// ===== Operation Processor =====
|
||||
pub const DEFAULT_TASK_TIMEOUT_SECS: u64 = 300; // 5 minutes
|
||||
#[deprecated(note = "use DEFAULT_TASK_TIMEOUT_MS; ttl values are milliseconds per the MCP spec")]
|
||||
pub const DEFAULT_TASK_TIMEOUT_SECS: u64 = 300;
|
||||
/// Default execution timeout (5 minutes), in milliseconds, applied when a
|
||||
/// descriptor does not specify a `ttl`.
|
||||
pub const DEFAULT_TASK_TIMEOUT_MS: u64 = 300_000;
|
||||
/// Operation processor that coordinates extractors and handlers
|
||||
pub struct OperationProcessor {
|
||||
/// Currently running tasks keyed by id
|
||||
|
|
@ -165,13 +170,13 @@ impl OperationProcessor {
|
|||
fn spawn_async_task(&mut self, message: OperationMessage) {
|
||||
let OperationMessage { descriptor, future } = message;
|
||||
let task_id = descriptor.operation_id.clone();
|
||||
let timeout_secs = descriptor.ttl.or(Some(DEFAULT_TASK_TIMEOUT_SECS));
|
||||
let timeout_ms = descriptor.ttl.or(Some(DEFAULT_TASK_TIMEOUT_MS));
|
||||
let sender = self.task_result_sender.clone();
|
||||
let descriptor_for_result = descriptor.clone();
|
||||
|
||||
let timed_future = async move {
|
||||
if let Some(secs) = timeout_secs {
|
||||
match timeout(Duration::from_secs(secs), future).await {
|
||||
if let Some(ms) = timeout_ms {
|
||||
match timeout(Duration::from_millis(ms), future).await {
|
||||
Ok(result) => result,
|
||||
Err(_) => Err(Error::TaskError("Operation timed out".to_string())),
|
||||
}
|
||||
|
|
@ -191,7 +196,7 @@ impl OperationProcessor {
|
|||
let running_task = RunningTask {
|
||||
task_handle: handle,
|
||||
started_at: std::time::Instant::now(),
|
||||
timeout: timeout_secs,
|
||||
timeout: timeout_ms,
|
||||
descriptor,
|
||||
};
|
||||
self.running_tasks.insert(task_id, running_task);
|
||||
|
|
@ -213,7 +218,7 @@ impl OperationProcessor {
|
|||
|
||||
for (task_id, task) in &self.running_tasks {
|
||||
if let Some(timeout_duration) = task.timeout {
|
||||
if now.duration_since(task.started_at).as_secs() > timeout_duration {
|
||||
if now.duration_since(task.started_at).as_millis() > u128::from(timeout_duration) {
|
||||
task.task_handle.abort();
|
||||
timed_out_tasks.push(task_id.clone());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1687,7 +1687,10 @@ impl AuthorizationManager {
|
|||
debug!("refresh token present, attempting refresh");
|
||||
|
||||
let refresh_token_value = RefreshToken::new(refresh_token.secret().to_string());
|
||||
let mut refresh_request = oauth_client.exchange_refresh_token(&refresh_token_value);
|
||||
let mut refresh_request = oauth_client
|
||||
.exchange_refresh_token(&refresh_token_value)
|
||||
// RFC 8707: the resource indicator is required on token requests, including refreshes
|
||||
.add_extra_param("resource", self.base_url.to_string());
|
||||
let mut refresh_scopes = stored_credentials.granted_scopes;
|
||||
self.add_offline_access_if_supported(&mut refresh_scopes);
|
||||
for scope in refresh_scopes {
|
||||
|
|
@ -5489,6 +5492,42 @@ mod tests {
|
|||
assert_eq!(scope_parts, vec!["read", "write"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_token_includes_resource_parameter() {
|
||||
let (base_url, captured) = start_token_server().await;
|
||||
|
||||
let mut manager = manager_with_metadata(Some(AuthorizationMetadata {
|
||||
authorization_endpoint: format!("{}/authorize", base_url),
|
||||
token_endpoint: format!("{}/token", base_url),
|
||||
..Default::default()
|
||||
}))
|
||||
.await;
|
||||
manager.configure_client(test_client_config()).unwrap();
|
||||
|
||||
let stored = StoredCredentials {
|
||||
client_id: "my-client".to_string(),
|
||||
token_response: Some(make_token_response_with_refresh(
|
||||
"old-token",
|
||||
"my-refresh-token",
|
||||
)),
|
||||
granted_scopes: vec![],
|
||||
token_received_at: Some(AuthorizationManager::now_epoch_secs()),
|
||||
};
|
||||
manager.credential_store.save(stored).await.unwrap();
|
||||
|
||||
manager.refresh_token().await.unwrap();
|
||||
|
||||
let body = captured.lock().unwrap().take().unwrap();
|
||||
let params: std::collections::HashMap<_, _> = url::form_urlencoded::parse(body.as_bytes())
|
||||
.into_owned()
|
||||
.collect();
|
||||
assert_eq!(
|
||||
params.get("resource").map(String::as_str),
|
||||
Some("http://localhost/"),
|
||||
"refresh requests must carry the RFC 8707 resource parameter, got body: {body}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_token_adds_offline_access_when_as_supports_it() {
|
||||
let (base_url, captured) = start_token_server().await;
|
||||
|
|
|
|||
|
|
@ -80,6 +80,38 @@ async fn rejects_duplicate_operation_ids() {
|
|||
assert!(format!("{err}").contains("already running"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ttl_is_interpreted_as_milliseconds() {
|
||||
let mut processor = OperationProcessor::new();
|
||||
let descriptor = OperationDescriptor::new("slow", "dummy").with_ttl(50);
|
||||
let future = Box::pin(async {
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
Ok(Box::new(DummyTransport {
|
||||
id: "slow".to_string(),
|
||||
value: 0,
|
||||
}) as Box<dyn OperationResultTransport>)
|
||||
});
|
||||
|
||||
processor
|
||||
.submit_operation(OperationMessage::new(descriptor, future))
|
||||
.expect("submit operation");
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let results = processor.peek_completed();
|
||||
assert_eq!(
|
||||
results.len(),
|
||||
1,
|
||||
"50ms ttl should have timed out the operation well within 200ms"
|
||||
);
|
||||
match &results[0].result {
|
||||
Err(err) => assert!(
|
||||
err.to_string().contains("timed out"),
|
||||
"unexpected error: {err}"
|
||||
),
|
||||
Ok(_) => panic!("expected the operation to time out, but it completed"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_status_notification_param_preserves_meta() {
|
||||
let raw = json!({
|
||||
|
|
|
|||
Loading…
Reference in a new issue