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

iceberg_catalog_rest/
client.rs
