starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-19 16:02:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(backend): pass all headers and query to webhook preprocessor (#6931)

Browse Source
This commit is contained in:
hugocasa
2025-10-24 21:40:44 +02:00
committed by GitHub
parent 6d403da05a
commit f0bbdd4aa8
3 changed files with 97 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/args.rs
+58 -79
View File
@@ -1,12 +1,12 @@
use std::collections::HashMap;
use axum::{
extract::{FromRequest, FromRequestParts, Multipart, Query, Request},
http::{HeaderMap, Uri},
extract::{FromRequest, Multipart, Query, Request},
http::HeaderMap,
response::{IntoResponse, Response},
};
use bytes::Bytes;
use http::{header::CONTENT_TYPE, request::Parts, StatusCode};
use http::{header::CONTENT_TYPE, StatusCode};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::types::JsonRawValue;
@@ -44,11 +44,13 @@ pub enum Body {
#[derive(Debug, Clone, Default)]
pub struct WebhookArgsMetadata {
pub raw_string: Option<String>,
pub headers: HashMap<String, Box<RawValue>>,
pub headers: HeaderMap,
pub query: Option<String>,
pub method: http::Method,
pub query: HashMap<String, Box<RawValue>>,
pub query_wrap_body: bool,
pub query_use_raw: bool,
pub query_include_header: Option<String>,
pub query_include_query: Option<String>,
}
pub struct RawWebhookArgs {
@@ -262,6 +264,18 @@ impl WebhookArgs {
self,
runnable_format: RunnableFormat,
) -> Result<PushArgsOwned, Error> {
let headers = build_headers(
&self.metadata.headers,
self.metadata.query_include_header,
runnable_format.has_preprocessor,
);
let query = build_query(
self.metadata.query.as_deref(),
self.metadata.query_include_query,
runnable_format.has_preprocessor,
);
match runnable_format {
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V2 } => {
let mut args = HashMap::new();
@@ -272,8 +286,8 @@ impl WebhookArgs {
kind: "webhook".to_string(),
body: to_raw_value(&self.body),
raw_string: self.metadata.raw_string,
headers: self.metadata.headers,
query: self.metadata.query,
headers,
query,
}),
);
@@ -282,8 +296,7 @@ impl WebhookArgs {
RunnableFormat { has_preprocessor, .. } => {
let mut extra = HashMap::new();
let WebhookArgsMetadata { query, query_wrap_body, headers, raw_string, .. } =
self.metadata;
let WebhookArgsMetadata { query_wrap_body, raw_string, .. } = self.metadata;
for (k, v) in headers {
extra.insert(k, v);
@@ -332,6 +345,7 @@ pub struct RequestQuery {
pub raw: Option<bool>,
pub wrap_body: Option<bool>,
pub include_header: Option<String>,
pub include_query: Option<String>,
}
async fn req_to_string<S: Send + Sync>(
@@ -359,23 +373,21 @@ where
let content_type = content_type_header.and_then(|value| value.to_str().ok());
let uri = request.uri();
let request_query = Query::<RequestQuery>::try_from_uri(uri).unwrap().0;
let headers = build_headers(&headers_map, request_query.include_header, is_http_trigger);
let query_decode = DecodeQueries::from_uri(uri, is_http_trigger);
let mut query = HashMap::new();
if let Some(DecodeQueries(queries)) = query_decode {
query.extend(queries);
}
let query = uri.query().map(|s| s.to_owned());
let raw = !is_http_trigger && request_query.raw.unwrap_or(false);
let wrap_body = !is_http_trigger && request_query.wrap_body.unwrap_or(false);
(
content_type,
WebhookArgsMetadata {
headers,
headers: headers_map.clone(),
query,
method: request.method().clone(),
raw_string: None,
query_wrap_body: wrap_body,
query_use_raw: raw,
query_include_header: request_query.include_header,
query_include_query: request_query.include_query,
},
)
};
@@ -460,11 +472,11 @@ lazy_static::lazy_static! {
pub fn build_headers(
headers: &HeaderMap,
include_header: Option<String>,
is_http_trigger: bool,
include_all_headers: bool,
) -> HashMap<String, Box<RawValue>> {
let mut selected_headers = HashMap::new();
if is_http_trigger {
if include_all_headers {
for (k, v) in headers.iter() {
selected_headers.insert(
k.to_string(),
@@ -490,73 +502,40 @@ pub fn build_headers(
selected_headers
}
#[derive(Deserialize)]
pub struct IncludeQuery {
pub include_query: Option<String>,
}
pub fn build_query(
query: Option<&str>,
include_query: Option<String>,
include_all_query: bool,
) -> HashMap<String, Box<RawValue>> {
let Some(query) = query else {
return HashMap::new();
};
pub struct DecodeQueries(pub HashMap<String, Box<RawValue>>);
#[axum::async_trait]
impl<S> FromRequestParts<S> for DecodeQueries
where
S: Send + Sync,
{
type Rejection = Response;
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
Ok(DecodeQueries::from_uri(&parts.uri, false)
.unwrap_or_else(|| DecodeQueries(HashMap::new())))
}
}
impl DecodeQueries {
pub fn from_uri(uri: &Uri, is_http_trigger: bool) -> Option<Self> {
let query = uri.query();
if query.is_none() {
return None;
}
let query = query.unwrap();
if is_http_trigger {
if include_all_query {
let queries =
serde_urlencoded::from_str::<HashMap<String, String>>(&query).unwrap_or_default();
queries
.into_iter()
.map(|(k, v)| (k, to_raw_value(&v)))
.collect()
} else {
let parse_query_args = include_query
.map(|s| s.split(",").map(|p| p.to_string()).collect::<Vec<_>>())
.unwrap_or_default();
let mut args = HashMap::new();
if !parse_query_args.is_empty() {
let queries =
serde_urlencoded::from_str::<HashMap<String, String>>(query).unwrap_or_default();
Some(DecodeQueries(
queries
.into_iter()
.map(|(k, v)| (k, to_raw_value(&v)))
.collect(),
))
} else {
let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
.map(|x| x.include_query)
.ok()
.flatten()
.unwrap_or_default();
let parse_query_args = include_query
.split(",")
.map(|s| s.to_string())
.collect::<Vec<_>>();
let mut args = HashMap::new();
if !parse_query_args.is_empty() {
let queries = serde_urlencoded::from_str::<HashMap<String, String>>(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))
serde_urlencoded::from_str::<HashMap<String, String>>(&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));
}
});
}
args
}
}
// impl<'c> PushArgs<'c> {
// pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
// self.extra.insert(k.into(), v.into());
// }
// }
fn restructure_cloudevents_metadata(
mut p: HashMap<String, Box<RawValue>>,
) -> Result<HashMap<String, Box<RawValue>>, Error> {
backend/windmill-api/src/capture.rs
+13 -2
View File
@@ -398,7 +398,7 @@ async fn set_gcp_trigger_config(
gcp_config.create_update,
false,
capture_config.is_flow,
gcp_config.ack_deadline
gcp_config.ack_deadline,
)
.await?;
gcp_config.create_update = Some(config);
@@ -966,6 +966,8 @@ async fn http_payload(
Path((w_id, runnable_kind, path, route_path)): Path<(String, RunnableKind, String, StripPath)>,
args: RawHttpTriggerArgs,
) -> std::result::Result<StatusCode, Response> {
use crate::args::{build_headers, build_query};
let path = path.replace(".", "/");
let is_flow = matches!(runnable_kind, RunnableKind::Flow);
let route_path = route_path.to_path();
@@ -1002,9 +1004,18 @@ async fn http_payload(
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let headers = build_headers(&args.0.metadata.headers, None, true);
let query = build_query(args.0.metadata.query.as_deref(), None, true);
let preprocessor_args = args
.clone()
.to_v2_preprocessor_args(&http_trigger_config.route_path, &route_path, &params)
.to_v2_preprocessor_args(
&http_trigger_config.route_path,
&route_path,
&params,
headers,
query,
)
.map_err(|e| e.into_response())?;
let main_args = args
backend/windmill-api/src/triggers/http/http_trigger_args.rs
+26 -11
View File
@@ -15,7 +15,10 @@ use windmill_common::{
use windmill_queue::PushArgsOwned;
use crate::{
args::{try_from_request_body, Body, RawWebhookArgs, WebhookArgs, WebhookArgsMetadata},
args::{
build_headers, build_query, try_from_request_body, Body, RawWebhookArgs, WebhookArgs,
WebhookArgsMetadata,
},
db::ApiAuthed,
};
@@ -102,8 +105,8 @@ struct HttpTriggerWmTrigger<'a> {
route: &'a str,
path: &'a str,
params: &'a HashMap<String, String>,
query: &'a HashMap<String, Box<RawValue>>,
headers: &'a HashMap<String, Box<RawValue>>,
query: HashMap<String, Box<RawValue>>,
headers: HashMap<String, Box<RawValue>>,
method: HttpMethod,
}
@@ -142,14 +145,22 @@ impl HttpTriggerArgs {
format: RunnableFormat,
wrap_body: bool,
) -> Result<PushArgsOwned, Error> {
let headers = build_headers(&self.0.metadata.headers, None, true);
let query = build_query(self.0.metadata.query.as_deref(), None, true);
match format {
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V2 } => {
// we don't care about wrap_body in v2
self.to_v2_preprocessor_args(route_path, called_path, params)
}
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V1 } => {
self.to_v1_preprocessor_args(route_path, called_path, params, wrap_body)
self.to_v2_preprocessor_args(route_path, called_path, params, headers, query)
}
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V1 } => self
.to_v1_preprocessor_args(
route_path,
called_path,
params,
wrap_body,
headers,
query,
),
RunnableFormat { has_preprocessor: false, .. } => self.to_main_args(wrap_body),
}
}
@@ -160,6 +171,8 @@ impl HttpTriggerArgs {
called_path: &str,
params: &HashMap<String, String>,
wrap_body: bool,
headers: HashMap<String, Box<RawValue>>,
query: HashMap<String, Box<RawValue>>,
) -> Result<PushArgsOwned, Error> {
let mut extra = HashMap::new();
let mut wm_trigger = HashMap::new();
@@ -171,8 +184,8 @@ impl HttpTriggerArgs {
path: called_path,
method: (&self.0.metadata.method).try_into()?,
params,
query: &self.0.metadata.query,
headers: &self.0.metadata.headers,
query,
headers,
}),
);
extra.insert("wm_trigger".to_string(), to_raw_value(&wm_trigger));
@@ -189,6 +202,8 @@ impl HttpTriggerArgs {
route_path: &str,
called_path: &str,
params: &HashMap<String, String>,
headers: HashMap<String, Box<RawValue>>,
query: HashMap<String, Box<RawValue>>,
) -> Result<PushArgsOwned, Error> {
let mut args = HashMap::new();
args.insert(
@@ -197,8 +212,8 @@ impl HttpTriggerArgs {
kind: "http".to_string(),
body: to_raw_value(&self.0.body),
raw_string: self.0.metadata.raw_string,
headers: self.0.metadata.headers,
query: self.0.metadata.query,
headers,
query,
method: (&self.0.metadata.method).try_into()?,
route: route_path,
path: called_path,