From 2dab9d666e5cc00c4e8a0e75cc680d97e535896e Mon Sep 17 00:00:00 2001 From: github-ops Date: Sun, 21 Jun 2026 01:48:27 +0000 Subject: [PATCH] Split admin import and runtime safety modules --- apps/admin-api/src/service.rs | 126 +------- apps/admin-api/src/service/import_export.rs | 137 +++++++++ crates/crank-runtime/src/confirmation.rs | 203 +++++++++++++ crates/crank-runtime/src/executor.rs | 318 +------------------- crates/crank-runtime/src/idempotency.rs | 104 +++++++ crates/crank-runtime/src/lib.rs | 2 + 6 files changed, 464 insertions(+), 426 deletions(-) create mode 100644 apps/admin-api/src/service/import_export.rs create mode 100644 crates/crank-runtime/src/confirmation.rs create mode 100644 crates/crank-runtime/src/idempotency.rs diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 5727eff..d0060fe 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -41,13 +41,14 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tracing::{info, instrument}; use uuid::Uuid; +mod import_export; + use crate::{ auth::{ AuthSettings, AuthenticatedSession, SessionCookie, create_session_cookie, hash_password, hash_session_secret, verify_password, }, error::ApiError, - import_guidance::import_guidance_warnings, storage::LocalArtifactStorage, }; @@ -2625,129 +2626,6 @@ impl AdminService { Ok(()) } - #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))] - pub async fn export_operation( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - query: ExportQuery, - ) -> Result { - let version = match query.version { - Some(version) => version, - None => { - self.get_operation(workspace_id, operation_id) - .await? - .current_draft_version - } - }; - let record = self - .get_operation_version(workspace_id, operation_id, version) - .await?; - let document = YamlOperationDocument { - format_version: "1".to_owned(), - kind: "operation".to_owned(), - operation: RegistryOperation { - config_export: Some(ConfigExport { - format_version: "1".to_owned(), - export_mode: query.mode, - }), - ..record.snapshot - }, - }; - - serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string())) - } - - #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))] - pub async fn import_operation( - &self, - workspace_id: &WorkspaceId, - query: ImportQuery, - yaml_document: &str, - ) -> Result { - let document: YamlOperationDocument = serde_yaml::from_str(yaml_document) - .map_err(|error| ApiError::validation(error.to_string()))?; - if document.kind != "operation" { - return Err(ApiError::validation("yaml kind must be operation")); - } - - let payload = OperationPayload { - name: document.operation.name.clone(), - display_name: document.operation.display_name.clone(), - category: document.operation.category.clone(), - protocol: document.operation.protocol, - security_level: document.operation.security_level, - target: document.operation.target.clone(), - input_schema: document.operation.input_schema.clone(), - output_schema: document.operation.output_schema.clone(), - input_mapping: document.operation.input_mapping.clone(), - output_mapping: document.operation.output_mapping.clone(), - execution_config: document.operation.execution_config.clone(), - tool_description: document.operation.tool_description.clone(), - wizard_state: document.operation.wizard_state.clone(), - }; - let warnings = import_guidance_warnings(&document.operation); - - match query.mode { - ImportMode::Create => { - let created = self.create_operation(workspace_id, payload).await?; - Ok(ImportResponse { - operation_id: created.operation_id, - workspace_id: created.workspace_id, - version: created.version, - import_mode: ImportMode::Create, - warnings, - }) - } - ImportMode::Upsert => { - if let Some(existing) = self - .find_operation_by_name(workspace_id, &document.operation.name) - .await? - { - let created = self - .create_version( - workspace_id, - &existing.id, - NewVersionPayload { - operation: payload, - change_note: Some("yaml upsert".to_owned()), - }, - ) - .await?; - - let response = ImportResponse { - operation_id: created.operation_id, - workspace_id: created.workspace_id, - version: created.version, - import_mode: ImportMode::Upsert, - warnings, - }; - info!( - operation_id = %response.operation_id, - version = response.version, - "operation imported by upsert" - ); - Ok(response) - } else { - let created = self.create_operation(workspace_id, payload).await?; - let response = ImportResponse { - operation_id: created.operation_id, - workspace_id: created.workspace_id, - version: created.version, - import_mode: ImportMode::Upsert, - warnings, - }; - info!( - operation_id = %response.operation_id, - version = response.version, - "operation imported by upsert" - ); - Ok(response) - } - } - } - } - #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))] pub async fn save_json_sample( &self, diff --git a/apps/admin-api/src/service/import_export.rs b/apps/admin-api/src/service/import_export.rs new file mode 100644 index 0000000..4de3e8e --- /dev/null +++ b/apps/admin-api/src/service/import_export.rs @@ -0,0 +1,137 @@ +use crank_core::{ConfigExport, WorkspaceId}; +use crank_registry::RegistryOperation; +use tracing::{info, instrument}; + +use crate::{ + error::ApiError, + import_guidance::import_guidance_warnings, + service::{ + AdminService, ExportQuery, ImportMode, ImportQuery, ImportResponse, NewVersionPayload, + OperationPayload, YamlOperationDocument, + }, +}; + +impl AdminService { + #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))] + pub async fn export_operation( + &self, + workspace_id: &WorkspaceId, + operation_id: &crank_core::OperationId, + query: ExportQuery, + ) -> Result { + let version = match query.version { + Some(version) => version, + None => { + self.get_operation(workspace_id, operation_id) + .await? + .current_draft_version + } + }; + let record = self + .get_operation_version(workspace_id, operation_id, version) + .await?; + let document = YamlOperationDocument { + format_version: "1".to_owned(), + kind: "operation".to_owned(), + operation: RegistryOperation { + config_export: Some(ConfigExport { + format_version: "1".to_owned(), + export_mode: query.mode, + }), + ..record.snapshot + }, + }; + + serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string())) + } + + #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))] + pub async fn import_operation( + &self, + workspace_id: &WorkspaceId, + query: ImportQuery, + yaml_document: &str, + ) -> Result { + let document: YamlOperationDocument = serde_yaml::from_str(yaml_document) + .map_err(|error| ApiError::validation(error.to_string()))?; + if document.kind != "operation" { + return Err(ApiError::validation("yaml kind must be operation")); + } + + let payload = OperationPayload { + name: document.operation.name.clone(), + display_name: document.operation.display_name.clone(), + category: document.operation.category.clone(), + protocol: document.operation.protocol, + security_level: document.operation.security_level, + target: document.operation.target.clone(), + input_schema: document.operation.input_schema.clone(), + output_schema: document.operation.output_schema.clone(), + input_mapping: document.operation.input_mapping.clone(), + output_mapping: document.operation.output_mapping.clone(), + execution_config: document.operation.execution_config.clone(), + tool_description: document.operation.tool_description.clone(), + wizard_state: document.operation.wizard_state.clone(), + }; + let warnings = import_guidance_warnings(&document.operation); + + match query.mode { + ImportMode::Create => { + let created = self.create_operation(workspace_id, payload).await?; + Ok(ImportResponse { + operation_id: created.operation_id, + workspace_id: created.workspace_id, + version: created.version, + import_mode: ImportMode::Create, + warnings, + }) + } + ImportMode::Upsert => { + if let Some(existing) = self + .find_operation_by_name(workspace_id, &document.operation.name) + .await? + { + let created = self + .create_version( + workspace_id, + &existing.id, + NewVersionPayload { + operation: payload, + change_note: Some("yaml upsert".to_owned()), + }, + ) + .await?; + + let response = ImportResponse { + operation_id: created.operation_id, + workspace_id: created.workspace_id, + version: created.version, + import_mode: ImportMode::Upsert, + warnings, + }; + info!( + operation_id = %response.operation_id, + version = response.version, + "operation imported by upsert" + ); + Ok(response) + } else { + let created = self.create_operation(workspace_id, payload).await?; + let response = ImportResponse { + operation_id: created.operation_id, + workspace_id: created.workspace_id, + version: created.version, + import_mode: ImportMode::Upsert, + warnings, + }; + info!( + operation_id = %response.operation_id, + version = response.version, + "operation imported by upsert" + ); + Ok(response) + } + } + } + } +} diff --git a/crates/crank-runtime/src/confirmation.rs b/crates/crank-runtime/src/confirmation.rs new file mode 100644 index 0000000..95109a4 --- /dev/null +++ b/crates/crank-runtime/src/confirmation.rs @@ -0,0 +1,203 @@ +use std::time::Duration; + +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use crank_core::{ + CacheScope, CoordinationStateStore, CoordinationStateValue, HttpMethod, OperationSafetyClass, + OperationSafetyPolicy, Target, +}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::{RuntimeError, RuntimeOperation, RuntimeRequestContext}; + +pub async fn confirm_operation( + store: Option<&dyn CoordinationStateStore>, + operation: &RuntimeOperation, + input: &Value, + request_context: Option<&RuntimeRequestContext>, +) -> Result<(), RuntimeError> { + let safety = effective_safety_policy(operation); + if !safety.class.requires_confirmation() { + return Ok(()); + } + + let Some(store) = store else { + return Err(RuntimeError::ConfirmationStoreUnavailable { + operation_id: operation.operation_id.as_str().to_owned(), + }); + }; + + let scope = confirmation_scope(operation, request_context)?; + let input_hash = hash_json(input)?; + let provided_token = request_context + .and_then(RuntimeRequestContext::confirmation_token) + .or_else(|| confirmation_token_from_input(input)); + + let Some(provided_token) = provided_token else { + let token = issue_confirmation_token(store, &scope, &input_hash, &safety).await?; + return Err(RuntimeError::ConfirmationRequired { + operation_id: operation.operation_id.as_str().to_owned(), + safety_class: safety.class, + confirmation_token: token, + expires_in_ms: confirmation_ttl_ms(&safety), + }); + }; + + consume_confirmation_token(store, &scope, provided_token, &input_hash).await +} + +fn effective_safety_policy(operation: &RuntimeOperation) -> OperationSafetyPolicy { + operation + .execution_config + .safety + .clone() + .unwrap_or_else(|| OperationSafetyPolicy { + class: infer_safety_class(operation), + confirmation: None, + }) +} + +fn infer_safety_class(operation: &RuntimeOperation) -> OperationSafetyClass { + match &operation.target { + Target::Rest(target) => match target.method { + HttpMethod::Get => OperationSafetyClass::Read, + HttpMethod::Delete => OperationSafetyClass::Destructive, + HttpMethod::Post | HttpMethod::Put | HttpMethod::Patch => OperationSafetyClass::Write, + }, + } +} + +fn confirmation_ttl_ms(policy: &OperationSafetyPolicy) -> u64 { + policy + .confirmation + .as_ref() + .map(|confirmation| confirmation.ttl_ms) + .filter(|ttl_ms| *ttl_ms > 0) + .unwrap_or(300_000) +} + +fn confirmation_scope( + operation: &RuntimeOperation, + request_context: Option<&RuntimeRequestContext>, +) -> Result { + let Some(scope) = request_context.and_then(RuntimeRequestContext::response_cache_scope) else { + return Err(RuntimeError::InvalidPreparedRequest { + field: "runtime_context.response_cache_scope".to_owned(), + reason: "confirmation requires workspace and agent scope".to_owned(), + }); + }; + + Ok(format!( + "workspace:{}:agent:{}:operation:{}:version:{}", + scope.workspace_key, + scope.agent_key, + operation.operation_id.as_str(), + operation.operation_version + )) +} + +async fn issue_confirmation_token( + store: &dyn CoordinationStateStore, + scope: &str, + input_hash: &str, + safety: &OperationSafetyPolicy, +) -> Result { + let token = format!("ct_{}", Uuid::now_v7().simple()); + let key = confirmation_cache_key(scope, &token); + let ttl_ms = confirmation_ttl_ms(safety); + store + .put_value( + CacheScope::Coordination, + &key, + CoordinationStateValue { + payload: json!({ + "scope": scope, + "input_hash": input_hash, + }), + }, + Duration::from_millis(ttl_ms), + ) + .await + .map_err(|error| RuntimeError::InvalidPreparedRequest { + field: "confirmation_token".to_owned(), + reason: error.to_string(), + })?; + Ok(token) +} + +async fn consume_confirmation_token( + store: &dyn CoordinationStateStore, + operation_scope: &str, + token: &str, + input_hash: &str, +) -> Result<(), RuntimeError> { + let key = confirmation_cache_key(operation_scope, token); + let stored = store + .get_value(CacheScope::Coordination, &key) + .await + .map_err(|error| RuntimeError::InvalidPreparedRequest { + field: "confirmation_token".to_owned(), + reason: error.to_string(), + })?; + let _ = store.delete_value(CacheScope::Coordination, &key).await; + + let Some(stored) = stored else { + return Err(RuntimeError::InvalidConfirmationToken { + operation_id: operation_id_from_scope(operation_scope), + }); + }; + + let matches_scope = + stored.payload.get("scope").and_then(Value::as_str) == Some(operation_scope); + let matches_input = + stored.payload.get("input_hash").and_then(Value::as_str) == Some(input_hash); + + if matches_scope && matches_input { + Ok(()) + } else { + Err(RuntimeError::InvalidConfirmationToken { + operation_id: operation_id_from_scope(operation_scope), + }) + } +} + +fn confirmation_cache_key(scope: &str, token: &str) -> String { + let token_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(token.as_bytes())); + format!("crank:confirmation:{scope}:token:{token_hash}") +} + +fn confirmation_token_from_input(input: &Value) -> Option<&str> { + input + .get("_crank_confirmation_token") + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) +} + +fn hash_json(input: &Value) -> Result { + let sanitized = input_without_confirmation_token(input); + let encoded = + serde_json::to_vec(&sanitized).map_err(|error| RuntimeError::InvalidPreparedRequest { + field: "confirmation_input".to_owned(), + reason: error.to_string(), + })?; + Ok(URL_SAFE_NO_PAD.encode(Sha256::digest(encoded))) +} + +fn input_without_confirmation_token(input: &Value) -> Value { + let Value::Object(object) = input else { + return input.clone(); + }; + let mut sanitized = object.clone(); + sanitized.remove("_crank_confirmation_token"); + Value::Object(sanitized) +} + +fn operation_id_from_scope(scope: &str) -> String { + scope + .split(":operation:") + .nth(1) + .and_then(|tail| tail.split(":version:").next()) + .unwrap_or("") + .to_owned() +} diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index 9c5333e..99f1fed 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -4,17 +4,15 @@ use std::time::{Duration, Instant}; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ - AdapterRegistry, CacheScope, CachedHeader, CachedResponse, CoordinationStateStore, - CoordinationStateValue, ExecutionMode, HttpMethod, IdempotencyMode, IdempotencyPolicy, - InvocationStatus, MeteringEvent, OperationSafetyClass, OperationSafetyPolicy, - ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, Target, + AdapterRegistry, CachedHeader, CachedResponse, CoordinationStateStore, ExecutionMode, + HttpMethod, InvocationStatus, MeteringEvent, ResponseCacheStore, SharedMeteringSink, + SharedProtocolAdapter, Target, }; use serde_json::{Map, Value, json}; use sha2::{Digest, Sha256}; use time::OffsetDateTime; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tracing::debug; -use uuid::Uuid; use crate::{ AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation, @@ -153,9 +151,15 @@ impl RuntimeExecutor { request_context: Option<&RuntimeRequestContext>, ) -> Result { let mut prepared_request = prepared_request; - let idempotency_key = self.prepare_idempotency(operation, input, &mut prepared_request)?; - self.confirm_destructive_operation(operation, input, request_context) - .await?; + let idempotency_key = + crate::idempotency::prepare_idempotency(operation, input, &mut prepared_request)?; + crate::confirmation::confirm_operation( + self.coordination_store.as_deref(), + operation, + input, + request_context, + ) + .await?; if let Some(response) = self .load_idempotent_adapter_response( operation, @@ -202,78 +206,6 @@ impl RuntimeExecutor { Ok(finalized_output) } - fn prepare_idempotency( - &self, - operation: &RuntimeOperation, - input: &Value, - prepared_request: &mut PreparedRequest, - ) -> Result, RuntimeError> { - let Some(policy) = idempotency_policy(operation) else { - return Ok(None); - }; - - let key = idempotency_key_from_policy(policy, input, prepared_request); - let Some(key) = key else { - if policy.mode == IdempotencyMode::Required { - return Err(RuntimeError::InvalidPreparedRequest { - field: "execution_config.idempotency".to_owned(), - reason: "required idempotency key was not provided".to_owned(), - }); - } - return Ok(None); - }; - - if let Some(header_name) = policy - .header_name - .as_deref() - .filter(|value| !value.is_empty()) - { - prepared_request - .headers - .entry(header_name.to_owned()) - .or_insert_with(|| key.clone()); - } - - Ok(Some(key)) - } - - async fn confirm_destructive_operation( - &self, - operation: &RuntimeOperation, - input: &Value, - request_context: Option<&RuntimeRequestContext>, - ) -> Result<(), RuntimeError> { - let safety = effective_safety_policy(operation); - if !safety.class.requires_confirmation() { - return Ok(()); - } - - let Some(store) = self.coordination_store.as_ref() else { - return Err(RuntimeError::ConfirmationStoreUnavailable { - operation_id: operation.operation_id.as_str().to_owned(), - }); - }; - - let scope = confirmation_scope(operation, request_context)?; - let input_hash = hash_json(input)?; - let provided_token = request_context - .and_then(RuntimeRequestContext::confirmation_token) - .or_else(|| confirmation_token_from_input(input)); - - let Some(provided_token) = provided_token else { - let token = - issue_confirmation_token(store.as_ref(), &scope, &input_hash, &safety).await?; - return Err(RuntimeError::ConfirmationRequired { - operation_id: operation.operation_id.as_str().to_owned(), - safety_class: safety.class, - confirmation_token: token, - expires_in_ms: confirmation_ttl_ms(&safety), - }); - }; - - consume_confirmation_token(store.as_ref(), &scope, provided_token, &input_hash).await - } - async fn record_metering( &self, operation: &RuntimeOperation, @@ -430,7 +362,8 @@ impl RuntimeExecutor { request_context: Option<&RuntimeRequestContext>, ) -> Option { let response_cache = self.response_cache.as_ref()?; - let cache_key = idempotency_cache_key(operation, idempotency_key?, request_context)?; + let cache_key = + crate::idempotency::cache_key(operation, idempotency_key?, request_context)?; let cached = match response_cache.get(&cache_key).await { Ok(cached) => cached?, Err(error) => { @@ -460,10 +393,10 @@ impl RuntimeExecutor { if !(200..=299).contains(&adapter_response.status_code) { return; } - let Some(policy) = idempotency_policy(operation) else { + let Some(policy) = crate::idempotency::policy(operation) else { return; }; - let Some(cache_key) = idempotency_cache_key( + let Some(cache_key) = crate::idempotency::cache_key( operation, idempotency_key.unwrap_or_default(), request_context, @@ -669,225 +602,6 @@ fn response_cache_ttl(operation: &RuntimeOperation) -> Option { Some(Duration::from_millis(policy.ttl_ms)) } -fn idempotency_policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> { - let is_mutating_rest = - matches!(&operation.target, Target::Rest(target) if target.method != HttpMethod::Get); - if !is_mutating_rest { - return None; - } - - let policy = operation.execution_config.idempotency.as_ref()?; - if policy.mode == IdempotencyMode::Disabled || policy.ttl_ms == 0 { - return None; - } - - Some(policy) -} - -fn effective_safety_policy(operation: &RuntimeOperation) -> OperationSafetyPolicy { - operation - .execution_config - .safety - .clone() - .unwrap_or_else(|| OperationSafetyPolicy { - class: infer_safety_class(operation), - confirmation: None, - }) -} - -fn infer_safety_class(operation: &RuntimeOperation) -> OperationSafetyClass { - match &operation.target { - Target::Rest(target) => match target.method { - HttpMethod::Get => OperationSafetyClass::Read, - HttpMethod::Delete => OperationSafetyClass::Destructive, - HttpMethod::Post | HttpMethod::Put | HttpMethod::Patch => OperationSafetyClass::Write, - }, - } -} - -fn confirmation_ttl_ms(policy: &OperationSafetyPolicy) -> u64 { - policy - .confirmation - .as_ref() - .map(|confirmation| confirmation.ttl_ms) - .filter(|ttl_ms| *ttl_ms > 0) - .unwrap_or(300_000) -} - -fn confirmation_scope( - operation: &RuntimeOperation, - request_context: Option<&RuntimeRequestContext>, -) -> Result { - let Some(scope) = request_context.and_then(RuntimeRequestContext::response_cache_scope) else { - return Err(RuntimeError::InvalidPreparedRequest { - field: "runtime_context.response_cache_scope".to_owned(), - reason: "confirmation requires workspace and agent scope".to_owned(), - }); - }; - - Ok(format!( - "workspace:{}:agent:{}:operation:{}:version:{}", - scope.workspace_key, - scope.agent_key, - operation.operation_id.as_str(), - operation.operation_version - )) -} - -async fn issue_confirmation_token( - store: &dyn CoordinationStateStore, - scope: &str, - input_hash: &str, - safety: &OperationSafetyPolicy, -) -> Result { - let token = format!("ct_{}", Uuid::now_v7().simple()); - let key = confirmation_cache_key(scope, &token); - let ttl_ms = confirmation_ttl_ms(safety); - store - .put_value( - CacheScope::Coordination, - &key, - CoordinationStateValue { - payload: json!({ - "scope": scope, - "input_hash": input_hash, - }), - }, - Duration::from_millis(ttl_ms), - ) - .await - .map_err(|error| RuntimeError::InvalidPreparedRequest { - field: "confirmation_token".to_owned(), - reason: error.to_string(), - })?; - Ok(token) -} - -async fn consume_confirmation_token( - store: &dyn CoordinationStateStore, - operation_scope: &str, - token: &str, - input_hash: &str, -) -> Result<(), RuntimeError> { - let key = confirmation_cache_key(operation_scope, token); - let stored = store - .get_value(CacheScope::Coordination, &key) - .await - .map_err(|error| RuntimeError::InvalidPreparedRequest { - field: "confirmation_token".to_owned(), - reason: error.to_string(), - })?; - let _ = store.delete_value(CacheScope::Coordination, &key).await; - - let Some(stored) = stored else { - return Err(RuntimeError::InvalidConfirmationToken { - operation_id: operation_id_from_scope(operation_scope), - }); - }; - - let matches_scope = - stored.payload.get("scope").and_then(Value::as_str) == Some(operation_scope); - let matches_input = - stored.payload.get("input_hash").and_then(Value::as_str) == Some(input_hash); - - if matches_scope && matches_input { - Ok(()) - } else { - Err(RuntimeError::InvalidConfirmationToken { - operation_id: operation_id_from_scope(operation_scope), - }) - } -} - -fn confirmation_cache_key(scope: &str, token: &str) -> String { - let token_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(token.as_bytes())); - format!("crank:confirmation:{scope}:token:{token_hash}") -} - -fn confirmation_token_from_input(input: &Value) -> Option<&str> { - input - .get("_crank_confirmation_token") - .and_then(Value::as_str) - .filter(|value| !value.trim().is_empty()) -} - -fn hash_json(input: &Value) -> Result { - let sanitized = input_without_confirmation_token(input); - let encoded = - serde_json::to_vec(&sanitized).map_err(|error| RuntimeError::InvalidPreparedRequest { - field: "confirmation_input".to_owned(), - reason: error.to_string(), - })?; - Ok(URL_SAFE_NO_PAD.encode(Sha256::digest(encoded))) -} - -fn input_without_confirmation_token(input: &Value) -> Value { - let Value::Object(object) = input else { - return input.clone(); - }; - let mut sanitized = object.clone(); - sanitized.remove("_crank_confirmation_token"); - Value::Object(sanitized) -} - -fn operation_id_from_scope(scope: &str) -> String { - scope - .split(":operation:") - .nth(1) - .and_then(|tail| tail.split(":version:").next()) - .unwrap_or("") - .to_owned() -} - -fn idempotency_key_from_policy( - policy: &IdempotencyPolicy, - input: &Value, - prepared_request: &PreparedRequest, -) -> Option { - if let Some(header_name) = policy - .header_name - .as_deref() - .filter(|value| !value.is_empty()) - { - if let Some(value) = prepared_request.headers.get(header_name) { - return Some(value.clone()); - } - } - - let field_name = policy.input_field.as_deref()?.trim(); - if field_name.is_empty() { - return None; - } - - input.get(field_name).and_then(|value| match value { - Value::String(value) if !value.trim().is_empty() => Some(value.clone()), - Value::Number(value) => Some(value.to_string()), - _ => None, - }) -} - -fn idempotency_cache_key( - operation: &RuntimeOperation, - idempotency_key: &str, - request_context: Option<&RuntimeRequestContext>, -) -> Option { - if idempotency_key.is_empty() { - return None; - } - - let scope = request_context?.response_cache_scope()?; - let key_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(idempotency_key.as_bytes())); - - Some(format!( - "crank:idempotency:workspace:{}:agent:{}:operation:{}:version:{}:key:{}", - scope.workspace_key, - scope.agent_key, - operation.operation_id.as_str(), - operation.operation_version, - key_hash - )) -} - fn response_cache_key( operation: &RuntimeOperation, prepared_request: &PreparedRequest, diff --git a/crates/crank-runtime/src/idempotency.rs b/crates/crank-runtime/src/idempotency.rs new file mode 100644 index 0000000..8fbe5c5 --- /dev/null +++ b/crates/crank-runtime/src/idempotency.rs @@ -0,0 +1,104 @@ +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use crank_core::{HttpMethod, IdempotencyMode, IdempotencyPolicy, Target}; +use serde_json::Value; +use sha2::{Digest, Sha256}; + +use crate::{PreparedRequest, RuntimeError, RuntimeOperation, RuntimeRequestContext}; + +pub fn prepare_idempotency( + operation: &RuntimeOperation, + input: &Value, + prepared_request: &mut PreparedRequest, +) -> Result, RuntimeError> { + let Some(policy) = policy(operation) else { + return Ok(None); + }; + + let key = key_from_policy(policy, input, prepared_request); + let Some(key) = key else { + if policy.mode == IdempotencyMode::Required { + return Err(RuntimeError::InvalidPreparedRequest { + field: "execution_config.idempotency".to_owned(), + reason: "required idempotency key was not provided".to_owned(), + }); + } + return Ok(None); + }; + + if let Some(header_name) = policy + .header_name + .as_deref() + .filter(|value| !value.is_empty()) + { + prepared_request + .headers + .entry(header_name.to_owned()) + .or_insert_with(|| key.clone()); + } + + Ok(Some(key)) +} + +pub fn policy(operation: &RuntimeOperation) -> Option<&IdempotencyPolicy> { + let is_mutating_rest = + matches!(&operation.target, Target::Rest(target) if target.method != HttpMethod::Get); + if !is_mutating_rest { + return None; + } + + let policy = operation.execution_config.idempotency.as_ref()?; + if policy.mode == IdempotencyMode::Disabled || policy.ttl_ms == 0 { + return None; + } + + Some(policy) +} + +pub fn cache_key( + operation: &RuntimeOperation, + idempotency_key: &str, + request_context: Option<&RuntimeRequestContext>, +) -> Option { + if idempotency_key.is_empty() { + return None; + } + + let scope = request_context?.response_cache_scope()?; + let key_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(idempotency_key.as_bytes())); + + Some(format!( + "crank:idempotency:workspace:{}:agent:{}:operation:{}:version:{}:key:{}", + scope.workspace_key, + scope.agent_key, + operation.operation_id.as_str(), + operation.operation_version, + key_hash + )) +} + +fn key_from_policy( + policy: &IdempotencyPolicy, + input: &Value, + prepared_request: &PreparedRequest, +) -> Option { + if let Some(header_name) = policy + .header_name + .as_deref() + .filter(|value| !value.is_empty()) + { + if let Some(value) = prepared_request.headers.get(header_name) { + return Some(value.clone()); + } + } + + let field_name = policy.input_field.as_deref()?.trim(); + if field_name.is_empty() { + return None; + } + + input.get(field_name).and_then(|value| match value { + Value::String(value) if !value.trim().is_empty() => Some(value.clone()), + Value::Number(value) => Some(value.to_string()), + _ => None, + }) +} diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index 3ce59c3..a1cf57b 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -1,9 +1,11 @@ mod auth; mod cache; mod cache_factory; +mod confirmation; mod error; mod executor; mod executor_builder; +mod idempotency; mod limits; mod model; mod rate_limit;