diff --git a/.env.example b/.env.example index 3d4a62f..8a04937 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,11 @@ CRANK_RUNTIME_MAX_CONCURRENT_UNARY=64 CRANK_RUNTIME_MAX_CONCURRENT_WINDOW=16 CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS=16 CRANK_RUNTIME_MAX_CONCURRENT_JOBS=16 +# Публичные узлы разрешены по умолчанию. Для внутренних API перечислите +# допустимые имена или IP через запятую. +CRANK_OUTBOUND_ALLOWED_HOSTS= +CRANK_OUTBOUND_DENIED_HOSTS= +CRANK_OUTBOUND_MAX_RESPONSE_BYTES=4194304 CRANK_LOG_LEVEL=info CRANK_MASTER_KEY=change-me-master-key CRANK_SESSION_SECRET=change-me-session-secret diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 649c0c5..7cdd053 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -302,6 +302,9 @@ jobs: append_if_set CRANK_ADMIN_BIND "$CRANK_ADMIN_BIND" append_if_set CRANK_MCP_BIND "$CRANK_MCP_BIND" append_if_set CRANK_MCP_REFRESH_MS "$CRANK_MCP_REFRESH_MS" + append_if_set CRANK_OUTBOUND_ALLOWED_HOSTS "${CRANK_OUTBOUND_ALLOWED_HOSTS:-}" + append_if_set CRANK_OUTBOUND_DENIED_HOSTS "${CRANK_OUTBOUND_DENIED_HOSTS:-}" + append_if_set CRANK_OUTBOUND_MAX_RESPONSE_BYTES "${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-}" append_if_set CRANK_LOG_LEVEL "$CRANK_LOG_LEVEL" append_if_set CRANK_MASTER_KEY "$CRANK_MASTER_KEY" append_if_set CRANK_BASE_URL "$CRANK_BASE_URL" diff --git a/Cargo.lock b/Cargo.lock index 99f3c80..46f1a0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -677,6 +677,7 @@ dependencies = [ "crank-test-support", "serde", "serde_json", + "sha2 0.10.9", "sqlx", "thiserror", "time", diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index 0d9e860..37213fd 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -10,7 +10,7 @@ use crank_community_auth::PasswordIdentityProvider; use crank_registry::{PostgresPoolConfig, PostgresRegistry}; use crank_runtime::{ RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores, - RuntimeLimits, SecretCrypto, community_default, + RuntimeLimits, SecretCrypto, }; use sqlx::postgres::PgConnectOptions; use tokio::net::TcpListener; @@ -56,7 +56,8 @@ async fn main() -> Result<(), Box> { let cache_stores = RuntimeCacheStores::from_config(&cache_config).await?; let api_rate_limit = admin_api_rate_limit_config_from_env()?; let secret_crypto = SecretCrypto::new(&env::var("CRANK_MASTER_KEY")?)?; - let runtime = community_default() + let outbound_http_policy = crank_runtime::OutboundHttpPolicy::from_env()?; + let runtime = crank_runtime::community_with_outbound_policy(outbound_http_policy.clone()) .with_limits(runtime_limits) .with_response_cache(cache_stores.response.clone()) .with_coordination_store(cache_stores.coordination.clone()) @@ -70,6 +71,7 @@ async fn main() -> Result<(), Box> { secret_crypto, runtime, ) + .with_outbound_http_policy(outbound_http_policy) .with_identity_provider(std::sync::Arc::new(identity_provider)) .build(); service.bootstrap_admin_user().await?; diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index df47a1e..466786e 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -15,7 +15,9 @@ use crank_registry::{ AgentSummary, CreateInvocationLogRequest, OperationAgentRef, OperationSummary, OperationUsageSummary, PostgresRegistry, RegistryOperation, UsageBucket, }; -use crank_runtime::{PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, SecretCrypto}; +use crank_runtime::{ + OutboundHttpPolicy, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, SecretCrypto, +}; use crank_schema::{Schema, SchemaKind}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; @@ -38,8 +40,8 @@ mod workspaces; use crate::{auth::AuthSettings, error::ApiError, storage::LocalArtifactStorage}; use operation_validation::{ - validate_approval_policy, validate_idempotency_policy, validate_protocol_target, - validate_response_cache_policy, + validate_approval_policy, validate_execution_timeout, validate_idempotency_policy, + validate_protocol_target, validate_response_cache_policy, }; #[derive(Clone)] @@ -53,6 +55,7 @@ pub struct AdminService { policy_engine: Arc, audit_sink: Arc, capability_profile: Arc, + outbound_http_policy: OutboundHttpPolicy, } pub struct AdminServiceBuilder { @@ -65,6 +68,7 @@ pub struct AdminServiceBuilder { policy_engine: Option>, audit_sink: Option>, capability_profile: Option>, + outbound_http_policy: OutboundHttpPolicy, } pub use crate::dto::*; @@ -139,6 +143,7 @@ impl AdminServiceBuilder { policy_engine: None, audit_sink: None, capability_profile: None, + outbound_http_policy: OutboundHttpPolicy::default(), } } @@ -147,6 +152,11 @@ impl AdminServiceBuilder { self } + pub fn with_outbound_http_policy(mut self, policy: OutboundHttpPolicy) -> Self { + self.outbound_http_policy = policy; + self + } + #[allow(dead_code)] pub fn with_policy_engine(mut self, policy_engine: Arc) -> Self { self.policy_engine = Some(policy_engine); @@ -183,6 +193,7 @@ impl AdminServiceBuilder { capability_profile: self .capability_profile .unwrap_or_else(|| Arc::new(CommunityCapabilityProfile)), + outbound_http_policy: self.outbound_http_policy, } } } @@ -297,6 +308,8 @@ impl AdminService { fn validate_operation_payload(&self, payload: &OperationPayload) -> Result<(), ApiError> { self.validate_operation_capabilities(payload.protocol, payload.security_level)?; validate_protocol_target(payload.protocol, &payload.target)?; + self.validate_outbound_target(&payload.target)?; + validate_execution_timeout(&payload.execution_config)?; validate_response_cache_policy(&payload.target, &payload.execution_config)?; validate_idempotency_policy(&payload.target, &payload.execution_config)?; validate_approval_policy(&payload.execution_config)?; @@ -308,6 +321,8 @@ impl AdminService { fn validate_registry_operation(&self, operation: &RegistryOperation) -> Result<(), ApiError> { self.validate_operation_capabilities(operation.protocol, operation.security_level)?; validate_protocol_target(operation.protocol, &operation.target)?; + self.validate_outbound_target(&operation.target)?; + validate_execution_timeout(&operation.execution_config)?; validate_response_cache_policy(&operation.target, &operation.execution_config)?; validate_idempotency_policy(&operation.target, &operation.execution_config)?; validate_approval_policy(&operation.execution_config)?; @@ -316,6 +331,15 @@ impl AdminService { Ok(()) } + fn validate_outbound_target(&self, target: &crank_core::Target) -> Result<(), ApiError> { + match target { + crank_core::Target::Rest(rest) => self + .outbound_http_policy + .validate_base_url(&rest.base_url) + .map_err(|error| ApiError::validation(error.to_string())), + } + } + fn validate_operation_capabilities( &self, protocol: Protocol, diff --git a/apps/admin-api/src/service/operation_validation.rs b/apps/admin-api/src/service/operation_validation.rs index a35b8a2..8f7de41 100644 --- a/apps/admin-api/src/service/operation_validation.rs +++ b/apps/admin-api/src/service/operation_validation.rs @@ -3,6 +3,8 @@ use serde_json::json; use crate::error::ApiError; +const MAX_OPERATION_TIMEOUT_MS: u64 = 300_000; + pub(super) fn validate_protocol_target( protocol: Protocol, target: &Target, @@ -16,6 +18,18 @@ pub(super) fn validate_protocol_target( Err(ApiError::validation("protocol and target kind must match")) } +pub(super) fn validate_execution_timeout( + execution_config: &crank_core::ExecutionConfig, +) -> Result<(), ApiError> { + if !(1..=MAX_OPERATION_TIMEOUT_MS).contains(&execution_config.timeout_ms) { + return Err(ApiError::validation_with_context( + format!("operation timeout must be between 1 and {MAX_OPERATION_TIMEOUT_MS} ms"), + json!({ "field": "execution_config.timeout_ms" }), + )); + } + Ok(()) +} + pub(super) fn validate_response_cache_policy( target: &Target, execution_config: &crank_core::ExecutionConfig, @@ -150,7 +164,8 @@ mod tests { }; use super::{ - validate_approval_policy, validate_idempotency_policy, validate_response_cache_policy, + validate_approval_policy, validate_execution_timeout, validate_idempotency_policy, + validate_response_cache_policy, }; fn cacheable_execution_config() -> ExecutionConfig { @@ -198,6 +213,19 @@ mod tests { assert!(result.is_ok()); } + #[test] + fn rejects_zero_and_excessive_execution_timeouts() { + let mut config = cacheable_execution_config(); + config.timeout_ms = 0; + assert!(validate_execution_timeout(&config).is_err()); + + config.timeout_ms = 300_001; + assert!(validate_execution_timeout(&config).is_err()); + + config.timeout_ms = 300_000; + assert!(validate_execution_timeout(&config).is_ok()); + } + #[test] fn rejects_response_cache_for_non_get_rest_operation() { let target = Target::Rest(RestTarget { diff --git a/apps/admin-api/tests/integration/common.rs b/apps/admin-api/tests/integration/common.rs index 4d1df4a..52a258e 100644 --- a/apps/admin-api/tests/integration/common.rs +++ b/apps/admin-api/tests/integration/common.rs @@ -114,13 +114,16 @@ pub(super) fn test_service( auth_settings: AuthSettings, secret_crypto: SecretCrypto, ) -> AdminService { + let outbound_policy = crank_runtime::OutboundHttpPolicy::allowing_hosts(["127.0.0.1"]); + let runtime = crank_runtime::community_with_outbound_policy(outbound_policy.clone()).build(); AdminServiceBuilder::new( registry, storage_root, auth_settings, secret_crypto, - crank_runtime::RuntimeExecutor::new(), + runtime, ) + .with_outbound_http_policy(outbound_policy) .build() } diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index c46951e..bcd797b 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -1,12 +1,13 @@ use std::{env, net::SocketAddr, time::Duration}; use crank_community_mcp::{ - auth::CommunityMachineCredentialVerifier, build_app, session::PostgresTransportSessionStore, + auth::CommunityMachineCredentialVerifier, build_app_with_background_workers, + session::PostgresTransportSessionStore, }; use crank_registry::{PostgresPoolConfig, PostgresRegistry}; use crank_runtime::{ RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores, - RuntimeLimits, SecretCrypto, community_default, + RuntimeLimits, SecretCrypto, }; use sqlx::postgres::PgConnectOptions; use tokio::net::TcpListener; @@ -46,12 +47,12 @@ async fn main() -> Result<(), Box> { ) .await?; let secret_crypto = SecretCrypto::new(&env::var("CRANK_MASTER_KEY")?)?; - let runtime = community_default() + let runtime = crank_runtime::community_from_env()? .with_limits(runtime_limits) .with_response_cache(cache_stores.response.clone()) .with_coordination_store(cache_stores.coordination.clone()) .build(); - let app = build_app( + let app = build_app_with_background_workers( registry, refresh_interval, base_url, diff --git a/apps/mcp-server/tests/integration/catalog_access.rs b/apps/mcp-server/tests/integration/catalog_access.rs index 63994ab..bee481a 100644 --- a/apps/mcp-server/tests/integration/catalog_access.rs +++ b/apps/mcp-server/tests/integration/catalog_access.rs @@ -142,7 +142,10 @@ fn build_test_app_with_store( refresh_interval, public_base_url, SecretCrypto::new("test-master-key").unwrap(), - RuntimeExecutor::new(), + crank_runtime::community_with_outbound_policy( + crank_runtime::OutboundHttpPolicy::allowing_hosts(["127.0.0.1"]), + ) + .build(), RequestRateLimiter::new(rate_limit_config), std::sync::Arc::new(InMemoryCoordinationStateStore::default()), sessions, diff --git a/apps/mcp-server/tests/integration/common.rs b/apps/mcp-server/tests/integration/common.rs index 193dd32..c1dd2fd 100644 --- a/apps/mcp-server/tests/integration/common.rs +++ b/apps/mcp-server/tests/integration/common.rs @@ -135,7 +135,10 @@ fn build_test_app_with_store( refresh_interval, public_base_url, SecretCrypto::new("test-master-key").unwrap(), - RuntimeExecutor::new(), + crank_runtime::community_with_outbound_policy( + crank_runtime::OutboundHttpPolicy::allowing_hosts(["127.0.0.1"]), + ) + .build(), RequestRateLimiter::new(rate_limit_config), std::sync::Arc::new(InMemoryCoordinationStateStore::default()), sessions, diff --git a/apps/mcp-server/tests/integration/transport_protocol.rs b/apps/mcp-server/tests/integration/transport_protocol.rs index 466f5fb..8004e3e 100644 --- a/apps/mcp-server/tests/integration/transport_protocol.rs +++ b/apps/mcp-server/tests/integration/transport_protocol.rs @@ -136,7 +136,10 @@ fn build_test_app_with_store( refresh_interval, public_base_url, SecretCrypto::new("test-master-key").unwrap(), - RuntimeExecutor::new(), + crank_runtime::community_with_outbound_policy( + crank_runtime::OutboundHttpPolicy::allowing_hosts(["127.0.0.1"]), + ) + .build(), RequestRateLimiter::new(rate_limit_config), std::sync::Arc::new(InMemoryCoordinationStateStore::default()), sessions, diff --git a/crates/crank-adapter-rest/src/client.rs b/crates/crank-adapter-rest/src/client.rs index 7af701e..f9b0d9f 100644 --- a/crates/crank-adapter-rest/src/client.rs +++ b/crates/crank-adapter-rest/src/client.rs @@ -1,9 +1,18 @@ -use std::{collections::BTreeMap, time::Duration}; +use std::{ + collections::BTreeMap, + env, io, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + sync::Arc, + time::Duration, +}; use crank_core::{HttpMethod, RestTarget}; +use futures_util::StreamExt; use reqwest::{ Client, + dns::{Addrs, Name, Resolve, Resolving}, header::{HeaderMap, HeaderName, HeaderValue}, + redirect, }; use serde_json::Value; @@ -11,9 +20,19 @@ use crate::{RestAdapterError, RestRequest, RestResponse}; #[derive(Clone, Debug)] pub struct RestAdapter { - client: Client, + client: Result>, + policy: OutboundHttpPolicy, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct OutboundHttpPolicy { + allowed_hosts: Vec, + denied_hosts: Vec, + max_response_bytes: usize, +} + +const DEFAULT_MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024; + impl Default for RestAdapter { fn default() -> Self { Self::new() @@ -22,9 +41,25 @@ impl Default for RestAdapter { impl RestAdapter { pub fn new() -> Self { - Self { - client: Client::new(), - } + Self::with_policy(OutboundHttpPolicy::default()) + } + + pub fn from_env() -> Result { + Ok(Self::with_policy(OutboundHttpPolicy::from_env()?)) + } + + pub fn with_policy(policy: OutboundHttpPolicy) -> Self { + let resolver = Arc::new(PolicyDnsResolver { + policy: policy.clone(), + }); + let client = Client::builder() + .redirect(redirect::Policy::none()) + .no_proxy() + .dns_resolver(resolver) + .build() + .map_err(|error| Arc::::from(error.to_string())); + + Self { client, policy } } pub async fn execute( @@ -33,9 +68,15 @@ impl RestAdapter { request: &RestRequest, ) -> Result { let url = build_url(target, request)?; + self.policy.validate_url(&url)?; let headers = build_headers(target, request)?; - let mut builder = self - .client + let client = + self.client + .as_ref() + .map_err(|details| RestAdapterError::InvalidConfiguration { + details: details.to_string(), + })?; + let mut builder = client .request(to_reqwest_method(target.method), url) .headers(headers) .timeout(Duration::from_millis(request.timeout_ms)); @@ -47,7 +88,7 @@ impl RestAdapter { let response = builder.send().await?; let status = response.status(); let headers = normalize_headers(response.headers()); - let body = decode_body(response).await?; + let body = decode_body(response, self.policy.max_response_bytes).await?; if !status.is_success() { return Err(RestAdapterError::UnexpectedStatus { @@ -64,6 +105,254 @@ impl RestAdapter { } } +impl Default for OutboundHttpPolicy { + fn default() -> Self { + Self { + allowed_hosts: Vec::new(), + denied_hosts: Vec::new(), + max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES, + } + } +} + +impl OutboundHttpPolicy { + pub fn from_env() -> Result { + let max_response_bytes = match env::var("CRANK_OUTBOUND_MAX_RESPONSE_BYTES") { + Ok(value) => { + value + .parse::() + .map_err(|_| RestAdapterError::InvalidConfiguration { + details: "CRANK_OUTBOUND_MAX_RESPONSE_BYTES must be a positive integer" + .to_owned(), + })? + } + Err(env::VarError::NotPresent) => DEFAULT_MAX_RESPONSE_BYTES, + Err(error) => { + return Err(RestAdapterError::InvalidConfiguration { + details: error.to_string(), + }); + } + }; + if max_response_bytes == 0 { + return Err(RestAdapterError::InvalidConfiguration { + details: "CRANK_OUTBOUND_MAX_RESPONSE_BYTES must be greater than zero".to_owned(), + }); + } + + Ok(Self { + allowed_hosts: host_patterns_from_env("CRANK_OUTBOUND_ALLOWED_HOSTS")?, + denied_hosts: host_patterns_from_env("CRANK_OUTBOUND_DENIED_HOSTS")?, + max_response_bytes, + }) + } + + pub fn allowing_hosts(hosts: impl IntoIterator>) -> Self { + Self { + allowed_hosts: hosts.into_iter().map(Into::into).collect(), + ..Self::default() + } + } + + pub fn with_max_response_bytes(mut self, max_response_bytes: usize) -> Self { + self.max_response_bytes = max_response_bytes; + self + } + + pub fn validate_base_url(&self, base_url: &str) -> Result<(), RestAdapterError> { + let url = reqwest::Url::parse(base_url).map_err(|_| RestAdapterError::InvalidBaseUrl { + url: base_url.to_owned(), + })?; + self.validate_url(&url) + } + + fn validate_url(&self, url: &reqwest::Url) -> Result<(), RestAdapterError> { + if !matches!(url.scheme(), "http" | "https") + || !url.username().is_empty() + || url.password().is_some() + { + return Err(RestAdapterError::TargetNotAllowed { + target: url.to_string(), + }); + } + let host = url + .host_str() + .ok_or_else(|| RestAdapterError::TargetNotAllowed { + target: url.to_string(), + })?; + self.validate_host(host)?; + if !self.is_explicitly_allowed(host) && is_local_hostname(host) { + return Err(RestAdapterError::TargetNotAllowed { + target: host.to_owned(), + }); + } + if let Ok(address) = host.parse::() + && !self.is_explicitly_allowed(host) + && !is_public_ip(address) + { + return Err(RestAdapterError::TargetNotAllowed { + target: host.to_owned(), + }); + } + Ok(()) + } + + fn validate_host(&self, host: &str) -> Result<(), RestAdapterError> { + let host = normalize_host(host); + let denied = self + .denied_hosts + .iter() + .any(|pattern| host_matches(pattern, &host)); + if denied { + return Err(RestAdapterError::TargetNotAllowed { target: host }); + } + Ok(()) + } + + fn is_explicitly_allowed(&self, host: &str) -> bool { + let host = normalize_host(host); + self.allowed_hosts + .iter() + .any(|pattern| host_matches(pattern, &host)) + } +} + +#[derive(Clone, Debug)] +struct PolicyDnsResolver { + policy: OutboundHttpPolicy, +} + +impl Resolve for PolicyDnsResolver { + fn resolve(&self, name: Name) -> Resolving { + let host = normalize_host(name.as_str()); + let policy = self.policy.clone(); + Box::pin(async move { + policy + .validate_host(&host) + .map_err(|error| boxed_io_error(error.to_string()))?; + let explicitly_allowed = policy.is_explicitly_allowed(&host); + let resolved = tokio::net::lookup_host((host.as_str(), 0)) + .await + .map_err(|error| Box::new(error) as Box)?; + let addresses = resolved + .filter(|address| explicitly_allowed || is_public_ip(address.ip())) + .collect::>(); + if addresses.is_empty() { + return Err(boxed_io_error(format!( + "outbound target {host} did not resolve to an allowed address" + ))); + } + Ok(Box::new(addresses.into_iter()) as Addrs) + }) + } +} + +fn boxed_io_error(message: String) -> Box { + Box::new(io::Error::new(io::ErrorKind::PermissionDenied, message)) +} + +fn host_patterns_from_env(name: &str) -> Result, RestAdapterError> { + let value = match env::var(name) { + Ok(value) => value, + Err(env::VarError::NotPresent) => return Ok(Vec::new()), + Err(error) => { + return Err(RestAdapterError::InvalidConfiguration { + details: error.to_string(), + }); + } + }; + value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(|value| { + let wildcard = value.starts_with("*."); + let normalized = normalize_host(value.trim_start_matches("*.")); + let valid_ip = !wildcard && normalized.parse::().is_ok(); + if normalized.is_empty() + || normalized.contains('/') + || (!valid_ip && normalized.contains(':')) + || (wildcard && normalized.parse::().is_ok()) + { + return Err(RestAdapterError::InvalidConfiguration { + details: format!("{name} contains an invalid host pattern: {value}"), + }); + } + Ok(if wildcard { + format!("*.{normalized}") + } else { + normalized + }) + }) + .collect() +} + +fn normalize_host(host: &str) -> String { + host.trim() + .trim_start_matches('[') + .trim_end_matches(']') + .trim_end_matches('.') + .to_ascii_lowercase() +} + +fn is_local_hostname(host: &str) -> bool { + let host = normalize_host(host); + host == "localhost" || host.ends_with(".localhost") +} + +fn host_matches(pattern: &str, host: &str) -> bool { + pattern.strip_prefix("*.").map_or_else( + || pattern == host, + |suffix| host != suffix && host.ends_with(&format!(".{suffix}")), + ) +} + +fn is_public_ip(address: IpAddr) -> bool { + match address { + IpAddr::V4(address) => is_public_ipv4(address), + IpAddr::V6(address) => is_public_ipv6(address), + } +} + +fn is_public_ipv4(address: Ipv4Addr) -> bool { + let octets = address.octets(); + !(address.is_private() + || address.is_loopback() + || address.is_link_local() + || address.is_broadcast() + || address.is_documentation() + || address.is_unspecified() + || address.is_multicast() + || octets[0] == 0 + || (octets[0] == 100 && (64..=127).contains(&octets[1])) + || (octets[0] == 192 && octets[1] == 0 && octets[2] == 0) + || (octets[0] == 198 && (18..=19).contains(&octets[1])) + || octets[0] >= 240) +} + +fn is_public_ipv6(address: Ipv6Addr) -> bool { + let segments = address.segments(); + if let Some(address) = address.to_ipv4_mapped() { + return is_public_ipv4(address); + } + if segments[..6].iter().all(|segment| *segment == 0) { + let [a, b] = segments[6].to_be_bytes(); + let [c, d] = segments[7].to_be_bytes(); + return is_public_ipv4(Ipv4Addr::new(a, b, c, d)); + } + !(address.is_unspecified() + || address.is_loopback() + || address.is_multicast() + || (segments[0] & 0xfe00) == 0xfc00 + || (segments[0] & 0xffc0) == 0xfe80 + || (segments[0] & 0xffc0) == 0xfec0 + || (segments[0] == 0x0064 + && segments[1] == 0xff9b + && segments[2..6].iter().all(|segment| *segment == 0)) + || (segments[0] == 0x0064 && segments[1] == 0xff9b && segments[2] == 1) + || segments[0] == 0x2002 + || (segments[0] == 0x2001 && matches!(segments[1], 0 | 0x0db8))) +} + fn build_url(target: &RestTarget, request: &RestRequest) -> Result { let base_url = reqwest::Url::parse(&target.base_url).map_err(|_| RestAdapterError::InvalidBaseUrl { @@ -127,8 +416,29 @@ fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> Result<(), Ok(()) } -async fn decode_body(response: reqwest::Response) -> Result { - let bytes = response.bytes().await?; +async fn decode_body( + response: reqwest::Response, + max_response_bytes: usize, +) -> Result { + if response + .content_length() + .is_some_and(|length| length > max_response_bytes as u64) + { + return Err(RestAdapterError::ResponseTooLarge { + limit_bytes: max_response_bytes, + }); + } + let mut stream = response.bytes_stream(); + let mut bytes = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + if bytes.len().saturating_add(chunk.len()) > max_response_bytes { + return Err(RestAdapterError::ResponseTooLarge { + limit_bytes: max_response_bytes, + }); + } + bytes.extend_from_slice(&chunk); + } if bytes.is_empty() { return Ok(Value::Null); diff --git a/crates/crank-adapter-rest/src/error.rs b/crates/crank-adapter-rest/src/error.rs index 64f732d..ae14b8e 100644 --- a/crates/crank-adapter-rest/src/error.rs +++ b/crates/crank-adapter-rest/src/error.rs @@ -13,6 +13,12 @@ pub enum RestAdapterError { InvalidHeaderName { header: String }, #[error("invalid header value for {header}")] InvalidHeaderValue { header: String }, + #[error("outbound target is not allowed: {target}")] + TargetNotAllowed { target: String }, + #[error("rest response exceeds the configured limit of {limit_bytes} bytes")] + ResponseTooLarge { limit_bytes: usize }, + #[error("invalid outbound HTTP configuration: {details}")] + InvalidConfiguration { details: String }, #[error("request failed")] Transport(#[from] reqwest::Error), #[error("sse collection window expired before stream completed")] diff --git a/crates/crank-adapter-rest/src/lib.rs b/crates/crank-adapter-rest/src/lib.rs index 22ca2df..0f9d5b1 100644 --- a/crates/crank-adapter-rest/src/lib.rs +++ b/crates/crank-adapter-rest/src/lib.rs @@ -8,7 +8,7 @@ use crank_core::{ ProtocolAdapterError, RestTarget, RuntimeRequestContext, Target, }; -pub use client::RestAdapter; +pub use client::{OutboundHttpPolicy, RestAdapter}; pub use error::RestAdapterError; pub use model::{RestRequest, RestResponse}; diff --git a/crates/crank-adapter-rest/tests/integration/client.rs b/crates/crank-adapter-rest/tests/integration/client.rs index 424c5ba..d15eee2 100644 --- a/crates/crank-adapter-rest/tests/integration/client.rs +++ b/crates/crank-adapter-rest/tests/integration/client.rs @@ -3,10 +3,11 @@ use std::collections::BTreeMap; use axum::{ Json, Router, extract::{Path, Query}, - http::HeaderMap, + http::{HeaderMap, StatusCode}, + response::Redirect, routing::{get, post}, }; -use crank_adapter_rest::{RestAdapter, RestAdapterError, RestRequest}; +use crank_adapter_rest::{OutboundHttpPolicy, RestAdapter, RestAdapterError, RestRequest}; use crank_core::{HttpMethod, RestTarget}; use serde_json::{Value, json}; use tokio::net::TcpListener; @@ -14,7 +15,7 @@ use tokio::net::TcpListener; #[tokio::test] async fn executes_rest_request_and_normalizes_json_response() { let base_url = spawn_test_server().await; - let adapter = RestAdapter::new(); + let adapter = test_adapter(); let target = RestTarget { base_url, method: HttpMethod::Post, @@ -47,7 +48,7 @@ async fn executes_rest_request_and_normalizes_json_response() { #[tokio::test] async fn returns_unexpected_status_with_normalized_body() { let base_url = spawn_test_server().await; - let adapter = RestAdapter::new(); + let adapter = test_adapter(); let target = RestTarget { base_url, method: HttpMethod::Get, @@ -73,10 +74,94 @@ async fn returns_unexpected_status_with_normalized_body() { )); } +#[test] +fn rejects_private_targets_by_default() { + let policy = OutboundHttpPolicy::default(); + + let error = policy + .validate_base_url("http://127.0.0.1:8080") + .unwrap_err(); + + assert!(matches!(error, RestAdapterError::TargetNotAllowed { .. })); +} + +#[test] +fn accepts_explicit_private_ipv4_and_ipv6_targets() { + let policy = OutboundHttpPolicy::allowing_hosts(["192.168.1.10", "::1"]); + + assert!(policy.validate_base_url("http://192.168.1.10:8080").is_ok()); + assert!(policy.validate_base_url("http://[::1]:8080").is_ok()); +} + +#[tokio::test] +async fn does_not_follow_redirects() { + let base_url = spawn_test_server().await; + let adapter = test_adapter(); + let target = RestTarget { + base_url, + method: HttpMethod::Get, + path_template: "/redirect".to_owned(), + static_headers: BTreeMap::new(), + }; + + let error = adapter + .execute(&target, &empty_request()) + .await + .unwrap_err(); + + assert!(matches!( + error, + RestAdapterError::UnexpectedStatus { status: 303, .. } + )); +} + +#[tokio::test] +async fn rejects_responses_over_the_configured_limit() { + let base_url = spawn_test_server().await; + let adapter = RestAdapter::with_policy( + OutboundHttpPolicy::allowing_hosts(["127.0.0.1"]).with_max_response_bytes(8), + ); + let target = RestTarget { + base_url, + method: HttpMethod::Get, + path_template: "/large".to_owned(), + static_headers: BTreeMap::new(), + }; + + let error = adapter + .execute(&target, &empty_request()) + .await + .unwrap_err(); + + assert!(matches!( + error, + RestAdapterError::ResponseTooLarge { limit_bytes: 8 } + )); +} + +fn empty_request() -> RestRequest { + RestRequest { + path_params: BTreeMap::new(), + query_params: BTreeMap::new(), + headers: BTreeMap::new(), + body: None, + timeout_ms: 1_000, + } +} + +fn test_adapter() -> RestAdapter { + RestAdapter::with_policy(OutboundHttpPolicy::allowing_hosts(["127.0.0.1"])) +} + async fn spawn_test_server() -> String { let app = Router::new() .route("/users/{user_id}", post(create_user)) - .route("/fail", get(fail)); + .route("/fail", get(fail)) + .route("/redirect", get(|| async { Redirect::to("/large") })) + .route( + "/large", + get(|| async { "response larger than eight bytes" }), + ); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); @@ -113,7 +198,7 @@ async fn create_user( async fn fail() -> (axum::http::StatusCode, Json) { ( - axum::http::StatusCode::BAD_GATEWAY, + StatusCode::BAD_GATEWAY, Json(json!({ "error": "upstream failed" })), ) } diff --git a/crates/crank-community-mcp/src/app.rs b/crates/crank-community-mcp/src/app.rs index 31cf7a3..ccf860e 100644 --- a/crates/crank-community-mcp/src/app.rs +++ b/crates/crank-community-mcp/src/app.rs @@ -18,9 +18,8 @@ use crank_core::{ OperationApprovalMode, PlatformApiKeyScope, SecretId, }; use crank_registry::{ - ApprovalRequestRecord, CreateApprovalRequest, CreateInvocationLogRequest, - DecideApprovalRequest, ExpireApprovalRequest, FinishApprovalRequest, PostgresRegistry, - PublishedAgentTool, + CreateApprovalRequest, CreateInvocationLogRequest, DecideApprovalRequest, + ExpireApprovalRequest, PostgresRegistry, PublishedAgentTool, }; use crank_runtime::{ RequestRateLimiter, ResolvedAuth, RuntimeError, RuntimeExecutionRequest, RuntimeExecutor, @@ -30,13 +29,14 @@ use futures_util::stream; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use time::OffsetDateTime; -use tracing::info; +use tracing::{info, warn}; use crate::{ access::{ credential_allows_security_level, require_approval_access, require_machine_access, serialize_machine_access_mode, serialize_security_level, }, + approval_execution::{execute_approved_request, spawn_approval_recovery}, auth::{SharedMachineCredentialVerifier, VerifiedMachineCredential}, catalog::PublishedToolCatalog, jsonrpc::{ @@ -66,8 +66,8 @@ const TRANSPORT_SESSION_TTL_MS: u64 = 86_400_000; #[derive(Clone)] pub(super) struct AppState { pub(super) registry: PostgresRegistry, - catalog: PublishedToolCatalog, - runtime: RuntimeExecutor, + pub(super) catalog: PublishedToolCatalog, + pub(super) runtime: RuntimeExecutor, pub(super) api_rate_limiter: RequestRateLimiter, secret_crypto: SecretCrypto, sessions: SharedSessionStore, @@ -132,6 +132,59 @@ pub fn build_app( coordination_store: Arc, sessions: SharedSessionStore, credential_verifier: SharedMachineCredentialVerifier, +) -> Router { + build_app_inner( + registry, + refresh_interval, + public_base_url, + secret_crypto, + runtime, + api_rate_limiter, + coordination_store, + sessions, + credential_verifier, + false, + ) +} + +#[allow(clippy::too_many_arguments)] +pub fn build_app_with_background_workers( + registry: PostgresRegistry, + refresh_interval: Duration, + public_base_url: Option, + secret_crypto: SecretCrypto, + runtime: RuntimeExecutor, + api_rate_limiter: RequestRateLimiter, + coordination_store: Arc, + sessions: SharedSessionStore, + credential_verifier: SharedMachineCredentialVerifier, +) -> Router { + build_app_inner( + registry, + refresh_interval, + public_base_url, + secret_crypto, + runtime, + api_rate_limiter, + coordination_store, + sessions, + credential_verifier, + true, + ) +} + +#[allow(clippy::too_many_arguments)] +fn build_app_inner( + registry: PostgresRegistry, + refresh_interval: Duration, + public_base_url: Option, + secret_crypto: SecretCrypto, + runtime: RuntimeExecutor, + api_rate_limiter: RequestRateLimiter, + coordination_store: Arc, + sessions: SharedSessionStore, + credential_verifier: SharedMachineCredentialVerifier, + start_background_workers: bool, ) -> Router { let state = Arc::new(AppState { registry: registry.clone(), @@ -143,6 +196,9 @@ pub fn build_app( credential_verifier, allowed_origins: AllowedOrigins::new(public_base_url), }); + if start_background_workers { + spawn_approval_recovery(Arc::clone(&state)); + } Router::new() .route("/health", get(health)) @@ -315,7 +371,21 @@ async fn decide_approval_request( .await { Ok(Some(record)) if status == ApprovalRequestStatus::Approved => { - match execute_approved_request(&state, &agent_path, record).await { + let claimed = match state + .registry + .claim_approval_request( + &record.approval.workspace_id, + &record.approval.agent_id, + &record.approval.id, + OffsetDateTime::now_utc(), + ) + .await + { + Ok(Some(claimed)) => claimed, + Ok(None) => return StatusCode::CONFLICT.into_response(), + Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(), + }; + match execute_approved_request(&state, &agent_path, claimed).await { Ok(record) => Json(json!(record)).into_response(), Err(response) => response, } @@ -408,103 +478,6 @@ async fn expire_approval_response( } } -async fn execute_approved_request( - state: &Arc, - path: &AgentRoutePath, - approval: ApprovalRequestRecord, -) -> Result { - let tools = state - .catalog - .list_tools(&path.workspace_slug, &path.agent_slug) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())?; - let Some(tool) = tools.into_iter().find(|tool| { - tool.operation.id == approval.approval.operation_id - && tool.operation.version == approval.approval.operation_version - }) else { - return Err(StatusCode::NOT_FOUND.into_response()); - }; - - let operation = runtime_operation(&tool); - let request_preview = build_request_preview( - &state.runtime, - &operation, - &approval.approval.request_payload, - ); - let started_at = Instant::now(); - let resolved_auth = - resolve_operation_auth(state, &tool.workspace_id, &operation.execution_config).await; - let result = match resolved_auth { - Ok(resolved_auth) => { - state - .runtime - .execute_request( - RuntimeExecutionRequest::new(&operation, &approval.approval.request_payload) - .with_optional_auth(resolved_auth.as_ref()), - ) - .await - } - Err(error) => Err(error), - }; - - let (status, response_payload, invocation_status, invocation_level, message, error_kind) = - match result { - Ok(output) => ( - ApprovalRequestStatus::Completed, - output, - InvocationStatus::Ok, - InvocationLevel::Info, - "approved tool call completed", - None, - ), - Err(error) => ( - ApprovalRequestStatus::Failed, - json!({ - "error": { - "code": runtime_error_code(&error), - "message": error.to_string(), - } - }), - InvocationStatus::Error, - InvocationLevel::Error, - "approved tool call failed", - Some(runtime_error_code(&error)), - ), - }; - - let _ = persist_invocation( - state, - &tool, - InvocationRecord { - request_id: Some(approval.approval.id.as_str()), - tool_name: &tool.tool_name, - status: invocation_status, - level: invocation_level, - message, - status_code: None, - error_kind, - duration: started_at.elapsed(), - request_preview, - response_preview: response_payload.clone(), - }, - ) - .await; - - state - .registry - .finish_approval_request(FinishApprovalRequest { - workspace_id: &approval.approval.workspace_id, - agent_id: &approval.approval.agent_id, - approval_id: &approval.approval.id, - status, - response_payload: Some(response_payload), - decision_note: None, - }) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())? - .ok_or_else(|| StatusCode::CONFLICT.into_response()) -} - async fn mcp_get( Path(path): Path, State(state): State>, @@ -857,7 +830,7 @@ async fn handle_tool_call( .await } -async fn resolve_operation_auth( +pub(super) async fn resolve_operation_auth( state: &Arc, workspace_id: &crank_core::WorkspaceId, execution_config: &crank_core::ExecutionConfig, @@ -1004,7 +977,7 @@ async fn handle_base_tool_call( match result { Ok(output) => { - let _ = persist_invocation( + if let Err(error) = persist_invocation( &state, &tool, InvocationRecord { @@ -1020,12 +993,15 @@ async fn handle_base_tool_call( response_preview: output.clone(), }, ) - .await; + .await + { + warn!(error = %error, "successful invocation log write failed"); + } success_tool_response(message, response_mode, &session.protocol_version, output) } Err(error) => { - let _ = persist_invocation( + if let Err(log_error) = persist_invocation( &state, &tool, InvocationRecord { @@ -1041,7 +1017,10 @@ async fn handle_base_tool_call( response_preview: Value::Null, }, ) - .await; + .await + { + warn!(error = %log_error, "failed invocation log write failed"); + } tool_error_response( message, @@ -1146,17 +1125,22 @@ async fn maybe_create_custom_pending_approval( decision_note: None, }; - if let Err(error) = state + let persisted_approval = match state .registry .create_approval_request(CreateApprovalRequest { approval: &approval, }) .await { - return Some(internal_jsonrpc_error(message, error)); - } + Ok(approval) => approval, + Err(error) => return Some(internal_jsonrpc_error(message, error)), + }; + let response_payload = persisted_approval + .approval + .response_payload + .unwrap_or(response_payload); - let _ = persist_invocation( + if let Err(error) = persist_invocation( state, tool, InvocationRecord { @@ -1172,7 +1156,10 @@ async fn maybe_create_custom_pending_approval( response_preview: response_payload.clone(), }, ) - .await; + .await + { + warn!(error = %error, "pending approval invocation log write failed"); + } Some(success_tool_response( message, @@ -1404,7 +1391,7 @@ fn take_confirmation_token(arguments: &mut Value) -> Option { .filter(|value| !value.trim().is_empty()) } -fn build_request_preview( +pub(super) fn build_request_preview( runtime: &RuntimeExecutor, operation: &RuntimeOperation, arguments: &Value, @@ -1420,20 +1407,20 @@ fn build_request_preview( } } -struct InvocationRecord<'a> { - request_id: Option<&'a str>, - tool_name: &'a str, - status: InvocationStatus, - level: InvocationLevel, - message: &'a str, - status_code: Option, - error_kind: Option<&'a str>, - duration: Duration, - request_preview: Value, - response_preview: Value, +pub(super) struct InvocationRecord<'a> { + pub(super) request_id: Option<&'a str>, + pub(super) tool_name: &'a str, + pub(super) status: InvocationStatus, + pub(super) level: InvocationLevel, + pub(super) message: &'a str, + pub(super) status_code: Option, + pub(super) error_kind: Option<&'a str>, + pub(super) duration: Duration, + pub(super) request_preview: Value, + pub(super) response_preview: Value, } -async fn persist_invocation( +pub(super) async fn persist_invocation( state: &Arc, tool: &PublishedAgentTool, record: InvocationRecord<'_>, @@ -1543,7 +1530,7 @@ fn resolve_generated_tool( None } -fn runtime_operation(tool: &PublishedAgentTool) -> RuntimeOperation { +pub(super) fn runtime_operation(tool: &PublishedAgentTool) -> RuntimeOperation { let mut operation = RuntimeOperation::from(tool.operation.clone()); operation.tool_name = tool.tool_name.clone(); operation.tool_description.title = tool.tool_title.clone(); diff --git a/crates/crank-community-mcp/src/approval_execution.rs b/crates/crank-community-mcp/src/approval_execution.rs new file mode 100644 index 0000000..e8e4e79 --- /dev/null +++ b/crates/crank-community-mcp/src/approval_execution.rs @@ -0,0 +1,237 @@ +use std::{sync::Arc, time::Instant}; + +use axum::{ + http::StatusCode, + response::{IntoResponse, Response}, +}; +use crank_core::{ApprovalRequestStatus, InvocationLevel, InvocationSource, InvocationStatus}; +use crank_registry::{ApprovalRequestRecord, FinishApprovalRequest}; +use crank_runtime::{RuntimeExecutionRequest, RuntimeRequestContext}; +use serde_json::json; +use time::OffsetDateTime; +use tracing::warn; + +use crate::{ + app::{ + AgentRoutePath, AppState, InvocationRecord, build_request_preview, persist_invocation, + resolve_operation_auth, runtime_operation, + }, + tool_error::runtime_error_code, +}; + +const RECOVERY_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); +const RECOVERY_GRACE: time::Duration = time::Duration::seconds(5); +const EXECUTION_LEASE: time::Duration = time::Duration::minutes(6); + +pub(super) fn spawn_approval_recovery(state: Arc) { + tokio::spawn(async move { + let mut interval = tokio::time::interval(RECOVERY_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + recover_approved_requests(&state).await; + } + }); +} + +async fn recover_approved_requests(state: &Arc) { + for _ in 0..32 { + let now = OffsetDateTime::now_utc(); + let approval = match state + .registry + .claim_next_recoverable_approval_request( + now, + now - RECOVERY_GRACE, + now - EXECUTION_LEASE, + ) + .await + { + Ok(Some(approval)) => approval, + Ok(None) => break, + Err(error) => { + warn!(error = %error, "approval recovery query failed"); + break; + } + }; + let Some(path) = approval_agent_path(state, &approval).await else { + continue; + }; + if execute_approved_request(state, &path, approval) + .await + .is_err() + { + warn!("recovered approval execution did not finish"); + } + } +} + +async fn approval_agent_path( + state: &Arc, + approval: &ApprovalRequestRecord, +) -> Option { + let workspace = match state + .registry + .get_workspace(&approval.approval.workspace_id) + .await + { + Ok(Some(workspace)) => workspace, + Ok(None) => return None, + Err(error) => { + warn!(error = %error, "approval workspace lookup failed"); + return None; + } + }; + let agent = match state + .registry + .get_agent_summary(&approval.approval.workspace_id, &approval.approval.agent_id) + .await + { + Ok(Some(agent)) => agent, + Ok(None) => return None, + Err(error) => { + warn!(error = %error, "approval agent lookup failed"); + return None; + } + }; + Some(AgentRoutePath { + workspace_slug: workspace.workspace.slug, + agent_slug: agent.slug, + }) +} + +pub(super) async fn execute_approved_request( + state: &Arc, + path: &AgentRoutePath, + approval: ApprovalRequestRecord, +) -> Result { + let tools = state + .catalog + .list_tools(&path.workspace_slug, &path.agent_slug) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())?; + let Some(tool) = tools.into_iter().find(|tool| { + tool.operation.id == approval.approval.operation_id + && tool.operation.version == approval.approval.operation_version + }) else { + return finish_unavailable_approval(state, &approval).await; + }; + + let operation = runtime_operation(&tool); + let request_preview = build_request_preview( + &state.runtime, + &operation, + &approval.approval.request_payload, + ); + let started_at = Instant::now(); + let runtime_request_context = + RuntimeRequestContext::from_request_id(approval.approval.id.as_str().to_owned()) + .with_response_cache_scope( + tool.workspace_id.as_str().to_owned(), + tool.agent_id.as_str().to_owned(), + ) + .with_metering_context( + tool.workspace_id.clone(), + Some(tool.agent_id.clone()), + InvocationSource::AgentToolCall, + ) + .with_approval_granted(); + let resolved_auth = + resolve_operation_auth(state, &tool.workspace_id, &operation.execution_config).await; + let result = match resolved_auth { + Ok(resolved_auth) => { + state + .runtime + .execute_request( + RuntimeExecutionRequest::new(&operation, &approval.approval.request_payload) + .with_optional_auth(resolved_auth.as_ref()) + .with_context(&runtime_request_context), + ) + .await + } + Err(error) => Err(error), + }; + + let (status, response_payload, invocation_status, invocation_level, message, error_kind) = + match result { + Ok(output) => ( + ApprovalRequestStatus::Completed, + output, + InvocationStatus::Ok, + InvocationLevel::Info, + "approved tool call completed", + None, + ), + Err(error) => ( + ApprovalRequestStatus::Failed, + json!({ + "error": { + "code": runtime_error_code(&error), + "message": error.to_string(), + } + }), + InvocationStatus::Error, + InvocationLevel::Error, + "approved tool call failed", + Some(runtime_error_code(&error)), + ), + }; + + if let Err(error) = persist_invocation( + state, + &tool, + InvocationRecord { + request_id: Some(approval.approval.id.as_str()), + tool_name: &tool.tool_name, + status: invocation_status, + level: invocation_level, + message, + status_code: None, + error_kind, + duration: started_at.elapsed(), + request_preview, + response_preview: response_payload.clone(), + }, + ) + .await + { + warn!(error = %error, "approved invocation log write failed"); + } + + state + .registry + .finish_approval_request(FinishApprovalRequest { + workspace_id: &approval.approval.workspace_id, + agent_id: &approval.approval.agent_id, + approval_id: &approval.approval.id, + status, + response_payload: Some(response_payload), + decision_note: None, + }) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())? + .ok_or_else(|| StatusCode::CONFLICT.into_response()) +} + +async fn finish_unavailable_approval( + state: &Arc, + approval: &ApprovalRequestRecord, +) -> Result { + state + .registry + .finish_approval_request(FinishApprovalRequest { + workspace_id: &approval.approval.workspace_id, + agent_id: &approval.approval.agent_id, + approval_id: &approval.approval.id, + status: ApprovalRequestStatus::Failed, + response_payload: Some(json!({ + "error": { + "code": "approved_operation_unavailable", + "message": "the approved operation version is no longer published" + } + })), + decision_note: None, + }) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())? + .ok_or_else(|| StatusCode::CONFLICT.into_response()) +} diff --git a/crates/crank-community-mcp/src/catalog.rs b/crates/crank-community-mcp/src/catalog.rs index f3f73bd..73d58e6 100644 --- a/crates/crank-community-mcp/src/catalog.rs +++ b/crates/crank-community-mcp/src/catalog.rs @@ -1,13 +1,13 @@ use std::{ collections::HashMap, sync::Arc, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; use crank_core::{CacheScope, CoordinationStateStore, CoordinationStateValue}; use crank_registry::{PostgresRegistry, PublishedAgentTool, RegistryError}; use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use tracing::info; #[derive(Clone)] @@ -16,6 +16,7 @@ pub struct PublishedToolCatalog { refresh_interval: Duration, coordination_store: Arc, cached: Arc>>, + refresh_locks: Arc>>>>, } #[derive(Clone, Debug, PartialEq, Eq, Hash)] @@ -33,6 +34,7 @@ struct CachedCatalog { #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] struct CatalogSnapshot { tools: Vec, + generated_at_ms: u64, } impl PublishedToolCatalog { @@ -46,6 +48,7 @@ impl PublishedToolCatalog { refresh_interval, coordination_store, cached: Arc::new(RwLock::new(HashMap::new())), + refresh_locks: Arc::new(Mutex::new(HashMap::new())), } } @@ -81,12 +84,32 @@ impl PublishedToolCatalog { return Ok(()); } - if let Some(tools) = self.load_shared_snapshot(workspace_slug, agent_slug).await { + let refresh_lock = { + let mut locks = self.refresh_locks.lock().await; + Arc::clone( + locks + .entry(key.clone()) + .or_insert_with(|| Arc::new(Mutex::new(()))), + ) + }; + let _refresh_guard = refresh_lock.lock().await; + let still_stale = { + let guard = self.cached.read().await; + match guard.get(&key).and_then(|entry| entry.loaded_at) { + Some(loaded_at) => loaded_at.elapsed() >= self.refresh_interval, + None => true, + } + }; + if !still_stale { + return Ok(()); + } + + if let Some((tools, age)) = self.load_shared_snapshot(workspace_slug, agent_slug).await { let mut guard = self.cached.write().await; guard.insert( key, CachedCatalog { - loaded_at: Some(Instant::now()), + loaded_at: Instant::now().checked_sub(age), tools, }, ); @@ -136,7 +159,7 @@ impl PublishedToolCatalog { &self, workspace_slug: &str, agent_slug: &str, - ) -> Option> { + ) -> Option<(Vec, Duration)> { if self.refresh_interval.is_zero() { return None; } @@ -150,9 +173,9 @@ impl PublishedToolCatalog { Ok(value) => value?, Err(_) => return None, }; - serde_json::from_value::(value.payload) - .ok() - .map(|snapshot| snapshot.tools) + let snapshot = serde_json::from_value::(value.payload).ok()?; + let age = Duration::from_millis(now_unix_ms().saturating_sub(snapshot.generated_at_ms)); + (age < self.refresh_interval).then_some((snapshot.tools, age)) } async fn store_shared_snapshot( @@ -167,6 +190,7 @@ impl PublishedToolCatalog { let payload = match serde_json::to_value(CatalogSnapshot { tools: tools.to_vec(), + generated_at_ms: now_unix_ms(), }) { Ok(payload) => payload, Err(_) => return, @@ -184,6 +208,13 @@ impl PublishedToolCatalog { } } +fn now_unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)) + .unwrap_or_default() +} + impl CatalogKey { fn new(workspace_slug: &str, agent_slug: &str) -> Self { Self { diff --git a/crates/crank-community-mcp/src/lib.rs b/crates/crank-community-mcp/src/lib.rs index db2dc09..bdf7deb 100644 --- a/crates/crank-community-mcp/src/lib.rs +++ b/crates/crank-community-mcp/src/lib.rs @@ -1,5 +1,6 @@ mod access; mod app; +mod approval_execution; pub mod auth; pub mod catalog; pub mod jsonrpc; @@ -9,4 +10,4 @@ pub mod session; pub mod tool_error; mod transport; -pub use app::build_app; +pub use app::{build_app, build_app_with_background_workers}; diff --git a/crates/crank-core/src/approval.rs b/crates/crank-core/src/approval.rs index e1f6863..152b8d2 100644 --- a/crates/crank-core/src/approval.rs +++ b/crates/crank-core/src/approval.rs @@ -11,6 +11,7 @@ pub enum ApprovalRequestStatus { #[default] Pending, Approved, + Executing, Denied, Expired, Completed, diff --git a/crates/crank-core/src/lib.rs b/crates/crank-core/src/lib.rs index 75ff337..f831e5b 100644 --- a/crates/crank-core/src/lib.rs +++ b/crates/crank-core/src/lib.rs @@ -38,8 +38,8 @@ pub mod domain { UserSessionId, WorkspaceId, }; pub use crate::observability::{ - InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod, - UsageRollup, + INVOCATION_PREVIEW_MAX_BYTES, InvocationLevel, InvocationLog, InvocationSource, + InvocationStatus, UsagePeriod, UsageRollup, sanitize_invocation_preview, }; pub use crate::operation::{ ConfigExport, ConfirmationPolicy, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, @@ -129,7 +129,8 @@ pub use ids::{ UserSessionId, WorkspaceId, }; pub use observability::{ - InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod, UsageRollup, + INVOCATION_PREVIEW_MAX_BYTES, InvocationLevel, InvocationLog, InvocationSource, + InvocationStatus, UsagePeriod, UsageRollup, sanitize_invocation_preview, }; pub use operation::{ ConfigExport, ConfirmationPolicy, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, diff --git a/crates/crank-core/src/observability.rs b/crates/crank-core/src/observability.rs index 8347f61..191e920 100644 --- a/crates/crank-core/src/observability.rs +++ b/crates/crank-core/src/observability.rs @@ -1,9 +1,68 @@ use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{Map, Value, json}; use time::OffsetDateTime; use crate::{AgentId, OperationId, WorkspaceId}; +pub const INVOCATION_PREVIEW_MAX_BYTES: usize = 16 * 1024; + +pub fn sanitize_invocation_preview(value: &Value) -> Value { + let redacted = redact_sensitive_fields(value); + let Ok(encoded) = serde_json::to_vec(&redacted) else { + return Value::Null; + }; + if encoded.len() <= INVOCATION_PREVIEW_MAX_BYTES { + return redacted; + } + let preview = String::from_utf8_lossy(&encoded[..INVOCATION_PREVIEW_MAX_BYTES]).into_owned(); + json!({ + "truncated": true, + "original_bytes": encoded.len(), + "preview": preview, + }) +} + +fn redact_sensitive_fields(value: &Value) -> Value { + match value { + Value::Object(object) => Value::Object( + object + .iter() + .map(|(key, value)| { + let value = if is_sensitive_key(key) { + Value::String("[REDACTED]".to_owned()) + } else { + redact_sensitive_fields(value) + }; + (key.clone(), value) + }) + .collect::>(), + ), + Value::Array(items) => Value::Array(items.iter().map(redact_sensitive_fields).collect()), + _ => value.clone(), + } +} + +fn is_sensitive_key(key: &str) -> bool { + let normalized = key.to_ascii_lowercase().replace('-', "_"); + let compact = normalized + .chars() + .filter(char::is_ascii_alphanumeric) + .collect::(); + [ + "authorization", + "cookie", + "password", + "passwd", + "secret", + "token", + ] + .iter() + .any(|sensitive| compact.contains(sensitive)) + || ["apikey", "accesskey", "privatekey"] + .iter() + .any(|sensitive| compact.contains(sensitive)) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum InvocationSource { @@ -87,7 +146,10 @@ mod tests { use serde_json::json; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; - use super::{InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod}; + use super::{ + INVOCATION_PREVIEW_MAX_BYTES, InvocationLevel, InvocationLog, InvocationSource, + InvocationStatus, UsagePeriod, sanitize_invocation_preview, + }; use crate::{AgentId, OperationId, WorkspaceId, ids::InvocationLogId}; fn timestamp(value: &str) -> OffsetDateTime { @@ -129,4 +191,33 @@ mod tests { assert_eq!(value, "7d"); } + + #[test] + fn redacts_sensitive_fields_recursively() { + let preview = sanitize_invocation_preview(&json!({ + "authorization": "Bearer secret", + "nested": { + "api_key": "key", + "refreshToken": "token", + "client_secret_value": "secret", + "value": 42 + } + })); + + assert_eq!(preview["authorization"], "[REDACTED]"); + assert_eq!(preview["nested"]["api_key"], "[REDACTED]"); + assert_eq!(preview["nested"]["refreshToken"], "[REDACTED]"); + assert_eq!(preview["nested"]["client_secret_value"], "[REDACTED]"); + assert_eq!(preview["nested"]["value"], 42); + } + + #[test] + fn truncates_large_previews() { + let preview = sanitize_invocation_preview(&json!({ + "body": "x".repeat(INVOCATION_PREVIEW_MAX_BYTES * 2) + })); + + assert_eq!(preview["truncated"], true); + assert!(preview["original_bytes"].as_u64().unwrap() > INVOCATION_PREVIEW_MAX_BYTES as u64); + } } diff --git a/crates/crank-registry/Cargo.toml b/crates/crank-registry/Cargo.toml index 58f3413..559efc1 100644 --- a/crates/crank-registry/Cargo.toml +++ b/crates/crank-registry/Cargo.toml @@ -11,6 +11,7 @@ crank-mapping = { path = "../crank-mapping" } crank-schema = { path = "../crank-schema" } serde.workspace = true serde_json.workspace = true +sha2.workspace = true sqlx.workspace = true thiserror.workspace = true time.workspace = true diff --git a/crates/crank-registry/src/migrations.rs b/crates/crank-registry/src/migrations.rs index d5bceac..47408b4 100644 --- a/crates/crank-registry/src/migrations.rs +++ b/crates/crank-registry/src/migrations.rs @@ -586,6 +586,22 @@ pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> { query("alter table approval_requests drop column if exists confirmation_body") .execute(pool) .await?; + query("alter table approval_requests add column if not exists execution_started_at timestamptz null") + .execute(pool) + .await?; + query("alter table approval_requests add column if not exists execution_attempts integer not null default 0") + .execute(pool) + .await?; + query("alter table approval_requests add column if not exists request_fingerprint text null") + .execute(pool) + .await?; + query( + "create unique index if not exists approval_requests_pending_fingerprint_idx + on approval_requests(agent_id, operation_id, operation_version, request_fingerprint) + where status = 'pending' and request_fingerprint is not null", + ) + .execute(pool) + .await?; query( "create index if not exists approval_requests_agent_status_idx on approval_requests(workspace_id, agent_id, status, expires_at)", diff --git a/crates/crank-registry/src/postgres/approval.rs b/crates/crank-registry/src/postgres/approval.rs index 2739ba7..4b553e9 100644 --- a/crates/crank-registry/src/postgres/approval.rs +++ b/crates/crank-registry/src/postgres/approval.rs @@ -1,11 +1,31 @@ use super::*; +use sha2::{Digest, Sha256}; impl PostgresRegistry { pub async fn create_approval_request( &self, request: CreateApprovalRequest<'_>, - ) -> Result<(), RegistryError> { + ) -> Result { + let fingerprint = approval_request_fingerprint(&request.approval.request_payload)?; + let mut transaction = self.pool.begin().await?; sqlx::query( + "update approval_requests + set status = 'expired' + where agent_id = $1 + and operation_id = $2 + and operation_version = $3 + and request_fingerprint = $4 + and status = 'pending' + and expires_at <= $5", + ) + .bind(request.approval.agent_id.as_str()) + .bind(request.approval.operation_id.as_str()) + .bind(to_db_version(request.approval.operation_version)) + .bind(&fingerprint) + .bind(request.approval.created_at) + .execute(&mut *transaction) + .await?; + let row = sqlx::query( "insert into approval_requests ( id, workspace_id, @@ -20,12 +40,20 @@ impl PostgresRegistry { expires_at, decided_at, decided_by_key_id, - decision_note + decision_note, + request_fingerprint ) values ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10::timestamptz, $11::timestamptz, $12::timestamptz, - $13, $14 - )", + $13, $14, $15 + ) + on conflict (agent_id, operation_id, operation_version, request_fingerprint) + where status = 'pending' and request_fingerprint is not null + do update set request_fingerprint = excluded.request_fingerprint + returning + id, workspace_id, agent_id, operation_id, operation_version, + status, risk_level, request_payload_json, response_payload_json, + created_at, expires_at, decided_at, decided_by_key_id, decision_note", ) .bind(request.approval.id.as_str()) .bind(request.approval.workspace_id.as_str()) @@ -53,10 +81,12 @@ impl PostgresRegistry { .map(PlatformApiKeyId::as_str), ) .bind(request.approval.decision_note.as_deref()) - .execute(&self.pool) + .bind(fingerprint) + .fetch_one(&mut *transaction) .await?; + transaction.commit().await?; - Ok(()) + map_approval_request_row(row) } pub async fn list_pending_approval_requests_for_agent( @@ -263,7 +293,7 @@ impl PostgresRegistry { where workspace_id = $4 and agent_id = $5 and id = $6 - and status = 'approved' + and status = 'executing' returning id, workspace_id, @@ -292,6 +322,75 @@ impl PostgresRegistry { row.map(map_approval_request_row).transpose() } + pub async fn claim_approval_request( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + approval_id: &ApprovalRequestId, + started_at: OffsetDateTime, + ) -> Result, RegistryError> { + let row = sqlx::query( + "update approval_requests + set status = 'executing', + execution_started_at = $1, + execution_attempts = execution_attempts + 1 + where workspace_id = $2 + and agent_id = $3 + and id = $4 + and status = 'approved' + returning + id, workspace_id, agent_id, operation_id, operation_version, + status, risk_level, request_payload_json, response_payload_json, + created_at, expires_at, decided_at, decided_by_key_id, decision_note", + ) + .bind(started_at) + .bind(workspace_id.as_str()) + .bind(agent_id.as_str()) + .bind(approval_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.map(map_approval_request_row).transpose() + } + + pub async fn claim_next_recoverable_approval_request( + &self, + started_at: OffsetDateTime, + approved_before: OffsetDateTime, + stale_before: OffsetDateTime, + ) -> Result, RegistryError> { + let row = sqlx::query( + "with candidate as ( + select id + from approval_requests + where (status = 'approved' and decided_at <= $1) + or (status = 'executing' and execution_started_at < $2) + order by decided_at asc nulls last, created_at asc + for update skip locked + limit 1 + ) + update approval_requests as approval + set status = 'executing', + execution_started_at = $3, + execution_attempts = approval.execution_attempts + 1 + from candidate + where approval.id = candidate.id + returning + approval.id, approval.workspace_id, approval.agent_id, + approval.operation_id, approval.operation_version, approval.status, + approval.risk_level, approval.request_payload_json, + approval.response_payload_json, approval.created_at, approval.expires_at, + approval.decided_at, approval.decided_by_key_id, approval.decision_note", + ) + .bind(approved_before) + .bind(stale_before) + .bind(started_at) + .fetch_optional(&self.pool) + .await?; + + row.map(map_approval_request_row).transpose() + } + pub async fn expire_approval_request( &self, request: ExpireApprovalRequest<'_>, @@ -331,6 +430,26 @@ impl PostgresRegistry { } } +fn approval_request_fingerprint(payload: &Value) -> Result { + let canonical = canonical_json(payload); + let encoded = serde_json::to_vec(&canonical)?; + Ok(format!("{:x}", Sha256::digest(encoded))) +} + +fn canonical_json(value: &Value) -> Value { + match value { + Value::Object(object) => { + let sorted = object + .iter() + .map(|(key, value)| (key.clone(), canonical_json(value))) + .collect::>(); + Value::Object(sorted.into_iter().collect()) + } + Value::Array(items) => Value::Array(items.iter().map(canonical_json).collect()), + _ => value.clone(), + } +} + fn map_approval_request_row(row: PgRow) -> Result { Ok(ApprovalRequestRecord { approval: ApprovalRequest { diff --git a/crates/crank-registry/src/postgres/observability.rs b/crates/crank-registry/src/postgres/observability.rs index f170af8..fb06911 100644 --- a/crates/crank-registry/src/postgres/observability.rs +++ b/crates/crank-registry/src/postgres/observability.rs @@ -5,6 +5,9 @@ impl PostgresRegistry { &self, request: CreateInvocationLogRequest<'_>, ) -> Result<(), RegistryError> { + let request_preview = crank_core::sanitize_invocation_preview(&request.log.request_preview); + let response_preview = + crank_core::sanitize_invocation_preview(&request.log.response_preview); sqlx::query( "insert into invocation_logs ( id, @@ -45,8 +48,8 @@ impl PostgresRegistry { } })?) .bind(&request.log.error_kind) - .bind(Json(request.log.request_preview.clone())) - .bind(Json(request.log.response_preview.clone())) + .bind(Json(request_preview)) + .bind(Json(response_preview)) .bind(request.log.created_at) .execute(&self.pool) .await?; diff --git a/crates/crank-registry/tests/integration/workspace_access.rs b/crates/crank-registry/tests/integration/workspace_access.rs index 2fffcb7..5e2a3e8 100644 --- a/crates/crank-registry/tests/integration/workspace_access.rs +++ b/crates/crank-registry/tests/integration/workspace_access.rs @@ -515,12 +515,23 @@ async fn manages_approval_request_lifecycle() { }) .await .unwrap(); - registry + let created = registry .create_approval_request(CreateApprovalRequest { approval: &approval, }) .await .unwrap(); + assert_eq!(created.approval.id, approval.id); + let mut duplicate = approval.clone(); + duplicate.id = ApprovalRequestId::new("approval_02"); + duplicate.request_payload = json!({"amount": 100}); + let deduplicated = registry + .create_approval_request(CreateApprovalRequest { + approval: &duplicate, + }) + .await + .unwrap(); + assert_eq!(deduplicated.approval.id, approval.id); let pending = registry .list_pending_approval_requests_for_agent(&workspace_id, &agent.id) @@ -572,6 +583,48 @@ async fn manages_approval_request_lifecycle() { Some(json!({"approve": "yes"})) ); + let approval_inside_recovery_grace = registry + .claim_next_recoverable_approval_request( + timestamp("2026-03-25T12:02:01Z"), + timestamp("2026-03-25T12:01:59Z"), + timestamp("2026-03-25T11:55:00Z"), + ) + .await + .unwrap(); + assert!(approval_inside_recovery_grace.is_none()); + + let claimed = registry + .claim_approval_request( + &workspace_id, + &agent.id, + &approval.id, + timestamp("2026-03-25T12:02:01Z"), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(claimed.approval.status, ApprovalRequestStatus::Executing); + let fresh_claim = registry + .claim_next_recoverable_approval_request( + timestamp("2026-03-25T12:02:10Z"), + timestamp("2026-03-25T12:02:09Z"), + timestamp("2026-03-25T12:01:59Z"), + ) + .await + .unwrap(); + assert!(fresh_claim.is_none()); + let recovered = registry + .claim_next_recoverable_approval_request( + timestamp("2026-03-25T12:03:00Z"), + timestamp("2026-03-25T12:02:59Z"), + timestamp("2026-03-25T12:02:30Z"), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(recovered.approval.id, approval.id); + assert_eq!(recovered.approval.status, ApprovalRequestStatus::Executing); + let completed = registry .finish_approval_request(FinishApprovalRequest { workspace_id: &workspace_id, @@ -626,5 +679,26 @@ async fn manages_approval_request_lifecycle() { .unwrap(); assert!(repeated_decision.is_none()); + let mut expired = approval.clone(); + expired.id = ApprovalRequestId::new("approval_expired_pending"); + expired.request_payload = json!({"amount": 200}); + expired.created_at = timestamp("2026-03-25T12:04:00Z"); + expired.expires_at = timestamp("2026-03-25T12:04:01Z"); + registry + .create_approval_request(CreateApprovalRequest { approval: &expired }) + .await + .unwrap(); + let mut replacement = expired.clone(); + replacement.id = ApprovalRequestId::new("approval_replacement_pending"); + replacement.created_at = timestamp("2026-03-25T12:05:00Z"); + replacement.expires_at = timestamp("2026-03-25T12:10:00Z"); + let replaced = registry + .create_approval_request(CreateApprovalRequest { + approval: &replacement, + }) + .await + .unwrap(); + assert_eq!(replaced.approval.id, replacement.id); + database.cleanup().await; } diff --git a/crates/crank-runtime/src/confirmation.rs b/crates/crank-runtime/src/confirmation.rs index 95109a4..8e58f39 100644 --- a/crates/crank-runtime/src/confirmation.rs +++ b/crates/crank-runtime/src/confirmation.rs @@ -21,6 +21,9 @@ pub async fn confirm_operation( if !safety.class.requires_confirmation() { return Ok(()); } + if request_context.is_some_and(RuntimeRequestContext::approval_granted) { + return Ok(()); + } let Some(store) = store else { return Err(RuntimeError::ConfirmationStoreUnavailable { diff --git a/crates/crank-runtime/src/executor_builder.rs b/crates/crank-runtime/src/executor_builder.rs index 3e8a75f..5d43804 100644 --- a/crates/crank-runtime/src/executor_builder.rs +++ b/crates/crank-runtime/src/executor_builder.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use crank_adapter_rest::RestAdapter; +use crank_adapter_rest::{OutboundHttpPolicy, RestAdapter, RestAdapterError}; use crank_core::{ AdapterRegistry, CoordinationStateStore, MeteringSink, NoopMeteringSink, ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, @@ -73,3 +73,13 @@ pub fn community_default() -> RuntimeExecutorBuilder { RuntimeExecutorBuilder::new() .register_adapter(Arc::new(RestAdapter::new()) as SharedProtocolAdapter) } + +pub fn community_from_env() -> Result { + Ok(RuntimeExecutorBuilder::new() + .register_adapter(Arc::new(RestAdapter::from_env()?) as SharedProtocolAdapter)) +} + +pub fn community_with_outbound_policy(policy: OutboundHttpPolicy) -> RuntimeExecutorBuilder { + RuntimeExecutorBuilder::new() + .register_adapter(Arc::new(RestAdapter::with_policy(policy)) as SharedProtocolAdapter) +} diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index a9dd7a6..e23b748 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -23,9 +23,12 @@ pub use cache::{ pub use cache_factory::{ BuiltinCacheBackendFactory, CacheBackendFactory, SharedCacheBackendFactory, }; +pub use crank_adapter_rest::OutboundHttpPolicy; pub use error::RuntimeError; pub use executor::{RuntimeExecutionRequest, RuntimeExecutor}; -pub use executor_builder::{RuntimeExecutorBuilder, community_default}; +pub use executor_builder::{ + RuntimeExecutorBuilder, community_default, community_from_env, community_with_outbound_policy, +}; pub use limits::{RuntimeLimits, RuntimeLimitsConfigError}; pub use model::{AdapterResponse, PreparedRequest, RuntimeOperation}; pub use rate_limit::{ diff --git a/crates/crank-runtime/src/request_context.rs b/crates/crank-runtime/src/request_context.rs index e5b15ad..b3b1583 100644 --- a/crates/crank-runtime/src/request_context.rs +++ b/crates/crank-runtime/src/request_context.rs @@ -15,6 +15,7 @@ pub struct RuntimeRequestContext { pub response_cache_scope: Option, pub metering_context: Option, pub confirmation_token: Option, + pub approval_granted: bool, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -32,6 +33,7 @@ impl RuntimeRequestContext { response_cache_scope: None, metering_context: None, confirmation_token: None, + approval_granted: false, } } @@ -89,6 +91,15 @@ impl RuntimeRequestContext { pub fn confirmation_token(&self) -> Option<&str> { self.confirmation_token.as_deref() } + + pub fn with_approval_granted(mut self) -> Self { + self.approval_granted = true; + self + } + + pub fn approval_granted(&self) -> bool { + self.approval_granted + } } impl From for crank_core::ResponseCacheScope { @@ -154,6 +165,7 @@ mod tests { assert_eq!(context.correlation_id, "req_123"); assert!(context.response_cache_scope.is_none()); assert!(context.confirmation_token.is_none()); + assert!(!context.approval_granted()); } #[test] @@ -187,4 +199,11 @@ mod tests { assert_eq!(context.confirmation_token(), Some("ct_123")); } + + #[test] + fn records_prior_human_approval() { + let context = RuntimeRequestContext::from_request_id("req_123").with_approval_granted(); + + assert!(context.approval_granted()); + } } diff --git a/crates/crank-runtime/tests/integration/confirmation.rs b/crates/crank-runtime/tests/integration/confirmation.rs index 6cba2c6..2a4c0d4 100644 --- a/crates/crank-runtime/tests/integration/confirmation.rs +++ b/crates/crank-runtime/tests/integration/confirmation.rs @@ -75,6 +75,18 @@ async fn destructive_operation_requires_single_use_confirmation() { RuntimeError::InvalidConfirmationToken { .. } )); assert_eq!(call_count.load(Ordering::SeqCst), 1); + + let approved_context = context.with_approval_granted(); + let approved = executor + .execute_with_context( + &operation, + &json!({ "order_id": "ord_123" }), + Some(&approved_context), + ) + .await + .unwrap(); + assert_eq!(approved, json!({ "deleted": true })); + assert_eq!(call_count.load(Ordering::SeqCst), 2); } struct CountingAdapter { diff --git a/deploy/community/.env.example b/deploy/community/.env.example index 8a09e4a..a201d07 100644 --- a/deploy/community/.env.example +++ b/deploy/community/.env.example @@ -24,6 +24,9 @@ CRANK_RUNTIME_MAX_CONCURRENT_UNARY=64 CRANK_RUNTIME_MAX_CONCURRENT_WINDOW=16 CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS=16 CRANK_RUNTIME_MAX_CONCURRENT_JOBS=16 +CRANK_OUTBOUND_ALLOWED_HOSTS= +CRANK_OUTBOUND_DENIED_HOSTS= +CRANK_OUTBOUND_MAX_RESPONSE_BYTES=4194304 CRANK_CACHE_BACKEND=memory CRANK_CACHE_URL= CRANK_CACHE_DEFAULT_TTL_MS= diff --git a/deploy/community/.env.images.example b/deploy/community/.env.images.example index edc2cd1..eb7c159 100644 --- a/deploy/community/.env.images.example +++ b/deploy/community/.env.images.example @@ -16,6 +16,9 @@ CRANK_PUBLISH_BIND=127.0.0.1 CRANK_ADMIN_BIND=0.0.0.0:3001 CRANK_MCP_BIND=0.0.0.0:3002 CRANK_MCP_REFRESH_MS=5000 +CRANK_OUTBOUND_ALLOWED_HOSTS= +CRANK_OUTBOUND_DENIED_HOSTS= +CRANK_OUTBOUND_MAX_RESPONSE_BYTES=4194304 CRANK_CACHE_BACKEND=memory CRANK_CACHE_URL= diff --git a/deploy/community/docker-compose.images.yml b/deploy/community/docker-compose.images.yml index c11d783..373300c 100644 --- a/deploy/community/docker-compose.images.yml +++ b/deploy/community/docker-compose.images.yml @@ -59,6 +59,9 @@ services: CRANK_BOOTSTRAP_ADMIN_PASSWORD: ${CRANK_BOOTSTRAP_ADMIN_PASSWORD} CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME: ${CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME:-Crank Owner} CRANK_DEMO_SEED: ${CRANK_DEMO_SEED:-false} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} volumes: - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} ports: @@ -87,6 +90,9 @@ services: CRANK_LOG_LEVEL: ${CRANK_LOG_LEVEL:-info} CRANK_MASTER_KEY: ${CRANK_MASTER_KEY} CRANK_BASE_URL: ${CRANK_BASE_URL:-http://localhost:3000} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} volumes: - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} ports: diff --git a/deploy/community/docker-compose.yml b/deploy/community/docker-compose.yml index f31dd38..ee11401 100644 --- a/deploy/community/docker-compose.yml +++ b/deploy/community/docker-compose.yml @@ -43,6 +43,9 @@ services: CRANK_BOOTSTRAP_ADMIN_PASSWORD: ${CRANK_BOOTSTRAP_ADMIN_PASSWORD} CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME: ${CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME:-Crank Owner} CRANK_DEMO_SEED: ${CRANK_DEMO_SEED:-false} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} volumes: - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} ports: @@ -74,6 +77,9 @@ services: CRANK_LOG_LEVEL: ${CRANK_LOG_LEVEL:-info} CRANK_MASTER_KEY: ${CRANK_MASTER_KEY} CRANK_BASE_URL: ${CRANK_BASE_URL:-http://localhost:3000} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} volumes: - artifact_storage:${CRANK_STORAGE_ROOT:-/var/lib/crank/storage} ports: diff --git a/docker-compose.yml b/docker-compose.yml index 084ef18..6597d08 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -41,6 +41,9 @@ services: CRANK_BOOTSTRAP_ADMIN_PASSWORD: ${CRANK_BOOTSTRAP_ADMIN_PASSWORD} CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME: ${CRANK_BOOTSTRAP_ADMIN_DISPLAY_NAME:-Crank Owner} CRANK_DEMO_SEED: ${CRANK_DEMO_SEED:-false} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} depends_on: postgres: condition: service_healthy @@ -72,6 +75,9 @@ services: CRANK_LOG_LEVEL: ${CRANK_LOG_LEVEL:-info} CRANK_MASTER_KEY: ${CRANK_MASTER_KEY} CRANK_BASE_URL: ${CRANK_BASE_URL:-http://localhost:3000} + CRANK_OUTBOUND_ALLOWED_HOSTS: ${CRANK_OUTBOUND_ALLOWED_HOSTS:-} + CRANK_OUTBOUND_DENIED_HOSTS: ${CRANK_OUTBOUND_DENIED_HOSTS:-} + CRANK_OUTBOUND_MAX_RESPONSE_BYTES: ${CRANK_OUTBOUND_MAX_RESPONSE_BYTES:-4194304} depends_on: postgres: condition: service_healthy diff --git a/docs/mcp-interface.md b/docs/mcp-interface.md index 9e7ece2..f58a4d9 100644 --- a/docs/mcp-interface.md +++ b/docs/mcp-interface.md @@ -199,6 +199,16 @@ curl https://crank.example.com/mcp/v1/default/sales/approvals//appr После подтверждения Crank выполняет исходный REST-запрос с тем payload, который был сохранен при первом `tools/call`. Если запрос прошел успешно, заявка получает статус `completed`, а результат сохраняется в `response_payload`. Если upstream вернул ошибку, заявка получает статус `failed`, а в `response_payload` сохраняется код и текст ошибки. +Перед обращением к upstream заявка атомарно переходит в статус `executing`. Если +`mcp-server` завершился во время выполнения, другой рабочий цикл повторно захватит +заявку после истечения аренды. Для изменяющих операций рекомендуется настроить +`execution_config.idempotency` и передавать поддерживаемый upstream заголовок +идемпотентности: универсальный HTTP-клиент не может гарантировать ровно одно внешнее +побочное действие при падении процесса между ответом upstream и записью результата. + +Повторный `tools/call` с теми же агентом, операцией, версией и JSON-аргументами +возвращает уже существующую активную заявку вместо создания дубликата. + Отклонение: ```bash diff --git a/docs/runtime-config.md b/docs/runtime-config.md index e9fb680..2a5287a 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -85,6 +85,32 @@ Demo seed идемпотентный: повторный старт не соз Эти настройки ограничивают параллельное выполнение операций и служебных задач. +## Исходящие HTTP-запросы + +По умолчанию Crank обращается только к публичным IP-адресам. Локальные, частные, +служебные и link-local сети блокируются после разрешения DNS-имени. Автоматические +HTTP-перенаправления и системный прокси отключены. + +- `CRANK_OUTBOUND_ALLOWED_HOSTS` - исключения для разрешённых внутренних узлов через + запятую. Публичные узлы разрешены независимо от этого списка. Для внутреннего API + укажите его имя или IP явно. Поддерживаются маски вида `*.example.internal`. +- `CRANK_OUTBOUND_DENIED_HOSTS` - список узлов, запрещённых независимо от списка + разрешённых. +- `CRANK_OUTBOUND_MAX_RESPONSE_BYTES` - максимальный размер ответа внешнего API; + по умолчанию `4194304` байт. + +Пример доступа только к двум внутренним API: + +```env +CRANK_OUTBOUND_ALLOWED_HOSTS=crm.example.internal,192.168.1.50 +CRANK_OUTBOUND_DENIED_HOSTS=metadata.example.internal +CRANK_OUTBOUND_MAX_RESPONSE_BYTES=4194304 +``` + +Одинаковые значения должны передаваться в `admin-api` и `mcp-server`: первый +проверяет операции при сохранении, второй применяет политику при каждом соединении. +Максимальный `execution_config.timeout_ms` операции равен `300000` мс. + ## Кэш По умолчанию Crank работает без внешнего кэша: @@ -107,6 +133,10 @@ CRANK_CACHE_DEFAULT_TTL_MS=60000 - `CRANK_LOG_LEVEL` - уровень логирования, например `info`, `debug`, `warn`. +Поля с паролями, токенами, ключами и заголовками авторизации удаляются из снимков +запросов и ответов. Один снимок ограничен 16 КиБ; более крупное значение хранится в +усечённом виде с исходным размером. + Пример: ```env diff --git a/justfile b/justfile index b06d8e9..03c8641 100644 --- a/justfile +++ b/justfile @@ -10,6 +10,12 @@ tooling-test: community-scope-check: scripts/check-community-scope.sh +rust-boundaries: + scripts/check-rust-boundaries.sh + +rust-code-health: + scripts/check-rust-code-health.sh + check: cargo check --workspace @@ -28,6 +34,8 @@ sqlx-check: verify: just tooling-test just community-scope-check + just rust-boundaries + just rust-code-health just fmt-check just clippy just test