featherbit/plugins/native/
request_validation.rs

1//! The `request-validation` node — validates request headers and/or body
2//! against JSON Schemas before the request reaches the upstream, rejecting
3//! non-conforming requests with a configurable status code.
4//!
5//! Port of APISIX's `request-validation` plugin. Schemas are compiled once at
6//! config load (invalid schemas fail policy compilation, not requests).
7//! Bodies are validated as JSON by default; `application/x-www-form-urlencoded`
8//! bodies are decoded into a flat object first, mirroring the Lua plugin's
9//! `ngx.decode_args` shape (duplicate keys become arrays, keys without `=`
10//! become `true`).
11
12use async_trait::async_trait;
13use bytes::Bytes;
14use std::collections::HashMap;
15
16use crate::context::Context;
17use crate::plugins::{Plugin, PluginOutput, PluginResult};
18use crate::vars::template::Template;
19
20/// Validates `context.request` headers and body against compiled JSON
21/// Schemas, and optionally evaluates boolean condition predicates on the
22/// request. On failure the request is rejected through the `denied` port
23/// with a JSON response using `rejected_code`.
24///
25/// Headers are validated as a single-value object (first value per header,
26/// names lowercased), matching the shape APISIX passes to its schema check.
27/// After a successful JSON-body validation the body is re-serialized from the
28/// parsed document, so the JSON that was validated is exactly the JSON the
29/// upstream receives (guards against JSON-interoperability smuggling).
30#[derive(Debug)]
31pub struct RequestValidationPlugin {
32    header_schema: Option<jsonschema::Validator>,
33    body_schema: Option<jsonschema::Validator>,
34    conditions: Option<crate::vars::Expr>,
35    rejected_code: u16,
36    /// Fixed message returned instead of the validator's error description.
37    /// Supports `{{namespace.path}}` references (no legacy `$var`
38    /// interpolation — this field never supported it, so this sweep must not
39    /// start).
40    rejected_msg: Option<Template>,
41}
42
43/// Compiles an optional schema key into a validator, failing fast with the
44/// key name on malformed schemas.
45fn compile_schema(
46    config: &HashMap<String, serde_json::Value>,
47    key: &str,
48) -> Result<Option<jsonschema::Validator>, String> {
49    match config.get(key) {
50        None => Ok(None),
51        Some(raw) => {
52            if !raw.is_object() {
53                return Err(format!(
54                    "request-validation: '{}' must be a JSON Schema object",
55                    key
56                ));
57            }
58            jsonschema::validator_for(raw)
59                .map(Some)
60                .map_err(|e| format!("request-validation: invalid '{}': {}", key, e))
61        }
62    }
63}
64
65/// Decodes an `application/x-www-form-urlencoded` body into a flat JSON
66/// object, mirroring `ngx.decode_args`: `a=1&a=2` becomes `{"a": ["1","2"]}`,
67/// a bare `flag` (no `=`) becomes `{"flag": true}`, values are
68/// percent-decoded and `+` maps to space.
69fn decode_urlencoded(body: &str) -> serde_json::Value {
70    let mut map = serde_json::Map::new();
71    for pair in body.split('&') {
72        if pair.is_empty() {
73            continue;
74        }
75        let (key, value) = match pair.split_once('=') {
76            Some((k, v)) => (urldecode(k), serde_json::Value::String(urldecode(v))),
77            None => (urldecode(pair), serde_json::Value::Bool(true)),
78        };
79        match map.get_mut(&key) {
80            None => {
81                map.insert(key, value);
82            }
83            Some(serde_json::Value::Array(arr)) => arr.push(value),
84            Some(existing) => {
85                let first = existing.take();
86                *existing = serde_json::Value::Array(vec![first, value]);
87            }
88        }
89    }
90    serde_json::Value::Object(map)
91}
92
93/// Percent-decodes a urlencoded component (`+` becomes space; malformed
94/// escapes pass through literally).
95fn urldecode(s: &str) -> String {
96    let bytes = s.as_bytes();
97    let mut out = Vec::with_capacity(bytes.len());
98    let mut i = 0;
99    while i < bytes.len() {
100        match bytes[i] {
101            b'+' => out.push(b' '),
102            b'%' if i + 2 < bytes.len() => {
103                if let (Some(h), Some(l)) = (hex_val(bytes[i + 1]), hex_val(bytes[i + 2])) {
104                    out.push(h * 16 + l);
105                    i += 3;
106                    continue;
107                }
108                out.push(b'%');
109            }
110            b => out.push(b),
111        }
112        i += 1;
113    }
114    String::from_utf8_lossy(&out).into_owned()
115}
116
117fn hex_val(b: u8) -> Option<u8> {
118    match b {
119        b'0'..=b'9' => Some(b - b'0'),
120        b'a'..=b'f' => Some(b - b'a' + 10),
121        b'A'..=b'F' => Some(b - b'A' + 10),
122        _ => None,
123    }
124}
125
126impl RequestValidationPlugin {
127    /// Builds the plugin from node config.
128    ///
129    /// Accepted keys:
130    /// - `header_schema` (object): JSON Schema applied to the request headers,
131    ///   seen as `{name: first_value}` with lowercase names.
132    /// - `body_schema` (object): JSON Schema applied to the parsed request
133    ///   body (JSON, or urlencoded decoded to a flat object).
134    /// - `conditions` (array): a condition expression (see [`crate::vars::Expr`])
135    ///   — rules ANDed at top level, nested `AND`/`OR`/`NOT` groups, JSONPath
136    ///   body subjects. Evaluated after the schemas; failure rejects like a
137    ///   schema failure with message "request conditions not satisfied".
138    /// - `rejected_code` (integer 200–599, default `400`): response status
139    ///   for rejected requests.
140    /// - `rejected_msg` (string): fixed message returned instead of the
141    ///   validator's error description. Supports `{{namespace.path}}`
142    ///   references.
143    ///
144    /// At least one of `header_schema`, `body_schema`, or `conditions` is
145    /// required; schemas are compiled here so malformed schemas fail at config
146    /// load.
147    ///
148    /// ```yaml
149    /// type: request-validation
150    /// config:
151    ///   rejected_code: 422
152    ///   body_schema:
153    ///     type: object
154    ///     required: [name]
155    ///     properties:
156    ///       name: { type: string, minLength: 1 }
157    /// ```
158    pub fn from_config(config: &HashMap<String, serde_json::Value>) -> Result<Self, String> {
159        let header_schema = compile_schema(config, "header_schema")?;
160        let body_schema = compile_schema(config, "body_schema")?;
161
162        let conditions = match config.get("conditions") {
163            None => None,
164            Some(v) => Some(
165                crate::vars::Expr::parse(v)
166                    .map_err(|e| format!("request-validation: invalid 'conditions': {}", e))?,
167            ),
168        };
169
170        if header_schema.is_none() && body_schema.is_none() && conditions.is_none() {
171            return Err(
172                "request-validation: at least one of 'header_schema', 'body_schema', or 'conditions' is required"
173                    .to_string(),
174            );
175        }
176
177        let rejected_code = match config.get("rejected_code") {
178            None => 400,
179            Some(v) => {
180                let code = v
181                    .as_u64()
182                    .filter(|c| (200..=599).contains(c))
183                    .ok_or("request-validation: 'rejected_code' must be an integer in 200..=599")?;
184                code as u16
185            }
186        };
187
188        let rejected_msg = config
189            .get("rejected_msg")
190            .and_then(|v| v.as_str())
191            // Discard warnings here — the compile-time walk (a later task)
192            // reports well-formed-but-unknown references; execution must not.
193            .map(|s| Template::parse(s).0);
194
195        Ok(Self {
196            header_schema,
197            body_schema,
198            conditions,
199            rejected_code,
200            rejected_msg,
201        })
202    }
203
204    /// Writes the rejection onto the response and routes the context through
205    /// the node's `denied` port.
206    fn reject(&self, mut ctx: Context, detail: String) -> PluginResult {
207        let message = self
208            .rejected_msg
209            .as_ref()
210            .map(|t| t.render(&ctx).into_owned())
211            .unwrap_or(detail);
212        ctx.response.status_code = self.rejected_code;
213        ctx.response.body = Bytes::from(
214            serde_json::json!({ "error": "validation_failed", "message": message }).to_string(),
215        );
216        ctx.response.headers.insert(
217            "content-type".to_string(),
218            vec!["application/json".to_string()],
219        );
220        Ok(PluginOutput::on_port(ctx, "denied"))
221    }
222}
223
224#[async_trait]
225impl Plugin for RequestValidationPlugin {
226    fn plugin_type(&self) -> &str {
227        "request-validation"
228    }
229
230    async fn execute(&self, mut ctx: Context) -> PluginResult {
231        if let Some(validator) = &self.header_schema {
232            let headers: serde_json::Map<String, serde_json::Value> = ctx
233                .request
234                .headers
235                .iter()
236                .filter_map(|(k, v)| {
237                    v.first()
238                        .map(|first| (k.clone(), serde_json::Value::String(first.clone())))
239                })
240                .collect();
241            if let Err(e) = validator.validate(&serde_json::Value::Object(headers)) {
242                return self.reject(ctx, format!("header validation failed: {}", e));
243            }
244        }
245
246        if let Some(validator) = &self.body_schema {
247            if ctx.request.body.is_empty() {
248                return self.reject(ctx, "request body is required".to_string());
249            }
250
251            let is_urlencoded = ctx
252                .request
253                .headers
254                .get("content-type")
255                .and_then(|v| v.first())
256                .map(|ct| {
257                    ct.to_lowercase()
258                        .starts_with("application/x-www-form-urlencoded")
259                })
260                .unwrap_or(false);
261
262            let (parsed, body_is_json) = if is_urlencoded {
263                let text = String::from_utf8_lossy(&ctx.request.body).into_owned();
264                (decode_urlencoded(&text), false)
265            } else {
266                match serde_json::from_slice::<serde_json::Value>(&ctx.request.body) {
267                    Ok(v) => (v, true),
268                    Err(e) => {
269                        return self
270                            .reject(ctx, format!("failed to decode the request body: {}", e));
271                    }
272                }
273            };
274
275            if let Err(e) = validator.validate(&parsed) {
276                return self.reject(ctx, format!("body validation failed: {}", e));
277            }
278
279            if body_is_json {
280                // Ensure the JSON we validated is the JSON the upstream sees
281                // (JSON interoperability hardening, mirroring APISIX).
282                // Body-mutation convention: drop the stale content-length.
283                ctx.request.body = Bytes::from(serde_json::to_vec(&parsed).unwrap_or_default());
284                ctx.request.headers.remove("content-length");
285            }
286        }
287
288        if let Some(expr) = &self.conditions {
289            if !expr.eval(&ctx) {
290                return self.reject(ctx, "request conditions not satisfied".to_string());
291            }
292        }
293
294        Ok(PluginOutput::success(ctx))
295    }
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
302
303    fn test_context(body: &str, content_type: Option<&str>) -> Context {
304        let mut headers = HashMap::new();
305        headers.insert("x-api-version".to_string(), vec!["2".to_string()]);
306        if let Some(ct) = content_type {
307            headers.insert("content-type".to_string(), vec![ct.to_string()]);
308        }
309        if !body.is_empty() {
310            headers.insert("content-length".to_string(), vec![body.len().to_string()]);
311        }
312
313        Context {
314            request: GatewayRequest {
315                method: "POST".to_string(),
316                path: "/api".to_string(),
317                host: "localhost".to_string(),
318                scheme: "http".to_string(),
319                headers,
320                query_params: HashMap::new(),
321                body: Bytes::from(body.to_string()),
322                remote_addr: "127.0.0.1:12345".to_string(),
323                protocol: Protocol::Http1,
324            },
325            response: GatewayResponse {
326                status_code: 0,
327                headers: HashMap::new(),
328                body: Bytes::new(),
329                stream: None,
330            },
331            message: HashMap::new(),
332            errors: Vec::new(),
333        }
334    }
335
336    fn body_plugin(schema: serde_json::Value) -> RequestValidationPlugin {
337        let mut config = HashMap::new();
338        config.insert("body_schema".to_string(), schema);
339        RequestValidationPlugin::from_config(&config).unwrap()
340    }
341
342    #[tokio::test]
343    async fn test_request_validation_body_accept_and_normalize() {
344        let p = body_plugin(serde_json::json!({
345            "type": "object",
346            "required": ["name"],
347            "properties": { "name": { "type": "string" } }
348        }));
349        let ctx = test_context(r#"{"name":  "jack"}"#, Some("application/json"));
350        let out = p.execute(ctx).await.unwrap();
351        // validated JSON is re-serialized (normalized) and content-length dropped
352        assert_eq!(out.context.request.body, Bytes::from(r#"{"name":"jack"}"#));
353        assert!(!out.context.request.headers.contains_key("content-length"));
354    }
355
356    #[tokio::test]
357    async fn test_request_validation_body_reject() {
358        let p = body_plugin(serde_json::json!({
359            "type": "object",
360            "required": ["name"]
361        }));
362        let ctx = test_context(r#"{"age": 3}"#, Some("application/json"));
363        let out = p.execute(ctx).await.unwrap();
364        assert_eq!(out.port, Some("denied"));
365        assert_eq!(out.context.response.status_code, 400);
366        let body: serde_json::Value = serde_json::from_slice(&out.context.response.body).unwrap();
367        assert_eq!(body["error"], "validation_failed");
368    }
369
370    #[tokio::test]
371    async fn test_request_validation_non_json_body_rejected() {
372        let p = body_plugin(serde_json::json!({ "type": "object" }));
373        let ctx = test_context("this is not json", Some("application/json"));
374        let out = p.execute(ctx).await.unwrap();
375        assert_eq!(out.port, Some("denied"));
376    }
377
378    #[tokio::test]
379    async fn test_request_validation_missing_body_rejected() {
380        let p = body_plugin(serde_json::json!({ "type": "object" }));
381        let out = p
382            .execute(test_context("", Some("application/json")))
383            .await
384            .unwrap();
385        assert_eq!(out.port, Some("denied"));
386    }
387
388    #[tokio::test]
389    async fn test_request_validation_urlencoded_body() {
390        let p = body_plugin(serde_json::json!({
391            "type": "object",
392            "required": ["user"],
393            "properties": { "user": { "type": "string", "minLength": 2 } }
394        }));
395        let ctx = test_context(
396            "user=jack&note=hello%20world",
397            Some("application/x-www-form-urlencoded"),
398        );
399        let out = p.execute(ctx).await.unwrap();
400        // urlencoded bodies are not rewritten
401        assert_eq!(
402            out.context.request.body,
403            Bytes::from("user=jack&note=hello%20world")
404        );
405
406        let ctx = test_context("note=only", Some("application/x-www-form-urlencoded"));
407        assert_eq!(p.execute(ctx).await.unwrap().port, Some("denied"));
408    }
409
410    #[tokio::test]
411    async fn test_request_validation_header_schema() {
412        let mut config = HashMap::new();
413        config.insert(
414            "header_schema".to_string(),
415            serde_json::json!({
416                "type": "object",
417                "required": ["x-api-version"],
418                "properties": { "x-api-version": { "type": "string", "enum": ["2"] } }
419            }),
420        );
421        let p = RequestValidationPlugin::from_config(&config).unwrap();
422
423        assert!(p
424            .execute(test_context("", None))
425            .await
426            .unwrap()
427            .port
428            .is_none());
429
430        let mut ctx = test_context("", None);
431        ctx.request.headers.remove("x-api-version");
432        let out = p.execute(ctx).await.unwrap();
433        assert_eq!(out.port, Some("denied"));
434    }
435
436    #[tokio::test]
437    async fn test_request_validation_rejected_code_and_msg() {
438        let mut config = HashMap::new();
439        config.insert(
440            "body_schema".to_string(),
441            serde_json::json!({ "type": "object" }),
442        );
443        config.insert("rejected_code".to_string(), serde_json::json!(422));
444        config.insert("rejected_msg".to_string(), serde_json::json!("bad payload"));
445        let p = RequestValidationPlugin::from_config(&config).unwrap();
446
447        let ctx = test_context("[1,2,3]", Some("application/json"));
448        let out = p.execute(ctx).await.unwrap();
449        assert_eq!(out.port, Some("denied"));
450        assert_eq!(out.context.response.status_code, 422);
451        let body: serde_json::Value = serde_json::from_slice(&out.context.response.body).unwrap();
452        assert_eq!(body["message"], "bad payload");
453    }
454
455    #[tokio::test]
456    async fn test_request_validation_rejected_msg_renders_template() {
457        let mut config = HashMap::new();
458        config.insert(
459            "body_schema".to_string(),
460            serde_json::json!({ "type": "object", "required": ["name"] }),
461        );
462        config.insert(
463            "rejected_msg".to_string(),
464            serde_json::json!("bad payload for {{request.path}}"),
465        );
466        let p = RequestValidationPlugin::from_config(&config).unwrap();
467
468        let ctx = test_context(r#"{"age": 3}"#, Some("application/json"));
469        let out = p.execute(ctx).await.unwrap();
470        assert_eq!(out.port, Some("denied"));
471        let body: serde_json::Value = serde_json::from_slice(&out.context.response.body).unwrap();
472        assert_eq!(body["message"], "bad payload for /api");
473    }
474
475    #[test]
476    fn test_request_validation_config_rejections() {
477        // no schema at all
478        assert!(RequestValidationPlugin::from_config(&HashMap::new()).is_err());
479
480        // invalid JSON Schema fails fast at config load
481        let mut config = HashMap::new();
482        config.insert(
483            "body_schema".to_string(),
484            serde_json::json!({ "type": "definitely-not-a-type" }),
485        );
486        assert!(RequestValidationPlugin::from_config(&config).is_err());
487
488        // schema must be an object
489        let mut config = HashMap::new();
490        config.insert("body_schema".to_string(), serde_json::json!("nope"));
491        assert!(RequestValidationPlugin::from_config(&config).is_err());
492
493        // rejected_code out of range
494        let mut config = HashMap::new();
495        config.insert(
496            "body_schema".to_string(),
497            serde_json::json!({ "type": "object" }),
498        );
499        config.insert("rejected_code".to_string(), serde_json::json!(199));
500        assert!(RequestValidationPlugin::from_config(&config).is_err());
501    }
502
503    #[test]
504    fn test_request_validation_decode_urlencoded_shape() {
505        let v = decode_urlencoded("a=1&a=2&flag&note=hello+world");
506        assert_eq!(v["a"], serde_json::json!(["1", "2"]));
507        assert_eq!(v["flag"], serde_json::json!(true));
508        assert_eq!(v["note"], serde_json::json!("hello world"));
509    }
510
511    fn conditions_plugin(conditions: serde_json::Value) -> RequestValidationPlugin {
512        let mut config = HashMap::new();
513        config.insert("conditions".to_string(), conditions);
514        RequestValidationPlugin::from_config(&config).unwrap()
515    }
516
517    #[tokio::test]
518    async fn test_request_validation_conditions_accept() {
519        let p = conditions_plugin(serde_json::json!([
520            ["http_authorization", "present"],
521            ["http_authorization", "contains", "Bearer"],
522            [
523                "OR",
524                ["$.user.email", "present"],
525                ["NOT", ["$.user.id", "is_null"]]
526            ]
527        ]));
528        let mut ctx = test_context(r#"{"user":{"email":"a@b.c"}}"#, Some("application/json"));
529        ctx.request
530            .headers
531            .insert("authorization".to_string(), vec!["Bearer tok".to_string()]);
532        let out = p.execute(ctx).await.unwrap();
533        assert!(out.port.is_none());
534    }
535
536    #[tokio::test]
537    async fn test_request_validation_conditions_reject() {
538        let p = conditions_plugin(serde_json::json!([[
539            "http_authorization",
540            "contains",
541            "Bearer"
542        ]]));
543        let out = p.execute(test_context("", None)).await.unwrap();
544        assert_eq!(out.port, Some("denied"));
545        assert_eq!(out.context.response.status_code, 400);
546        let body: serde_json::Value = serde_json::from_slice(&out.context.response.body).unwrap();
547        assert_eq!(body["error"], "validation_failed");
548        assert_eq!(body["message"], "request conditions not satisfied");
549    }
550
551    #[tokio::test]
552    async fn test_request_validation_conditions_after_schema() {
553        // both body_schema and conditions: schema normalizes, conditions still run
554        let mut config = HashMap::new();
555        config.insert(
556            "body_schema".to_string(),
557            serde_json::json!({ "type": "object" }),
558        );
559        config.insert(
560            "conditions".to_string(),
561            serde_json::json!([["$.name", "==", "jack"]]),
562        );
563        let p = RequestValidationPlugin::from_config(&config).unwrap();
564
565        let ctx = test_context(r#"{"name": "jack"}"#, Some("application/json"));
566        assert!(p.execute(ctx).await.unwrap().port.is_none());
567
568        let ctx = test_context(r#"{"name": "jill"}"#, Some("application/json"));
569        assert_eq!(p.execute(ctx).await.unwrap().port, Some("denied"));
570    }
571
572    #[test]
573    fn test_request_validation_conditions_config() {
574        // conditions alone satisfies the at-least-one requirement
575        let mut config = HashMap::new();
576        config.insert(
577            "conditions".to_string(),
578            serde_json::json!([["http_x", "present"]]),
579        );
580        assert!(RequestValidationPlugin::from_config(&config).is_ok());
581
582        // malformed conditions fail at config load, with plugin-prefixed message
583        let mut config = HashMap::new();
584        config.insert(
585            "conditions".to_string(),
586            serde_json::json!([["$.a", "bogus_op", 1]]),
587        );
588        let err = RequestValidationPlugin::from_config(&config).unwrap_err();
589        assert!(
590            err.starts_with("request-validation: invalid 'conditions'"),
591            "{err}"
592        );
593    }
594}
featherbit/plugins/native/request_validation.rs

featherbit/plugins/native/
request_validation.rs
