vector/sinks/util/
uri.rs

1use std::{fmt, str::FromStr};
2
3use http::uri::{Authority, PathAndQuery, Scheme, Uri};
4use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, percent_decode_str, utf8_percent_encode};
5use snafu::{ResultExt, Snafu};
6use vector_lib::configurable::configurable_component;
7
8use crate::http::Auth;
9
10/// Characters that must be percent-encoded in a URI userinfo.
11/// RFC 3986 unreserved characters are left as-is.
12const USERINFO: &AsciiSet = &NON_ALPHANUMERIC
13    .remove(b'-')
14    .remove(b'.')
15    .remove(b'_')
16    .remove(b'~');
17
18/// A wrapper for `http::Uri` that implements `Deserialize` and `Serialize`.
19///
20/// Authorization credentials, if they exist, will be removed from the URI and stored separately in `auth`.
21#[configurable_component]
22#[configurable(title = "The URI component of a request.", description = "")]
23#[derive(Default, Debug, Clone)]
24#[serde(try_from = "String", into = "String")]
25pub struct UriSerde {
26    pub uri: Uri,
27    pub auth: Option<Auth>,
28}
29
30impl UriSerde {
31    /// `Uri` supports incomplete URIs such as "/test", "example.com", etc.
32    /// This function fills in empty scheme with HTTP,
33    /// and empty authority with "127.0.0.1".
34    pub fn with_default_parts(&self) -> Self {
35        let mut parts = self.uri.clone().into_parts();
36        if parts.scheme.is_none() {
37            parts.scheme = Some(Scheme::HTTP);
38        }
39        if parts.authority.is_none() {
40            parts.authority = Some(Authority::from_static("127.0.0.1"));
41        }
42        if parts.path_and_query.is_none() {
43            // just an empty `path_and_query`,
44            // but `from_parts` will fail without this.
45            parts.path_and_query = Some(PathAndQuery::from_static(""));
46        }
47        let uri = Uri::from_parts(parts).expect("invalid parts");
48        Self {
49            uri,
50            auth: self.auth.clone(),
51        }
52    }
53
54    /// Creates a new instance of `UriSerde` by appending a path to the existing one.
55    pub fn append_path(&self, path: &str) -> crate::Result<Self> {
56        let uri = self.uri.to_string();
57        let self_path = uri.trim_end_matches('/');
58        let other_path = path.trim_start_matches('/');
59        let path = format!("{self_path}/{other_path}");
60        let uri = path.parse::<Uri>()?;
61        Ok(Self {
62            uri,
63            auth: self.auth.clone(),
64        })
65    }
66
67    #[allow(clippy::missing_const_for_fn)] // constant functions cannot evaluate destructors
68    pub fn with_auth(mut self, auth: Option<Auth>) -> Self {
69        self.auth = auth;
70        self
71    }
72}
73
74impl TryFrom<String> for UriSerde {
75    type Error = crate::Error;
76
77    fn try_from(value: String) -> Result<Self, Self::Error> {
78        value.as_str().parse()
79    }
80}
81
82impl From<UriSerde> for String {
83    fn from(uri: UriSerde) -> Self {
84        uri.to_string()
85    }
86}
87
88impl fmt::Display for UriSerde {
89    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90        match (self.uri.authority(), &self.auth) {
91            (Some(authority), Some(Auth::Basic { user, password })) => {
92                let user = utf8_percent_encode(user, USERINFO);
93                let password = utf8_percent_encode(password.inner(), USERINFO);
94                let authority = format!("{user}:{password}@{authority}");
95                let authority =
96                    Authority::from_maybe_shared(authority).map_err(|_| std::fmt::Error)?;
97                let mut parts = self.uri.clone().into_parts();
98                parts.authority = Some(authority);
99                Uri::from_parts(parts).unwrap().fmt(f)
100            }
101            _ => self.uri.fmt(f),
102        }
103    }
104}
105
106impl FromStr for UriSerde {
107    type Err = crate::Error;
108
109    fn from_str(s: &str) -> Result<Self, Self::Err> {
110        let uri: Uri = s.parse()?;
111        uri.try_into()
112    }
113}
114
115impl TryFrom<Uri> for UriSerde {
116    type Error = crate::Error;
117
118    /// Fallible construction from a parsed `Uri`, extracting any basic auth
119    /// credentials from the authority.
120    ///
121    /// This can fail: `http::Uri` accepts authorities that `url::Url` rejects
122    /// (e.g. a non-numeric port), in which case the basic auth cannot be
123    /// extracted.
124    fn try_from(uri: Uri) -> Result<Self, Self::Error> {
125        match uri.authority() {
126            None => Ok(Self { uri, auth: None }),
127            Some(authority) => {
128                let (authority, auth) = get_basic_auth(authority)?;
129
130                let mut parts = uri.into_parts();
131                parts.authority = Some(authority);
132                let uri = Uri::from_parts(parts)?;
133
134                Ok(Self { uri, auth })
135            }
136        }
137    }
138}
139
140fn get_basic_auth(authority: &Authority) -> crate::Result<(Authority, Option<Auth>)> {
141    // `http::Uri` accepts authorities that `url::Url` rejects (e.g. a
142    // non-numeric port), so this parse can fail; propagate the error instead
143    // of panicking.
144    let url = url::Url::parse(&format!("http://{authority}"))?;
145    let Some((_, host_port)) = authority.as_str().rsplit_once('@') else {
146        return Ok((authority.clone(), None));
147    };
148
149    let has_auth = !url.username().is_empty() || url.password().is_some();
150    let auth = has_auth.then(|| {
151        let user = percent_decode_str(url.username())
152            .decode_utf8_lossy()
153            .into_owned();
154        let password = percent_decode_str(url.password().unwrap_or(""))
155            .decode_utf8_lossy()
156            .into_owned();
157        Auth::Basic {
158            user,
159            password: password.into(),
160        }
161    });
162
163    // Rebuild the authority from the parsed URL so the host is normalized
164    // (e.g. host case), while retaining an explicit port from the raw
165    // authority (`url::Url` drops the port when it matches the scheme default).
166    let host = url
167        .host_str()
168        .ok_or_else(|| "unexpected empty authority".to_string())?;
169    let port = host_port
170        .rsplit_once(':')
171        .and_then(|(_, port)| port.parse::<u16>().ok());
172    let authority = match port {
173        Some(port) => format!("{host}:{port}"),
174        None => host.to_string(),
175    };
176
177    Ok((authority.parse()?, auth))
178}
179
180/// Simplify the URI into a protocol and endpoint by removing the
181/// "query" portion of the `path_and_query`.
182pub fn protocol_endpoint(uri: Uri) -> (String, String) {
183    let mut parts = uri.into_parts();
184
185    // Drop any username and password
186    parts.authority = parts.authority.map(|auth| {
187        let host = auth.host();
188        match auth.port() {
189            None => host.to_string(),
190            Some(port) => format!("{host}:{port}"),
191        }
192        .parse()
193        .unwrap_or_else(|_| unreachable!())
194    });
195
196    // Drop the query and fragment
197    parts.path_and_query = parts.path_and_query.map(|pq| {
198        pq.path()
199            .parse::<PathAndQuery>()
200            .unwrap_or_else(|_| unreachable!())
201    });
202
203    (
204        parts.scheme.clone().unwrap_or(Scheme::HTTP).as_str().into(),
205        Uri::from_parts(parts)
206            .unwrap_or_else(|_| unreachable!())
207            .to_string(),
208    )
209}
210
211/// Error returned when a configured endpoint cannot be used as an absolute HTTP URL.
212#[derive(Debug, Snafu)]
213pub enum HttpEndpointError {
214    #[snafu(display("endpoint `{endpoint}` is not a valid URI: {source}"))]
215    InvalidUri {
216        endpoint: String,
217        source: http::uri::InvalidUri,
218    },
219
220    #[snafu(display("endpoint `{endpoint}` has an invalid path `{path}`: {source}"))]
221    InvalidPath {
222        endpoint: String,
223        path: String,
224        source: http::uri::InvalidUri,
225    },
226
227    #[snafu(display("endpoint `{endpoint}` cannot be reassembled from its parts: {source}"))]
228    InvalidUriParts {
229        endpoint: String,
230        source: http::uri::InvalidUriParts,
231    },
232
233    #[snafu(display(
234        "endpoint must be an absolute http(s) URL, for example `https://example.com`; got `{endpoint}`"
235    ))]
236    NotAbsoluteHttp { endpoint: String },
237
238    #[snafu(display("endpoint `{endpoint}` has an invalid port"))]
239    InvalidPort { endpoint: String },
240}
241
242/// A `Uri` proven to be an absolute `http`/`https` URL.
243///
244/// Constructing an `HttpEndpoint` is the only way to obtain one: both
245/// [`HttpEndpoint::new`] and [`HttpEndpoint::parse`] reject URIs without an
246/// `http`/`https` scheme or without an authority. Sinks that issue requests
247/// through `HttpClient` need this invariant, since `HttpClient` rejects such
248/// URIs at request time, deferring a pure configuration error to runtime.
249///
250/// As a configuration type it deserializes from a string, so an invalid
251/// endpoint is rejected at config load time with the config path in the error.
252///
253/// Path composition goes through [`HttpEndpoint::append_path`], which
254/// manipates `Uri` parts directly instead of string-concatenating and
255/// re-parsing, so the scheme and authority are preserved and the result is
256/// still an absolute `http(s)` URL.
257#[configurable_component]
258#[configurable(title = "An absolute HTTP(S) URL.", description = "")]
259#[derive(Debug, Clone, PartialEq, Eq)]
260#[serde(try_from = "String", into = "String")]
261pub struct HttpEndpoint(Uri);
262
263fn redact_uri(uri: &Uri) -> String {
264    if uri
265        .authority()
266        .is_some_and(|authority| authority.as_str().contains('@'))
267    {
268        "<redacted endpoint>".to_owned()
269    } else {
270        uri.to_string()
271    }
272}
273
274fn redact_unparsed_endpoint(endpoint: &str) -> String {
275    if endpoint.contains('@') {
276        "<redacted endpoint>".to_owned()
277    } else {
278        endpoint.to_owned()
279    }
280}
281
282impl TryFrom<String> for HttpEndpoint {
283    type Error = HttpEndpointError;
284
285    fn try_from(value: String) -> Result<Self, Self::Error> {
286        Self::parse(&value)
287    }
288}
289
290impl From<HttpEndpoint> for String {
291    fn from(value: HttpEndpoint) -> Self {
292        value.to_string()
293    }
294}
295
296impl HttpEndpoint {
297    /// Requires `uri` to be an absolute `http`/`https` URL with a host and a
298    /// usable port.
299    ///
300    /// The authority check alone is not enough: `http://:8080` parses as a
301    /// valid `http::Uri` with an authority but an empty host, and
302    /// `http://localhost:notaport` parses with a nonempty host but a port that
303    /// cannot be dialed. Both are checked explicitly.
304    pub fn new(uri: Uri) -> Result<Self, HttpEndpointError> {
305        let has_valid_scheme_and_host = matches!(uri.scheme_str(), Some("http" | "https"))
306            && uri.host().is_some_and(|host| !host.is_empty());
307        if !has_valid_scheme_and_host {
308            return Err(HttpEndpointError::NotAbsoluteHttp {
309                endpoint: redact_uri(&uri),
310            });
311        }
312        if authority_has_invalid_port(&uri) {
313            return Err(HttpEndpointError::InvalidPort {
314                endpoint: redact_uri(&uri),
315            });
316        }
317        Ok(Self(uri))
318    }
319
320    /// Parses `endpoint` and requires it to be an absolute `http`/`https` URL.
321    ///
322    /// A missing scheme is defaulted to `https`, so `example.com:8080` becomes
323    /// `https://example.com:8080`. An explicit `http`/`https` scheme is
324    /// preserved. Endpoints that still lack a host after defaulting (for
325    /// example `/path`) are rejected.
326    pub fn parse(endpoint: &str) -> Result<Self, HttpEndpointError> {
327        // Default a missing scheme to https. `http::Uri` cannot parse
328        // `host:port/path` without a scheme (it reads `host` as a scheme), so
329        // the scheme is added up front rather than relying on the parser to
330        // accept authority-form input.
331        let parse = |value: &str| {
332            value
333                .parse::<Uri>()
334                .map_err(|source| HttpEndpointError::InvalidUri {
335                    endpoint: redact_unparsed_endpoint(endpoint),
336                    source,
337                })
338        };
339        let uri = if has_scheme(endpoint) {
340            parse(endpoint)?
341        } else {
342            parse(&format!("https://{endpoint}"))?
343        };
344        Self::new(uri)
345    }
346
347    /// Returns the underlying `Uri`.
348    pub const fn as_uri(&self) -> &Uri {
349        &self.0
350    }
351
352    /// Consumes the endpoint, returning the underlying `Uri`.
353    pub fn into_uri(self) -> Uri {
354        self.0
355    }
356
357    /// Extracts basic-auth credentials embedded in the authority, returning a
358    /// credential-free endpoint alongside the credentials.
359    pub fn extract_basic_auth(self) -> crate::Result<(Self, Option<Auth>)> {
360        if !self
361            .as_uri()
362            .authority()
363            .is_some_and(|authority| authority.as_str().contains('@'))
364        {
365            return Ok((self, None));
366        }
367
368        let UriSerde { uri, auth } = self.into_uri().try_into()?;
369        Ok((Self::new(uri)?, auth))
370    }
371
372    /// Returns the URL scheme (`http` or `https`) of this endpoint.
373    ///
374    /// [`HttpEndpoint::new`] guarantees the scheme is an absolute `http`/`https`
375    /// scheme, so this is infallible.
376    pub fn protocol(&self) -> &str {
377        self.0.scheme_str().unwrap_or("https")
378    }
379
380    /// Appends `path` to this endpoint, preserving the scheme and authority.
381    ///
382    /// `path` may include a leading slash and a query. The existing query, if
383    /// any, is dropped (as with `UriSerde::append_path`), but the scheme and
384    /// authority are preserved and the result is still an absolute `http(s)` URL.
385    pub fn append_path(&self, path: &str) -> Result<Self, HttpEndpointError> {
386        if path.is_empty() {
387            return Ok(self.clone());
388        }
389        let mut parts = self.0.clone().into_parts();
390        let base_path = parts
391            .path_and_query
392            .as_ref()
393            .map(PathAndQuery::path)
394            .unwrap_or_default();
395        let joined = if base_path.is_empty() {
396            path.to_string()
397        } else if base_path.ends_with('/') {
398            format!("{base_path}{}", path.strip_prefix('/').unwrap_or(path))
399        } else {
400            format!("{base_path}/{}", path.strip_prefix('/').unwrap_or(path))
401        };
402        parts.path_and_query =
403            Some(
404                joined
405                    .parse::<PathAndQuery>()
406                    .with_context(|_| InvalidPathSnafu {
407                        endpoint: redact_uri(&self.0),
408                        path: joined,
409                    })?,
410            );
411        let uri = Uri::from_parts(parts).with_context(|_| InvalidUriPartsSnafu {
412            endpoint: redact_uri(&self.0),
413        })?;
414        Self::new(uri)
415    }
416
417    /// Appends `suffix` directly to the path without inserting a separator.
418    ///
419    /// Unlike [`HttpEndpoint::append_path`], this does not add a `/`. It is for
420    /// API method suffixes that attach directly to a resource path, such as
421    /// Google's `:publish` convention.
422    pub fn append_raw_suffix(&self, suffix: &str) -> Result<Self, HttpEndpointError> {
423        if suffix.is_empty() {
424            return Ok(self.clone());
425        }
426        let mut parts = self.0.clone().into_parts();
427        let base_path = parts
428            .path_and_query
429            .as_ref()
430            .map(PathAndQuery::path)
431            .unwrap_or_default();
432        let joined = format!("{base_path}{suffix}");
433        parts.path_and_query =
434            Some(
435                joined
436                    .parse::<PathAndQuery>()
437                    .with_context(|_| InvalidPathSnafu {
438                        endpoint: redact_uri(&self.0),
439                        path: joined,
440                    })?,
441            );
442        let uri = Uri::from_parts(parts).with_context(|_| InvalidUriPartsSnafu {
443            endpoint: redact_uri(&self.0),
444        })?;
445        Self::new(uri)
446    }
447}
448
449impl fmt::Display for HttpEndpoint {
450    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
451        self.0.fmt(f)
452    }
453}
454
455/// Returns `true` if the URI's authority contains a port that is not a valid
456/// `u16`.
457///
458/// `http::Uri` accepts non-numeric ports (for example
459/// `http://localhost:notaport`), which `HttpClient` cannot dial. `Authority::port`
460/// returns `None` for both a missing port and an invalid one, so the raw
461/// authority is inspected instead.
462fn authority_has_invalid_port(uri: &Uri) -> bool {
463    let Some(authority) = uri.authority() else {
464        return false;
465    };
466    let auth = authority.as_str();
467    // Strip any userinfo (everything up to the last `@`).
468    let host_port = auth
469        .rsplit_once('@')
470        .map(|(_, host_port)| host_port)
471        .unwrap_or(auth);
472    // An IPv6 host is bracketed; the port follows the closing `]`.
473    let host_end = host_port.rfind(']').map_or(0, |i| i + 1);
474    let Some(host_port) = host_port.get(host_end..) else {
475        return false;
476    };
477    host_port.rfind(':').is_some_and(|i| {
478        host_port
479            .get(i + 1..)
480            .is_some_and(|port| port.parse::<u16>().is_err())
481    })
482}
483
484/// Returns `true` if `endpoint` starts with a URI scheme (`[a-zA-Z][a-zA-Z0-9+.-]*://`).
485///
486/// The scheme must be at the very start: a `://` later in the path or query
487/// (for example `localhost:8080/write?target=http://upstream`) is not a scheme
488/// marker, so the endpoint is still defaulted to `https`.
489fn has_scheme(endpoint: &str) -> bool {
490    let Some(scheme_end) = endpoint.find("://") else {
491        return false;
492    };
493    let Some(scheme) = endpoint.get(..scheme_end) else {
494        return false;
495    };
496    let mut chars = scheme.chars();
497    matches!(chars.next(), Some(c) if c.is_ascii_alphabetic())
498        && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.'))
499}
500
501#[cfg(test)]
502mod tests {
503    use super::*;
504    use rstest::rstest;
505
506    fn test_parse(input: &str, expected_uri: &'static str, expected_auth: Option<(&str, &str)>) {
507        let UriSerde { uri, auth } = input.parse().unwrap();
508        assert_eq!(uri, Uri::from_static(expected_uri));
509        assert_eq!(
510            auth,
511            expected_auth.map(|(user, password)| {
512                Auth::Basic {
513                    user: user.to_owned(),
514                    password: password.to_owned().into(),
515                }
516            })
517        );
518    }
519
520    #[test]
521    fn parse_endpoint() {
522        test_parse(
523            "http://user:pass@example.com/test",
524            "http://example.com/test",
525            Some(("user", "pass")),
526        );
527
528        test_parse("localhost:8080", "localhost:8080", None);
529
530        test_parse("/api/test", "/api/test", None);
531
532        test_parse(
533            "http://user:pass;@example.com",
534            "http://example.com",
535            Some(("user", "pass;")),
536        );
537
538        test_parse(
539            "user:pass@example.com",
540            "example.com",
541            Some(("user", "pass")),
542        );
543
544        test_parse("user@example.com", "example.com", Some(("user", "")));
545
546        test_parse(
547            "https://user:pass@example.com:80/api",
548            "https://example.com:80/api",
549            Some(("user", "pass")),
550        );
551
552        test_parse(
553            "https://:secret@example.com/api",
554            "https://example.com/api",
555            Some(("", "secret")),
556        );
557    }
558
559    #[test]
560    fn parse_rejects_malformed_authority_without_panicking() {
561        // `http::Uri` accepts a non-numeric port in the authority, but
562        // `url::Url` rejects it. This must be a parse error, not a panic.
563        let result = "http://user:pass@localhost:notaport/path".parse::<UriSerde>();
564        assert!(result.is_err());
565    }
566
567    #[test]
568    fn protocol_endpoint_parses_urls() {
569        let parse = |uri: &str| protocol_endpoint(uri.parse().unwrap());
570
571        assert_eq!(
572            parse("http://example.com/"),
573            ("http".into(), "http://example.com/".into())
574        );
575        assert_eq!(
576            parse("https://user:pass@example.org:123/path?query"),
577            ("https".into(), "https://example.org:123/path".into())
578        );
579        assert_eq!(
580            parse("gopher://example.net:123/path?query#frag,emt"),
581            ("gopher".into(), "gopher://example.net:123/path".into())
582        );
583    }
584
585    #[test]
586    fn http_endpoint_accepts_absolute_http_urls() {
587        for endpoint in [
588            "http://example.com",
589            "https://example.com",
590            "https://example.com:8088/services/collector",
591            "http://127.0.0.1:9000/endpoint?query=1",
592            "https://user:pass@example.com/path",
593            // IPv6 hosts are returned bracketed (`[::1]`) and must be accepted.
594            "http://[::1]:8080",
595            "https://[::1]/path",
596            // A missing scheme is defaulted to https.
597            "example.com",
598            "example.com:8088/services/collector",
599            "localhost:8080",
600            "[::1]:8080",
601            // A `://` later in the path or query is not a scheme marker.
602            "localhost:8080/write?target=http://upstream",
603        ] {
604            let endpoint =
605                HttpEndpoint::parse(endpoint).expect("should accept absolute http(s) URL");
606            assert!(matches!(
607                endpoint.as_uri().scheme_str(),
608                Some("http" | "https")
609            ));
610            assert!(endpoint.as_uri().authority().is_some());
611        }
612    }
613
614    #[test]
615    fn http_endpoint_extracts_basic_auth() {
616        let endpoint = HttpEndpoint::parse("http://user:pass@example.com:8080/path").unwrap();
617        let (endpoint, auth) = endpoint.extract_basic_auth().unwrap();
618
619        assert_eq!(endpoint.to_string(), "http://example.com:8080/path");
620        assert!(matches!(auth, Some(Auth::Basic { user, .. }) if user == "user"));
621    }
622
623    #[rstest]
624    #[case::explicit_port(
625        "user:pass@example.com:80/path",
626        "https://example.com:80/path",
627        "user",
628        "pass"
629    )]
630    #[case::empty_username(":secret@example.com/path", "https://example.com/path", "", "secret")]
631    fn http_endpoint_auth_extraction_handles_userinfo(
632        #[case] endpoint: &str,
633        #[case] expected_endpoint: &str,
634        #[case] expected_user: &str,
635        #[case] expected_password: &str,
636    ) {
637        let endpoint = HttpEndpoint::parse(endpoint).unwrap();
638        let (endpoint, auth) = endpoint.extract_basic_auth().unwrap();
639
640        assert_eq!(endpoint.to_string(), expected_endpoint);
641        assert!(matches!(
642            auth,
643            Some(Auth::Basic { user, password })
644                if user == expected_user && password.inner() == expected_password
645        ));
646    }
647
648    #[test]
649    fn http_endpoint_auth_extraction_normalizes_host() {
650        let endpoint = HttpEndpoint::parse("user:pass@EXAMPLE.com/path").unwrap();
651        let (endpoint, auth) = endpoint.extract_basic_auth().unwrap();
652
653        assert_eq!(endpoint.to_string(), "https://example.com/path");
654        assert!(matches!(auth, Some(Auth::Basic { user, .. }) if user == "user"));
655    }
656
657    #[test]
658    fn http_endpoint_auth_extraction_round_trips_encoded_password() {
659        let endpoint = HttpEndpoint::parse(":%2F@example.com/path").unwrap();
660        let (endpoint, auth) = endpoint.extract_basic_auth().unwrap();
661
662        let uri_serde = UriSerde {
663            uri: endpoint.into_uri(),
664            auth,
665        };
666        assert_eq!(uri_serde.to_string(), "https://:%2F@example.com/path");
667    }
668
669    #[test]
670    fn http_endpoint_defaults_missing_scheme_to_https() {
671        for endpoint in [
672            "example.com",
673            "example.com:8080",
674            "localhost:8080/path",
675            "[::1]:8080",
676        ] {
677            let endpoint =
678                HttpEndpoint::parse(endpoint).expect("should default a missing scheme to https");
679            assert_eq!(endpoint.as_uri().scheme_str(), Some("https"));
680            assert!(
681                endpoint
682                    .as_uri()
683                    .host()
684                    .is_some_and(|host| !host.is_empty())
685            );
686        }
687        // An explicit scheme is preserved.
688        assert_eq!(
689            HttpEndpoint::parse("http://example.com")
690                .unwrap()
691                .as_uri()
692                .scheme_str(),
693            Some("http")
694        );
695    }
696
697    #[test]
698    fn http_endpoint_rejects_non_absolute_http_urls() {
699        for endpoint in [
700            // No scheme and no host: `http::Uri` parses these as a path.
701            "/services/collector",
702            "",
703            // Absolute, but not a scheme `HttpClient` can dial.
704            "gopher://example.com",
705            "unix:///var/run/vector.sock",
706            // Scheme but no authority.
707            "http:///path",
708            // Authority with a port but an empty host: `http::Uri` parses this
709            // with `authority() == Some` and `host() == Some("")`.
710            "http://:8080",
711            "http://:8080/path",
712            // A non-numeric port parses with a nonempty host but cannot be dialed.
713            "http://localhost:notaport",
714            "https://example.com:notaport/path",
715            // Multiple port separators are rejected by the URI parser.
716            "http://localhost:notaport:8080",
717        ] {
718            assert!(
719                matches!(
720                    HttpEndpoint::parse(endpoint),
721                    Err(HttpEndpointError::NotAbsoluteHttp { .. })
722                        | Err(HttpEndpointError::InvalidUri { .. })
723                        | Err(HttpEndpointError::InvalidUriParts { .. })
724                        | Err(HttpEndpointError::InvalidPort { .. })
725                ),
726                "expected `{endpoint}` to be rejected"
727            );
728        }
729    }
730
731    #[test]
732    fn http_endpoint_reports_unparseable_endpoints() {
733        let endpoint = "http://exa mple.com";
734        let error = HttpEndpoint::parse(endpoint).unwrap_err();
735        assert!(matches!(error, HttpEndpointError::InvalidUri { .. }));
736        assert!(error.to_string().contains(endpoint));
737    }
738
739    #[test]
740    fn http_endpoint_rejects_malformed_ports() {
741        for endpoint in [
742            "http://localhost:notaport",
743            "https://example.com:notaport/path",
744        ] {
745            assert!(matches!(
746                HttpEndpoint::parse(endpoint),
747                Err(HttpEndpointError::InvalidPort { .. })
748            ));
749        }
750    }
751
752    #[test]
753    fn http_endpoint_errors_redact_userinfo() {
754        for endpoint in [
755            "http://user:secret@localhost:notaport",
756            "http://user:secret@exa mple.com",
757        ] {
758            let message = HttpEndpoint::parse(endpoint).unwrap_err().to_string();
759            assert!(message.contains("<redacted endpoint>"), "{message}");
760            assert!(!message.contains("secret"), "{message}");
761        }
762    }
763
764    #[test]
765    fn http_endpoint_append_errors_redact_userinfo() {
766        let endpoint = HttpEndpoint::parse("https://user:secret@example.com/base").unwrap();
767
768        for error in [
769            endpoint.append_path("invalid path").unwrap_err(),
770            endpoint.append_raw_suffix(" invalid suffix").unwrap_err(),
771        ] {
772            assert!(matches!(&error, HttpEndpointError::InvalidPath { .. }));
773            let message = error.to_string();
774            assert!(message.contains("<redacted endpoint>"), "{message}");
775            assert!(!message.contains("secret"), "{message}");
776        }
777    }
778
779    #[test]
780    fn http_endpoint_append_path_joins_without_string_concatenation() {
781        let base = HttpEndpoint::parse("https://example.com").unwrap();
782
783        assert_eq!(
784            base.append_path("vector/events").unwrap().to_string(),
785            "https://example.com/vector/events"
786        );
787        assert_eq!(
788            base.append_path("/api/v1/series").unwrap().to_string(),
789            "https://example.com/api/v1/series"
790        );
791        assert_eq!(
792            HttpEndpoint::parse("https://example.com/")
793                .unwrap()
794                .append_path("vector/events")
795                .unwrap()
796                .to_string(),
797            "https://example.com/vector/events"
798        );
799        // The query is carried in the appended path.
800        assert_eq!(
801            base.append_path("/write?db=mydb").unwrap().to_string(),
802            "https://example.com/write?db=mydb"
803        );
804        // The scheme and authority survive appending.
805        let appended = HttpEndpoint::parse("https://user:pass@example.com:8088/base")
806            .unwrap()
807            .append_path("sub/path")
808            .unwrap();
809        assert_eq!(
810            appended.to_string(),
811            "https://user:pass@example.com:8088/base/sub/path"
812        );
813        assert!(matches!(appended.as_uri().scheme_str(), Some("https")));
814        // A non-root base path with a leading-slash appended path must not
815        // produce a double slash.
816        assert_eq!(
817            HttpEndpoint::parse("https://proxy/prefix")
818                .unwrap()
819                .append_path("/api/v1/series")
820                .unwrap()
821                .to_string(),
822            "https://proxy/prefix/api/v1/series"
823        );
824        // Only the single boundary slash is removed; significant leading
825        // slashes in the appended path are preserved (GCS object keys).
826        assert_eq!(
827            HttpEndpoint::parse("https://storage.googleapis.com/bucket/")
828                .unwrap()
829                .append_path("//archive/")
830                .unwrap()
831                .to_string(),
832            "https://storage.googleapis.com/bucket//archive/"
833        );
834    }
835
836    #[test]
837    fn http_endpoint_append_raw_suffix_attaches_without_separator() {
838        let base = HttpEndpoint::parse("https://example.com/v1/projects/p/topics/t").unwrap();
839        assert_eq!(
840            base.append_raw_suffix(":publish").unwrap().to_string(),
841            "https://example.com/v1/projects/p/topics/t:publish"
842        );
843        // An empty suffix returns the endpoint unchanged.
844        assert_eq!(
845            base.append_raw_suffix("").unwrap().to_string(),
846            base.to_string()
847        );
848    }
849}
vector/sinks/util/uri.rs

vector/sinks/util/
uri.rs
