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};
26use crate::vars::template::Template;
27
28use super::faas;
29
30/// Forwards the request to an OpenFunction endpoint and maps its reply into
31/// `Context.response`.
32pub struct OpenFunctionPlugin {
33    /// Supports `{{namespace.path}}` template references, rendered per
34    /// request.
35    function_uri: Template,
36    /// Pre-computed `Basic <base64(service_token)>` header value, if configured.
37    authorization: Option<String>,
38    ssl_verify: bool,
39    timeout: Duration,
40    client: Arc<OutboundClient>,
41}
42
43impl OpenFunctionPlugin {
44    /// Builds the plugin from node config.
45    ///
46    /// Accepted keys:
47    /// - `function_uri` (string, **required**): the OpenFunction URL. A
48    ///   missing/empty value is a config-load error.
49    /// - `authorization` (object, optional):
50    ///   - `service_token` (string) — sent as `Authorization: Basic
51    ///     <base64(service_token)>`.
52    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
53    /// - `timeout` (integer ms, default `3000`): whole-call deadline.
54    ///
55    /// ```yaml
56    /// type: openfunction
57    /// config:
58    ///   function_uri: http://openfunction.svc/default/hello
59    ///   authorization:
60    ///     service_token: ${OPENFUNCTION_TOKEN}
61    ///   ssl_verify: true
62    ///   timeout: 3000
63    /// ```
64    pub fn from_config(
65        config: &HashMap<String, serde_json::Value>,
66        resources: &Arc<PluginResources>,
67    ) -> Result<Self, String> {
68        let function_uri = config
69            .get("function_uri")
70            .and_then(|v| v.as_str())
71            .filter(|s| !s.is_empty())
72            .ok_or_else(|| "openfunction plugin requires 'function_uri'".to_string())?;
73        // Discard warnings here — the compile-time walk (a later task)
74        // reports well-formed-but-unknown references; execution must not.
75        let function_uri = Template::parse(function_uri).0;
76
77        let authorization = config
78            .get("authorization")
79            .and_then(|v| v.as_object())
80            .and_then(|a| a.get("service_token"))
81            .and_then(|v| v.as_str())
82            .filter(|s| !s.is_empty())
83            .map(|token| format!("Basic {}", BASE64.encode(token)));
84
85        let ssl_verify = config
86            .get("ssl_verify")
87            .and_then(|v| v.as_bool())
88            .unwrap_or(true);
89
90        let timeout = Duration::from_millis(
91            config
92                .get("timeout")
93                .and_then(|v| v.as_u64())
94                .unwrap_or(3000),
95        );
96
97        Ok(Self {
98            function_uri,
99            authorization,
100            ssl_verify,
101            timeout,
102            client: resources.outbound.clone(),
103        })
104    }
105
106    /// Builds the outbound request: the client's method/body/query forwarded to
107    /// `function_uri` plus the optional `Authorization: Basic` header.
108    fn build_request(&self, ctx: &Context) -> Result<OutboundRequest, String> {
109        let function_uri = self.function_uri.render(ctx);
110        let (mut headers, url, method) = faas::forward_parts(&function_uri, ctx)?;
111        if let Some(authz) = &self.authorization {
112            headers.retain(|(k, _)| k != "authorization");
113            headers.push(("authorization".to_string(), authz.clone()));
114        }
115        Ok(OutboundRequest {
116            method,
117            url,
118            headers,
119            body: ctx.request.body.clone(),
120            timeout: self.timeout,
121            ssl_verify: self.ssl_verify,
122            tls: None,
123        })
124    }
125}
126
127/// Copies the FaaS reply into `Context.response`.
128fn apply_response(ctx: &mut Context, response: OutboundResponse) {
129    ctx.response.status_code = response.status;
130    ctx.response.headers = response.headers;
131    ctx.response.body = response.body;
132}
133
134#[async_trait]
135impl Plugin for OpenFunctionPlugin {
136    fn plugin_type(&self) -> &str {
137        "openfunction"
138    }
139
140    async fn execute(&self, mut ctx: Context) -> PluginResult {
141        let request = match self.build_request(&ctx) {
142            Ok(req) => req,
143            Err(message) => return Err(reject(ctx, 502, message)),
144        };
145
146        match self.client.request(request).await {
147            Ok(response) => {
148                apply_response(&mut ctx, response);
149                Ok(PluginOutput::success(ctx))
150            }
151            Err(e) => {
152                let (status, message) = faas::classify_error("openfunction", &e);
153                Err(reject(ctx, status, message))
154            }
155        }
156    }
157}
158
159/// Builds the `OPENFUNCTION_CALLOUT_ERROR` rejection carrying the context.
160fn reject(mut ctx: Context, status: u16, message: String) -> PluginExecutionError {
161    ctx.response.status_code = status;
162    PluginExecutionError {
163        context: ctx,
164        error: GatewayError {
165            node_id: String::new(),
166            code: "OPENFUNCTION_CALLOUT_ERROR".to_string(),
167            message,
168            metadata: HashMap::new(),
169        },
170    }
171}
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
177    use bytes::Bytes;
178
179    fn ctx() -> Context {
180        Context {
181            request: GatewayRequest {
182                method: "POST".to_string(),
183                path: "/orig".to_string(),
184                host: "gw".to_string(),
185                scheme: "http".to_string(),
186                headers: HashMap::new(),
187                query_params: HashMap::new(),
188                body: Bytes::from_static(b"payload"),
189                remote_addr: "1.2.3.4:5".to_string(),
190                protocol: Protocol::Http1,
191            },
192            response: GatewayResponse {
193                status_code: 0,
194                headers: HashMap::new(),
195                body: Bytes::new(),
196                stream: None,
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_function_uri_renders_template() {
235        let p = plugin(serde_json::json!({
236            "function_uri": "http://of.svc/{{request.headers.x-fn}}"
237        }));
238        let mut c = ctx();
239        c.request
240            .headers
241            .insert("x-fn".to_string(), vec!["hello".to_string()]);
242        let req = p.build_request(&c).unwrap();
243        assert_eq!(req.url, "http://of.svc/hello");
244    }
245
246    #[test]
247    fn test_no_auth_when_absent() {
248        let p = plugin(serde_json::json!({ "function_uri": "http://of.svc/fn" }));
249        let req = p.build_request(&ctx()).unwrap();
250        assert!(!req.headers.iter().any(|(k, _)| k == "authorization"));
251    }
252
253    #[tokio::test]
254    async fn test_callout_failure_routes_error() {
255        let p = plugin(serde_json::json!({
256            "function_uri": "http://127.0.0.1:1/fn",
257            "timeout": 200
258        }));
259        let err = p.execute(ctx()).await.unwrap_err();
260        assert_eq!(err.error.code, "OPENFUNCTION_CALLOUT_ERROR");
261        assert!(err.context.response.status_code >= 502);
262    }
263}
featherbit/plugins/native/openfunction.rs

featherbit/plugins/native/
openfunction.rs
