featherbit/plugins/native/
opa.rs

1//! Open Policy Agent authorization plugin (`opa`).
2//!
3//! Delegates the access decision for each request to an external OPA server.
4//! The plugin builds an OPA *input document* describing the request (and,
5//! optionally, the matched consumer), POSTs it to
6//! `<host>/v1/data/<policy>`, and interprets the decision under `result`:
7//! `allow: true` continues (optionally copying selected OPA-provided headers
8//! onto the request forwarded upstream), while `allow: false`/missing rejects,
9//! honoring OPA-supplied status, headers, and reason. A callout or
10//! response-parse failure blocks the request by default.
11//!
12//! Ports the APISIX `opa` plugin (and its `opa/helper.lua` input builder) onto
13//! featherbit's shared outbound HTTP client.
14//!
15//! ## Deviations from APISIX
16//!
17//! featherbit has no route/service objects, so `with_route` and `with_service`
18//! are accepted for config compatibility but are **no-ops** (no `route` /
19//! `service` object is added to the input document). The input `var` block
20//! omits `server_addr` / `server_port`, which featherbit does not track in the
21//! request context.
22
23use async_trait::async_trait;
24use bytes::Bytes;
25use std::collections::HashMap;
26use std::sync::Arc;
27use std::time::{Duration, SystemTime, UNIX_EPOCH};
28
29use crate::context::{Context, GatewayError};
30use crate::outbound::{OutboundClient, OutboundRequest};
31use crate::plugins::resources::PluginResources;
32use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
33
34/// Sends an OPA input document to an external policy server and routes on the
35/// returned decision: `allow: true` continues (success port), otherwise
36/// rejects (error port).
37pub struct OpaPlugin {
38    /// OPA base URL (e.g. `http://opa:8181`).
39    host: String,
40    /// Decision path appended as `/v1/data/<policy>`.
41    policy: String,
42    /// TLS certificate verification for `https` callouts.
43    ssl_verify: bool,
44    /// Whole-call callout deadline.
45    timeout: Duration,
46    /// Include the matched consumer object in the input document.
47    with_consumer: bool,
48    /// OPA-response header names copied onto the request forwarded upstream on
49    /// allow (lowercased). Empty means none are copied.
50    send_headers_upstream: Vec<String>,
51    /// Shared pooled HTTP client (from [`PluginResources`]).
52    client: Arc<OutboundClient>,
53}
54
55/// A parsed OPA decision (the `result` object of the response).
56#[derive(Debug, PartialEq)]
57struct OpaDecision {
58    allow: bool,
59    /// Deny status from `result.status_code` (or `result.status`), if present.
60    status: Option<u16>,
61    /// Stringified `result.reason`, if present (objects are JSON-encoded).
62    reason: Option<String>,
63    /// `result.headers` as lowercased name→value pairs.
64    headers: HashMap<String, String>,
65}
66
67/// Why an OPA response could not be interpreted as a decision.
68#[derive(Debug, PartialEq)]
69enum OpaParseError {
70    /// Body was not valid JSON.
71    NotJson,
72    /// Body parsed but had no `result` object.
73    MissingResult,
74}
75
76impl OpaPlugin {
77    /// Builds the plugin from node config.
78    ///
79    /// Accepted keys:
80    /// - `host` (string, **required**): OPA base URL. Missing → config error.
81    /// - `policy` (string, **required**): decision path appended as
82    ///   `/v1/data/<policy>`. Missing → config error.
83    /// - `ssl_verify` (bool, default `true`): verify TLS certificates for
84    ///   `https` callouts.
85    /// - `timeout` (integer ms, default `3000`): whole-call callout deadline.
86    /// - `with_consumer` (bool, default `false`): include a `consumer` object
87    ///   (built from `context.message`'s `consumer.*` keys) in the input
88    ///   document.
89    /// - `with_route` / `with_service` (bool, default `false`): accepted for
90    ///   APISIX compatibility but **no-ops** — featherbit has no route/service
91    ///   objects.
92    /// - `send_headers_upstream` (array of strings, optional): OPA-response
93    ///   header names copied onto the request forwarded upstream on allow. A
94    ///   configured name absent from the OPA response removes any
95    ///   client-supplied value.
96    ///
97    /// ```yaml
98    /// type: opa
99    /// config:
100    ///   host: http://opa:8181
101    ///   policy: example/allow
102    ///   with_consumer: true
103    ///   send_headers_upstream: [x-user-id]
104    ///   ssl_verify: true
105    ///   timeout: 3000
106    /// ```
107    pub fn from_config(
108        config: &HashMap<String, serde_json::Value>,
109        resources: &Arc<PluginResources>,
110    ) -> Result<Self, String> {
111        let host = config
112            .get("host")
113            .and_then(|v| v.as_str())
114            .filter(|s| !s.is_empty())
115            .ok_or_else(|| "opa plugin requires 'host'".to_string())?
116            .trim_end_matches('/')
117            .to_string();
118
119        let policy = config
120            .get("policy")
121            .and_then(|v| v.as_str())
122            .filter(|s| !s.is_empty())
123            .ok_or_else(|| "opa plugin requires 'policy'".to_string())?
124            .trim_start_matches('/')
125            .to_string();
126
127        let ssl_verify = config
128            .get("ssl_verify")
129            .and_then(|v| v.as_bool())
130            .unwrap_or(true);
131
132        let timeout = Duration::from_millis(
133            config
134                .get("timeout")
135                .and_then(|v| v.as_u64())
136                .unwrap_or(3000),
137        );
138
139        let with_consumer = config
140            .get("with_consumer")
141            .and_then(|v| v.as_bool())
142            .unwrap_or(false);
143
144        let send_headers_upstream = config
145            .get("send_headers_upstream")
146            .and_then(|v| v.as_array())
147            .map(|seq| {
148                seq.iter()
149                    .filter_map(|v| v.as_str().map(|s| s.to_lowercase()))
150                    .collect()
151            })
152            .unwrap_or_default();
153
154        Ok(Self {
155            host,
156            policy,
157            ssl_verify,
158            timeout,
159            with_consumer,
160            send_headers_upstream,
161            client: resources.outbound.clone(),
162        })
163    }
164
165    /// Builds the OPA input document from the context, mirroring
166    /// `opa/helper.lua`'s `build_opa_input` as closely as featherbit's
167    /// [`Context`] allows. `now_unix` is injected so the (otherwise
168    /// wall-clock) `var.timestamp` is testable.
169    fn build_opa_input(&self, ctx: &Context, now_unix: u64) -> serde_json::Value {
170        let (host, port) = split_host_port(&ctx.request.host, &ctx.request.scheme);
171        let (remote_addr, remote_port) = split_addr_port(&ctx.request.remote_addr);
172
173        let request = serde_json::json!({
174            "scheme": ctx.request.scheme,
175            "method": ctx.request.method,
176            "host": host,
177            "port": port,
178            "path": ctx.request.path,
179            "headers": collapse_multi(&ctx.request.headers),
180            "query": collapse_multi(&ctx.request.query_params),
181        });
182
183        let mut var = serde_json::Map::new();
184        var.insert("remote_addr".to_string(), serde_json::json!(remote_addr));
185        if let Some(p) = remote_port {
186            var.insert("remote_port".to_string(), serde_json::json!(p));
187        }
188        var.insert("timestamp".to_string(), serde_json::json!(now_unix));
189
190        let mut input = serde_json::Map::new();
191        input.insert("type".to_string(), serde_json::json!("http"));
192        input.insert("request".to_string(), request);
193        input.insert("var".to_string(), serde_json::Value::Object(var));
194
195        if self.with_consumer {
196            if let Some(consumer) = build_consumer(ctx) {
197                input.insert("consumer".to_string(), consumer);
198            }
199        }
200
201        serde_json::json!({ "input": serde_json::Value::Object(input) })
202    }
203
204    /// On an allow decision, copies the configured `send_headers_upstream`
205    /// from the OPA response onto the request forwarded upstream. A configured
206    /// header absent from the OPA response removes any client-supplied value.
207    fn apply_allow(&self, ctx: &mut Context, decision: &OpaDecision) {
208        for name in &self.send_headers_upstream {
209            match decision.headers.get(name) {
210                Some(value) => {
211                    ctx.request
212                        .headers
213                        .insert(name.clone(), vec![value.clone()]);
214                }
215                None => {
216                    ctx.request.headers.remove(name);
217                }
218            }
219        }
220    }
221
222    /// Builds the `OPA_DENIED` rejection from a deny decision, honoring
223    /// OPA-supplied status (default `403`), headers, and reason (as the body).
224    fn build_deny(&self, mut ctx: Context, decision: &OpaDecision) -> PluginExecutionError {
225        let status = decision.status.unwrap_or(403);
226        ctx.response.status_code = status;
227        if let Some(reason) = &decision.reason {
228            ctx.response.body = Bytes::from(reason.clone());
229        }
230        for (name, value) in &decision.headers {
231            ctx.response
232                .headers
233                .insert(name.clone(), vec![value.clone()]);
234        }
235        PluginExecutionError {
236            context: ctx,
237            error: GatewayError {
238                node_id: String::new(),
239                code: "OPA_DENIED".to_string(),
240                message: format!("OPA denied the request ({})", status),
241                metadata: HashMap::new(),
242            },
243        }
244    }
245
246    /// Builds the `OPA_ERROR` rejection used when the callout fails or the
247    /// response cannot be interpreted as a decision.
248    fn build_error(&self, mut ctx: Context, status: u16, message: String) -> PluginExecutionError {
249        ctx.response.status_code = status;
250        PluginExecutionError {
251            context: ctx,
252            error: GatewayError {
253                node_id: String::new(),
254                code: "OPA_ERROR".to_string(),
255                message,
256                metadata: HashMap::new(),
257            },
258        }
259    }
260}
261
262/// Parses an OPA response body into an [`OpaDecision`]. Pure and network-free
263/// so the decision-mapping logic is unit-testable.
264fn parse_opa_response(body: &[u8]) -> Result<OpaDecision, OpaParseError> {
265    let data: serde_json::Value =
266        serde_json::from_slice(body).map_err(|_| OpaParseError::NotJson)?;
267    let result = data.get("result").ok_or(OpaParseError::MissingResult)?;
268
269    let allow = result
270        .get("allow")
271        .and_then(|v| v.as_bool())
272        .unwrap_or(false);
273
274    let status = result
275        .get("status_code")
276        .or_else(|| result.get("status"))
277        .and_then(|v| v.as_u64())
278        .map(|n| n as u16);
279
280    let reason = result.get("reason").and_then(|v| match v {
281        serde_json::Value::Null => None,
282        serde_json::Value::String(s) => Some(s.clone()),
283        other => Some(other.to_string()),
284    });
285
286    let headers = result
287        .get("headers")
288        .and_then(|v| v.as_object())
289        .map(|obj| {
290            obj.iter()
291                .filter_map(|(k, v)| {
292                    let value = match v {
293                        serde_json::Value::String(s) => s.clone(),
294                        serde_json::Value::Number(n) => n.to_string(),
295                        serde_json::Value::Bool(b) => b.to_string(),
296                        _ => return None,
297                    };
298                    Some((k.to_lowercase(), value))
299                })
300                .collect()
301        })
302        .unwrap_or_default();
303
304    Ok(OpaDecision {
305        allow,
306        status,
307        reason,
308        headers,
309    })
310}
311
312/// Collapses a multi-valued header/query map into single strings where there
313/// is one value and arrays where there are several — mirroring OpenResty's
314/// `core.request.headers` / `get_uri_args` shape.
315fn collapse_multi(map: &HashMap<String, Vec<String>>) -> serde_json::Value {
316    let mut out = serde_json::Map::new();
317    for (k, values) in map {
318        let value = match values.as_slice() {
319            [single] => serde_json::json!(single),
320            _ => serde_json::json!(values),
321        };
322        out.insert(k.clone(), value);
323    }
324    serde_json::Value::Object(out)
325}
326
327/// Builds the `consumer` object from `context.message`'s `consumer.*` keys,
328/// stripping the prefix. Returns `None` when no consumer is attached.
329fn build_consumer(ctx: &Context) -> Option<serde_json::Value> {
330    let mut obj = serde_json::Map::new();
331    for (key, value) in &ctx.message {
332        if let Some(field) = key.strip_prefix("consumer.") {
333            obj.insert(field.to_string(), value.clone());
334        }
335    }
336    if obj.is_empty() {
337        None
338    } else {
339        Some(serde_json::Value::Object(obj))
340    }
341}
342
343/// Splits a `Host` header into `(hostname, port)`, defaulting the port from
344/// the scheme when the host carries none.
345fn split_host_port(host: &str, scheme: &str) -> (String, u16) {
346    let default_port = if scheme.eq_ignore_ascii_case("https") {
347        443
348    } else {
349        80
350    };
351    // Bracketed IPv6 literal: [::1]:8080
352    if let Some(stripped) = host.strip_prefix('[') {
353        if let Some((h, rest)) = stripped.split_once(']') {
354            let port = rest
355                .strip_prefix(':')
356                .and_then(|p| p.parse().ok())
357                .unwrap_or(default_port);
358            return (h.to_string(), port);
359        }
360    }
361    match host.rsplit_once(':') {
362        // Bare IPv6 (multiple colons) — no port.
363        Some((h, port)) if !h.contains(':') => {
364            (h.to_string(), port.parse().unwrap_or(default_port))
365        }
366        _ => (host.to_string(), default_port),
367    }
368}
369
370/// Splits a client `ip:port` into `(ip, port)`, tolerating bare IPs and
371/// bracketed IPv6.
372fn split_addr_port(addr: &str) -> (String, Option<u16>) {
373    if let Some(stripped) = addr.strip_prefix('[') {
374        if let Some((ip, rest)) = stripped.split_once(']') {
375            return (
376                ip.to_string(),
377                rest.strip_prefix(':').and_then(|p| p.parse().ok()),
378            );
379        }
380    }
381    match addr.rsplit_once(':') {
382        Some((ip, port)) if !ip.contains(':') => (ip.to_string(), port.parse().ok()),
383        _ => (addr.to_string(), None),
384    }
385}
386
387#[async_trait]
388impl Plugin for OpaPlugin {
389    fn plugin_type(&self) -> &str {
390        "opa"
391    }
392
393    async fn execute(
394        &self,
395        mut ctx: Context,
396        _named_inputs: &HashMap<String, serde_json::Value>,
397    ) -> PluginResult {
398        let now = SystemTime::now()
399            .duration_since(UNIX_EPOCH)
400            .map(|d| d.as_secs())
401            .unwrap_or(0);
402        let input = self.build_opa_input(&ctx, now);
403        let body = match serde_json::to_vec(&input) {
404            Ok(b) => Bytes::from(b),
405            Err(e) => {
406                return Err(self.build_error(
407                    ctx,
408                    503,
409                    format!("failed to encode OPA input: {}", e),
410                ))
411            }
412        };
413
414        let url = format!("{}/v1/data/{}", self.host, self.policy);
415        let request = OutboundRequest {
416            method: http::Method::POST,
417            url,
418            headers: vec![("Content-Type".to_string(), "application/json".to_string())],
419            body,
420            timeout: self.timeout,
421            ssl_verify: self.ssl_verify,
422            tls: None,
423        };
424
425        // Block by default when the decision is unavailable.
426        let response = match self.client.request(request).await {
427            Ok(resp) => resp,
428            Err(e) => return Err(self.build_error(ctx, 403, format!("OPA callout failed: {}", e))),
429        };
430
431        let decision = match parse_opa_response(&response.body) {
432            Ok(d) => d,
433            Err(e) => {
434                let message = match e {
435                    OpaParseError::NotJson => "OPA response was not valid JSON".to_string(),
436                    OpaParseError::MissingResult => {
437                        "OPA response missing 'result' field".to_string()
438                    }
439                };
440                return Err(self.build_error(ctx, 503, message));
441            }
442        };
443
444        if !decision.allow {
445            return Err(self.build_deny(ctx, &decision));
446        }
447
448        self.apply_allow(&mut ctx, &decision);
449        Ok(PluginOutput {
450            context: ctx,
451            named_outputs: HashMap::new(),
452        })
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
460
461    fn test_ctx() -> Context {
462        let mut headers = HashMap::new();
463        headers.insert("accept".to_string(), vec!["application/json".to_string()]);
464        headers.insert(
465            "x-multi".to_string(),
466            vec!["a".to_string(), "b".to_string()],
467        );
468        let mut query = HashMap::new();
469        query.insert("q".to_string(), vec!["1".to_string()]);
470        let mut message = HashMap::new();
471        message.insert("consumer.name".to_string(), serde_json::json!("alice"));
472        message.insert(
473            "consumer.auth_type".to_string(),
474            serde_json::json!("key-auth"),
475        );
476        Context {
477            request: GatewayRequest {
478                method: "GET".to_string(),
479                path: "/api/users".to_string(),
480                host: "example.com:8080".to_string(),
481                scheme: "http".to_string(),
482                headers,
483                query_params: query,
484                body: Bytes::new(),
485                remote_addr: "10.0.0.7:5555".to_string(),
486                protocol: Protocol::Http1,
487            },
488            response: GatewayResponse {
489                status_code: 0,
490                headers: HashMap::new(),
491                body: Bytes::new(),
492            },
493            message,
494            errors: Vec::new(),
495        }
496    }
497
498    fn plugin(config: serde_json::Value) -> OpaPlugin {
499        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
500        OpaPlugin::from_config(&map, &PluginResources::empty()).unwrap()
501    }
502
503    #[test]
504    fn test_requires_host_and_policy() {
505        assert!(OpaPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err());
506        let mut map = HashMap::new();
507        map.insert("host".to_string(), serde_json::json!("http://opa"));
508        assert!(OpaPlugin::from_config(&map, &PluginResources::empty()).is_err());
509        map.insert("policy".to_string(), serde_json::json!("p/allow"));
510        assert!(OpaPlugin::from_config(&map, &PluginResources::empty()).is_ok());
511    }
512
513    #[test]
514    fn test_host_and_policy_trimmed() {
515        let p = plugin(serde_json::json!({ "host": "http://opa/", "policy": "/p/allow" }));
516        assert_eq!(p.host, "http://opa");
517        assert_eq!(p.policy, "p/allow");
518    }
519
520    #[test]
521    fn test_build_input_shape() {
522        let p = plugin(serde_json::json!({
523            "host": "http://opa",
524            "policy": "p/allow",
525            "with_consumer": true
526        }));
527        let input = p.build_opa_input(&test_ctx(), 1234);
528        let req = &input["input"]["request"];
529        assert_eq!(req["scheme"], serde_json::json!("http"));
530        assert_eq!(req["method"], serde_json::json!("GET"));
531        assert_eq!(req["host"], serde_json::json!("example.com"));
532        assert_eq!(req["port"], serde_json::json!(8080));
533        assert_eq!(req["path"], serde_json::json!("/api/users"));
534        // single header collapses to string, multi stays array
535        assert_eq!(
536            req["headers"]["accept"],
537            serde_json::json!("application/json")
538        );
539        assert_eq!(req["headers"]["x-multi"], serde_json::json!(["a", "b"]));
540        assert_eq!(req["query"]["q"], serde_json::json!("1"));
541        assert_eq!(
542            input["input"]["var"]["remote_addr"],
543            serde_json::json!("10.0.0.7")
544        );
545        assert_eq!(
546            input["input"]["var"]["remote_port"],
547            serde_json::json!(5555)
548        );
549        assert_eq!(input["input"]["var"]["timestamp"], serde_json::json!(1234));
550        // consumer built from message consumer.* keys
551        assert_eq!(
552            input["input"]["consumer"]["name"],
553            serde_json::json!("alice")
554        );
555        assert_eq!(
556            input["input"]["consumer"]["auth_type"],
557            serde_json::json!("key-auth")
558        );
559    }
560
561    #[test]
562    fn test_build_input_omits_consumer_when_disabled() {
563        let p = plugin(serde_json::json!({ "host": "http://opa", "policy": "p" }));
564        let input = p.build_opa_input(&test_ctx(), 0);
565        assert!(input["input"].get("consumer").is_none());
566    }
567
568    #[test]
569    fn test_parse_allow() {
570        let d = parse_opa_response(br#"{"result": {"allow": true}}"#).unwrap();
571        assert!(d.allow);
572        assert_eq!(d.status, None);
573    }
574
575    #[test]
576    fn test_parse_deny_with_details() {
577        let body = br#"{"result": {"allow": false, "status_code": 401,
578            "reason": "no", "headers": {"WWW-Authenticate": "Bearer"}}}"#;
579        let d = parse_opa_response(body).unwrap();
580        assert!(!d.allow);
581        assert_eq!(d.status, Some(401));
582        assert_eq!(d.reason.as_deref(), Some("no"));
583        assert_eq!(
584            d.headers.get("www-authenticate"),
585            Some(&"Bearer".to_string())
586        );
587    }
588
589    #[test]
590    fn test_parse_object_reason_encoded() {
591        let d =
592            parse_opa_response(br#"{"result": {"allow": false, "reason": {"m": "x"}}}"#).unwrap();
593        assert_eq!(d.reason.as_deref(), Some("{\"m\":\"x\"}"));
594    }
595
596    #[test]
597    fn test_parse_errors() {
598        assert_eq!(parse_opa_response(b"not json"), Err(OpaParseError::NotJson));
599        assert_eq!(
600            parse_opa_response(br#"{"foo": 1}"#),
601            Err(OpaParseError::MissingResult)
602        );
603        // missing allow defaults to false (deny)
604        let d = parse_opa_response(br#"{"result": {}}"#).unwrap();
605        assert!(!d.allow);
606    }
607
608    #[test]
609    fn test_build_deny_applies_status_headers_reason() {
610        let p = plugin(serde_json::json!({ "host": "http://opa", "policy": "p" }));
611        let decision = OpaDecision {
612            allow: false,
613            status: Some(401),
614            reason: Some("denied".to_string()),
615            headers: HashMap::from([("x-why".to_string(), "policy".to_string())]),
616        };
617        let err = p.build_deny(test_ctx(), &decision);
618        assert_eq!(err.error.code, "OPA_DENIED");
619        assert_eq!(err.context.response.status_code, 401);
620        assert_eq!(err.context.response.body, Bytes::from_static(b"denied"));
621        assert_eq!(
622            err.context.response.headers.get("x-why"),
623            Some(&vec!["policy".to_string()])
624        );
625    }
626
627    #[test]
628    fn test_apply_allow_sets_and_removes() {
629        let p = plugin(serde_json::json!({
630            "host": "http://opa",
631            "policy": "p",
632            "send_headers_upstream": ["X-User-Id", "X-Absent"]
633        }));
634        let mut ctx = test_ctx();
635        ctx.request
636            .headers
637            .insert("x-absent".to_string(), vec!["stale".to_string()]);
638        let decision = OpaDecision {
639            allow: true,
640            status: None,
641            reason: None,
642            headers: HashMap::from([("x-user-id".to_string(), "u42".to_string())]),
643        };
644        p.apply_allow(&mut ctx, &decision);
645        assert_eq!(
646            ctx.request.headers.get("x-user-id"),
647            Some(&vec!["u42".to_string()])
648        );
649        assert!(!ctx.request.headers.contains_key("x-absent"));
650    }
651
652    #[test]
653    fn test_split_host_port() {
654        assert_eq!(split_host_port("h:8080", "http"), ("h".to_string(), 8080));
655        assert_eq!(split_host_port("h", "https"), ("h".to_string(), 443));
656        assert_eq!(split_host_port("h", "http"), ("h".to_string(), 80));
657        assert_eq!(
658            split_host_port("[::1]:9000", "http"),
659            ("::1".to_string(), 9000)
660        );
661    }
662}
featherbit/plugins/native/opa.rs

featherbit/plugins/native/
opa.rs
