use crate::pipeline::model::{
DslFlag, DslFlagKind, LayoutItem, NodeFieldDataSource, NodeFieldDef, NodeFieldType,
SelectOptionDef,
};
use crate::pipeline::{
NodeDefinition, PipelineError,
nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
pub const NODE_KIND: &str = "n.trigger.webhook";
pub const OUTPUT_PIN_OUT: &str = "out";
pub fn definition() -> NodeDefinition {
NodeDefinition {
kind: NODE_KIND.to_string(),
title: "Webhook Trigger".to_string(),
description: "Start pipeline run from inbound HTTP path + method. \
User-submitted data is namespaced under input.body to prevent collisions with request context: \
application/json → parsed value at input.body; \
application/x-www-form-urlencoded → form fields at input.body (percent-decoded); \
multipart/form-data → text fields at input.body, files under input.files.{field} as FileRef metadata; repeated fields, field[], and field[0] become arrays. \
GET requests → input.body is null. \
Request context is at root: input.query (URL query params), input.params (path params), \
input.path (request path), input.method (HTTP method). \
Also available via $trigger: $trigger.query, $trigger.params, $trigger.auth, $trigger.headers. \
Use --auth-type jwt/hmac/api_key and --auth-credential <id> to protect the route. \
jwt auth checks Authorization: Bearer header first, then Cookie: zebflow_session fallback — \
verified claims are injected into input.auth. \
Streaming: clients that send Accept: text/event-stream receive an SSE stream \
instead of a single response. Nodes that emit signals via the ExecutionBus \
(or return __signal in their output) are forwarded as event: signal messages. \
The final pipeline result is sent as event: done (or event: error) and the \
stream closes. The pipeline definition is identical for both modes — the \
Accept header is the only switch.".to_string(),
input_schema: serde_json::json!({
"type": "object",
"description": "Structured request payload. User body at input.body (JSON/form/multipart text). Request context at root: input.query, input.params, input.path, input.method. FileRef uploads at input.files.{field}; repeated fields and field[]/field[0] names become arrays. JWT claims at input.auth."
}),
output_schema: serde_json::json!({
"type":"object",
"description":"Unmodified request payload for downstream nodes."
}),
input_pins: vec![],
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
script_available: false,
script_bridge: None,
config_schema: serde_json::json!({
"type": "object",
"required": ["path"],
"properties": {
"path": {
"type": "string",
"description": "HTTP path this webhook listens on, e.g. '/blog' or '/api/users/:id'. Must start with /."
},
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "PATCH", "DELETE"],
"description": "HTTP method. Defaults to GET."
},
"auth_type": {
"type": "string",
"enum": ["none", "jwt", "hmac", "api_key"],
"description": "Authentication mode. none = open (default). jwt/hmac/api_key require auth_credential."
},
"auth_credential": {
"type": "string",
"description": "Credential ID used for auth verification. Required when auth_type is not none."
},
"auth_required_role": {
"type": "array",
"items": { "type": "string" },
"description": "Required roles for this route. JWT claim 'role' must match one of these. Empty = any authenticated user."
}
}
}),
fields: vec![
NodeFieldDef { name: "method".to_string(), label: "Method".to_string(), field_type: NodeFieldType::MethodButtons, options: vec!["GET","POST","PUT","PATCH","DELETE"].iter().map(|m| SelectOptionDef { value: m.to_string(), label: m.to_string() }).collect(), help: Some("HTTP method accepted by webhook trigger.".to_string()), ..Default::default() },
NodeFieldDef { name: "path".to_string(), label: "Path".to_string(), field_type: NodeFieldType::Text, help: Some("Webhook relative path under /wh/{owner}/{project}.".to_string()), ..Default::default() },
NodeFieldDef { name: "__webhook_public_url".to_string(), label: "Public URL".to_string(), field_type: NodeFieldType::CopyUrl, help: Some("Copy-ready URL for this trigger.".to_string()), ..Default::default() },
NodeFieldDef { name: "auth_type".to_string(), label: "Auth Type".to_string(), field_type: NodeFieldType::Select, options: vec![
SelectOptionDef { value: "none".to_string(), label: "None (public)".to_string() },
SelectOptionDef { value: "jwt".to_string(), label: "JWT Bearer".to_string() },
SelectOptionDef { value: "hmac".to_string(), label: "HMAC Signature".to_string() },
SelectOptionDef { value: "api_key".to_string(), label: "API Key (X-API-Key)".to_string() },
], help: Some("Trigger-level auth. On failure returns 401.".to_string()), ..Default::default() },
NodeFieldDef { name: "auth_credential".to_string(), label: "Auth Credential".to_string(), field_type: NodeFieldType::Select, data_source: Some(NodeFieldDataSource::CredentialsWebhookAuth), help: Some("Credential for signing key / secret / api_key.".to_string()), ..Default::default() },
NodeFieldDef { name: "auth_required_role".to_string(), label: "Required Role".to_string(), field_type: NodeFieldType::MultiCheckbox, data_source: Some(NodeFieldDataSource::CredentialJwtRoles), help: Some("Roles allowed to access this route. Populated from the selected JWT credential's registered roles. Empty = any authenticated user.".to_string()), ..Default::default() },
],
dsl_flags: vec![
DslFlag {
flag: "--path".to_string(),
config_key: "path".to_string(),
description: "HTTP path this webhook listens on. Must start with /. Examples: /blog, /api/users/:id.".to_string(),
kind: DslFlagKind::Scalar,
required: true,
},
DslFlag {
flag: "--method".to_string(),
config_key: "method".to_string(),
description: "HTTP method: GET (default), POST, PUT, PATCH, DELETE.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
},
DslFlag {
flag: "--auth-type".to_string(),
config_key: "auth_type".to_string(),
description: "Authentication mode: none (default), jwt, hmac, api_key.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
},
DslFlag {
flag: "--auth-credential".to_string(),
config_key: "auth_credential".to_string(),
description: "Credential ID for auth verification. Required when auth_type != none.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
},
DslFlag {
flag: "--auth-required-role".to_string(),
config_key: "auth_required_role".to_string(),
description: "Comma-separated roles required for this route. JWT claim 'role' must match one. E.g. lecturer,student. Empty = any authenticated user.".to_string(),
kind: DslFlagKind::CommaSeparatedList,
required: false,
},
],
layout: vec![
LayoutItem::Field("path".to_string()),
LayoutItem::Field("method".to_string()),
LayoutItem::Field("__webhook_public_url".to_string()),
LayoutItem::Row { row: vec![LayoutItem::Field("auth_type".to_string()), LayoutItem::Field("auth_credential".to_string())] },
LayoutItem::Field("auth_required_role".to_string()),
],
ai_tool: Default::default(),
..Default::default()
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Config {
pub path: String,
#[serde(default = "default_method")]
pub method: String,
#[serde(default)]
pub auth_type: String,
#[serde(default)]
pub auth_credential: String,
#[serde(default)]
pub auth_required_role: Vec<String>,
}
fn default_method() -> String {
"GET".to_string()
}
pub struct Node {
config: Config,
}
impl Node {
pub fn new(config: Config) -> Self {
Self { config }
}
}
fn collect_trace_private_tokens(payload: &Value) -> Vec<String> {
const SENSITIVE_KEYS: &[&str] = &[
"password",
"passwd",
"pwd",
"secret",
"token",
"access_token",
"refresh_token",
"api_key",
"apikey",
"authorization",
];
let Some(map) = payload.as_object() else {
return Vec::new();
};
let mut out = Vec::new();
for (key, value) in map {
if !SENSITIVE_KEYS
.iter()
.any(|candidate| key.eq_ignore_ascii_case(candidate))
{
continue;
}
let Some(text) = value.as_str() else {
continue;
};
let trimmed = text.trim();
if trimmed.is_empty() || out.iter().any(|existing| existing == trimmed) {
continue;
}
out.push(trimmed.to_string());
}
out
}
#[async_trait]
impl NodeHandler for Node {
fn kind(&self) -> &'static str {
NODE_KIND
}
fn input_pins(&self) -> &'static [&'static str] {
&[]
}
fn output_pins(&self) -> &'static [&'static str] {
&[OUTPUT_PIN_OUT]
}
async fn execute_async(
&self,
input: NodeExecutionInput,
) -> Result<NodeExecutionOutput, PipelineError> {
let mut payload = input.payload;
let trace_private_tokens = collect_trace_private_tokens(&payload);
if !trace_private_tokens.is_empty() {
if let Value::Object(map) = &mut payload {
map.insert(
"__zf_private_trace_redact".to_string(),
Value::Array(
trace_private_tokens
.into_iter()
.map(Value::String)
.collect(),
),
);
}
}
Ok(NodeExecutionOutput {
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
payload,
trace: vec![
format!("node_kind={NODE_KIND}"),
format!("method={}", self.config.method),
format!("path={}", self.config.path),
],
})
}
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::collect_trace_private_tokens;
#[test]
fn collects_common_sensitive_form_fields_for_trace_redaction() {
let tokens = collect_trace_private_tokens(&json!({
"username": "wawan dot wan",
"password": "toryoto",
"api_key": "abc123"
}));
assert_eq!(tokens, vec!["toryoto", "abc123"]);
}
}