diff --git a/backend/windmill-api/src/args.rs b/backend/windmill-api/src/args.rs new file mode 100644 index 0000000000..5952a5ee5f --- /dev/null +++ b/backend/windmill-api/src/args.rs @@ -0,0 +1,469 @@ +use std::collections::HashMap; + +#[cfg(feature = "parquet")] +use crate::job_helpers_ee::get_workspace_s3_resource; +use axum::{ + extract::{FromRequest, FromRequestParts, Query, Request}, + http::{HeaderMap, Uri}, + response::{IntoResponse, Response}, +}; +use bytes::Bytes; +use http::{header::CONTENT_TYPE, request::Parts}; +#[cfg(feature = "parquet")] +use object_store::{Attribute, Attributes}; +use serde::Deserialize; +use serde_json::value::RawValue; +use sqlx::types::JsonRawValue; +#[cfg(feature = "parquet")] +use windmill_common::s3_helpers::build_object_store_client; +use windmill_common::{error::Error, worker::to_raw_value, DB}; +use windmill_queue::PushArgsOwned; + +use crate::db::ApiAuthed; +#[cfg(feature = "parquet")] +use crate::job_helpers_ee::{get_random_file_name, upload_file_internal}; + +pub struct WebhookArgs { + pub args: PushArgsOwned, + pub file_req: Option, +} + +impl WebhookArgs { + #[cfg(not(feature = "parquet"))] + pub async fn to_push_args_owned( + self, + _authed: &ApiAuthed, + _db: &DB, + _w_id: &str, + ) -> Result { + return Err(Error::BadRequest(format!( + "Uploading files requires the parquet feature" + ))); + } + + #[cfg(feature = "parquet")] + pub async fn to_push_args_owned( + mut self, + authed: &ApiAuthed, + db: &DB, + w_id: &str, + ) -> Result { + if let Some(req) = self.file_req { + { + let (_, s3_resource) = + get_workspace_s3_resource(authed, db, None, "", w_id, None).await?; + + if let Some(s3_resource) = s3_resource { + let s3_client = build_object_store_client(&s3_resource).await?; + + let content_type = req + .headers() + .get(CONTENT_TYPE) + .map(|x| x.to_str().ok().map(|x| x.to_string())) + .flatten(); + + let file_extension = content_type + .as_ref() + .map(|mime_str| { + mime_guess::get_mime_extensions_str(mime_str) + .map(|x| x.first().map(|x| x.to_string())) + }) + .flatten() + .flatten(); + + let file_key = get_random_file_name(file_extension); + + let options = Attributes::from_iter(vec![ + ( + Attribute::ContentType, + content_type.unwrap_or("application/octet-stream".to_string()), + ), + (Attribute::ContentDisposition, "inline".to_string()), + ]) + .into(); + + upload_file_internal(s3_client, &file_key, req, options).await?; + + self.args.args.insert( + "body".to_string(), + to_raw_value(&serde_json::json!({ + "s3": &file_key + })), + ); + + return Ok(self.args); + } + } + + return Err(Error::BadRequest(format!( + "You need to connect your workspace to an S3 bucket to upload files" + ))); + } + + Ok(self.args) + } +} + +#[derive(Deserialize)] +pub struct RequestQuery { + pub raw: Option, + pub wrap_body: Option, + pub include_header: Option, +} + +#[axum::async_trait] +impl FromRequest for WebhookArgs +where + S: Send + Sync, +{ + type Rejection = Response; + + async fn from_request( + req: Request, + _state: &S, + ) -> Result { + let (content_type, mut extra, use_raw, wrap_body) = { + let headers_map = req.headers(); + let content_type_header = headers_map.get(CONTENT_TYPE); + let content_type = content_type_header.and_then(|value| value.to_str().ok()); + let uri = req.uri(); + let query = Query::::try_from_uri(uri).unwrap().0; + let mut extra = build_extra(&headers_map, query.include_header); + let query_decode = DecodeQueries::from_uri(uri); + if let Some(DecodeQueries(queries)) = query_decode { + extra.extend(queries); + } + let raw = query.raw.as_ref().is_some_and(|x| *x); + let wrap_body = query.wrap_body.as_ref().is_some_and(|x| *x); + (content_type, extra, raw, wrap_body) + }; + + let no_content_type = content_type.is_none(); + if no_content_type || content_type.unwrap().starts_with("application/json") { + let bytes = Bytes::from_request(req, _state) + .await + .map_err(IntoResponse::into_response)?; + if no_content_type && bytes.is_empty() { + if use_raw { + extra.insert("raw_string".to_string(), to_raw_value(&"".to_string())); + } + let mut args = HashMap::new(); + if wrap_body { + args.insert("body".to_string(), to_raw_value(&serde_json::json!({}))); + } + return Ok(Self { + args: PushArgsOwned { extra: Some(extra), args: args }, + file_req: None, + }); + } + let str = String::from_utf8(bytes.to_vec()) + .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; + + PushArgsOwned::from_json(extra, use_raw, wrap_body, str) + .await + .map(|args| Self { args, file_req: None }) + } else if content_type + .unwrap() + .starts_with("application/cloudevents+json") + { + let bytes = Bytes::from_request(req, _state) + .await + .map_err(IntoResponse::into_response)?; + let str = String::from_utf8(bytes.to_vec()) + .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; + + PushArgsOwned::from_ce_json(extra, use_raw, str) + .await + .map(|args| Self { args, file_req: None }) + } else if content_type + .unwrap() + .starts_with("application/cloudevents-batch+json") + { + Err( + Error::BadRequest(format!("Cloud events batching is not supported yet")) + .into_response(), + ) + } else if content_type.unwrap().starts_with("text/plain") { + let bytes = Bytes::from_request(req, _state) + .await + .map_err(IntoResponse::into_response)?; + let str = String::from_utf8(bytes.to_vec()) + .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; + extra.insert("raw_string".to_string(), to_raw_value(&str)); + Ok(Self { + args: PushArgsOwned { extra: Some(extra), args: HashMap::new() }, + file_req: None, + }) + } else if content_type + .unwrap() + .starts_with("application/x-www-form-urlencoded") + { + let bytes = Bytes::from_request(req, _state) + .await + .map_err(IntoResponse::into_response)?; + + if use_raw { + let raw_string = String::from_utf8(bytes.to_vec()).map_err(|e| { + Error::BadRequest(format!("invalid utf8: {}", e)).into_response() + })?; + extra.insert("raw_string".to_string(), to_raw_value(&raw_string)); + } + + let payload: HashMap> = serde_urlencoded::from_bytes(&bytes) + .map_err(|e| { + Error::BadRequest(format!("invalid urlencoded data: {}", e)).into_response() + })?; + let payload = payload + .into_iter() + .map(|(k, v)| (k, to_raw_value(&v))) + .collect::>(); + + return Ok(Self { + args: PushArgsOwned { extra: Some(extra), args: payload }, + file_req: None, + }); + } else { + return Ok(Self { + args: PushArgsOwned { extra: None, args: HashMap::new() }, + file_req: Some(req), + }); + + // Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) + } + } +} + +lazy_static::lazy_static! { + static ref INCLUDE_HEADERS: Vec = std::env::var("INCLUDE_HEADERS") + .ok().map(|x| x + .split(',') + .map(|s| s.to_string()) + .collect()).unwrap_or_default(); +} + +pub fn build_extra( + headers: &HeaderMap, + include_header: Option, +) -> HashMap> { + let mut args = HashMap::new(); + let whitelist = include_header + .map(|s| s.split(",").map(|s| s.to_string()).collect::>()) + .unwrap_or_default(); + + whitelist + .iter() + .chain(INCLUDE_HEADERS.iter()) + .for_each(|h| { + if let Some(v) = headers.get(h) { + args.insert( + h.to_string().to_lowercase().replace('-', "_"), + to_raw_value(&v.to_str().unwrap().to_string()), + ); + } + }); + args +} + +#[derive(Deserialize)] +pub struct IncludeQuery { + pub include_query: Option, +} + +pub struct DecodeQueries(pub HashMap>); + +#[axum::async_trait] +impl FromRequestParts for DecodeQueries +where + S: Send + Sync, +{ + type Rejection = Response; + + async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { + Ok(DecodeQueries::from_uri(&parts.uri).unwrap_or_else(|| DecodeQueries(HashMap::new()))) + } +} + +impl DecodeQueries { + pub fn from_uri(uri: &Uri) -> Option { + let query = uri.query(); + if query.is_none() { + return None; + } + let query = query.unwrap(); + let include_query = serde_urlencoded::from_str::(query) + .map(|x| x.include_query) + .ok() + .flatten() + .unwrap_or_default(); + let parse_query_args = include_query + .split(",") + .map(|s| s.to_string()) + .collect::>(); + let mut args = HashMap::new(); + if !parse_query_args.is_empty() { + let queries = + serde_urlencoded::from_str::>(query).unwrap_or_default(); + parse_query_args.iter().for_each(|h| { + if let Some(v) = queries.get(h) { + args.insert(h.to_string(), to_raw_value(v)); + } + }); + } + Some(DecodeQueries(args)) + } +} + +// impl<'c> PushArgs<'c> { +// pub fn insert, V: Into>>(&mut self, k: K, v: V) { +// self.extra.insert(k.into(), v.into()); +// } +// } + +fn restructure_cloudevents_metadata( + mut p: HashMap>, +) -> Result>, Error> { + let data = p + .remove("data") + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); + let str = data.to_string(); + + let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{'; + + if wrap_body { + let args = serde_json::from_str::>>(&str) + .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); + let mut hm = HashMap::new(); + hm.insert("body".to_string(), args); + hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); + Ok(hm) + } else { + let mut hm = serde_json::from_str::>>>(&str) + .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? + .unwrap_or_else(HashMap::new); + hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); + Ok(hm) + } +} + +trait PushArgsOwnedExt: Sized { + async fn from_json( + extra: HashMap>, + use_raw: bool, + force_wrap_body: bool, + str: String, + ) -> Result; + + async fn from_ce_json( + extra: HashMap>, + use_raw: bool, + str: String, + ) -> Result; +} + +impl PushArgsOwnedExt for PushArgsOwned { + async fn from_json( + mut extra: HashMap>, + use_raw: bool, + force_wrap_body: bool, + str: String, + ) -> Result { + if use_raw { + extra.insert("raw_string".to_string(), to_raw_value(&str)); + } + + let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{'; + + if wrap_body { + let args = serde_json::from_str::>>(&str) + .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())? + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); + let mut hm = HashMap::new(); + hm.insert("body".to_string(), args); + Ok(PushArgsOwned { extra: Some(extra), args: hm }) + } else { + let hm = serde_json::from_str::>>>(&str) + .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())? + .unwrap_or_else(HashMap::new); + Ok(PushArgsOwned { extra: Some(extra), args: hm }) + } + } + + async fn from_ce_json( + mut extra: HashMap>, + use_raw: bool, + str: String, + ) -> Result { + if use_raw { + extra.insert("raw_string".to_string(), to_raw_value(&str)); + } + + let hm = serde_json::from_str::>>(&str).map_err(|e| { + Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response() + })?; + let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?; + Ok(PushArgsOwned { extra: Some(extra), args: hm }) + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + + #[tokio::test] + async fn test_cloudevents_json_payload() { + let r1 = r#" + { + "specversion" : "1.0", + "type" : "com.example.someevent", + "source" : "/mycontext", + "subject": null, + "id" : "C234-1234-1234", + "time" : "2018-04-05T17:31:00Z", + "comexampleextension1" : "value", + "comexampleothervalue" : 5, + "datacontenttype" : "application/json", + "data" : { + "appinfoA" : "abc", + "appinfoB" : 123, + "appinfoC" : true + } + } + "#; + let r2 = r#" + { + "specversion" : "1.0", + "type" : "com.example.someevent", + "source" : "/mycontext", + "subject": null, + "id" : "C234-1234-1234", + "time" : "2018-04-05T17:31:00Z", + "comexampleextension1" : "value", + "comexampleothervalue" : 5, + "datacontenttype" : "application/json", + "data" : 1.5 + } + "#; + let extra = HashMap::new(); + + let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string()) + .await + .expect("Failed to parse the cloudevent"); + let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string()) + .await + .expect("Failed to parse the cloudevent"); + + a1.args.get("WEBHOOK__METADATA__").expect( + "CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs", + ); + assert_eq!( + a2.args + .get("body") + .expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs") + .to_string(), + "1.5" + ); + } +} diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 241469d05a..7043a5c58a 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -18,9 +18,12 @@ use windmill_common::{ error::{JsonResult, Result}, utils::{not_found_if_none, StripPath}, }; -use windmill_queue::{PushArgs, PushArgsOwned}; +use windmill_queue::PushArgs; -use crate::db::{ApiAuthed, DB}; +use crate::{ + args::WebhookArgs, + db::{ApiAuthed, DB}, +}; const KEEP_LAST: i64 = 8; @@ -86,8 +89,10 @@ pub async fn new_payload( pub async fn update_payload( Extension(db): Extension, Path((w_id, path)): Path<(String, StripPath)>, - args: PushArgsOwned, + args: WebhookArgs, ) -> Result { + let args = args.args; + let mut tx = db.begin().await?; sqlx::query!( diff --git a/backend/windmill-api/src/http_triggers.rs b/backend/windmill-api/src/http_triggers.rs index ca03d5803c..529692fe3b 100644 --- a/backend/windmill-api/src/http_triggers.rs +++ b/backend/windmill-api/src/http_triggers.rs @@ -1,6 +1,7 @@ #[cfg(feature = "parquet")] use crate::job_helpers_ee::get_workspace_s3_resource; use crate::{ + args::WebhookArgs, db::{ApiAuthed, DB}, jobs::{ run_flow_by_path_inner, run_script_by_path_inner, run_wait_result_flow_by_path_internal, @@ -32,7 +33,6 @@ use windmill_common::{ utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath}, worker::{to_raw_value, CLOUD_HOSTED}, }; -use windmill_queue::PushArgsOwned; lazy_static::lazy_static! { static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/:\w+").unwrap(); @@ -531,7 +531,7 @@ async fn route_job( Query(query): Query>, method: http::Method, headers: HeaderMap, - mut args: PushArgsOwned, + args: WebhookArgs, ) -> impl IntoResponse { let route_path = route_path.to_path(); let (trigger, called_path, params, authed) = match get_http_route_trigger( @@ -547,6 +547,14 @@ async fn route_job( Err(e) => return e.into_response(), }; + let mut args = match args + .to_push_args_owned(&authed, &db, &trigger.workspace_id) + .await + { + Ok(args) => args, + Err(e) => return e.into_response(), + }; + #[cfg(not(feature = "parquet"))] if trigger.static_asset_config.is_some() { return error::Error::InternalErr( diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 034aaa932c..622a08f985 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -36,6 +36,7 @@ use crate::db::ApiAuthed; use crate::users::get_scope_tags; use crate::utils::content_plain; use crate::{ + args::{DecodeQueries, WebhookArgs}, db::DB, users::{check_scopes, require_owner_of_path, OptAuthed}, utils::require_super_admin, @@ -83,7 +84,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED}; use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL}; use windmill_queue::{ cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push, - DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel, + PushArgs, PushArgsOwned, PushIsolationLevel, }; #[cfg(feature = "prometheus")] @@ -631,7 +632,10 @@ async fn get_flow_job_debug_info( } } for job_id in job_ids { - let job = GetQuery::new().with_auth(&opt_authed).fetch(&db, job_id, &w_id).await; + let job = GetQuery::new() + .with_auth(&opt_authed) + .fetch(&db, job_id, &w_id) + .await; if let Ok(job) = job { jobs.insert(job.id().to_string(), job); } @@ -832,7 +836,10 @@ impl<'a> GetQuery<'a> { self.check_auth(cjob.as_ref().map(|job| job.created_by.as_str()))?; if self.with_flow { - cjob = resolve_maybe_value(db, workspace_id, self.with_code, cjob, |job| job.raw_flow.as_mut()).await?; + cjob = resolve_maybe_value(db, workspace_id, self.with_code, cjob, |job| { + job.raw_flow.as_mut() + }) + .await?; } if let Some(mut cjob) = cjob { cjob.inner = format_completed_job_result(cjob.inner); @@ -842,14 +849,16 @@ impl<'a> GetQuery<'a> { } async fn fetch(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result { - let cjob = self.fetch_completed(db, job_id, workspace_id) + let cjob = self + .fetch_completed(db, job_id, workspace_id) .await? .map(Job::CompletedJob); match cjob { Some(cjob) => Ok(cjob), None => { - let job_maybe = self.fetch_queued(db, job_id, workspace_id) + let job_maybe = self + .fetch_queued(db, job_id, workspace_id) .await? .map(Job::QueuedJob); not_found_if_none(job_maybe, "Job", job_id.to_string()) @@ -2940,8 +2949,10 @@ pub async fn run_flow_by_path( Extension(user_db): Extension, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result<(StatusCode, String)> { + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; + run_flow_by_path_inner(authed, db, user_db, w_id, flow_path, run_query, args, None).await } @@ -3116,8 +3127,9 @@ pub async fn run_script_by_path( Extension(user_db): Extension, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result<(StatusCode, String)> { + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; run_script_by_path_inner( authed, db, @@ -3218,7 +3230,10 @@ pub async fn run_workflow_as_code( i += 1; } - let job = GetQuery::new().without_logs().fetch_queued(&db, job_id, &w_id).await?; + let job = GetQuery::new() + .without_logs() + .fetch_queued(&db, job_id, &w_id) + .await?; if *CLOUD_HOSTED { tracing::info!("workflow_as_code_tracing id {i} "); @@ -3779,11 +3794,13 @@ pub async fn run_wait_result_script_by_path( Extension(db): Extension, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; + run_wait_result_script_by_path_internal( db, run_query, @@ -3861,11 +3878,13 @@ pub async fn run_wait_result_script_by_hash( Extension(db): Extension, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; + check_queue_too_long(&db, run_query.queue_limit).await?; let hash = script_hash.0; @@ -3945,11 +3964,13 @@ pub async fn run_wait_result_flow_by_path( Extension(db): Extension, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result { #[cfg(feature = "enterprise")] check_license_key_valid().await?; + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; + run_wait_result_flow_by_path_internal( db, run_query, flow_path, authed, user_db, args, w_id, None, ) @@ -4692,12 +4713,12 @@ async fn run_preview_flow_job( pub async fn run_job_by_hash( authed: ApiAuthed, Extension(db): Extension, - Extension(user_db): Extension, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, - args: PushArgsOwned, + args: WebhookArgs, ) -> error::Result<(StatusCode, String)> { + let args = args.to_push_args_owned(&authed, &db, &w_id).await?; run_job_by_hash_inner( authed, db, @@ -4747,6 +4768,7 @@ pub async fn run_job_by_hash_inner( let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&w_id, &tag, &authed).await?; + let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into()); let (uuid, tx) = push( diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 88243789b4..088707327b 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -46,6 +46,7 @@ use windmill_common::error::AppError; mod ai; mod apps; +mod args; mod audit; mod capture; mod concurrency_groups; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4b442588ed..e88a7c198c 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -10,23 +10,14 @@ use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec}; use anyhow::Context; use async_recursion::async_recursion; -use axum::{ - body::Bytes, - extract::{FromRequest, FromRequestParts, Query}, - http::{request::Parts, Request, Uri}, - response::{IntoResponse, Response}, -}; use chrono::{DateTime, Duration, Utc}; use futures::future::TryFutureExt; use itertools::Itertools; #[cfg(feature = "prometheus")] use prometheus::IntCounter; use regex::Regex; -use reqwest::{ - header::{HeaderMap, CONTENT_TYPE}, - Client, StatusCode, -}; -use serde::{ser::SerializeMap, Deserialize, Serialize}; +use reqwest::Client; +use serde::{ser::SerializeMap, Serialize}; use serde_json::{json, value::RawValue}; use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction}; #[cfg(feature = "benchmark")] @@ -2569,340 +2560,6 @@ impl<'c> Serialize for PushArgs<'c> { } } -#[derive(Deserialize)] -pub struct DecodeQuery { - pub include_query: Option, -} - -#[derive(Deserialize)] -pub struct IncludeQuery { - pub include_query: Option, -} - -pub struct DecodeQueries(pub HashMap>); - -#[axum::async_trait] -impl FromRequestParts for DecodeQueries -where - S: Send + Sync, -{ - type Rejection = Response; - - async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { - Ok(DecodeQueries::from_uri(&parts.uri).unwrap_or_else(|| DecodeQueries(HashMap::new()))) - } -} - -impl DecodeQueries { - fn from_uri(uri: &Uri) -> Option { - let query = uri.query(); - if query.is_none() { - return None; - } - let query = query.unwrap(); - let include_query = serde_urlencoded::from_str::(query) - .map(|x| x.include_query) - .ok() - .flatten() - .unwrap_or_default(); - let parse_query_args = include_query - .split(",") - .map(|s| s.to_string()) - .collect::>(); - let mut args = HashMap::new(); - if !parse_query_args.is_empty() { - let queries = - serde_urlencoded::from_str::>(query).unwrap_or_default(); - parse_query_args.iter().for_each(|h| { - if let Some(v) = queries.get(h) { - args.insert(h.to_string(), to_raw_value(v)); - } - }); - } - Some(DecodeQueries(args)) - } -} - -// impl<'c> PushArgs<'c> { -// pub fn insert, V: Into>>(&mut self, k: K, v: V) { -// self.extra.insert(k.into(), v.into()); -// } -// } - -#[derive(Deserialize)] -pub struct RequestQuery { - pub raw: Option, - pub wrap_body: Option, - pub include_header: Option, -} - -fn restructure_cloudevents_metadata( - mut p: HashMap>, -) -> Result>, Error> { - let data = p - .remove("data") - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - let str = data.to_string(); - - let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{'; - - if wrap_body { - let args = serde_json::from_str::>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - let mut hm = HashMap::new(); - hm.insert("body".to_string(), args); - hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); - Ok(hm) - } else { - let mut hm = serde_json::from_str::>>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(HashMap::new); - hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); - Ok(hm) - } -} - -impl PushArgsOwned { - async fn from_json( - mut extra: HashMap>, - use_raw: bool, - force_wrap_body: bool, - str: String, - ) -> Result { - if use_raw { - extra.insert("raw_string".to_string(), to_raw_value(&str)); - } - - let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{'; - - if wrap_body { - let args = serde_json::from_str::>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())? - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - let mut hm = HashMap::new(); - hm.insert("body".to_string(), args); - Ok(PushArgsOwned { extra: Some(extra), args: hm }) - } else { - let hm = serde_json::from_str::>>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())? - .unwrap_or_else(HashMap::new); - Ok(PushArgsOwned { extra: Some(extra), args: hm }) - } - } - - async fn from_ce_json( - mut extra: HashMap>, - use_raw: bool, - str: String, - ) -> Result { - if use_raw { - extra.insert("raw_string".to_string(), to_raw_value(&str)); - } - - let hm = serde_json::from_str::>>(&str).map_err(|e| { - Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response() - })?; - let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?; - Ok(PushArgsOwned { extra: Some(extra), args: hm }) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_cloudevents_json_payload() { - let r1 = r#" - { - "specversion" : "1.0", - "type" : "com.example.someevent", - "source" : "/mycontext", - "subject": null, - "id" : "C234-1234-1234", - "time" : "2018-04-05T17:31:00Z", - "comexampleextension1" : "value", - "comexampleothervalue" : 5, - "datacontenttype" : "application/json", - "data" : { - "appinfoA" : "abc", - "appinfoB" : 123, - "appinfoC" : true - } - } - "#; - let r2 = r#" - { - "specversion" : "1.0", - "type" : "com.example.someevent", - "source" : "/mycontext", - "subject": null, - "id" : "C234-1234-1234", - "time" : "2018-04-05T17:31:00Z", - "comexampleextension1" : "value", - "comexampleothervalue" : 5, - "datacontenttype" : "application/json", - "data" : 1.5 - } - "#; - let extra = HashMap::new(); - - let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string()) - .await - .expect("Failed to parse the cloudevent"); - let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string()) - .await - .expect("Failed to parse the cloudevent"); - - a1.args.get("WEBHOOK__METADATA__").expect( - "CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs", - ); - assert_eq!( - a2.args - .get("body") - .expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs") - .to_string(), - "1.5" - ); - } -} - -#[axum::async_trait] -impl FromRequest for PushArgsOwned -where - S: Send + Sync, -{ - type Rejection = Response; - - async fn from_request( - req: Request, - _state: &S, - ) -> Result { - let (content_type, mut extra, use_raw, wrap_body) = { - let headers_map = req.headers(); - let content_type_header = headers_map.get(CONTENT_TYPE); - let content_type = content_type_header.and_then(|value| value.to_str().ok()); - let uri = req.uri(); - let query = Query::::try_from_uri(uri).unwrap().0; - let mut extra = build_extra(&headers_map, query.include_header); - let query_decode = DecodeQueries::from_uri(uri); - if let Some(DecodeQueries(queries)) = query_decode { - extra.extend(queries); - } - let raw = query.raw.as_ref().is_some_and(|x| *x); - let wrap_body = query.wrap_body.as_ref().is_some_and(|x| *x); - (content_type, extra, raw, wrap_body) - }; - - let no_content_type = content_type.is_none(); - if no_content_type || content_type.unwrap().starts_with("application/json") { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - if no_content_type && bytes.is_empty() { - if use_raw { - extra.insert("raw_string".to_string(), to_raw_value(&"".to_string())); - } - let mut args = HashMap::new(); - if wrap_body { - args.insert("body".to_string(), to_raw_value(&serde_json::json!({}))); - } - return Ok(PushArgsOwned { extra: Some(extra), args: args }); - } - let str = String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - - PushArgsOwned::from_json(extra, use_raw, wrap_body, str).await - } else if content_type - .unwrap() - .starts_with("application/cloudevents+json") - { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - let str = String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - - PushArgsOwned::from_ce_json(extra, use_raw, str).await - } else if content_type - .unwrap() - .starts_with("application/cloudevents-batch+json") - { - Err( - Error::BadRequest(format!("Cloud events batching is not supported yet")) - .into_response(), - ) - } else if content_type.unwrap().starts_with("text/plain") { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - let str = String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - extra.insert("raw_string".to_string(), to_raw_value(&str)); - Ok(PushArgsOwned { extra: Some(extra), args: HashMap::new() }) - } else if content_type - .unwrap() - .starts_with("application/x-www-form-urlencoded") - { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - - if use_raw { - let raw_string = String::from_utf8(bytes.to_vec()).map_err(|e| { - Error::BadRequest(format!("invalid utf8: {}", e)).into_response() - })?; - extra.insert("raw_string".to_string(), to_raw_value(&raw_string)); - } - - let payload: HashMap> = serde_urlencoded::from_bytes(&bytes) - .map_err(|e| { - Error::BadRequest(format!("invalid urlencoded data: {}", e)).into_response() - })?; - let payload = payload - .into_iter() - .map(|(k, v)| (k, to_raw_value(&v))) - .collect::>(); - - return Ok(PushArgsOwned { extra: Some(extra), args: payload }); - } else { - Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) - } - } -} - -lazy_static::lazy_static! { - static ref INCLUDE_HEADERS: Vec = std::env::var("INCLUDE_HEADERS") - .ok().map(|x| x - .split(',') - .map(|s| s.to_string()) - .collect()).unwrap_or_default(); -} - -pub fn build_extra( - headers: &HeaderMap, - include_header: Option, -) -> HashMap> { - let mut args = HashMap::new(); - let whitelist = include_header - .map(|s| s.split(",").map(|s| s.to_string()).collect::>()) - .unwrap_or_default(); - - whitelist - .iter() - .chain(INCLUDE_HEADERS.iter()) - .for_each(|h| { - if let Some(v) = headers.get(h) { - args.insert( - h.to_string().to_lowercase().replace('-', "_"), - to_raw_value(&v.to_str().unwrap().to_string()), - ); - } - }); - args -} - impl PushArgsOwned { pub fn empty() -> Self { PushArgsOwned { extra: None, args: HashMap::new() }