iceberg_catalog_rest/
client.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::fmt::{Debug, Formatter};
20
21use iceberg::{Error, ErrorKind, Result};
22use reqwest::header::HeaderMap;
23use reqwest::{Client, IntoUrl, Method, RequestBuilder};
24use serde::de::DeserializeOwned;
25
26use crate::RestCatalogConfig;
27use crate::auth::AuthSession;
28use crate::request::HttpRequest;
29use crate::response::HttpResponse;
30
31/// The catalog's HTTP client, handed to an [`AuthManager`] so its own
32/// requests share the catalog's connection pool and configuration.
33///
34/// [`AuthManager`]: crate::auth::AuthManager
35#[derive(Clone)]
36pub struct HttpClient {
37    client: Client,
38
39    /// Extra headers to be added to each request.
40    extra_headers: HeaderMap,
41    /// Whether to disable header redaction in error logs (defaults to false for security).
42    disable_header_redaction: bool,
43}
44
45impl Debug for HttpClient {
46    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
47        // The inner client is omitted: an injected one may carry secrets in
48        // its default headers.
49        f.debug_struct("HttpClient")
50            .field(
51                // `header.*` values may hold secrets (e.g. `authorization`).
52                "extra_headers",
53                &format_headers_redacted(&self.extra_headers, self.disable_header_redaction),
54            )
55            .finish_non_exhaustive()
56    }
57}
58
59impl HttpClient {
60    /// Sends an unauthenticated form-encoded POST and returns the response,
61    /// which is what an [`AuthManager`] needs to exchange a credential for a
62    /// token.
63    ///
64    /// Only `headers` are sent: neither the catalog's extra headers nor
65    /// any session's authentication is applied, so a token request carries
66    /// exactly the credentials its caller puts into `headers` or `form`.
67    ///
68    /// [`AuthManager`]: crate::auth::AuthManager
69    pub async fn post_form(
70        &self,
71        url: &str,
72        headers: &HeaderMap,
73        form: &HashMap<&str, &str>,
74    ) -> Result<HttpResponse> {
75        let mut request = HttpRequest::build(
76            self.client
77                .request(Method::POST, url)
78                .headers(headers.clone())
79                .form(form),
80        )?;
81        // `headers` may carry a `content-type: application/json` that `form`
82        // leaves in place.
83        request.headers_mut().insert(
84            reqwest::header::CONTENT_TYPE,
85            reqwest::header::HeaderValue::from_static("application/x-www-form-urlencoded"),
86        );
87
88        let response = self.client.execute(request.into_inner()).await?;
89        HttpResponse::read(response).await
90    }
91
92    /// Create a new http client.
93    pub(crate) fn new(cfg: &RestCatalogConfig) -> Result<Self> {
94        Ok(HttpClient {
95            client: cfg.client(),
96            extra_headers: cfg.extra_headers()?,
97            disable_header_redaction: cfg.disable_header_redaction(),
98        })
99    }
100
101    /// Update the http client with new configuration.
102    ///
103    /// If cfg carries new value, we will use cfg instead.
104    /// Otherwise, we will keep the old value.
105    pub(crate) fn update_with(self, cfg: &RestCatalogConfig) -> Result<Self> {
106        let extra_headers = (!cfg.extra_headers()?.is_empty())
107            .then(|| cfg.extra_headers())
108            .transpose()?
109            .unwrap_or(self.extra_headers);
110        Ok(HttpClient {
111            // `cfg.client()` returns the same shared client.
112            client: cfg.client(),
113            extra_headers,
114            disable_header_redaction: cfg.disable_header_redaction(),
115        })
116    }
117
118    /// Testing only: the bearer token `session` would attach.
119    ///
120    /// Authenticates a throwaway request (never sent) and reads the header
121    /// back, so it works for any [`AuthSession`].
122    #[cfg(test)]
123    pub(crate) async fn token(&self, auth_session: &dyn AuthSession) -> Option<String> {
124        let mut request = HttpRequest::build(
125            self.client
126                .request(Method::GET, "http://localhost/token-probe"),
127        )
128        .ok()?;
129        auth_session.authenticate(&mut request).await.ok()?;
130        let request = request.into_inner();
131        request
132            .headers()
133            .get(reqwest::header::AUTHORIZATION)?
134            .to_str()
135            .ok()?
136            .strip_prefix("Bearer ")
137            .map(str::to_string)
138    }
139
140    #[inline]
141    pub(crate) fn request<U: IntoUrl>(&self, method: Method, url: U) -> RequestBuilder {
142        self.client
143            .request(method, url)
144            .headers(self.extra_headers.clone())
145    }
146
147    /// Sends `request` to the Iceberg REST catalog, authenticated by this
148    /// client's session.
149    pub(crate) async fn query_catalog(
150        &self,
151        auth_session: &dyn AuthSession,
152        mut request: HttpRequest,
153    ) -> Result<HttpResponse> {
154        // Authenticate first, then apply extra headers, so a configured
155        // `header.authorization` keeps overriding a token (unchanged behavior).
156        auth_session.authenticate(&mut request).await?;
157
158        let mut request = request.into_inner();
159        request.headers_mut().extend(self.extra_headers.clone());
160        HttpResponse::read(self.client.execute(request).await?).await
161    }
162
163    /// Returns whether header redaction is disabled for this client.
164    pub(crate) fn disable_header_redaction(&self) -> bool {
165        self.disable_header_redaction
166    }
167}
168
169/// Deserializes a catalog response into the given [`DeserializeOwned`] type.
170///
171/// Returns an error if unable to parse the response bytes.
172pub(crate) fn deserialize_catalog_response<R: DeserializeOwned>(
173    response: HttpResponse,
174) -> Result<R> {
175    let bytes = response.body();
176
177    serde_json::from_slice::<R>(bytes).map_err(|e| {
178        Error::new(
179            ErrorKind::Unexpected,
180            "Failed to parse response from rest catalog server",
181        )
182        .with_context("json", String::from_utf8_lossy(bytes))
183        .with_source(e)
184    })
185}
186
187/// Returns true if the header may carry a secret (matched by substring, so
188/// e.g. `x-client-secret` is covered along with `authorization`).
189fn is_sensitive_header(name: &str) -> bool {
190    let name_lower = name.to_lowercase();
191    [
192        "auth",
193        "token",
194        "secret",
195        "key",
196        "password",
197        "cookie",
198        "credential",
199    ]
200    .iter()
201    .any(|pattern| name_lower.contains(pattern))
202}
203
204/// Redacts sensitive headers and returns a debug-formatted string.
205///
206/// If `disable_redaction` is true, returns all headers without redaction.
207/// Otherwise, replaces sensitive header values with `[REDACTED]`.
208pub(crate) fn format_headers_redacted(headers: &HeaderMap, disable_redaction: bool) -> String {
209    if disable_redaction {
210        // Return all headers as-is without redaction
211        let all: HashMap<&str, &str> = headers
212            .iter()
213            .filter_map(|(name, value)| value.to_str().ok().map(|v| (name.as_str(), v)))
214            .collect();
215        return format!("{all:?}");
216    }
217
218    // Redact sensitive headers by replacing their values with "[REDACTED]"
219    let redacted: HashMap<&str, &str> = headers
220        .iter()
221        .filter_map(|(name, value)| {
222            if is_sensitive_header(name.as_str()) {
223                Some((name.as_str(), "[REDACTED]"))
224            } else {
225                value.to_str().ok().map(|v| (name.as_str(), v))
226            }
227        })
228        .collect();
229    format!("{redacted:?}")
230}
231
232/// Deserializes a unexpected catalog response into an error.
233pub(crate) fn deserialize_unexpected_catalog_error(
234    response: HttpResponse,
235    disable_header_redaction: bool,
236) -> Error {
237    let err = Error::new(
238        ErrorKind::Unexpected,
239        "Received response with unexpected status code",
240    )
241    .with_context("status", response.status().to_string())
242    .with_context(
243        "headers",
244        format_headers_redacted(response.headers(), disable_header_redaction),
245    );
246
247    let bytes = response.body();
248    if bytes.is_empty() {
249        return err;
250    }
251    err.with_context("json", String::from_utf8_lossy(bytes))
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257
258    #[tokio::test]
259    async fn test_a_truncated_body_error_names_the_url() {
260        // `bytes()` builds its error without a URL, so `read` attaches the one
261        // the response came from; otherwise a failure can't be attributed to a
262        // catalog. Needs a raw socket: the body has to be cut short.
263        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
264        let addr = listener.local_addr().unwrap();
265        std::thread::spawn(move || {
266            let (mut stream, _) = listener.accept().unwrap();
267            let mut buf = [0u8; 1024];
268            let _ = std::io::Read::read(&mut stream, &mut buf);
269            // Promise more than is sent, then hang up.
270            let _ = std::io::Write::write_all(
271                &mut stream,
272                b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nshort",
273            );
274        });
275
276        let url = format!("http://{addr}/token");
277        let err = HttpClient::new(&RestCatalogConfig::builder().uri(url.clone()).build())
278            .unwrap()
279            .post_form(&url, &HeaderMap::new(), &HashMap::new())
280            .await
281            .unwrap_err();
282
283        assert!(format!("{err:?}").contains(&url), "{err:?}");
284    }
285
286    #[tokio::test]
287    async fn test_reading_a_response_keeps_status_headers_and_body() {
288        let mut server = mockito::Server::new_async().await;
289        let mock = server
290            .mock("POST", "/token")
291            .with_status(418)
292            .with_header("x-request-id", "abc123")
293            .with_body("brewing")
294            .create_async()
295            .await;
296
297        let response = HttpClient::new(&RestCatalogConfig::builder().uri(server.url()).build())
298            .unwrap()
299            .post_form(
300                &format!("{}/token", server.url()),
301                &HeaderMap::new(),
302                &HashMap::new(),
303            )
304            .await
305            .unwrap();
306
307        assert_eq!(response.status(), 418);
308        assert_eq!(response.headers().get("x-request-id").unwrap(), "abc123");
309        assert_eq!(response.body(), b"brewing");
310        mock.assert_async().await;
311    }
312
313    #[test]
314    fn test_unexpected_error_carries_status_headers_and_body() {
315        // Everything a user needs to diagnose an unexpected status, with the
316        // sensitive headers held back.
317        let mut headers = HeaderMap::new();
318        headers.insert("authorization", "Bearer leaked".parse().unwrap());
319        headers.insert("x-request-id", "abc123".parse().unwrap());
320        let response = HttpResponse::new(
321            http::StatusCode::IM_A_TEAPOT,
322            headers,
323            br#"{"error": "nope"}"#.to_vec(),
324        );
325
326        let err = format!(
327            "{:?}",
328            deserialize_unexpected_catalog_error(response, false)
329        );
330
331        assert!(err.contains("418"), "{err}");
332        assert!(err.contains("x-request-id"), "{err}");
333        assert!(err.contains("abc123"), "{err}");
334        assert!(err.contains("nope"), "{err}");
335        assert!(!err.contains("leaked"), "{err}");
336    }
337
338    #[tokio::test]
339    async fn test_post_form_is_never_authenticated() {
340        // A token request carries no authentication, not even a
341        // user-configured `header.authorization` meant for the catalog.
342        let mut server = mockito::Server::new_async().await;
343        let mock = server
344            .mock("POST", "/token")
345            .match_header("authorization", mockito::Matcher::Missing)
346            .with_status(200)
347            .create_async()
348            .await;
349
350        HttpClient::new(
351            &RestCatalogConfig::builder()
352                .uri(server.url())
353                .props(HashMap::from([(
354                    "header.authorization".to_string(),
355                    "Basic xyz".to_string(),
356                )]))
357                .build(),
358        )
359        .unwrap()
360        .post_form(
361            &format!("{}/token", server.url()),
362            &HeaderMap::new(),
363            &HashMap::new(),
364        )
365        .await
366        .unwrap();
367
368        mock.assert_async().await;
369    }
370
371    #[test]
372    fn test_format_headers_redacted_empty() {
373        let headers = HeaderMap::new();
374        let result = format_headers_redacted(&headers, false);
375        assert_eq!(result, "{}");
376    }
377
378    #[test]
379    fn test_format_headers_redacted_non_sensitive() {
380        let mut headers = HeaderMap::new();
381        headers.insert("content-type", "application/json".parse().unwrap());
382        headers.insert("x-request-id", "abc123".parse().unwrap());
383
384        let result = format_headers_redacted(&headers, false);
385
386        assert!(result.contains("content-type"));
387        assert!(result.contains("application/json"));
388        assert!(result.contains("x-request-id"));
389        assert!(result.contains("abc123"));
390    }
391
392    #[tokio::test]
393    async fn test_http_client_debug_redacts_headers() {
394        let config = RestCatalogConfig::builder()
395            .uri("http://localhost".to_string())
396            .props(HashMap::from([
397                ("header.authorization".to_string(), "Basic xyz".to_string()),
398                (
399                    "header.x-client-secret".to_string(),
400                    "shh-secret".to_string(),
401                ),
402                (
403                    "header.x-client-credential".to_string(),
404                    "cred-value".to_string(),
405                ),
406            ]))
407            .build();
408        let client = HttpClient::new(&config).unwrap();
409
410        let out = format!("{client:?}");
411        assert!(!out.contains("Basic xyz"));
412        assert!(!out.contains("shh-secret"));
413        assert!(!out.contains("cred-value"));
414        assert!(out.contains("[REDACTED]"));
415    }
416
417    #[test]
418    fn test_format_headers_redacted_filters_sensitive() {
419        let mut headers = HeaderMap::new();
420        headers.insert("authorization", "Bearer secret-token".parse().unwrap());
421        headers.insert("content-type", "application/json".parse().unwrap());
422
423        let result = format_headers_redacted(&headers, false);
424
425        // Sensitive header should be present but with redacted value
426        assert!(result.contains("authorization"));
427        assert!(result.contains("[REDACTED]"));
428        // Sensitive value should NOT be present
429        assert!(!result.contains("secret-token"));
430        // Non-sensitive header should be present with actual value
431        assert!(result.contains("content-type"));
432        assert!(result.contains("application/json"));
433    }
434
435    #[test]
436    fn test_format_headers_redacted_filters_set_cookie() {
437        let mut headers = HeaderMap::new();
438        headers.insert(
439            "set-cookie",
440            "CF_Authorization=sensitive-session-token; Path=/; Secure;"
441                .parse()
442                .unwrap(),
443        );
444        headers.insert("server", "cloudflare".parse().unwrap());
445
446        let result = format_headers_redacted(&headers, false);
447
448        // Sensitive header should be present but with redacted value
449        assert!(result.contains("set-cookie"));
450        assert!(result.contains("[REDACTED]"));
451        // Sensitive value should NOT be present
452        assert!(!result.contains("sensitive-session-token"));
453        // Non-sensitive header should be present with actual value
454        assert!(result.contains("server"));
455        assert!(result.contains("cloudflare"));
456    }
457
458    #[test]
459    fn test_format_headers_redacted_filters_all_sensitive() {
460        let mut headers = HeaderMap::new();
461        headers.insert("authorization", "Bearer token".parse().unwrap());
462        headers.insert("proxy-authorization", "Basic creds".parse().unwrap());
463        headers.insert("set-cookie", "session=abc".parse().unwrap());
464        headers.insert("cookie", "session=abc".parse().unwrap());
465        headers.insert("x-api-key", "api-key-123".parse().unwrap());
466        headers.insert("x-auth-token", "auth-token-456".parse().unwrap());
467        headers.insert("x-request-id", "req-123".parse().unwrap());
468
469        let result = format_headers_redacted(&headers, false);
470
471        // All sensitive headers should be present but with redacted values
472        assert!(result.contains("authorization"));
473        assert!(result.contains("proxy-authorization"));
474        assert!(result.contains("set-cookie"));
475        assert!(result.contains("cookie"));
476        assert!(result.contains("x-api-key"));
477        assert!(result.contains("x-auth-token"));
478        assert!(result.contains("[REDACTED]"));
479
480        // Ensure no sensitive values leaked
481        assert!(!result.contains("Bearer token"));
482        assert!(!result.contains("Basic creds"));
483        assert!(!result.contains("session=abc"));
484        assert!(!result.contains("api-key-123"));
485        assert!(!result.contains("auth-token-456"));
486
487        // Non-sensitive header should be present with actual value
488        assert!(result.contains("x-request-id"));
489        assert!(result.contains("req-123"));
490    }
491
492    #[test]
493    fn test_format_headers_with_redaction_disabled() {
494        let mut headers = HeaderMap::new();
495        headers.insert("authorization", "Bearer secret-token".parse().unwrap());
496        headers.insert("x-api-key", "api-key-123".parse().unwrap());
497        headers.insert("content-type", "application/json".parse().unwrap());
498
499        let result = format_headers_redacted(&headers, true);
500
501        // When redaction is disabled, all headers and values should be present
502        assert!(result.contains("authorization"));
503        assert!(result.contains("Bearer secret-token"));
504        assert!(result.contains("x-api-key"));
505        assert!(result.contains("api-key-123"));
506        assert!(result.contains("content-type"));
507        assert!(result.contains("application/json"));
508        // [REDACTED] should NOT be present when redaction is disabled
509        assert!(!result.contains("[REDACTED]"));
510    }
511}
iceberg_catalog_rest/client.rs

iceberg_catalog_rest/
client.rs
