mz_storage_operators/oneshot_source/
http_source.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Generic HTTP oneshot source that will fetch a file from the public internet.
11
12use std::net::SocketAddr;
13use std::sync::Arc;
14
15use bytes::Bytes;
16use derivative::Derivative;
17use futures::TryStreamExt;
18use futures::stream::{BoxStream, StreamExt};
19use mz_ore::url::SensitiveUrl;
20use reqwest::Client;
21use reqwest::dns::{Addrs, Name, Resolve, Resolving};
22use serde::{Deserialize, Serialize};
23
24use crate::oneshot_source::util::IntoRangeHeaderValue;
25use crate::oneshot_source::{
26    Encoding, OneshotObject, OneshotSource, StorageErrorX, StorageErrorXContext, StorageErrorXKind,
27};
28
29/// reqwest DNS resolver that delegates to [`mz_ore::netio::resolve_address`].
30///
31/// Only the IP resolution step is overridden — reqwest still uses the URL's
32/// original hostname for SNI and TLS certificate validation, so HTTPS works
33/// normally.
34#[derive(Debug)]
35struct MzHttpResolver {
36    enforce_external_addresses: bool,
37}
38
39impl Resolve for MzHttpResolver {
40    fn resolve(&self, name: Name) -> Resolving {
41        let enforce = self.enforce_external_addresses;
42        Box::pin(async move {
43            let ips = mz_ore::netio::resolve_address(name.as_str(), enforce).await?;
44            // reqwest substitutes the conventional port (80/443) when the
45            // SocketAddr's port is 0 and no explicit port was given in the URL.
46            let addrs: Addrs = Box::new(
47                ips.into_iter()
48                    .map(|ip| SocketAddr::new(ip, 0))
49                    .collect::<Vec<_>>()
50                    .into_iter(),
51            );
52            Ok(addrs)
53        })
54    }
55}
56
57/// Build a reqwest [`Client`] for fetching `COPY FROM` URLs. This uses
58/// [`mz_ore::netio::resolve_address`] for DNS resolution.
59///
60/// Redirects are disabled: the custom DNS resolver re-validates hostnames on
61/// every hop, but reqwest skips DNS for IP-literal targets, so a redirect to
62/// `http://127.0.0.1/` would bypass the SSRF check. Refusing to follow
63/// redirects closes that hole.
64pub fn build_http_client(enforce_external_addresses: bool) -> Result<Client, reqwest::Error> {
65    Client::builder()
66        .dns_resolver(Arc::new(MzHttpResolver {
67            enforce_external_addresses,
68        }))
69        .redirect(reqwest::redirect::Policy::none())
70        .build()
71}
72
73/// Returns an error unless `response` has a 2xx status. A 3xx gets a
74/// dedicated error because redirects are disabled on the HTTP client (see
75/// `build_http_client`). Any other status would otherwise let an error page
76/// body be ingested as data.
77fn check_success(response: &reqwest::Response) -> Result<(), StorageErrorX> {
78    let status = response.status();
79    if status.is_redirection() {
80        return Err(StorageErrorXKind::Redirect(status.as_u16()).into());
81    }
82    if !status.is_success() {
83        return Err(StorageErrorXKind::HttpStatus(status.as_u16()).into());
84    }
85    Ok(())
86}
87
88/// Generic oneshot source that fetches a file from a URL on the public internet.
89#[derive(Clone, Derivative)]
90#[derivative(Debug)]
91pub struct HttpOneshotSource {
92    #[derivative(Debug = "ignore")]
93    client: Client,
94    origin: SensitiveUrl,
95}
96
97impl HttpOneshotSource {
98    pub fn new(client: Client, origin: SensitiveUrl) -> Self {
99        HttpOneshotSource { client, origin }
100    }
101}
102
103/// Object returned from an [`HttpOneshotSource`].
104#[derive(Debug, Clone, Serialize, Deserialize)]
105pub struct HttpObject {
106    /// [`SensitiveUrl`] to access the file.
107    url: SensitiveUrl,
108    /// Name of the file.
109    filename: String,
110    /// Size of this file reported by the [`Content-Length`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Length) header
111    size: usize,
112    /// Any values reporting from the [`Content-Encoding`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Encoding) header.
113    content_encoding: Vec<Encoding>,
114}
115
116impl OneshotObject for HttpObject {
117    fn name(&self) -> &str {
118        &self.filename
119    }
120
121    fn path(&self) -> &str {
122        &self.filename
123    }
124
125    fn size(&self) -> usize {
126        self.size
127    }
128
129    fn encodings(&self) -> &[Encoding] {
130        &self.content_encoding
131    }
132}
133
134#[derive(Debug, Clone, Serialize, Deserialize)]
135pub enum HttpChecksum {
136    /// No checksumming is requested.
137    None,
138    /// The HTTP [`ETag`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag) header.
139    ETag(String),
140    /// The HTTP [`Last-Modified`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Last-Modified) header.
141    LastModified(String),
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    /// `reqwest::dns::Name` has no public constructor, so we exercise
149    /// [`MzHttpResolver`] through a fully-built [`Client`]. `localhost`
150    /// resolves via /etc/hosts on supported platforms and stays inside the
151    /// resolver path (an IP literal would short-circuit DNS entirely).
152    #[mz_ore::test(tokio::test)]
153    #[cfg_attr(miri, ignore)]
154    async fn build_http_client_rejects_localhost_when_enforced() {
155        let client = build_http_client(true).expect("build client");
156        let err = client
157            .get("http://localhost:1/")
158            .send()
159            .await
160            .expect_err("request must fail at DNS resolution");
161        // Walk the error chain for the resolver's PrivateAddress message —
162        // reqwest wraps it inside its connect error, so a `to_string()` on
163        // the top-level error is not enough.
164        let mut found = false;
165        let mut current: &dyn std::error::Error = &err;
166        loop {
167            if current.to_string().to_lowercase().contains("private") {
168                found = true;
169                break;
170            }
171            match current.source() {
172                Some(src) => current = src,
173                None => break,
174            }
175        }
176        assert!(found, "expected private-address rejection, got: {err:?}");
177    }
178
179    /// With enforcement off the resolver returns the loopback IP, so the
180    /// request reaches the connect stage and fails for a different reason
181    /// (port 1 is not listening). The point is that DNS does *not* fail.
182    #[mz_ore::test(tokio::test)]
183    #[cfg_attr(miri, ignore)]
184    async fn build_http_client_allows_localhost_when_not_enforced() {
185        let client = build_http_client(false).expect("build client");
186        let err = client
187            .get("http://localhost:1/")
188            .send()
189            .await
190            .expect_err("port 1 should not be listening");
191        let mut current: &dyn std::error::Error = &err;
192        loop {
193            assert!(
194                !current.to_string().to_lowercase().contains("private"),
195                "expected a connect error, not a DNS rejection: {err:?}"
196            );
197            match current.source() {
198                Some(src) => current = src,
199                None => break,
200            }
201        }
202    }
203}
204
205impl OneshotSource for HttpOneshotSource {
206    type Object = HttpObject;
207    type Checksum = HttpChecksum;
208
209    async fn list<'a>(&'a self) -> Result<Vec<(Self::Object, Self::Checksum)>, StorageErrorX> {
210        // TODO(cf3): Support listing files from a directory index.
211
212        // To get metadata about a file we'll first try issuing a `HEAD` request, which
213        // canonically is the right thing do.
214        let response = self
215            .client
216            .head(self.origin.0.clone())
217            .send()
218            .await
219            .context("HEAD request")?;
220
221        // Other HEAD failures fall back to GET below, which must succeed.
222        if response.status().is_redirection() {
223            return Err(StorageErrorXKind::Redirect(response.status().as_u16()).into());
224        }
225
226        // Not all servers accept `HEAD` requests though, so we'll fallback to a `GET`
227        // request and skip fetching the body.
228        let headers = match response.error_for_status() {
229            Ok(response) => response.headers().clone(),
230            Err(err) => {
231                tracing::warn!(status = ?err.status(), "HEAD request failed");
232
233                let response = self
234                    .client
235                    .get(self.origin.0.clone())
236                    .send()
237                    .await
238                    .context("GET request")?;
239
240                check_success(&response)?;
241
242                let headers = response.headers().clone();
243
244                // Immediately drop the response so we don't attempt to fetch the body.
245                drop(response);
246
247                headers
248            }
249        };
250
251        let get_header = |name: &reqwest::header::HeaderName| {
252            let header = headers.get(name)?;
253            match header.to_str() {
254                Err(e) => {
255                    tracing::warn!("failed to deserialize header '{name}', err: {e}");
256                    None
257                }
258                Ok(value) => Some(value),
259            }
260        };
261
262        // Get a checksum from the content.
263        let checksum = if let Some(etag) = get_header(&reqwest::header::ETAG) {
264            HttpChecksum::ETag(etag.to_string())
265        } else if let Some(last_modified) = get_header(&reqwest::header::LAST_MODIFIED) {
266            let last_modified = last_modified.to_string();
267            HttpChecksum::LastModified(last_modified.to_string())
268        } else {
269            HttpChecksum::None
270        };
271
272        // Get the size of the object from the Conent-Length header.
273        let size = get_header(&reqwest::header::CONTENT_LENGTH)
274            .ok_or(StorageErrorXKind::MissingSize)
275            .and_then(|s| s.parse::<usize>().map_err(StorageErrorXKind::generic))
276            .context("content-length header")?;
277
278        // TODO(cf1): We should probably check the content-type as well. At least for advisory purposes.
279
280        let filename = self
281            .origin
282            .path_segments()
283            .and_then(|segments| segments.rev().next())
284            .map(|s| s.to_string())
285            .unwrap_or_default();
286        let object = HttpObject {
287            url: self.origin.clone(),
288            filename,
289            size,
290            content_encoding: vec![],
291        };
292        tracing::info!(?object, "found objects");
293
294        Ok(vec![(object, checksum)])
295    }
296
297    fn get<'s>(
298        &'s self,
299        object: Self::Object,
300        _checksum: Self::Checksum,
301        range: Option<std::ops::RangeInclusive<usize>>,
302    ) -> BoxStream<'s, Result<Bytes, StorageErrorX>> {
303        // TODO(cf1): Validate our checksum.
304
305        let initial_response = async move {
306            let mut request = self.client.get(object.url.0);
307
308            if let Some(range) = &range {
309                let value = range.into_range_header_value();
310                request = request.header(&reqwest::header::RANGE, value);
311            }
312
313            // TODO(parkmycar): We should probably assert that the response contains
314            // an appropriate Content-Range header in the response, and maybe that we
315            // got back an HTTP 206?
316
317            let response = request.send().await.context("get")?;
318            check_success(&response)?;
319            let bytes_stream = response.bytes_stream().err_into();
320
321            Ok::<_, StorageErrorX>(bytes_stream)
322        };
323
324        futures::stream::once(initial_response)
325            .try_flatten()
326            .boxed()
327    }
328}
mz_storage_operators/oneshot_source/http_source.rs

mz_storage_operators/oneshot_source/
http_source.rs
