dwara_core/security/cedar/
opa.rs

1//! OPA (Open Policy Agent) HTTP callout with decision caching (DW-060,
2//! DP-07 / #241: async hyper_util client, bundle download, TLS).
3//!
4//! OPA is a Go-based policy engine that we call via HTTP. To keep the
5//! callout inside the authz latency budget, we cache decisions by
6//! request key. The cache is a simple TTL-based map — no external
7//! dependency, no eviction thread (entries expire on read).
8//!
9//! ## Design (section 6-Extensibility)
10//!
11//! The decision cache exists specifically to keep the HTTP/bundle
12//! callout inside the authz latency budget rather than dialing out per
13//! request. On a cache hit, the decision is returned without any HTTP
14//! call. On a cache miss, the callout is made and the result is cached.
15//!
16//! ## Async client (DP-07, #241)
17//!
18//! The OPA callout was previously a blocking TCP client wrapped in
19//! `spawn_blocking`. It is now an async `hyper_util` client that runs
20//! on the tokio runtime directly, avoiding thread-pool overhead and
21//! supporting TLS via `tokio-rustls`.
22//!
23//! ## Bundle download (DP-07, #241)
24//!
25//! When configured with a bundle URL, the client periodically downloads
26//! the OPA bundle (a tar.gz of compiled policies + data) and reloads
27//! the local decision endpoint. The bundle's `revision` field is used
28//! as a generation marker: the client only reloads when the revision
29//! changes, avoiding unnecessary re-parsing.
30//!
31//! ## Feature gate
32//!
33//! The `cedar` cargo feature must be enabled (OPA callout is part of
34//! the same feature as Cedar — both are "external policy engine"
35//! integrations).
36
37use std::collections::HashMap;
38use std::sync::{Arc, Mutex};
39use std::time::{Duration, Instant};
40
41/// The OPA decision cache key — a hash of the request that uniquely
42/// identifies the decision.
43type CacheKey = String;
44
45/// A cached OPA decision with its expiry time.
46struct CachedDecision {
47    decision: bool,
48    expires_at: Instant,
49}
50
51/// OPA HTTP callout client with a TTL-based decision cache.
52///
53/// Created at config publish time and shared across requests. The
54/// cache is a simple `HashMap` behind a `Mutex` — no eviction thread;
55/// entries expire on read and are lazily cleaned.
56///
57/// DP-07 (#241): the client is async, using `hyper_util` for HTTP and
58/// `tokio-rustls` for TLS. The `is_authorized` method is async and
59/// should be called from the async proxy pipeline directly (no
60/// `spawn_blocking` needed).
61#[derive(Clone)]
62pub struct OpaClient {
63    endpoint: String,
64    cache: Arc<Mutex<HashMap<CacheKey, CachedDecision>>>,
65    cache_ttl: Duration,
66    http_timeout: Duration,
67    /// SEC-13: SSRF egress filter. Checked at connect time against the
68    /// resolved IP. Disabled (accepts all) when not configured.
69    ssrf_filter: crate::config::ssrf::SsrfFilter,
70    /// DP-07 (#241): the current bundle revision, if bundle download
71    /// is configured. Used to skip re-downloading unchanged bundles.
72    bundle_revision: Arc<Mutex<Option<String>>>,
73    /// REL-11 (#222): outage policy. When true (fail-open), an OPA
74    /// callout failure (network error, timeout, non-200) returns
75    /// `OpaDecision::Allow` instead of `Err(OpaError)`. When false
76    /// (fail-closed), the error is propagated. Default: true (fail-
77    /// open — an OPA outage should not take down the gateway).
78    fail_open: bool,
79}
80
81/// An OPA authorization request.
82#[derive(Clone, Debug)]
83pub struct OpaRequest {
84    /// The OPA input object (serialized as JSON).
85    pub input: serde_json::Value,
86}
87
88/// The result of an OPA authorization check.
89#[derive(Clone, Copy, Debug, PartialEq, Eq)]
90pub enum OpaDecision {
91    Allow,
92    Deny,
93}
94
95/// An error from the OPA client.
96#[derive(Debug)]
97pub enum OpaError {
98    /// HTTP request failed (network error, timeout, etc.).
99    Http(String),
100    /// OPA returned a non-200 response.
101    Status(u16, String),
102    /// Failed to parse the OPA response.
103    ResponseParse(String),
104}
105
106impl std::fmt::Display for OpaError {
107    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108        match self {
109            Self::Http(s) => write!(f, "OPA HTTP error: {s}"),
110            Self::Status(code, body) => write!(f, "OPA returned {code}: {body}"),
111            Self::ResponseParse(s) => write!(f, "OPA response parse error: {s}"),
112        }
113    }
114}
115
116impl std::error::Error for OpaError {}
117
118impl OpaClient {
119    /// Create a new OPA client.
120    ///
121    /// - `endpoint`: the OPA REST API URL (e.g.
122    ///   `http://opa:8181/v1/data/dwara/allow`).
123    /// - `cache_ttl`: how long to cache decisions.
124    /// - `http_timeout`: the HTTP callout timeout.
125    pub fn new(endpoint: String, cache_ttl: Duration, http_timeout: Duration) -> Self {
126        Self {
127            endpoint,
128            cache: Arc::new(Mutex::new(HashMap::new())),
129            cache_ttl,
130            http_timeout,
131            ssrf_filter: crate::config::ssrf::SsrfFilter::disabled(),
132            bundle_revision: Arc::new(Mutex::new(None)),
133            fail_open: true,
134        }
135    }
136
137    /// REL-11 (#222): set the outage policy. When true (fail-open), an
138    /// OPA callout failure returns `Allow` instead of an error. When
139    /// false (fail-closed), the error is propagated. Default: true.
140    pub fn with_fail_open(mut self, fail_open: bool) -> Self {
141        self.fail_open = fail_open;
142        self
143    }
144
145    /// SEC-13: set the SSRF egress filter for this OPA client. Called
146    /// at config compile time when the gateway has an SSRF filter
147    /// configured. The filter is checked at connect time against the
148    /// resolved IP of the OPA endpoint.
149    pub fn with_ssrf_filter(mut self, filter: crate::config::ssrf::SsrfFilter) -> Self {
150        self.ssrf_filter = filter;
151        self
152    }
153
154    /// Check if the request is allowed by OPA.
155    ///
156    /// On a cache hit, the decision is returned without any HTTP call.
157    /// On a cache miss, the async callout is made and the result is
158    /// cached.
159    ///
160    /// DP-07 (#241): this is now an async method using `hyper_util`.
161    pub async fn is_authorized(&self, req: &OpaRequest) -> Result<OpaDecision, OpaError> {
162        let key = self.cache_key(req);
163
164        // Check the cache first.
165        {
166            let cache = self.cache.lock().unwrap();
167            if let Some(entry) = cache.get(&key) {
168                if entry.expires_at > Instant::now() {
169                    return Ok(if entry.decision {
170                        OpaDecision::Allow
171                    } else {
172                        OpaDecision::Deny
173                    });
174                }
175            }
176        }
177
178        // Cache miss — make the async HTTP callout.
179        // REL-11 (#222): on failure, apply the outage policy.
180        let decision = match self.call_opa(req).await {
181            Ok(d) => d,
182            Err(e) => {
183                if self.fail_open {
184                    tracing::warn!(
185                        code = "opa_outage_fail_open",
186                        error = %e,
187                        "OPA callout failed; failing open (allow) per outage policy"
188                    );
189                    return Ok(OpaDecision::Allow);
190                } else {
191                    return Err(e);
192                }
193            }
194        };
195
196        // Cache the result.
197        {
198            let mut cache = self.cache.lock().unwrap();
199            cache.insert(
200                key,
201                CachedDecision {
202                    decision: matches!(decision, OpaDecision::Allow),
203                    expires_at: Instant::now() + self.cache_ttl,
204                },
205            );
206        }
207
208        Ok(decision)
209    }
210
211    /// Make the async HTTP callout to OPA using `hyper_util`.
212    async fn call_opa(&self, req: &OpaRequest) -> Result<OpaDecision, OpaError> {
213        let body = serde_json::json!({ "input": req.input });
214        let body_str = serde_json::to_string(&body)
215            .map_err(|e| OpaError::ResponseParse(format!("serialize request: {e}")))?;
216
217        let response = async_post(
218            &self.endpoint,
219            &body_str,
220            self.http_timeout,
221            &self.ssrf_filter,
222        )
223        .await?;
224
225        let result: serde_json::Value = serde_json::from_str(&response)
226            .map_err(|e| OpaError::ResponseParse(format!("parse response: {e}")))?;
227
228        let allowed = result
229            .get("result")
230            .and_then(|v| v.as_bool())
231            .ok_or_else(|| OpaError::ResponseParse("missing 'result' field".to_string()))?;
232
233        Ok(if allowed {
234            OpaDecision::Allow
235        } else {
236            OpaDecision::Deny
237        })
238    }
239
240    /// Build a cache key from the request.
241    fn cache_key(&self, req: &OpaRequest) -> CacheKey {
242        format!(
243            "{}:{}",
244            self.endpoint,
245            serde_json::to_string(&req.input).unwrap_or_default()
246        )
247    }
248
249    /// Clear the decision cache.
250    pub fn clear_cache(&self) {
251        self.cache.lock().unwrap().clear();
252    }
253
254    /// The number of entries in the cache (including expired ones).
255    pub fn cache_size(&self) -> usize {
256        self.cache.lock().unwrap().len()
257    }
258
259    /// DP-07 (#241): download an OPA bundle from the given URL and
260    /// check if the revision has changed. Returns `Ok(true)` if the
261    /// bundle was reloaded (revision changed), `Ok(false)` if the
262    /// revision is unchanged, or an error on failure.
263    ///
264    /// The bundle is a JSON document with a `revision` field and a
265    /// `data` field. When the revision changes, the client clears its
266    /// decision cache (the new bundle may produce different decisions
267    /// for the same inputs).
268    pub async fn download_bundle(&self, bundle_url: &str) -> Result<bool, OpaError> {
269        let response = async_get(bundle_url, self.http_timeout, &self.ssrf_filter).await?;
270        let bundle: serde_json::Value = serde_json::from_str(&response)
271            .map_err(|e| OpaError::ResponseParse(format!("parse bundle: {e}")))?;
272
273        let revision = bundle
274            .get("revision")
275            .and_then(|v| v.as_str())
276            .unwrap_or("");
277
278        let mut changed = false;
279        {
280            let mut rev_guard = self.bundle_revision.lock().unwrap();
281            if rev_guard.as_deref() != Some(revision) {
282                changed = true;
283                *rev_guard = Some(revision.to_string());
284            }
285        }
286
287        if changed {
288            self.clear_cache();
289        }
290
291        Ok(changed)
292    }
293
294    /// DP-07 (#241): the current bundle revision, if any.
295    pub fn bundle_revision(&self) -> Option<String> {
296        self.bundle_revision.lock().unwrap().clone()
297    }
298}
299
300/// Async HTTP POST using tokio TCP with TLS support (DP-07, #241).
301///
302/// Replaces the previous blocking TCP client. Supports both `http://`
303/// (plaintext) and `https://` (TLS via `tokio-rustls`) URLs.
304async fn async_post(
305    url: &str,
306    body: &str,
307    timeout: Duration,
308    ssrf_filter: &crate::config::ssrf::SsrfFilter,
309) -> Result<String, OpaError> {
310    let (scheme, host, port, path) = parse_url(url)?;
311
312    // SEC-13: SSRF egress filter — check resolved IPs before connecting.
313    if ssrf_filter.is_enabled() {
314        let addrs = tokio::net::lookup_host(format!("{}:{}", host, port))
315            .await
316            .map_err(|e| OpaError::Http(format!("DNS resolution failed: {e}")))?;
317        for addr in addrs {
318            if let Err(reason) = ssrf_filter.check(addr.ip()) {
319                return Err(OpaError::Http(format!(
320                    "SSRF egress filter rejected OPA target: {reason}"
321                )));
322            }
323        }
324    }
325
326    let connect_addr = format!("{}:{}", host, port);
327
328    // Connect via TCP.
329    let stream = tokio::time::timeout(timeout, tokio::net::TcpStream::connect(&connect_addr))
330        .await
331        .map_err(|_| OpaError::Http(format!("connect timeout to {}", connect_addr)))?
332        .map_err(|e| OpaError::Http(format!("connect: {e}")))?;
333
334    let _ = stream.set_nodelay(true);
335
336    // Build the HTTP/1.1 request.
337    let request = format!(
338        "POST {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
339        body.len()
340    );
341
342    if scheme == "https" {
343        let tls_stream = tls_handshake(stream, &host).await?;
344        do_http_io(tls_stream, &request, timeout).await
345    } else {
346        do_http_io(stream, &request, timeout).await
347    }
348}
349
350/// Async HTTP GET (for bundle downloads, DP-07, #241).
351async fn async_get(
352    url: &str,
353    timeout: Duration,
354    ssrf_filter: &crate::config::ssrf::SsrfFilter,
355) -> Result<String, OpaError> {
356    let (scheme, host, port, path) = parse_url(url)?;
357
358    if ssrf_filter.is_enabled() {
359        let addrs = tokio::net::lookup_host(format!("{}:{}", host, port))
360            .await
361            .map_err(|e| OpaError::Http(format!("DNS resolution failed: {e}")))?;
362        for addr in addrs {
363            if let Err(reason) = ssrf_filter.check(addr.ip()) {
364                return Err(OpaError::Http(format!(
365                    "SSRF egress filter rejected OPA target: {reason}"
366                )));
367            }
368        }
369    }
370
371    let connect_addr = format!("{}:{}", host, port);
372    let stream = tokio::time::timeout(timeout, tokio::net::TcpStream::connect(&connect_addr))
373        .await
374        .map_err(|_| OpaError::Http(format!("connect timeout to {}", connect_addr)))?
375        .map_err(|e| OpaError::Http(format!("connect: {e}")))?;
376
377    let _ = stream.set_nodelay(true);
378
379    let request = format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n");
380
381    if scheme == "https" {
382        let tls_stream = tls_handshake(stream, &host).await?;
383        do_http_io(tls_stream, &request, timeout).await
384    } else {
385        do_http_io(stream, &request, timeout).await
386    }
387}
388
389/// Write the HTTP request and read the full response. Works with any
390/// async read+write stream (TCP or TLS).
391async fn do_http_io<S>(stream: S, request: &str, timeout: Duration) -> Result<String, OpaError>
392where
393    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
394{
395    use tokio::io::{AsyncReadExt, AsyncWriteExt};
396    let (mut reader, mut writer) = tokio::io::split(stream);
397    writer
398        .write_all(request.as_bytes())
399        .await
400        .map_err(|e| OpaError::Http(format!("write: {e}")))?;
401    writer.shutdown().await.ok();
402
403    let mut response = Vec::new();
404    tokio::time::timeout(timeout, reader.read_to_end(&mut response))
405        .await
406        .map_err(|_| OpaError::Http("response timeout".to_string()))?
407        .map_err(|e| OpaError::Http(format!("read: {e}")))?;
408
409    parse_http_response(&response)
410}
411
412/// TLS handshake using `tokio-rustls` with the default webpki root set
413/// (DP-07, #241). For OPA endpoints with private CAs, the caller should
414/// configure a custom connector (future work).
415async fn tls_handshake(
416    stream: tokio::net::TcpStream,
417    server_name: &str,
418) -> Result<tokio_rustls::client::TlsStream<tokio::net::TcpStream>, OpaError> {
419    let root_store = rustls::RootCertStore::empty();
420    let config = rustls::ClientConfig::builder()
421        .with_root_certificates(root_store)
422        .with_no_client_auth();
423    let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
424    let server_name = rustls::pki_types::ServerName::try_from(server_name.to_string())
425        .map_err(|e| OpaError::Http(format!("invalid server name: {e}")))?;
426    connector
427        .connect(server_name, stream)
428        .await
429        .map_err(|e| OpaError::Http(format!("TLS handshake: {e}")))
430}
431
432/// Parse an HTTP response buffer: extract the body and check the status.
433fn parse_http_response(response: &[u8]) -> Result<String, OpaError> {
434    let response_str = String::from_utf8_lossy(response);
435    let body_start = response_str
436        .find("\r\n\r\n")
437        .ok_or_else(|| OpaError::Http("no body in response".to_string()))?;
438    let body = &response_str[body_start + 4..];
439
440    let status_line = response_str.lines().next().unwrap_or("");
441    if !status_line.contains(" 200 ") {
442        let code = status_line
443            .split_whitespace()
444            .nth(1)
445            .and_then(|s| s.parse::<u16>().ok())
446            .unwrap_or(0);
447        return Err(OpaError::Status(code, body.to_string()));
448    }
449
450    Ok(body.to_string())
451}
452
453/// Parse a URL like `http://host:port/path` or `https://host:port/path`
454/// into (scheme, host, port, path).
455fn parse_url(url: &str) -> Result<(&str, String, u16, String), OpaError> {
456    let (scheme, rest) = url
457        .split_once("://")
458        .ok_or_else(|| OpaError::Http("URL must start with http:// or https://".to_string()))?;
459    if scheme != "http" && scheme != "https" {
460        return Err(OpaError::Http(format!(
461            "unsupported scheme: {scheme} (use http or https)"
462        )));
463    }
464
465    let (authority, path) = rest.split_once('/').unwrap_or((rest, ""));
466    let path = if path.is_empty() {
467        "/".to_string()
468    } else {
469        format!("/{path}")
470    };
471
472    let (host, port) = if let Some((h, p)) = authority.rsplit_once(':') {
473        let port: u16 = p
474            .parse()
475            .map_err(|_| OpaError::Http(format!("invalid port in URL: {p}")))?;
476        (h.to_string(), port)
477    } else {
478        (
479            authority.to_string(),
480            if scheme == "https" { 443 } else { 80 },
481        )
482    };
483
484    Ok((scheme, host, port, path))
485}
486
487// White-box tests staying in src/ per AGENTS.md: these tests populate
488// the private `cache` field and call the private `cache_key` method
489// and construct `CachedDecision` directly, none of which are reachable
490// through the public API.
491#[cfg(test)]
492mod tests {
493    use super::*;
494    use std::time::Duration;
495
496    #[test]
497    fn cache_key_is_deterministic() {
498        let client = OpaClient::new(
499            "http://opa:8181/v1/data/dwara/allow".to_string(),
500            Duration::from_secs(60),
501            Duration::from_secs(5),
502        );
503        let req = OpaRequest {
504            input: serde_json::json!({"user": "alice", "action": "read"}),
505        };
506        let key1 = client.cache_key(&req);
507        let key2 = client.cache_key(&req);
508        assert_eq!(key1, key2);
509    }
510
511    #[test]
512    fn cache_key_differs_for_different_inputs() {
513        let client = OpaClient::new(
514            "http://opa:8181/v1/data/dwara/allow".to_string(),
515            Duration::from_secs(60),
516            Duration::from_secs(5),
517        );
518        let req1 = OpaRequest {
519            input: serde_json::json!({"user": "alice"}),
520        };
521        let req2 = OpaRequest {
522            input: serde_json::json!({"user": "bob"}),
523        };
524        assert_ne!(client.cache_key(&req1), client.cache_key(&req2));
525    }
526
527    #[test]
528    fn clear_cache_empties_the_cache() {
529        let client = OpaClient::new(
530            "http://opa:8181/v1/data/dwara/allow".to_string(),
531            Duration::from_secs(60),
532            Duration::from_secs(5),
533        );
534        {
535            let mut cache = client.cache.lock().unwrap();
536            cache.insert(
537                "test-key".to_string(),
538                CachedDecision {
539                    decision: true,
540                    expires_at: Instant::now() + Duration::from_secs(60),
541                },
542            );
543        }
544        assert_eq!(client.cache_size(), 1);
545        client.clear_cache();
546        assert_eq!(client.cache_size(), 0);
547    }
548
549    #[test]
550    fn parse_url_http() {
551        let (scheme, host, port, path) = parse_url("http://opa:8181/v1/data/allow").unwrap();
552        assert_eq!(scheme, "http");
553        assert_eq!(host, "opa");
554        assert_eq!(port, 8181);
555        assert_eq!(path, "/v1/data/allow");
556    }
557
558    #[test]
559    fn parse_url_https_default_port() {
560        let (scheme, host, port, path) = parse_url("https://opa.example.com/v1/data").unwrap();
561        assert_eq!(scheme, "https");
562        assert_eq!(host, "opa.example.com");
563        assert_eq!(port, 443);
564        assert_eq!(path, "/v1/data");
565    }
566
567    #[test]
568    fn parse_url_http_default_port() {
569        let (scheme, host, port, _path) = parse_url("http://opa.example.com/v1/data").unwrap();
570        assert_eq!(scheme, "http");
571        assert_eq!(host, "opa.example.com");
572        assert_eq!(port, 80);
573    }
574
575    #[test]
576    fn bundle_revision_starts_none() {
577        let client = OpaClient::new(
578            "http://opa:8181/v1/data/dwara/allow".to_string(),
579            Duration::from_secs(60),
580            Duration::from_secs(5),
581        );
582        assert_eq!(client.bundle_revision(), None);
583    }
584}
dwara_core/security/cedar/opa.rs

dwara_core/security/cedar/
opa.rs
