relay_knowledge/net/
http.rs

1//! HTTP runtime policy owned by the network boundary.
2//!
3//! This module intentionally models configuration and validation only. Future
4//! HTTP server/client adapters must use a mature async runtime underneath this
5//! boundary and keep QoS admission in `net::qos`.
6
7use std::{
8    convert::Infallible,
9    error::Error,
10    fmt,
11    future::{Future, IntoFuture, Ready, ready},
12    io,
13    net::IpAddr,
14    pin::Pin,
15    sync::{
16        Arc,
17        atomic::{AtomicU64, Ordering},
18    },
19    task::{Context, Poll},
20    time::Duration,
21};
22
23use axum::{
24    Router,
25    extract::Request,
26    http::StatusCode,
27    response::{IntoResponse, Response},
28    serve::{IncomingStream, Listener},
29};
30use serde_json::Value;
31use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
32use tower::Service;
33
34use crate::{
35    env::NetworkEnvOverrides,
36    net::qos::{QosPermit, QosPolicy, QosRuntime, RejectReason},
37};
38
39pub const DEFAULT_HTTP_BIND: &str = "127.0.0.1:8791";
40pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
41pub const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
42pub const DEFAULT_MAX_BODY_BYTES: u64 = 1_048_576;
43pub const DEFAULT_SSL_VERIFY: bool = true;
44
45/// Event-driven HTTP configuration for future inbound and outbound adapters.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct HttpConfig {
48    pub bind_address: HttpBindAddress,
49    pub request_timeout: Duration,
50    pub graceful_shutdown_timeout: Duration,
51    pub max_request_body_bytes: u64,
52    pub proxy: HttpProxyConfig,
53}
54
55/// Validated HTTP bind address in `host:port` form.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct HttpBindAddress {
58    value: String,
59    port: u16,
60}
61
62impl HttpBindAddress {
63    /// Parses a host or IP literal with an explicit non-zero port.
64    pub fn parse(value: &str) -> Result<Self, HttpConfigError> {
65        let trimmed = value.trim();
66        if trimmed.is_empty() {
67            return Err(HttpConfigError::InvalidBindAddress {
68                value: value.to_owned(),
69            });
70        }
71
72        if let Ok(socket_addr) = trimmed.parse::<std::net::SocketAddr>() {
73            return Self::from_parts(trimmed.to_owned(), socket_addr.port());
74        }
75
76        let Some((host, port)) = trimmed.rsplit_once(':') else {
77            return Err(HttpConfigError::InvalidBindAddress {
78                value: value.to_owned(),
79            });
80        };
81
82        if host.is_empty() || host.contains('/') || host.contains(char::is_whitespace) {
83            return Err(HttpConfigError::InvalidBindAddress {
84                value: value.to_owned(),
85            });
86        }
87
88        let port = port
89            .parse::<u16>()
90            .map_err(|_| HttpConfigError::InvalidBindAddress {
91                value: value.to_owned(),
92            })?;
93
94        Self::from_parts(trimmed.to_owned(), port)
95    }
96
97    /// Returns the explicit TCP port.
98    pub const fn port(&self) -> u16 {
99        self.port
100    }
101
102    fn from_parts(value: String, port: u16) -> Result<Self, HttpConfigError> {
103        if port == 0 {
104            return Err(HttpConfigError::EphemeralPort);
105        }
106
107        Ok(Self { value, port })
108    }
109}
110
111impl fmt::Display for HttpBindAddress {
112    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
113        formatter.write_str(&self.value)
114    }
115}
116
117/// Returns whether a listener may accept non-local clients under the access policy.
118pub fn remote_clients_allowed(config: &HttpConfig, allow_remote_clients: bool) -> bool {
119    allow_remote_clients || is_local_bind(&config.bind_address.to_string())
120}
121
122/// Outbound HTTP proxy and TLS verification policy.
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct HttpProxyConfig {
125    pub proxy: Option<String>,
126    pub no_proxy_rules: Vec<String>,
127    pub ssl_verify: bool,
128}
129
130/// Builds an async outbound JSON client from validated network policy.
131pub fn outbound_json_client(config: &HttpConfig) -> Result<reqwest::Client, OutboundClientError> {
132    outbound_json_client_with_policy(config, None, None)
133}
134
135/// Builds an async outbound JSON client with request-scoped transport policy.
136pub fn outbound_json_client_with_policy(
137    config: &HttpConfig,
138    ssl_verify: Option<bool>,
139    connect_timeout: Option<Duration>,
140) -> Result<reqwest::Client, OutboundClientError> {
141    let mut builder = reqwest::Client::builder()
142        .timeout(config.request_timeout)
143        .danger_accept_invalid_certs(!ssl_verify.unwrap_or(config.proxy.ssl_verify));
144    if let Some(timeout) = connect_timeout {
145        builder = builder.connect_timeout(timeout);
146    }
147    if let Some(proxy_url) = &config.proxy.proxy {
148        let no_proxy = reqwest::NoProxy::from_string(&config.proxy.no_proxy_rules.join(","));
149        let proxy = reqwest::Proxy::all(proxy_url)
150            .map_err(|error| OutboundClientError {
151                message: error.to_string(),
152            })?
153            .no_proxy(no_proxy);
154        builder = builder.proxy(proxy);
155    }
156
157    builder.build().map_err(|error| OutboundClientError {
158        message: error.to_string(),
159    })
160}
161
162#[derive(Debug, Clone, PartialEq, Eq)]
163pub struct OutboundClientError {
164    pub message: String,
165}
166impl fmt::Display for OutboundClientError {
167    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
168        self.message.fmt(formatter)
169    }
170}
171
172impl Error for OutboundClientError {}
173
174impl HttpProxyConfig {
175    /// Validates proxy URL shape and no-proxy entries without exposing credentials.
176    pub fn new(
177        proxy: Option<String>,
178        no_proxy_rules: Vec<String>,
179        ssl_verify: bool,
180    ) -> Result<Self, HttpConfigError> {
181        if let Some(proxy_url) = proxy.as_deref() {
182            validate_proxy_url(proxy_url)?;
183        }
184
185        for rule in &no_proxy_rules {
186            if rule.trim().is_empty() {
187                return Err(HttpConfigError::EmptyNoProxyRule);
188            }
189        }
190
191        Ok(Self {
192            proxy,
193            no_proxy_rules,
194            ssl_verify,
195        })
196    }
197
198    /// Applies proxy, no-proxy, and TLS verification environment overrides.
199    pub fn from_overrides(overrides: &NetworkEnvOverrides) -> Result<Self, HttpConfigError> {
200        Self::new(
201            overrides.proxy.clone(),
202            parse_no_proxy_rules(overrides.no_proxy.as_deref())?,
203            overrides.ssl_verify.unwrap_or(DEFAULT_SSL_VERIFY),
204        )
205    }
206
207    /// Returns whether outbound HTTP should use a proxy.
208    pub fn is_proxy_configured(&self) -> bool {
209        self.proxy.is_some()
210    }
211}
212
213impl HttpConfig {
214    /// Builds HTTP config while enforcing bounded request and shutdown behavior.
215    pub fn new(
216        bind_address: HttpBindAddress,
217        request_timeout: Duration,
218        graceful_shutdown_timeout: Duration,
219        max_request_body_bytes: u64,
220        proxy: HttpProxyConfig,
221    ) -> Result<Self, HttpConfigError> {
222        if request_timeout.is_zero() {
223            return Err(HttpConfigError::ZeroDuration {
224                field: "request_timeout",
225            });
226        }
227
228        if graceful_shutdown_timeout.is_zero() {
229            return Err(HttpConfigError::ZeroDuration {
230                field: "graceful_shutdown_timeout",
231            });
232        }
233
234        if max_request_body_bytes == 0 {
235            return Err(HttpConfigError::ZeroMaxBodyBytes);
236        }
237
238        Ok(Self {
239            bind_address,
240            request_timeout,
241            graceful_shutdown_timeout,
242            max_request_body_bytes,
243            proxy,
244        })
245    }
246
247    /// Applies environment overrides to the default local HTTP policy.
248    pub fn from_overrides(overrides: &NetworkEnvOverrides) -> Result<Self, HttpConfigError> {
249        let bind_value = overrides.http_bind.as_deref().unwrap_or(DEFAULT_HTTP_BIND);
250        let bind_address = HttpBindAddress::parse(bind_value)?;
251        let request_timeout = overrides
252            .http_request_timeout_ms
253            .map(Duration::from_millis)
254            .unwrap_or(DEFAULT_REQUEST_TIMEOUT);
255        let shutdown_timeout = overrides
256            .http_shutdown_timeout_ms
257            .map(Duration::from_millis)
258            .unwrap_or(DEFAULT_SHUTDOWN_TIMEOUT);
259        let max_body_bytes = overrides
260            .http_max_body_bytes
261            .unwrap_or(DEFAULT_MAX_BODY_BYTES);
262        let proxy = HttpProxyConfig::from_overrides(overrides)?;
263
264        Self::new(
265            bind_address,
266            request_timeout,
267            shutdown_timeout,
268            max_body_bytes,
269            proxy,
270        )
271    }
272}
273
274/// HTTP configuration validation error.
275#[derive(Debug, Clone, PartialEq, Eq)]
276pub enum HttpConfigError {
277    InvalidBindAddress { value: String },
278    EphemeralPort,
279    ZeroDuration { field: &'static str },
280    ZeroMaxBodyBytes,
281    InvalidProxyUrl,
282    EmptyNoProxyRule,
283}
284
285impl fmt::Display for HttpConfigError {
286    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
287        match self {
288            Self::InvalidBindAddress { value } => {
289                write!(formatter, "bind address '{value}' is not host:port")
290            }
291            Self::EphemeralPort => write!(formatter, "bind address must use an explicit port"),
292            Self::ZeroDuration { field } => write!(formatter, "{field} must be greater than zero"),
293            Self::ZeroMaxBodyBytes => write!(
294                formatter,
295                "max request body bytes must be greater than zero"
296            ),
297            Self::InvalidProxyUrl => write!(
298                formatter,
299                "proxy must use http:// or https:// and include a host"
300            ),
301            Self::EmptyNoProxyRule => write!(formatter, "no-proxy entries must not be empty"),
302        }
303    }
304}
305
306impl Error for HttpConfigError {}
307
308/// Error raised while serving an event-driven HTTP adapter.
309#[derive(Debug)]
310pub enum HttpServeError {
311    Bind(std::io::Error),
312    Serve(std::io::Error),
313    ShutdownTimeout,
314}
315
316impl fmt::Display for HttpServeError {
317    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
318        match self {
319            Self::Bind(error) => write!(formatter, "failed to bind HTTP listener: {error}"),
320            Self::Serve(error) => write!(formatter, "HTTP server failed: {error}"),
321            Self::ShutdownTimeout => write!(formatter, "HTTP graceful shutdown timed out"),
322        }
323    }
324}
325
326impl Error for HttpServeError {}
327
328/// Error raised by bounded outbound JSON HTTP calls.
329#[derive(Debug)]
330pub enum HttpClientError {
331    InvalidUrl(String),
332    Io(io::Error),
333    Timeout,
334    InvalidResponse,
335    ResponseStatus(u16),
336    ResponseJson(serde_json::Error),
337}
338
339impl fmt::Display for HttpClientError {
340    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
341        match self {
342            Self::InvalidUrl(value) => write!(formatter, "invalid HTTP worker URL: {value}"),
343            Self::Io(error) => write!(formatter, "HTTP worker request failed: {error}"),
344            Self::Timeout => write!(formatter, "HTTP worker request timed out"),
345            Self::InvalidResponse => write!(formatter, "HTTP worker returned invalid response"),
346            Self::ResponseStatus(status) => {
347                write!(formatter, "HTTP worker returned status {status}")
348            }
349            Self::ResponseJson(error) => {
350                write!(formatter, "HTTP worker returned invalid JSON: {error}")
351            }
352        }
353    }
354}
355
356impl Error for HttpClientError {}
357
358/// Posts a JSON payload through the network boundary using the configured timeout.
359pub async fn post_json(
360    config: &HttpConfig,
361    url: &str,
362    payload: &Value,
363) -> Result<Value, HttpClientError> {
364    let request = JsonHttpRequest::parse(url)?;
365    let body = serde_json::to_vec(payload).map_err(HttpClientError::ResponseJson)?;
366    let response = tokio::time::timeout(config.request_timeout, send_json_request(request, body))
367        .await
368        .map_err(|_| HttpClientError::Timeout)??;
369
370    serde_json::from_slice(&response).map_err(HttpClientError::ResponseJson)
371}
372
373struct JsonHttpRequest {
374    host: String,
375    port: u16,
376    path: String,
377}
378
379impl JsonHttpRequest {
380    fn parse(value: &str) -> Result<Self, HttpClientError> {
381        let remainder = value
382            .strip_prefix("http://")
383            .ok_or_else(|| HttpClientError::InvalidUrl(value.to_owned()))?;
384        let (authority, path) = remainder
385            .split_once('/')
386            .map_or((remainder, "/"), |(authority, path)| {
387                (authority, path.trim_start_matches('/'))
388            });
389        if authority.is_empty() {
390            return Err(HttpClientError::InvalidUrl(value.to_owned()));
391        }
392        let (host, port) = authority
393            .rsplit_once(':')
394            .map(|(host, port)| {
395                let parsed_port = port
396                    .parse::<u16>()
397                    .map_err(|_| HttpClientError::InvalidUrl(value.to_owned()))?;
398                Ok((host.to_owned(), parsed_port))
399            })
400            .unwrap_or_else(|| Ok((authority.to_owned(), 80)))?;
401        if host.is_empty() || port == 0 {
402            return Err(HttpClientError::InvalidUrl(value.to_owned()));
403        }
404        let path = if path.is_empty() {
405            "/".to_owned()
406        } else {
407            format!("/{path}")
408        };
409
410        Ok(Self { host, port, path })
411    }
412}
413
414async fn send_json_request(
415    request: JsonHttpRequest,
416    body: Vec<u8>,
417) -> Result<Vec<u8>, HttpClientError> {
418    let mut stream = tokio::net::TcpStream::connect((request.host.as_str(), request.port))
419        .await
420        .map_err(HttpClientError::Io)?;
421    let head = format!(
422        "POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nAccept: application/json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n",
423        request.path,
424        request.host,
425        body.len()
426    );
427    stream
428        .write_all(head.as_bytes())
429        .await
430        .map_err(HttpClientError::Io)?;
431    stream.write_all(&body).await.map_err(HttpClientError::Io)?;
432    stream.shutdown().await.map_err(HttpClientError::Io)?;
433    let mut response = Vec::new();
434    stream
435        .read_to_end(&mut response)
436        .await
437        .map_err(HttpClientError::Io)?;
438    parse_http_response(response)
439}
440
441fn parse_http_response(response: Vec<u8>) -> Result<Vec<u8>, HttpClientError> {
442    let Some(header_end) = response.windows(4).position(|window| window == b"\r\n\r\n") else {
443        return Err(HttpClientError::InvalidResponse);
444    };
445    let headers = std::str::from_utf8(&response[..header_end])
446        .map_err(|_| HttpClientError::InvalidResponse)?;
447    let status = headers
448        .lines()
449        .next()
450        .and_then(|line| line.split_whitespace().nth(1))
451        .and_then(|value| value.parse::<u16>().ok())
452        .ok_or(HttpClientError::InvalidResponse)?;
453    if !(200..300).contains(&status) {
454        return Err(HttpClientError::ResponseStatus(status));
455    }
456
457    Ok(response[header_end + 4..].to_vec())
458}
459
460/// Stable identifier assigned to an accepted HTTP connection.
461#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
462pub struct HttpConnectionId(u64);
463
464impl HttpConnectionId {
465    /// Returns the numeric connection identifier for request correlation.
466    pub const fn get(self) -> u64 {
467        self.0
468    }
469}
470
471/// Starts an async HTTP server with graceful shutdown under the network boundary.
472pub async fn serve_router(
473    router: Router,
474    config: HttpConfig,
475    shutdown: impl Future<Output = ()> + Send + 'static,
476) -> Result<(), HttpServeError> {
477    let listener = tokio::net::TcpListener::bind(config.bind_address.to_string())
478        .await
479        .map_err(HttpServeError::Bind)?;
480
481    serve_listener(listener, router, config, shutdown).await
482}
483
484/// Starts an async HTTP server whose accepted connections consume QoS permits.
485pub async fn serve_router_with_qos(
486    router: Router,
487    config: HttpConfig,
488    qos: QosRuntime,
489    policy: QosPolicy,
490    shutdown: impl Future<Output = ()> + Send + 'static,
491) -> Result<(), HttpServeError> {
492    let listener = tokio::net::TcpListener::bind(config.bind_address.to_string())
493        .await
494        .map_err(HttpServeError::Bind)?;
495    let listener = QosTcpListener::new(listener, qos, policy);
496
497    serve_listener(listener, router, config, shutdown).await
498}
499
500async fn serve_listener<L>(
501    listener: L,
502    router: Router,
503    config: HttpConfig,
504    shutdown: impl Future<Output = ()> + Send + 'static,
505) -> Result<(), HttpServeError>
506where
507    L: Listener,
508    L::Addr: fmt::Debug,
509{
510    let (shutdown_started, mut shutdown_observed) = tokio::sync::watch::channel(false);
511    let graceful_shutdown = async move {
512        shutdown.await;
513        let _ = shutdown_started.send(true);
514    };
515    let server = axum::serve(
516        listener,
517        HttpMakeService::new(router, config.request_timeout),
518    )
519    .with_graceful_shutdown(graceful_shutdown)
520    .into_future();
521
522    tokio::pin!(server);
523    tokio::select! {
524        result = &mut server => result.map_err(HttpServeError::Serve),
525        changed = shutdown_observed.changed() => {
526            let _ = changed;
527            match tokio::time::timeout(config.graceful_shutdown_timeout, &mut server).await {
528                Ok(result) => result.map_err(HttpServeError::Serve),
529                Err(_) => Err(HttpServeError::ShutdownTimeout),
530            }
531        }
532    }
533}
534
535struct HttpMakeService {
536    router: Router,
537    request_timeout: Duration,
538    next_connection_id: Arc<AtomicU64>,
539}
540
541impl HttpMakeService {
542    fn new(router: Router, request_timeout: Duration) -> Self {
543        Self {
544            router,
545            request_timeout,
546            next_connection_id: Arc::new(AtomicU64::new(1)),
547        }
548    }
549}
550
551impl<'a, L> Service<IncomingStream<'a, L>> for HttpMakeService
552where
553    L: Listener,
554{
555    type Response = HttpConnectionService<Router>;
556    type Error = Infallible;
557    type Future = Ready<Result<Self::Response, Self::Error>>;
558
559    fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
560        Poll::Ready(Ok(()))
561    }
562
563    fn call(&mut self, _target: IncomingStream<'a, L>) -> Self::Future {
564        let connection_id =
565            HttpConnectionId(self.next_connection_id.fetch_add(1, Ordering::Relaxed));
566        ready(Ok(HttpConnectionService::new(
567            self.router.clone(),
568            connection_id,
569            self.request_timeout,
570        )))
571    }
572}
573
574struct HttpConnectionService<S> {
575    inner: S,
576    connection_id: HttpConnectionId,
577    request_timeout: Duration,
578}
579
580impl<S> HttpConnectionService<S> {
581    fn new(inner: S, connection_id: HttpConnectionId, request_timeout: Duration) -> Self {
582        Self {
583            inner,
584            connection_id,
585            request_timeout,
586        }
587    }
588}
589
590impl<S> Clone for HttpConnectionService<S>
591where
592    S: Clone,
593{
594    fn clone(&self) -> Self {
595        Self {
596            inner: self.inner.clone(),
597            connection_id: self.connection_id,
598            request_timeout: self.request_timeout,
599        }
600    }
601}
602
603impl<S> Service<Request> for HttpConnectionService<S>
604where
605    S: Service<Request, Response = Response, Error = Infallible> + Send,
606    S::Future: Send + 'static,
607{
608    type Response = Response;
609    type Error = Infallible;
610    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
611
612    fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
613        self.inner.poll_ready(context)
614    }
615
616    fn call(&mut self, mut request: Request) -> Self::Future {
617        request.extensions_mut().insert(self.connection_id);
618        let future = self.inner.call(request);
619        let request_timeout = self.request_timeout;
620        Box::pin(async move {
621            match tokio::time::timeout(request_timeout, future).await {
622                Ok(result) => result,
623                Err(_) => Ok((StatusCode::REQUEST_TIMEOUT, "request timed out").into_response()),
624            }
625        })
626    }
627}
628
629struct QosTcpListener {
630    inner: tokio::net::TcpListener,
631    qos: QosRuntime,
632    policy: QosPolicy,
633}
634
635impl QosTcpListener {
636    fn new(inner: tokio::net::TcpListener, qos: QosRuntime, policy: QosPolicy) -> Self {
637        Self { inner, qos, policy }
638    }
639}
640
641impl Listener for QosTcpListener {
642    type Io = QosTcpStream;
643    type Addr = std::net::SocketAddr;
644
645    async fn accept(&mut self) -> (Self::Io, Self::Addr) {
646        loop {
647            match self.inner.accept().await {
648                Ok((stream, address)) => match self.qos.admit_connection(&self.policy) {
649                    Ok(permit) => {
650                        return (
651                            QosTcpStream {
652                                inner: stream,
653                                _permit: permit,
654                            },
655                            address,
656                        );
657                    }
658                    Err(RejectReason::ConnectionBudgetExceeded) => drop(stream),
659                    Err(_) => drop(stream),
660                },
661                Err(_) => tokio::time::sleep(Duration::from_secs(1)).await,
662            }
663        }
664    }
665
666    fn local_addr(&self) -> io::Result<Self::Addr> {
667        self.inner.local_addr()
668    }
669}
670
671struct QosTcpStream {
672    inner: tokio::net::TcpStream,
673    _permit: QosPermit,
674}
675
676impl AsyncRead for QosTcpStream {
677    fn poll_read(
678        mut self: Pin<&mut Self>,
679        context: &mut Context<'_>,
680        buffer: &mut ReadBuf<'_>,
681    ) -> Poll<io::Result<()>> {
682        Pin::new(&mut self.inner).poll_read(context, buffer)
683    }
684}
685
686impl AsyncWrite for QosTcpStream {
687    fn poll_write(
688        mut self: Pin<&mut Self>,
689        context: &mut Context<'_>,
690        buffer: &[u8],
691    ) -> Poll<io::Result<usize>> {
692        Pin::new(&mut self.inner).poll_write(context, buffer)
693    }
694
695    fn poll_flush(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
696        Pin::new(&mut self.inner).poll_flush(context)
697    }
698
699    fn poll_shutdown(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
700        Pin::new(&mut self.inner).poll_shutdown(context)
701    }
702}
703
704fn validate_proxy_url(value: &str) -> Result<(), HttpConfigError> {
705    let Some((scheme, remainder)) = value.split_once("://") else {
706        return Err(HttpConfigError::InvalidProxyUrl);
707    };
708
709    if !matches!(scheme, "http" | "https") {
710        return Err(HttpConfigError::InvalidProxyUrl);
711    }
712
713    let authority = remainder.split('/').next().unwrap_or_default();
714    if authority.is_empty() || authority.starts_with('@') {
715        return Err(HttpConfigError::InvalidProxyUrl);
716    }
717    let host_port = authority
718        .rsplit_once('@')
719        .map_or(authority, |(_, host)| host);
720    let host = if let Some(remainder) = host_port.strip_prefix('[') {
721        remainder.split_once(']').map_or("", |(host, _)| host)
722    } else {
723        host_port.split(':').next().unwrap_or_default()
724    };
725    if host.is_empty() {
726        return Err(HttpConfigError::InvalidProxyUrl);
727    }
728
729    Ok(())
730}
731
732fn parse_no_proxy_rules(value: Option<&str>) -> Result<Vec<String>, HttpConfigError> {
733    value
734        .map(|rules| {
735            rules
736                .split(',')
737                .map(str::trim)
738                .map(|rule| {
739                    if rule.is_empty() {
740                        Err(HttpConfigError::EmptyNoProxyRule)
741                    } else {
742                        Ok(rule.to_owned())
743                    }
744                })
745                .collect()
746        })
747        .unwrap_or_else(|| Ok(Vec::new()))
748}
749
750fn is_local_bind(bind: &str) -> bool {
751    is_loopback_host(authority_host(bind))
752}
753
754fn authority_host(authority: &str) -> &str {
755    if let Some(remainder) = authority.strip_prefix('[') {
756        return remainder
757            .find(']')
758            .map_or(authority, |index| &remainder[..index]);
759    }
760
761    authority
762        .rsplit_once(':')
763        .map_or(authority, |(host, _)| host)
764}
765
766fn is_loopback_host(host: &str) -> bool {
767    host.eq_ignore_ascii_case("localhost")
768        || host
769            .parse::<IpAddr>()
770            .is_ok_and(|address| address.is_loopback())
771}
772
773#[cfg(test)]
774#[path = "http_tests.rs"]
775mod tests;
relay_knowledge/net/http.rs

relay_knowledge/net/
http.rs
