featherbit/plugins/native/
openfunction.rs

1//! OpenFunction serverless-upstream plugin (`openfunction`).
2//!
3//! Port of APISIX's `openfunction` plugin (3.17). Forwards the request to an
4//! OpenFunction endpoint and returns the function's reply as the gateway
5//! response — it **replaces the upstream**, so the node's `success` port should
6//! be wired straight to `client.in`.
7//!
8//! Authorization: when `authorization.service_token` is set, it is sent as an
9//! `Authorization: Basic <base64(service_token)>` header (matching APISIX,
10//! which base64-encodes the raw token verbatim).
11//!
12//! On a callout failure the node rejects through its `error` port with
13//! `OPENFUNCTION_CALLOUT_ERROR` (a 502/503/504 depending on the failure).
14
15use async_trait::async_trait;
16use base64::engine::general_purpose::STANDARD as BASE64;
17use base64::Engine;
18use std::collections::HashMap;
19use std::sync::Arc;
20use std::time::Duration;
21
22use crate::context::{Context, GatewayError};
23use crate::outbound::{OutboundClient, OutboundRequest, OutboundResponse};
24use crate::plugins::resources::PluginResources;
25use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
26
27use super::faas;
28
29/// Forwards the request to an OpenFunction endpoint and maps its reply into
30/// `Context.response`.
31pub struct OpenFunctionPlugin {
32    function_uri: String,
33    /// Pre-computed `Basic <base64(service_token)>` header value, if configured.
34    authorization: Option<String>,
35    ssl_verify: bool,
36    timeout: Duration,
37    client: Arc<OutboundClient>,
38}
39
40impl OpenFunctionPlugin {
41    /// Builds the plugin from node config.
42    ///
43    /// Accepted keys:
44    /// - `function_uri` (string, **required**): the OpenFunction URL. A
45    ///   missing/empty value is a config-load error.
46    /// - `authorization` (object, optional):
47    ///   - `service_token` (string) — sent as `Authorization: Basic
48    ///     <base64(service_token)>`.
49    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
50    /// - `timeout` (integer ms, default `3000`): whole-call deadline.
51    ///
52    /// ```yaml
53    /// type: openfunction
54    /// config:
55    ///   function_uri: http://openfunction.svc/default/hello
56    ///   authorization:
57    ///     service_token: ${OPENFUNCTION_TOKEN}
58    ///   ssl_verify: true
59    ///   timeout: 3000
60    /// ```
61    pub fn from_config(
62        config: &HashMap<String, serde_json::Value>,
63        resources: &Arc<PluginResources>,
64    ) -> Result<Self, String> {
65        let function_uri = config
66            .get("function_uri")
67            .and_then(|v| v.as_str())
68            .filter(|s| !s.is_empty())
69            .ok_or_else(|| "openfunction plugin requires 'function_uri'".to_string())?
70            .to_string();
71
72        let authorization = config
73            .get("authorization")
74            .and_then(|v| v.as_object())
75            .and_then(|a| a.get("service_token"))
76            .and_then(|v| v.as_str())
77            .filter(|s| !s.is_empty())
78            .map(|token| format!("Basic {}", BASE64.encode(token)));
79
80        let ssl_verify = config
81            .get("ssl_verify")
82            .and_then(|v| v.as_bool())
83            .unwrap_or(true);
84
85        let timeout = Duration::from_millis(
86            config
87                .get("timeout")
88                .and_then(|v| v.as_u64())
89                .unwrap_or(3000),
90        );
91
92        Ok(Self {
93            function_uri,
94            authorization,
95            ssl_verify,
96            timeout,
97            client: resources.outbound.clone(),
98        })
99    }
100
101    /// Builds the outbound request: the client's method/body/query forwarded to
102    /// `function_uri` plus the optional `Authorization: Basic` header.
103    fn build_request(&self, ctx: &Context) -> Result<OutboundRequest, String> {
104        let (mut headers, url, method) = faas::forward_parts(&self.function_uri, ctx)?;
105        if let Some(authz) = &self.authorization {
106            headers.retain(|(k, _)| k != "authorization");
107            headers.push(("authorization".to_string(), authz.clone()));
108        }
109        Ok(OutboundRequest {
110            method,
111            url,
112            headers,
113            body: ctx.request.body.clone(),
114            timeout: self.timeout,
115            ssl_verify: self.ssl_verify,
116            tls: None,
117        })
118    }
119}
120
121/// Copies the FaaS reply into `Context.response`.
122fn apply_response(ctx: &mut Context, response: OutboundResponse) {
123    ctx.response.status_code = response.status;
124    ctx.response.headers = response.headers;
125    ctx.response.body = response.body;
126}
127
128#[async_trait]
129impl Plugin for OpenFunctionPlugin {
130    fn plugin_type(&self) -> &str {
131        "openfunction"
132    }
133
134    async fn execute(
135        &self,
136        mut ctx: Context,
137        _named_inputs: &HashMap<String, serde_json::Value>,
138    ) -> PluginResult {
139        let request = match self.build_request(&ctx) {
140            Ok(req) => req,
141            Err(message) => return Err(reject(ctx, 502, message)),
142        };
143
144        match self.client.request(request).await {
145            Ok(response) => {
146                apply_response(&mut ctx, response);
147                Ok(PluginOutput {
148                    context: ctx,
149                    named_outputs: HashMap::new(),
150                })
151            }
152            Err(e) => {
153                let (status, message) = faas::classify_error("openfunction", &e);
154                Err(reject(ctx, status, message))
155            }
156        }
157    }
158}
159
160/// Builds the `OPENFUNCTION_CALLOUT_ERROR` rejection carrying the context.
161fn reject(mut ctx: Context, status: u16, message: String) -> PluginExecutionError {
162    ctx.response.status_code = status;
163    PluginExecutionError {
164        context: ctx,
165        error: GatewayError {
166            node_id: String::new(),
167            code: "OPENFUNCTION_CALLOUT_ERROR".to_string(),
168            message,
169            metadata: HashMap::new(),
170        },
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
178    use bytes::Bytes;
179
180    fn ctx() -> Context {
181        Context {
182            request: GatewayRequest {
183                method: "POST".to_string(),
184                path: "/orig".to_string(),
185                host: "gw".to_string(),
186                scheme: "http".to_string(),
187                headers: HashMap::new(),
188                query_params: HashMap::new(),
189                body: Bytes::from_static(b"payload"),
190                remote_addr: "1.2.3.4:5".to_string(),
191                protocol: Protocol::Http1,
192            },
193            response: GatewayResponse {
194                status_code: 0,
195                headers: HashMap::new(),
196                body: Bytes::new(),
197            },
198            message: HashMap::new(),
199            errors: Vec::new(),
200        }
201    }
202
203    fn plugin(config: serde_json::Value) -> OpenFunctionPlugin {
204        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
205        OpenFunctionPlugin::from_config(&map, &PluginResources::empty()).unwrap()
206    }
207
208    #[test]
209    fn test_requires_function_uri() {
210        assert!(
211            OpenFunctionPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err()
212        );
213    }
214
215    #[test]
216    fn test_basic_auth_header_base64() {
217        let p = plugin(serde_json::json!({
218            "function_uri": "http://of.svc/default/hello",
219            "authorization": { "service_token": "user:pass" }
220        }));
221        let req = p.build_request(&ctx()).unwrap();
222        let authz = req
223            .headers
224            .iter()
225            .find(|(k, _)| k == "authorization")
226            .map(|(_, v)| v.clone())
227            .unwrap();
228        // "user:pass" base64 = dXNlcjpwYXNz
229        assert_eq!(authz, "Basic dXNlcjpwYXNz");
230        assert_eq!(req.url, "http://of.svc/default/hello");
231    }
232
233    #[test]
234    fn test_no_auth_when_absent() {
235        let p = plugin(serde_json::json!({ "function_uri": "http://of.svc/fn" }));
236        let req = p.build_request(&ctx()).unwrap();
237        assert!(!req.headers.iter().any(|(k, _)| k == "authorization"));
238    }
239
240    #[tokio::test]
241    async fn test_callout_failure_routes_error() {
242        let p = plugin(serde_json::json!({
243            "function_uri": "http://127.0.0.1:1/fn",
244            "timeout": 200
245        }));
246        let err = p.execute(ctx(), &HashMap::new()).await.unwrap_err();
247        assert_eq!(err.error.code, "OPENFUNCTION_CALLOUT_ERROR");
248        assert!(err.context.response.status_code >= 502);
249    }
250}
featherbit/plugins/native/openfunction.rs

featherbit/plugins/native/
openfunction.rs
