relay_knowledge/net/
http.rs

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

relay_knowledge/net/
http.rs
