featherbit/plugins/native/
openwhisk.rs

1//! Apache OpenWhisk serverless-upstream plugin (`openwhisk`).
2//!
3//! Port of APISIX's `openwhisk` plugin (3.17). Invokes an OpenWhisk action
4//! (blocking, with the result inlined) and returns the action's reply as the
5//! gateway response — it **replaces the upstream**, so the node's `success`
6//! port should be wired straight to `client.in`.
7//!
8//! The request body is POSTed as the action parameters to
9//! `<api_host>/api/v1/namespaces/<namespace>/actions/<package/><action>?blocking=true&result=<result>&timeout=<ms>`
10//! with an `Authorization: Basic <base64(service_token)>` header.
11//!
12//! `api_host`, `namespace`, `package`, and `action` support
13//! `{{namespace.path}}` references (plain render, no legacy `$var`
14//! interpolation) — the endpoint is rebuilt per request so a value like
15//! `{{request.headers.x-tenant}}` in `namespace` routes each request to a
16//! different OpenWhisk namespace. `authorization` (built from
17//! `service_token`) is never templated — it is a credential, not an
18//! endpoint field. `namespace`/`package`/`action` are percent-encoded at
19//! render time before insertion into the URL path (see
20//! [`encode_path_segment`]); `api_host` is not encoded (see its field doc).
21//!
22//! # Response mapping
23//!
24//! OpenWhisk returns a JSON envelope. An action may return just a body, or set
25//! `statusCode` and `headers` explicitly. [`map_response`] mirrors APISIX:
26//! `statusCode` (when present) becomes the response status, `headers` are
27//! applied, and `body` (or the raw envelope when absent) becomes the body. A
28//! non-JSON envelope fails the node with `503` through the error port.
29
30use async_trait::async_trait;
31use base64::engine::general_purpose::STANDARD as BASE64;
32use base64::Engine;
33use bytes::Bytes;
34use std::collections::HashMap;
35use std::sync::Arc;
36use std::time::Duration;
37
38use crate::context::{Context, GatewayError};
39use crate::outbound::{OutboundClient, OutboundRequest, OutboundResponse};
40use crate::plugins::resources::PluginResources;
41use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
42use crate::vars::template::Template;
43
44use super::faas;
45
46/// Invokes an OpenWhisk action and maps its reply into `Context.response`.
47pub struct OpenWhiskPlugin {
48    /// Supports `{{namespace.path}}` references — resolved per request when
49    /// building the endpoint URL. No legacy `$var` interpolation (endpoint
50    /// fields never supported it, so this sweep must not start).
51    ///
52    /// **Warning**: unlike `namespace`/`package`/`action`, the rendered value
53    /// is spliced into the URL unencoded (it's the origin + scheme, not a
54    /// path segment — percent-encoding would corrupt it). Templating this
55    /// field from request-controlled data lets a client redirect the entire
56    /// outbound call to an arbitrary host (SSRF); prefer a static value or
57    /// one derived only from `{{env.*}}`.
58    api_host: Template,
59    /// Pre-computed `Basic <base64(service_token)>` header value. Never
60    /// templated — a credential, not an endpoint field.
61    authorization: String,
62    /// Supports `{{namespace.path}}` references, same as `api_host`. Unlike
63    /// `api_host`, the rendered value is percent-encoded
64    /// ([`encode_path_segment`]) before insertion into the URL path — safe
65    /// to template from request data.
66    namespace: Template,
67    /// Supports `{{namespace.path}}` references, same as `namespace`
68    /// (percent-encoded at render time).
69    package: Option<Template>,
70    /// Supports `{{namespace.path}}` references, same as `namespace`
71    /// (percent-encoded at render time).
72    action: Template,
73    result: bool,
74    ssl_verify: bool,
75    timeout: Duration,
76    /// `timeout` in milliseconds, passed through as the action query param.
77    timeout_ms: u64,
78    client: Arc<OutboundClient>,
79}
80
81impl OpenWhiskPlugin {
82    /// Builds the plugin from node config.
83    ///
84    /// Accepted keys:
85    /// - `api_host` (string, **required**): the OpenWhisk API host
86    ///   (e.g. `https://ow.example.com`). Missing/empty is a config error.
87    ///   Supports `{{namespace.path}}` references, rendered per request when
88    ///   building the endpoint URL (a trailing `/` is trimmed after render).
89    /// - `service_token` (string, **required**): `user:pass` action token, sent
90    ///   base64-encoded as HTTP Basic auth. Missing/empty is a config error.
91    ///   Never templated — a credential, not an endpoint field.
92    /// - `action` (string, **required**): the action name to invoke; supports
93    ///   `{{namespace.path}}` references.
94    /// - `namespace` (string, default `_`): the OpenWhisk namespace; supports
95    ///   `{{namespace.path}}` references.
96    /// - `package` (string, optional): the package the action belongs to;
97    ///   supports `{{namespace.path}}` references.
98    /// - `result` (bool, default `true`): request `result=true` (inline the
99    ///   action result rather than the full activation record).
100    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
101    /// - `timeout` (integer ms, default `3000`): whole-call deadline, also
102    ///   passed as the action `timeout` query parameter.
103    ///
104    /// ```yaml
105    /// type: openwhisk
106    /// config:
107    ///   api_host: https://ow.example.com
108    ///   service_token: ${OPENWHISK_TOKEN}
109    ///   namespace: guest
110    ///   action: hello
111    ///   result: true
112    ///   ssl_verify: true
113    ///   timeout: 3000
114    /// ```
115    pub fn from_config(
116        config: &HashMap<String, serde_json::Value>,
117        resources: &Arc<PluginResources>,
118    ) -> Result<Self, String> {
119        let api_host = config
120            .get("api_host")
121            .and_then(|v| v.as_str())
122            .filter(|s| !s.is_empty())
123            .ok_or_else(|| "openwhisk plugin requires 'api_host'".to_string())?
124            .to_string();
125        // Discard warnings here — the compile-time walk (a later task)
126        // reports well-formed-but-unknown references; execution must not.
127        // The trailing-slash trim moves to `endpoint()` (post-render) since a
128        // templated value's trailing slash can't be known at load time.
129        let api_host = Template::parse(&api_host).0;
130
131        let service_token = config
132            .get("service_token")
133            .and_then(|v| v.as_str())
134            .filter(|s| !s.is_empty())
135            .ok_or_else(|| "openwhisk plugin requires 'service_token'".to_string())?;
136        let authorization = format!("Basic {}", BASE64.encode(service_token));
137
138        let action = config
139            .get("action")
140            .and_then(|v| v.as_str())
141            .filter(|s| !s.is_empty())
142            .ok_or_else(|| "openwhisk plugin requires 'action'".to_string())?
143            .to_string();
144        let action = Template::parse(&action).0;
145
146        let namespace = config
147            .get("namespace")
148            .and_then(|v| v.as_str())
149            .filter(|s| !s.is_empty())
150            .unwrap_or("_")
151            .to_string();
152        let namespace = Template::parse(&namespace).0;
153
154        let package = config
155            .get("package")
156            .and_then(|v| v.as_str())
157            .filter(|s| !s.is_empty())
158            .map(|s| Template::parse(s).0);
159
160        let result = config
161            .get("result")
162            .and_then(|v| v.as_bool())
163            .unwrap_or(true);
164
165        let ssl_verify = config
166            .get("ssl_verify")
167            .and_then(|v| v.as_bool())
168            .unwrap_or(true);
169
170        let timeout_ms = config
171            .get("timeout")
172            .and_then(|v| v.as_u64())
173            .unwrap_or(3000);
174
175        Ok(Self {
176            api_host,
177            authorization,
178            namespace,
179            package,
180            action,
181            result,
182            ssl_verify,
183            timeout: Duration::from_millis(timeout_ms),
184            timeout_ms,
185            client: resources.outbound.clone(),
186        })
187    }
188
189    /// Builds the OpenWhisk action-invocation URL, rendering `api_host` /
190    /// `namespace` / `package` / `action` against `ctx` so each field's
191    /// `{{namespace.path}}` references resolve per request.
192    ///
193    /// `namespace`/`package`/`action` are percent-encoded
194    /// ([`encode_path_segment`]) after rendering: since these fields are
195    /// templatable from request data (e.g. a header), an unescaped `/` or
196    /// `?` in the rendered value would otherwise break out of its path
197    /// segment or inject extra query parameters ahead of the plugin's own
198    /// `blocking`/`result`/`timeout`. `api_host` is deliberately not
199    /// encoded — see the warning on its field doc.
200    fn endpoint(&self, ctx: &Context) -> String {
201        let api_host = self.api_host.render(ctx);
202        let api_host = api_host.trim_end_matches('/');
203        let namespace = encode_path_segment(&self.namespace.render(ctx));
204        let action = encode_path_segment(&self.action.render(ctx));
205        let package = self
206            .package
207            .as_ref()
208            .map(|p| format!("{}/", encode_path_segment(&p.render(ctx))))
209            .unwrap_or_default();
210        format!(
211            "{}/api/v1/namespaces/{}/actions/{}{}?blocking=true&result={}&timeout={}",
212            api_host, namespace, package, action, self.result, self.timeout_ms
213        )
214    }
215
216    /// Builds the outbound POST request carrying the client body as the action
217    /// parameters.
218    fn build_request(&self, ctx: &Context) -> OutboundRequest {
219        let headers = vec![
220            ("authorization".to_string(), self.authorization.clone()),
221            ("content-type".to_string(), "application/json".to_string()),
222        ];
223        OutboundRequest {
224            method: http::Method::POST,
225            url: self.endpoint(ctx),
226            headers,
227            body: ctx.request.body.clone(),
228            timeout: self.timeout,
229            ssl_verify: self.ssl_verify,
230            tls: None,
231        }
232    }
233}
234
235/// Percent-encodes `s` for safe insertion as a single URL path segment.
236///
237/// Byte-wise: any byte outside the RFC 3986 `unreserved` set (`ALPHA` /
238/// `DIGIT` / `-` / `.` / `_` / `~`) becomes `%XX` (uppercase hex). This is
239/// deliberately stricter than a full path-escaper (it also encodes `/`),
240/// which is exactly the point here — `namespace`/`package`/`action` are each
241/// meant to occupy exactly one path segment, and a rendered value from
242/// request data (header, query, ...) must not be able to smuggle in a `/`
243/// (segment breakout, e.g. `foo/actions/bar`) or a `?`/`&` (query-string
244/// injection ahead of the plugin's own `blocking`/`result`/`timeout`
245/// params). A literal, already-safe value (`guest`, `my_pkg`) round-trips
246/// byte-identical since every one of its characters is in `unreserved`.
247fn encode_path_segment(s: &str) -> String {
248    let mut out = String::with_capacity(s.len());
249    for b in s.bytes() {
250        match b {
251            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
252                out.push(b as char)
253            }
254            _ => out.push_str(&format!("%{:02X}", b)),
255        }
256    }
257    out
258}
259
260/// The status, headers, and body mapped out of an OpenWhisk envelope.
261struct MappedResponse {
262    status: u16,
263    headers: HashMap<String, Vec<String>>,
264    body: Bytes,
265}
266
267/// Maps an OpenWhisk activation reply into a `(status, headers, body)` triple.
268///
269/// An empty body passes the transport status/body through untouched. A
270/// well-formed JSON envelope may override the status via `statusCode`, set
271/// response `headers`, and provide `body` (string or nested JSON); when
272/// `statusCode`/`body` are absent the transport status / raw envelope are used.
273/// A non-JSON, non-empty body is an error (mapped to `503` by the caller).
274fn map_response(status: u16, raw: &Bytes) -> Result<MappedResponse, String> {
275    if raw.is_empty() {
276        return Ok(MappedResponse {
277            status,
278            headers: HashMap::new(),
279            body: raw.clone(),
280        });
281    }
282
283    let envelope: serde_json::Value = serde_json::from_slice(raw)
284        .map_err(|e| format!("failed to parse openwhisk response: {}", e))?;
285
286    let mut headers: HashMap<String, Vec<String>> = HashMap::new();
287    if let Some(hdrs) = envelope.get("headers").and_then(|v| v.as_object()) {
288        for (name, value) in hdrs {
289            let value = match value {
290                serde_json::Value::String(s) => s.clone(),
291                other => other.to_string(),
292            };
293            headers
294                .entry(name.to_ascii_lowercase())
295                .or_default()
296                .push(value);
297        }
298    }
299
300    let code = envelope
301        .get("statusCode")
302        .and_then(|v| v.as_u64())
303        .map(|c| c as u16)
304        .unwrap_or(status);
305
306    let body = match envelope.get("body") {
307        Some(serde_json::Value::String(s)) => Bytes::from(s.clone().into_bytes()),
308        Some(other) => Bytes::from(other.to_string().into_bytes()),
309        None => raw.clone(),
310    };
311
312    Ok(MappedResponse {
313        status: code,
314        headers,
315        body,
316    })
317}
318
319#[async_trait]
320impl Plugin for OpenWhiskPlugin {
321    fn plugin_type(&self) -> &str {
322        "openwhisk"
323    }
324
325    async fn execute(&self, mut ctx: Context) -> PluginResult {
326        let request = self.build_request(&ctx);
327
328        let response: OutboundResponse = match self.client.request(request).await {
329            Ok(resp) => resp,
330            Err(e) => {
331                let (status, message) = faas::classify_error("openwhisk", &e);
332                return Err(reject(ctx, status, message));
333            }
334        };
335
336        match map_response(response.status, &response.body) {
337            Ok(mapped) => {
338                ctx.response.status_code = mapped.status;
339                ctx.response.headers = mapped.headers;
340                ctx.response.body = mapped.body;
341                Ok(PluginOutput::success(ctx))
342            }
343            Err(message) => Err(reject(ctx, 503, message)),
344        }
345    }
346}
347
348/// Builds the `OPENWHISK_CALLOUT_ERROR` rejection carrying the context.
349fn reject(mut ctx: Context, status: u16, message: String) -> PluginExecutionError {
350    ctx.response.status_code = status;
351    PluginExecutionError {
352        context: ctx,
353        error: GatewayError {
354            node_id: String::new(),
355            code: "OPENWHISK_CALLOUT_ERROR".to_string(),
356            message,
357            metadata: HashMap::new(),
358        },
359    }
360}
361
362#[cfg(test)]
363mod tests {
364    use super::*;
365    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
366
367    fn ctx() -> Context {
368        Context {
369            request: GatewayRequest {
370                method: "POST".to_string(),
371                path: "/orig".to_string(),
372                host: "gw".to_string(),
373                scheme: "http".to_string(),
374                headers: HashMap::new(),
375                query_params: HashMap::new(),
376                body: Bytes::from_static(b"{\"name\":\"x\"}"),
377                remote_addr: "1.2.3.4:5".to_string(),
378                protocol: Protocol::Http1,
379            },
380            response: GatewayResponse {
381                status_code: 0,
382                headers: HashMap::new(),
383                body: Bytes::new(),
384                stream: None,
385            },
386            message: HashMap::new(),
387            errors: Vec::new(),
388        }
389    }
390
391    fn plugin(config: serde_json::Value) -> OpenWhiskPlugin {
392        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
393        OpenWhiskPlugin::from_config(&map, &PluginResources::empty()).unwrap()
394    }
395
396    fn base_config() -> serde_json::Value {
397        serde_json::json!({
398            "api_host": "https://ow.example.com",
399            "service_token": "user:pass",
400            "namespace": "guest",
401            "action": "hello"
402        })
403    }
404
405    #[test]
406    fn test_requires_api_host_and_token_and_action() {
407        assert!(OpenWhiskPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err());
408        let mut cfg: HashMap<String, serde_json::Value> =
409            serde_json::from_value(serde_json::json!({ "api_host": "https://x" })).unwrap();
410        assert!(OpenWhiskPlugin::from_config(&cfg, &PluginResources::empty()).is_err());
411        cfg.insert("service_token".to_string(), serde_json::json!("t"));
412        // still missing action
413        assert!(OpenWhiskPlugin::from_config(&cfg, &PluginResources::empty()).is_err());
414    }
415
416    #[test]
417    fn test_endpoint_and_auth() {
418        let p = plugin(base_config());
419        let req = p.build_request(&ctx());
420        assert_eq!(
421            req.url,
422            "https://ow.example.com/api/v1/namespaces/guest/actions/hello?blocking=true&result=true&timeout=3000"
423        );
424        let authz = req
425            .headers
426            .iter()
427            .find(|(k, _)| k == "authorization")
428            .map(|(_, v)| v.clone())
429            .unwrap();
430        // "user:pass" base64 = dXNlcjpwYXNz
431        assert_eq!(authz, "Basic dXNlcjpwYXNz");
432        // body forwarded as action params
433        assert_eq!(req.body, Bytes::from_static(b"{\"name\":\"x\"}"));
434    }
435
436    #[test]
437    fn test_endpoint_with_package_and_result_false() {
438        let mut cfg = base_config();
439        cfg["package"] = serde_json::json!("mypkg");
440        cfg["result"] = serde_json::json!(false);
441        let p = plugin(cfg);
442        assert_eq!(
443            p.endpoint(&ctx()),
444            "https://ow.example.com/api/v1/namespaces/guest/actions/mypkg/hello?blocking=true&result=false&timeout=3000"
445        );
446    }
447
448    #[test]
449    fn test_endpoint_renders_namespace_template_per_request() {
450        // CONTROLLER-APPROVED SCOPE ADDITION: `namespace` (and `api_host`,
451        // `package`, `action`) render `{{namespace.path}}` references per
452        // request, closing the FaaS-family templating gap found in the
453        // Task 3 review.
454        let mut cfg = base_config();
455        cfg["namespace"] = serde_json::json!("tenant-{{request.headers.x-tenant}}");
456        let p = plugin(cfg);
457
458        let mut request_ctx = ctx();
459        request_ctx
460            .request
461            .headers
462            .insert("x-tenant".to_string(), vec!["acme".to_string()]);
463        assert_eq!(
464            p.endpoint(&request_ctx),
465            "https://ow.example.com/api/v1/namespaces/tenant-acme/actions/hello?blocking=true&result=true&timeout=3000"
466        );
467
468        // A different request routes to a different namespace.
469        let mut other_ctx = ctx();
470        other_ctx
471            .request
472            .headers
473            .insert("x-tenant".to_string(), vec!["globex".to_string()]);
474        assert_eq!(
475            p.endpoint(&other_ctx),
476            "https://ow.example.com/api/v1/namespaces/tenant-globex/actions/hello?blocking=true&result=true&timeout=3000"
477        );
478    }
479
480    #[test]
481    fn test_endpoint_literal_namespace_is_byte_identical() {
482        // A literal (non-templated) config must render byte-identically
483        // across requests — no accidental per-request drift from the
484        // templating change.
485        let p = plugin(base_config());
486        let expected =
487            "https://ow.example.com/api/v1/namespaces/guest/actions/hello?blocking=true&result=true&timeout=3000";
488        assert_eq!(p.endpoint(&ctx()), expected);
489
490        let mut other_ctx = ctx();
491        other_ctx.request.host = "totally-different-host".to_string();
492        assert_eq!(p.endpoint(&other_ctx), expected);
493    }
494
495    #[test]
496    fn test_encode_path_segment_passes_literal_values_byte_identical() {
497        // Existing literal configs (already-safe unreserved characters) must
498        // round-trip unchanged.
499        assert_eq!(encode_path_segment("guest"), "guest");
500        assert_eq!(encode_path_segment("my_pkg"), "my_pkg");
501        assert_eq!(encode_path_segment("my-pkg.v2"), "my-pkg.v2");
502        assert_eq!(encode_path_segment("_-~.Az09"), "_-~.Az09");
503    }
504
505    #[test]
506    fn test_encode_path_segment_escapes_reserved_bytes() {
507        assert_eq!(
508            encode_path_segment("foo/actions/bar"),
509            "foo%2Factions%2Fbar"
510        );
511        assert_eq!(encode_path_segment("hello?x=1"), "hello%3Fx%3D1");
512        assert_eq!(encode_path_segment("a&b"), "a%26b");
513        assert_eq!(encode_path_segment(" "), "%20");
514    }
515
516    #[test]
517    fn test_endpoint_namespace_slash_cannot_break_out_of_path_segment() {
518        // SECURITY: a templated `namespace` rendered from request data must
519        // not be able to inject an extra path segment via an unescaped `/`.
520        let mut cfg = base_config();
521        cfg["namespace"] = serde_json::json!("{{request.headers.x-tenant}}");
522        let p = plugin(cfg);
523
524        let mut request_ctx = ctx();
525        request_ctx
526            .request
527            .headers
528            .insert("x-tenant".to_string(), vec!["foo/actions/bar".to_string()]);
529        assert_eq!(
530            p.endpoint(&request_ctx),
531            "https://ow.example.com/api/v1/namespaces/foo%2Factions%2Fbar/actions/hello?blocking=true&result=true&timeout=3000"
532        );
533    }
534
535    #[test]
536    fn test_endpoint_action_question_mark_cannot_inject_query_params() {
537        // SECURITY: a templated `action` rendered from request data must not
538        // be able to inject extra query parameters ahead of the plugin's own
539        // `blocking`/`result`/`timeout`.
540        let mut cfg = base_config();
541        cfg["action"] = serde_json::json!("{{request.headers.x-action}}");
542        let p = plugin(cfg);
543
544        let mut request_ctx = ctx();
545        request_ctx
546            .request
547            .headers
548            .insert("x-action".to_string(), vec!["hello?x=1".to_string()]);
549        let endpoint = p.endpoint(&request_ctx);
550        assert_eq!(
551            endpoint,
552            "https://ow.example.com/api/v1/namespaces/guest/actions/hello%3Fx%3D1?blocking=true&result=true&timeout=3000"
553        );
554        // Exactly one '?' in the whole URL — the plugin's own query string.
555        assert_eq!(endpoint.matches('?').count(), 1);
556    }
557
558    #[test]
559    fn test_map_response_status_code_and_headers_and_body() {
560        let raw = Bytes::from_static(
561            br#"{"statusCode":201,"headers":{"Content-Type":"application/json"},"body":"hi"}"#,
562        );
563        let mapped = map_response(200, &raw).unwrap();
564        assert_eq!(mapped.status, 201);
565        assert_eq!(mapped.body, Bytes::from_static(b"hi"));
566        assert_eq!(
567            mapped.headers.get("content-type"),
568            Some(&vec!["application/json".to_string()])
569        );
570    }
571
572    #[test]
573    fn test_map_response_falls_back_to_transport_status() {
574        let raw = Bytes::from_static(br#"{"greeting":"hello"}"#);
575        let mapped = map_response(200, &raw).unwrap();
576        assert_eq!(mapped.status, 200);
577        // no `body` field -> raw envelope passed through
578        assert_eq!(mapped.body, raw);
579    }
580
581    #[test]
582    fn test_map_response_empty_body_passthrough() {
583        let mapped = map_response(204, &Bytes::new()).unwrap();
584        assert_eq!(mapped.status, 204);
585        assert!(mapped.body.is_empty());
586    }
587
588    #[test]
589    fn test_map_response_invalid_json_errors() {
590        assert!(map_response(200, &Bytes::from_static(b"not json")).is_err());
591    }
592
593    #[tokio::test]
594    async fn test_callout_failure_routes_error() {
595        let mut cfg = base_config();
596        cfg["api_host"] = serde_json::json!("http://127.0.0.1:1");
597        cfg["timeout"] = serde_json::json!(200);
598        let p = plugin(cfg);
599        let err = p.execute(ctx()).await.unwrap_err();
600        assert_eq!(err.error.code, "OPENWHISK_CALLOUT_ERROR");
601        assert!(err.context.response.status_code >= 502);
602    }
603}
featherbit/plugins/native/openwhisk.rs

featherbit/plugins/native/
openwhisk.rs
