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

relay_knowledge/net/
http.rs
