ferrin_policy/
http.rs

1//! Policy client for the OPA REST Data API.
2
3use std::fmt;
4use std::time::Duration;
5
6use bytes::Bytes;
7use ferrin_provider_util::http::HttpRequest;
8use ferrin_provider_util::http::HttpTransport;
9use ferrin_provider_util::http::RequestBody;
10use ferrin_provider_util::http::SharedTransport;
11use ferrin_provider_util::http::TransportError;
12use ferrin_provider_util::http::default_transport;
13use ferrin_provider_util::http::read_body;
14use ferrin_provider_util::secure_url::UrlPolicy;
15use ferrin_provider_util::secure_url::validate_url;
16use ferrin_spec::BoxFuture;
17use ferrin_spec::Headers;
18use ferrin_spec::JsonValue;
19use http::Method;
20use serde_json::json;
21use url::Url;
22
23use crate::client::PolicyClient;
24use crate::error::PolicyError;
25use crate::path::PolicyPath;
26
27/// Default limit on the size of a decision response body.
28pub const DEFAULT_MAX_RESPONSE_BYTES: u64 = 1024 * 1024;
29
30const USER_AGENT: &str = concat!("ferrin-policy/", env!("CARGO_PKG_VERSION"));
31const BODY_EXCERPT_BYTES: usize = 1024;
32
33/// Evaluates policies through a policy server implementing the OPA REST
34/// Data API: `POST <base>/v1/data/<path>` with `{"input": ..}`, answering
35/// `{"result": ..}`.
36///
37/// A missing `result` (an undefined rule) yields `JsonValue::Null`. The
38/// server URL is validated with the configured [`UrlPolicy`] on every call;
39/// the default policy requires HTTPS and a public host, so a local sidecar
40/// needs `UrlPolicy::new().allow_http().allow_private_networks()`.
41pub struct HttpPolicyClient {
42    base_url: Url,
43    headers: Headers,
44    transport: SharedTransport,
45    url_policy: UrlPolicy,
46    timeout: Option<Duration>,
47    max_response_bytes: u64,
48}
49
50impl fmt::Debug for HttpPolicyClient {
51    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52        f.debug_struct("HttpPolicyClient")
53            .field("base_url", &self.base_url.origin().ascii_serialization())
54            .field("headers", &self.headers.masked())
55            .field("timeout", &self.timeout)
56            .field("max_response_bytes", &self.max_response_bytes)
57            .finish_non_exhaustive()
58    }
59}
60
61impl HttpPolicyClient {
62    /// Starts building a client for the server at `base_url`.
63    #[must_use]
64    pub fn builder(base_url: Url) -> HttpPolicyClientBuilder {
65        HttpPolicyClientBuilder {
66            base_url,
67            headers: Headers::new(),
68            transport: None,
69            url_policy: UrlPolicy::new(),
70            timeout: None,
71            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
72        }
73    }
74
75    /// Creates a client with the default transport and URL policy.
76    ///
77    /// # Errors
78    ///
79    /// Returns [`PolicyError::Transport`] when the default transport cannot
80    /// be created.
81    pub fn new(base_url: Url) -> Result<Self, PolicyError> {
82        Self::builder(base_url).build()
83    }
84
85    /// The server URL.
86    #[must_use]
87    pub fn base_url(&self) -> &Url {
88        &self.base_url
89    }
90
91    fn data_url(&self, path: &PolicyPath) -> Result<Url, PolicyError> {
92        let mut url = self.base_url.clone();
93        {
94            let mut segments = url
95                .path_segments_mut()
96                .map_err(|()| PolicyError::InvalidUrl {
97                    message: "policy server url cannot be a base".to_owned(),
98                })?;
99            segments.pop_if_empty();
100            segments.extend(["v1", "data"]);
101            segments.extend(path.segments().iter().map(String::as_str));
102        }
103        url.set_query(None);
104        url.set_fragment(None);
105        Ok(url)
106    }
107
108    #[tracing::instrument(skip_all, fields(path))]
109    async fn evaluate_inner(&self, path: &str, input: JsonValue) -> Result<JsonValue, PolicyError> {
110        let path = PolicyPath::parse(path)?;
111        let url = self.data_url(&path)?;
112        let validated = validate_url(&url, &self.url_policy)
113            .await
114            .map_err(|error| PolicyError::InvalidUrl {
115                message: error.to_string(),
116            })?;
117        let body = serde_json::to_vec(&json!({ "input": input })).map_err(|error| {
118            PolicyError::InvalidInput {
119                message: error.to_string(),
120            }
121        })?;
122        let mut headers = self.headers.clone();
123        if !headers.contains("content-type") {
124            let _ = headers.insert("content-type", "application/json");
125        }
126        if !headers.contains("accept") {
127            let _ = headers.insert("accept", "application/json");
128        }
129        let mut request = HttpRequest::new(Method::POST, url.clone())
130            .with_headers(headers.with_user_agent_suffix([USER_AGENT]))
131            .with_body(RequestBody::json(Bytes::from(body)))
132            .with_pinned_addresses(validated.addresses);
133        if let Some(timeout) = self.timeout {
134            request = request.with_timeout(timeout);
135        }
136        let response = self
137            .transport
138            .execute(request)
139            .await
140            .map_err(|error| transport_error(&url, &error))?;
141        let status = response.status;
142        let bytes = read_body(&response.headers, response.body, self.max_response_bytes)
143            .await
144            .map_err(|error| transport_error(&url, &error))?;
145        if !status.is_success() {
146            return Err(PolicyError::Status {
147                status,
148                body: excerpt(&bytes),
149            });
150        }
151        let document: JsonValue =
152            serde_json::from_slice(&bytes).map_err(|error| PolicyError::InvalidResponse {
153                message: error.to_string(),
154            })?;
155        match document {
156            JsonValue::Object(mut object) => Ok(object.remove("result").unwrap_or(JsonValue::Null)),
157            _ => Err(PolicyError::InvalidResponse {
158                message: "expected a JSON object with a `result` member".to_owned(),
159            }),
160        }
161    }
162}
163
164impl PolicyClient for HttpPolicyClient {
165    fn evaluate<'a>(
166        &'a self,
167        path: &'a str,
168        input: JsonValue,
169    ) -> BoxFuture<'a, Result<JsonValue, PolicyError>> {
170        Box::pin(self.evaluate_inner(path, input))
171    }
172}
173
174fn transport_error(url: &Url, error: &TransportError) -> PolicyError {
175    PolicyError::Transport {
176        host: url.host_str().unwrap_or_default().to_owned(),
177        message: error.to_string(),
178    }
179}
180
181fn excerpt(bytes: &[u8]) -> String {
182    let end = bytes.len().min(BODY_EXCERPT_BYTES);
183    String::from_utf8_lossy(&bytes[..end]).into_owned()
184}
185
186/// Builder of an [`HttpPolicyClient`].
187pub struct HttpPolicyClientBuilder {
188    base_url: Url,
189    headers: Headers,
190    transport: Option<SharedTransport>,
191    url_policy: UrlPolicy,
192    timeout: Option<Duration>,
193    max_response_bytes: u64,
194}
195
196impl fmt::Debug for HttpPolicyClientBuilder {
197    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
198        f.debug_struct("HttpPolicyClientBuilder")
199            .field("base_url", &self.base_url.origin().ascii_serialization())
200            .field("headers", &self.headers.masked())
201            .field("custom_transport", &self.transport.is_some())
202            .field("timeout", &self.timeout)
203            .field("max_response_bytes", &self.max_response_bytes)
204            .finish_non_exhaustive()
205    }
206}
207
208impl HttpPolicyClientBuilder {
209    /// Sets the headers sent with every request (for example an
210    /// `authorization` header). Never logged unmasked.
211    #[must_use]
212    pub fn headers(mut self, headers: Headers) -> Self {
213        self.headers = headers;
214        self
215    }
216
217    /// Adds one header; invalid names or values are skipped.
218    #[must_use]
219    pub fn header(mut self, name: &str, value: &str) -> Self {
220        self.headers = self.headers.with(name, value);
221        self
222    }
223
224    /// Uses `transport` instead of the shared default transport.
225    #[must_use]
226    pub fn transport(mut self, transport: SharedTransport) -> Self {
227        self.transport = Some(transport);
228        self
229    }
230
231    /// Sets the URL policy applied to the server URL (default: HTTPS and
232    /// public networks only).
233    #[must_use]
234    pub fn url_policy(mut self, policy: UrlPolicy) -> Self {
235        self.url_policy = policy;
236        self
237    }
238
239    /// Sets the per-request timeout (default: the transport's own timeout).
240    #[must_use]
241    pub fn timeout(mut self, timeout: Duration) -> Self {
242        self.timeout = Some(timeout);
243        self
244    }
245
246    /// Sets the response body limit (default 1 MiB).
247    #[must_use]
248    pub fn max_response_bytes(mut self, max_response_bytes: u64) -> Self {
249        self.max_response_bytes = max_response_bytes;
250        self
251    }
252
253    /// Builds the client.
254    ///
255    /// # Errors
256    ///
257    /// Returns [`PolicyError::Transport`] when no transport was given and
258    /// the default transport cannot be created.
259    pub fn build(self) -> Result<HttpPolicyClient, PolicyError> {
260        let transport = match self.transport {
261            Some(transport) => transport,
262            None => default_transport().map_err(|error| transport_error(&self.base_url, &error))?,
263        };
264        Ok(HttpPolicyClient {
265            base_url: self.base_url,
266            headers: self.headers,
267            transport,
268            url_policy: self.url_policy,
269            timeout: self.timeout,
270            max_response_bytes: self.max_response_bytes,
271        })
272    }
273}
ferrin_policy/http.rs

ferrin_policy/
http.rs
