drasi_source_dataverse/
client.rs

1// Copyright 2026 The Drasi Authors.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Dataverse HTTP client for the OData Web API.
16//!
17//! Implements the REST equivalent of `RetrieveEntityChangesRequest` using
18//! OData change tracking with `Prefer: odata.track-changes` headers and
19//! delta links.
20
21use anyhow::{Context, Result};
22use std::sync::Arc;
23
24use crate::types::ODataDeltaResponse;
25use drasi_lib::identity::{Credentials, IdentityProvider};
26
27/// HTTP client for the Dataverse OData Web API.
28///
29/// Provides methods for:
30/// - Initial change tracking requests (equivalent to `RetrieveEntityChangesRequest` without `DataVersion`)
31/// - Delta polling (equivalent to `RetrieveEntityChangesRequest` with `DataVersion`)
32/// - Standard data retrieval for bootstrap
33pub struct DataverseClient {
34    base_url: String,
35    api_version: String,
36    identity_provider: Arc<dyn IdentityProvider>,
37    http_client: reqwest::Client,
38    /// Dataverse token scope derived from the environment URL (e.g., `https://myorg.crm.dynamics.com/.default`).
39    scope: String,
40}
41
42impl DataverseClient {
43    /// Create a new Dataverse client.
44    ///
45    /// # Arguments
46    ///
47    /// * `environment_url` - Dataverse environment URL (e.g., `https://myorg.crm.dynamics.com`)
48    /// * `api_version` - Web API version (e.g., `v9.2`)
49    /// * `identity_provider` - Identity provider for authentication
50    pub fn new(
51        environment_url: &str,
52        api_version: &str,
53        identity_provider: Arc<dyn IdentityProvider>,
54    ) -> Self {
55        let base_url = environment_url.trim_end_matches('/').to_string();
56        // Derive scope from the environment URL: https://myorg.crm.dynamics.com -> https://myorg.crm.dynamics.com/.default
57        let scope = crate::DataverseSource::dataverse_scope(&base_url);
58        Self {
59            base_url,
60            api_version: api_version.to_string(),
61            identity_provider,
62            http_client: reqwest::Client::new(),
63            scope,
64        }
65    }
66
67    /// Get a bearer token from the identity provider.
68    async fn get_token(&self) -> Result<String> {
69        let context =
70            drasi_lib::identity::CredentialContext::new().with_property("scope", &self.scope);
71        let creds = self.identity_provider.get_credentials(&context).await?;
72        match creds {
73            Credentials::Token { token, .. } => Ok(token),
74            _ => {
75                anyhow::bail!("Dataverse client requires Token credentials from identity provider")
76            }
77        }
78    }
79
80    /// Perform an initial change tracking request for an entity.
81    ///
82    /// This is the Web API equivalent of executing `RetrieveEntityChangesRequest`
83    /// without a `DataVersion` -- it returns all current records plus an initial
84    /// delta token for subsequent change tracking.
85    ///
86    /// Equivalent to the platform's `GetCurrentDeltaToken()` method which pages
87    /// through all initial data.
88    ///
89    /// # Arguments
90    ///
91    /// * `entity_set_name` - The entity set name (plural, e.g., `accounts`)
92    /// * `select` - Optional `$select` clause for column filtering
93    pub async fn initial_change_tracking(
94        &self,
95        entity_set_name: &str,
96        select: Option<&str>,
97    ) -> Result<ODataDeltaResponse> {
98        let token = self.get_token().await?;
99
100        let mut url = format!(
101            "{}/api/data/{}/{}",
102            self.base_url, self.api_version, entity_set_name
103        );
104
105        if let Some(select_str) = select {
106            url = format!("{url}?$select={select_str}");
107        }
108
109        let response = self
110            .http_client
111            .get(&url)
112            .header("Authorization", format!("Bearer {token}"))
113            .header("OData-MaxVersion", "4.0")
114            .header("OData-Version", "4.0")
115            .header("Prefer", "odata.track-changes,odata.maxpagesize=1000")
116            .send()
117            .await
118            .context("Failed to send initial change tracking request")?;
119
120        if !response.status().is_success() {
121            let status = response.status();
122            let body = response.text().await.unwrap_or_default();
123            anyhow::bail!("Initial change tracking request failed with status {status}: {body}",);
124        }
125
126        let delta_response: ODataDeltaResponse = response
127            .json()
128            .await
129            .context("Failed to parse initial change tracking response")?;
130
131        Ok(delta_response)
132    }
133
134    /// Follow a delta link to poll for changes.
135    ///
136    /// This is the Web API equivalent of executing `RetrieveEntityChangesRequest`
137    /// with `DataVersion` set to a previously obtained delta token.
138    ///
139    /// Maps directly to the platform's `GetChanges(deltaToken)` method.
140    ///
141    /// # Arguments
142    ///
143    /// * `delta_link` - The delta link URL from a previous response
144    pub async fn follow_delta_link(&self, delta_link: &str) -> Result<ODataDeltaResponse> {
145        // Reject cross-origin delta links to prevent SSRF via attacker-controlled
146        // OData responses (see `ensure_same_origin` for rationale).
147        self.ensure_same_origin(delta_link, "delta_link")?;
148
149        let token = self.get_token().await?;
150
151        let response = self
152            .http_client
153            .get(delta_link)
154            .header("Authorization", format!("Bearer {token}"))
155            .header("OData-MaxVersion", "4.0")
156            .header("OData-Version", "4.0")
157            .header("Prefer", "odata.maxpagesize=1000")
158            .send()
159            .await
160            .context("Failed to send delta link request")?;
161
162        if !response.status().is_success() {
163            let status = response.status();
164            let body = response.text().await.unwrap_or_default();
165            anyhow::bail!("Delta link request failed with status {status}: {body}",);
166        }
167
168        let delta_response: ODataDeltaResponse = response
169            .json()
170            .await
171            .context("Failed to parse delta response")?;
172
173        Ok(delta_response)
174    }
175
176    /// Follow a next link for pagination within a response.
177    ///
178    /// Equivalent to following the platform's `PagingCookie` for additional pages.
179    ///
180    /// # Arguments
181    ///
182    /// * `next_link` - The next link URL from the current page
183    pub async fn follow_next_link(&self, next_link: &str) -> Result<ODataDeltaResponse> {
184        // Reject cross-origin next links to prevent SSRF via attacker-controlled
185        // OData responses (see `ensure_same_origin` for rationale).
186        self.ensure_same_origin(next_link, "next_link")?;
187
188        let token = self.get_token().await?;
189
190        let response = self
191            .http_client
192            .get(next_link)
193            .header("Authorization", format!("Bearer {token}"))
194            .header("OData-MaxVersion", "4.0")
195            .header("OData-Version", "4.0")
196            .send()
197            .await
198            .context("Failed to send next link request")?;
199
200        if !response.status().is_success() {
201            let status = response.status();
202            let body = response.text().await.unwrap_or_default();
203            anyhow::bail!("Next link request failed with status {status}: {body}",);
204        }
205
206        let delta_response: ODataDeltaResponse = response
207            .json()
208            .await
209            .context("Failed to parse next link response")?;
210
211        Ok(delta_response)
212    }
213
214    /// Retrieve all records from an entity (for bootstrap).
215    ///
216    /// Sends a standard GET request without change tracking headers.
217    /// Used by the bootstrap provider to load initial data.
218    ///
219    /// # Arguments
220    ///
221    /// * `entity_set_name` - The entity set name (plural, e.g., `accounts`)
222    /// * `select` - Optional `$select` clause for column filtering
223    pub async fn get_entity_data(
224        &self,
225        entity_set_name: &str,
226        select: Option<&str>,
227    ) -> Result<ODataDeltaResponse> {
228        let token = self.get_token().await?;
229
230        let mut url = format!(
231            "{}/api/data/{}/{}",
232            self.base_url, self.api_version, entity_set_name
233        );
234
235        if let Some(select_str) = select {
236            url = format!("{url}?$select={select_str}");
237        }
238
239        let response = self
240            .http_client
241            .get(&url)
242            .header("Authorization", format!("Bearer {token}"))
243            .header("OData-MaxVersion", "4.0")
244            .header("OData-Version", "4.0")
245            .header("Prefer", "odata.maxpagesize=5000")
246            .send()
247            .await
248            .context("Failed to send entity data request")?;
249
250        if !response.status().is_success() {
251            let status = response.status();
252            let body = response.text().await.unwrap_or_default();
253            anyhow::bail!("Entity data request failed with status {status}: {body}",);
254        }
255
256        let data_response: ODataDeltaResponse = response
257            .json()
258            .await
259            .context("Failed to parse entity data response")?;
260
261        Ok(data_response)
262    }
263
264    /// Get the base URL for this client.
265    pub fn base_url(&self) -> &str {
266        &self.base_url
267    }
268
269    /// Get the API version.
270    pub fn api_version(&self) -> &str {
271        &self.api_version
272    }
273
274    /// Reject delta/next link URLs that point outside the configured Dataverse
275    /// environment. The OData server controls the contents of `@odata.deltaLink`
276    /// and `@odata.nextLink`; without this check, a compromised, misconfigured,
277    /// or proxied response could redirect the client (with its bearer token)
278    /// to an arbitrary host — e.g., the cloud metadata service at
279    /// `169.254.169.254` (SSRF).
280    fn ensure_same_origin(&self, link: &str, kind: &str) -> Result<()> {
281        let parsed =
282            url::Url::parse(link).with_context(|| format!("{kind} is not a valid URL: {link}"))?;
283        let base = url::Url::parse(&self.base_url)
284            .with_context(|| format!("client base_url is not a valid URL: {}", self.base_url))?;
285
286        // Compare scheme + host + port. We accept any path/query as long as the
287        // origin matches the configured environment.
288        if parsed.scheme() != base.scheme()
289            || parsed.host_str() != base.host_str()
290            || parsed.port_or_known_default() != base.port_or_known_default()
291        {
292            anyhow::bail!(
293                "{kind} origin '{}://{}{}' does not match configured Dataverse environment '{}'",
294                parsed.scheme(),
295                parsed.host_str().unwrap_or(""),
296                parsed.port().map(|p| format!(":{p}")).unwrap_or_default(),
297                self.base_url,
298            );
299        }
300        Ok(())
301    }
302}
303
304#[cfg(test)]
305mod tests {
306    //! HTTP error-path tests for `DataverseClient`.
307    //!
308    //! These tests stand the client up against `wiremock` and verify the
309    //! observable behaviour of failure modes that the production code is
310    //! expected to handle:
311    //! - Non-2xx HTTP responses (`401`, `429`, `503`) surface as errors with
312    //!   diagnostic context rather than panicking.
313    //! - Malformed OData responses (missing `@odata.deltaLink`) deserialize
314    //!   without crashing — `delta_link` is `None`.
315
316    use super::*;
317    use async_trait::async_trait;
318    use drasi_lib::identity::{CredentialContext, Credentials, IdentityProvider};
319    use serde_json::json;
320    use wiremock::matchers::{method, path};
321    use wiremock::{Mock, MockServer, ResponseTemplate};
322
323    /// Static-token identity provider used to bypass real auth in unit tests.
324    struct StaticToken(&'static str);
325
326    #[async_trait]
327    impl IdentityProvider for StaticToken {
328        async fn get_credentials(&self, _ctx: &CredentialContext) -> anyhow::Result<Credentials> {
329            Ok(Credentials::Token {
330                username: "test".to_string(),
331                token: self.0.to_string(),
332            })
333        }
334
335        fn clone_box(&self) -> Box<dyn IdentityProvider> {
336            Box::new(StaticToken(self.0))
337        }
338    }
339
340    fn client_for(server: &MockServer) -> DataverseClient {
341        DataverseClient::new(&server.uri(), "v9.2", Arc::new(StaticToken("test-token")))
342    }
343
344    #[tokio::test]
345    async fn initial_change_tracking_propagates_401() {
346        let server = MockServer::start().await;
347        Mock::given(method("GET"))
348            .and(path("/api/data/v9.2/accounts"))
349            .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
350            .mount(&server)
351            .await;
352
353        let client = client_for(&server);
354        let err = client
355            .initial_change_tracking("accounts", None)
356            .await
357            .expect_err("401 must surface as an error");
358        let msg = format!("{err:?}");
359        assert!(msg.contains("401"), "error should mention status: {msg}");
360    }
361
362    #[tokio::test]
363    async fn delta_link_propagates_429() {
364        let server = MockServer::start().await;
365        Mock::given(method("GET"))
366            .and(path("/api/data/v9.2/accounts"))
367            .respond_with(
368                ResponseTemplate::new(429)
369                    .insert_header("Retry-After", "5")
370                    .set_body_string("too many requests"),
371            )
372            .mount(&server)
373            .await;
374
375        let client = client_for(&server);
376        let delta_url = format!("{}/api/data/v9.2/accounts?$deltatoken=abc", server.uri());
377        let err = client
378            .follow_delta_link(&delta_url)
379            .await
380            .expect_err("429 must surface as an error");
381        assert!(format!("{err:?}").contains("429"));
382    }
383
384    #[tokio::test]
385    async fn delta_link_propagates_503() {
386        let server = MockServer::start().await;
387        Mock::given(method("GET"))
388            .and(path("/api/data/v9.2/accounts"))
389            .respond_with(ResponseTemplate::new(503).set_body_string("unavailable"))
390            .mount(&server)
391            .await;
392
393        let client = client_for(&server);
394        let delta_url = format!("{}/api/data/v9.2/accounts?$deltatoken=abc", server.uri());
395        let err = client
396            .follow_delta_link(&delta_url)
397            .await
398            .expect_err("503 must surface as an error");
399        assert!(format!("{err:?}").contains("503"));
400    }
401
402    #[tokio::test]
403    async fn delta_response_without_delta_link_parses_with_none() {
404        // A delta response that only contains `value` (no `@odata.deltaLink` and
405        // no `@odata.nextLink`) must still deserialize. Callers can then decide
406        // how to handle the missing token.
407        let server = MockServer::start().await;
408        Mock::given(method("GET"))
409            .and(path("/api/data/v9.2/accounts"))
410            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
411                "@odata.context": "context",
412                "value": []
413            })))
414            .mount(&server)
415            .await;
416
417        let client = client_for(&server);
418        let url = format!("{}/api/data/v9.2/accounts?$deltatoken=abc", server.uri());
419        let resp = client
420            .follow_delta_link(&url)
421            .await
422            .expect("malformed-but-valid OData response should deserialize");
423        assert!(resp.delta_link.is_none(), "delta_link should be absent");
424        assert!(resp.next_link.is_none(), "next_link should be absent");
425        assert!(resp.value.is_empty());
426    }
427
428    #[tokio::test]
429    async fn malformed_json_body_surfaces_as_parse_error() {
430        let server = MockServer::start().await;
431        Mock::given(method("GET"))
432            .and(path("/api/data/v9.2/accounts"))
433            .respond_with(ResponseTemplate::new(200).set_body_string("not-json"))
434            .mount(&server)
435            .await;
436
437        let client = client_for(&server);
438        let err = client
439            .initial_change_tracking("accounts", None)
440            .await
441            .expect_err("non-JSON body must surface as a parse error");
442        let msg = format!("{err:?}");
443        assert!(
444            msg.contains("parse") || msg.contains("decoding"),
445            "error should reference parsing: {msg}"
446        );
447    }
448
449    #[tokio::test]
450    async fn next_link_propagates_500() {
451        let server = MockServer::start().await;
452        Mock::given(method("GET"))
453            .and(path("/api/data/v9.2/accounts"))
454            .respond_with(ResponseTemplate::new(500).set_body_string("server error"))
455            .mount(&server)
456            .await;
457
458        let client = client_for(&server);
459        let next_url = format!("{}/api/data/v9.2/accounts?$skiptoken=xyz", server.uri());
460        let err = client
461            .follow_next_link(&next_url)
462            .await
463            .expect_err("500 must surface as an error");
464        assert!(format!("{err:?}").contains("500"));
465    }
466
467    /// SSRF guard: `follow_delta_link` must reject a URL that points outside
468    /// the configured Dataverse environment. Without this, an attacker who
469    /// influences the OData response body can redirect the bearer-token
470    /// request to e.g. the cloud metadata service.
471    #[tokio::test]
472    async fn follow_delta_link_rejects_cross_origin() {
473        let server = MockServer::start().await;
474        // No mocks are mounted: the request must be blocked before it leaves.
475        let client = client_for(&server);
476
477        let err = client
478            .follow_delta_link("http://169.254.169.254/latest/meta-data/")
479            .await
480            .expect_err("cross-origin delta_link must be rejected");
481        let msg = format!("{err:?}");
482        assert!(
483            msg.contains("origin") && msg.contains("does not match"),
484            "error should explain origin mismatch: {msg}"
485        );
486    }
487
488    #[tokio::test]
489    async fn follow_next_link_rejects_cross_origin() {
490        let server = MockServer::start().await;
491        let client = client_for(&server);
492
493        let err = client
494            .follow_next_link("https://attacker.example.com/api/data/v9.2/accounts")
495            .await
496            .expect_err("cross-origin next_link must be rejected");
497        assert!(format!("{err:?}").contains("does not match"));
498    }
499
500    #[tokio::test]
501    async fn follow_delta_link_rejects_different_scheme() {
502        let server = MockServer::start().await;
503        let client = client_for(&server);
504
505        // Same host, different scheme — still a different origin.
506        let server_host = url::Url::parse(&server.uri())
507            .unwrap()
508            .host_str()
509            .unwrap()
510            .to_string();
511        let err = client
512            .follow_delta_link(&format!("ftp://{server_host}/api/data/v9.2/accounts"))
513            .await
514            .expect_err("scheme mismatch must be rejected");
515        assert!(format!("{err:?}").contains("does not match"));
516    }
517
518    #[tokio::test]
519    async fn follow_delta_link_rejects_malformed_url() {
520        let server = MockServer::start().await;
521        let client = client_for(&server);
522
523        let err = client
524            .follow_delta_link("not a url")
525            .await
526            .expect_err("malformed URL must be rejected");
527        let msg = format!("{err:?}");
528        assert!(
529            msg.contains("not a valid URL") || msg.contains("delta_link"),
530            "error should explain parse failure: {msg}"
531        );
532    }
533}
drasi_source_dataverse/client.rs

drasi_source_dataverse/
client.rs
