use std::{
    convert::Infallible,
    net::{IpAddr, SocketAddr},
    sync::Arc,
    time::Instant,
};

use bytes::Bytes;
use http::{
    HeaderValue, Method, Request, Response, StatusCode, Uri,
    header::{HOST, PROXY_AUTHENTICATE},
    uri::Authority,
};
use http_body_util::{BodyExt, Empty, Full, combinators::UnsyncBoxBody};
use hyper::{body::Incoming, server::conn::http1, service::service_fn};
use hyper_util::{
    client::legacy::{Client, connect::HttpConnector},
    rt::{TokioExecutor, TokioIo},
};
use opentelemetry::global;
use tokio::{
    net::{TcpListener, TcpStream},
    task::{JoinHandle, JoinSet},
    time::timeout,
};
use tracing::{Instrument, Span, debug, error, field, info, info_span, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;

use crate::{
    BoxError,
    acl::{Acl, Decision, DestinationHost},
    config::Config,
    metrics::Metrics,
    socks5::{self, SocksState},
    telemetry::{HeaderExtractor, HeaderInjector, remove_hop_by_hop_headers},
    tunnel::{self, TunnelProtocol},
};

type ProxyBody = UnsyncBoxBody<Bytes, BoxError>;
type ProxyResponse = Response<ProxyBody>;
type HttpClient = Client<HttpConnector, Incoming>;

#[derive(Clone)]
struct ProxyState {
    client: HttpClient,
    acl: Arc<Acl>,
    metrics: Metrics,
    connect_timeout: std::time::Duration,
    upstream_timeout: std::time::Duration,
}

impl ProxyState {
    fn new(config: &Config) -> Self {
        let mut connector = HttpConnector::new();
        connector.enforce_http(true);
        connector.set_connect_timeout(Some(config.connect_timeout));

        Self {
            client: Client::builder(TokioExecutor::new()).build(connector),
            acl: config.acl.clone(),
            metrics: Metrics::new(global::meter("hermes")),
            connect_timeout: config.connect_timeout,
            upstream_timeout: config.upstream_timeout,
        }
    }
}

pub async fn run(config: Config) -> Result<(), BoxError> {
    let listener = TcpListener::bind(config.listen_addr).await?;
    let state = ProxyState::new(&config);
    let mut socks_task = match config.socks5_listen_addr {
        Some(address) => {
            let listener = TcpListener::bind(address).await?;
            let state = SocksState::new(&config);
            Some(tokio::spawn(socks5::run(listener, state)))
        }
        None => None,
    };
    let mut connections = JoinSet::new();
    let shutdown = tokio::signal::ctrl_c();
    tokio::pin!(shutdown);

    info!(listen.address = %config.listen_addr, "proxy listening");

    loop {
        tokio::select! {
            _ = &mut shutdown => {
                info!("shutdown signal received");
                break;
            }
            accepted = listener.accept() => {
                match accepted {
                    Ok((stream, peer)) => {
                        let state = state.clone();
                        connections.spawn(async move {
                            serve_connection(stream, peer, state).await;
                        });
                    }
                    Err(error) => {
                        error!(error.message = %error, "failed to accept downstream connection");
                    }
                }
            }
            completed = connections.join_next(), if !connections.is_empty() => {
                if let Some(Err(error)) = completed {
                    error!(error.message = %error, "connection task failed");
                }
            }
            completed = wait_for_socks(&mut socks_task) => {
                match completed {
                    Ok(Ok(())) => error!("SOCKS5 listener stopped unexpectedly"),
                    Ok(Err(error)) => error!(error.message = %error, "SOCKS5 listener failed"),
                    Err(error) => error!(error.message = %error, "SOCKS5 listener task failed"),
                }
                break;
            }
        }
    }

    if let Some(task) = socks_task {
        task.abort();
    }
    info!("proxy listener stopped");
    Ok(())
}

async fn wait_for_socks(
    task: &mut Option<JoinHandle<Result<(), BoxError>>>,
) -> Result<Result<(), BoxError>, tokio::task::JoinError> {
    match task {
        Some(task) => task.await,
        None => std::future::pending().await,
    }
}

async fn serve_connection(stream: TcpStream, peer: SocketAddr, state: ProxyState) {
    let peer_ip = peer.ip();
    let service = service_fn(move |request| {
        let state = state.clone();
        async move { Ok::<_, Infallible>(handle_request(request, peer_ip, state).await) }
    });

    let connection = http1::Builder::new()
        .serve_connection(TokioIo::new(stream), service)
        .with_upgrades();

    if let Err(error) = connection.await {
        debug!(
            client.address = %peer,
            error.message = %error,
            "downstream connection ended with an error"
        );
    }
}

async fn handle_request(
    request: Request<Incoming>,
    peer: IpAddr,
    state: ProxyState,
) -> ProxyResponse {
    if request.method() == Method::CONNECT {
        handle_connect(request, peer, state).await
    } else {
        handle_http(request, peer, state).await
    }
}

async fn handle_http(request: Request<Incoming>, peer: IpAddr, state: ProxyState) -> ProxyResponse {
    let method = request.method().to_string();
    let uri = request.uri().clone();
    let parent = global::get_text_map_propagator(|propagator| {
        propagator.extract(&HeaderExtractor(request.headers()))
    });
    let span = info_span!(
        "proxy.request",
        otel.name = %format!("HTTP {method}"),
        otel.kind = "server",
        http.request.method = %method,
        url.scheme = %uri.scheme_str().unwrap_or_default(),
        url.path = %uri.path(),
        server.address = %uri.authority().map(Authority::host).unwrap_or_default(),
        http.response.status_code = field::Empty,
        error.type = field::Empty,
        acl.decision = field::Empty,
        acl.reason = field::Empty,
        acl.rule = field::Empty,
        otel.status_code = field::Empty,
    );
    let _ = span.set_parent(parent);

    async move {
        let _active = state.metrics.request_started();
        let started = Instant::now();
        let result = match authorize_request(&request, peer, &state) {
            Ok(Some(response)) => Ok(response),
            Ok(None) => forward_http(request, &state).await,
            Err(error) => Err(error),
        };
        finish_request(result, &method, started, &state.metrics)
    }
    .instrument(span)
    .await
}

fn authorize_request(
    request: &Request<Incoming>,
    peer: IpAddr,
    state: &ProxyState,
) -> Result<Option<ProxyResponse>, ProxyError> {
    let destination = request.uri().authority().and_then(|authority| {
        if authority.as_str().contains('@') {
            None
        } else {
            DestinationHost::parse(authority.host())
        }
    });
    let outcome = state
        .acl
        .authorize(peer, request.headers(), destination.as_ref());

    let protocol = if request.method() == Method::CONNECT {
        "http_connect"
    } else {
        "http"
    };
    state
        .metrics
        .acl_decision(protocol, outcome.decision.as_str(), outcome.reason);
    Span::current().record("acl.decision", outcome.decision.as_str());
    Span::current().record("acl.reason", outcome.reason);
    if let Some(rule) = &outcome.rule {
        Span::current().record("acl.rule", rule);
    }

    match outcome.decision {
        Decision::Allow => {
            debug!(
                acl.reason = outcome.reason,
                acl.rule = outcome.rule.as_deref().unwrap_or_default(),
                "ACL allowed request"
            );
            Ok(None)
        }
        Decision::Deny => {
            warn!(
                acl.reason = outcome.reason,
                acl.rule = outcome.rule.as_deref().unwrap_or_default(),
                "ACL denied request"
            );
            Ok(Some(error_response(StatusCode::FORBIDDEN)))
        }
        Decision::Challenge => {
            info!(acl.reason = outcome.reason, "ACL challenged request");
            let mut response = error_response(StatusCode::PROXY_AUTHENTICATION_REQUIRED);
            response.headers_mut().insert(
                PROXY_AUTHENTICATE,
                HeaderValue::from_static("Basic realm=\"hermes\", charset=\"UTF-8\""),
            );
            Ok(Some(response))
        }
    }
}

async fn forward_http(
    mut request: Request<Incoming>,
    state: &ProxyState,
) -> Result<ProxyResponse, ProxyError> {
    let authority = validate_http_target(request.uri())?;
    let host = HeaderValue::from_str(authority.as_str())
        .map_err(|_| ProxyError::bad_request("invalid_target", "invalid upstream Host header"))?;

    remove_hop_by_hop_headers(request.headers_mut());
    request.headers_mut().insert(HOST, host);

    let response = send_upstream(request, state).await?;
    let (parts, body) = response.into_parts();
    let mut response = Response::from_parts(parts, box_incoming_body(body));
    remove_hop_by_hop_headers(response.headers_mut());
    Ok(response)
}

async fn send_upstream(
    mut request: Request<Incoming>,
    state: &ProxyState,
) -> Result<Response<Incoming>, ProxyError> {
    let method = request.method().to_string();
    let host = request
        .uri()
        .authority()
        .map(Authority::host)
        .unwrap_or_default()
        .to_owned();
    let port = request.uri().port_u16().unwrap_or(80);
    let span = info_span!(
        "proxy.upstream",
        otel.name = %format!("HTTP {method}"),
        otel.kind = "client",
        http.request.method = %method,
        server.address = %host,
        server.port = port,
        http.response.status_code = field::Empty,
        error.type = field::Empty,
        otel.status_code = field::Empty,
    );

    async move {
        let context = Span::current().context();
        global::get_text_map_propagator(|propagator| {
            propagator.inject_context(&context, &mut HeaderInjector(request.headers_mut()));
        });

        let started = Instant::now();
        let result = match timeout(state.upstream_timeout, state.client.request(request)).await {
            Ok(Ok(response)) => Ok(response),
            Ok(Err(error)) => Err(ProxyError::new(
                StatusCode::BAD_GATEWAY,
                "upstream_request",
                "upstream",
                error.to_string(),
            )),
            Err(_) => Err(ProxyError::new(
                StatusCode::GATEWAY_TIMEOUT,
                "upstream_timeout",
                "upstream",
                "upstream request timed out".to_owned(),
            )),
        };

        let (status, error_type) = match &result {
            Ok(response) => {
                let status = response.status();
                Span::current().record("http.response.status_code", status.as_u16());
                if status.is_server_error() {
                    Span::current().record("otel.status_code", "ERROR");
                }
                (Some(status.as_u16()), None)
            }
            Err(error) => {
                Span::current().record("error.type", error.error_type);
                Span::current().record("otel.status_code", "ERROR");
                (None, Some(error.error_type))
            }
        };
        state
            .metrics
            .upstream_finished(&method, status, error_type, started.elapsed());

        result
    }
    .instrument(span)
    .await
}

async fn handle_connect(
    request: Request<Incoming>,
    peer: IpAddr,
    state: ProxyState,
) -> ProxyResponse {
    let authority = request.uri().authority().cloned();
    let parent = global::get_text_map_propagator(|propagator| {
        propagator.extract(&HeaderExtractor(request.headers()))
    });
    let span = info_span!(
        "proxy.connect",
        otel.name = "CONNECT",
        otel.kind = "server",
        http.request.method = "CONNECT",
        server.address = %authority.as_ref().map(Authority::host).unwrap_or_default(),
        server.port = authority.as_ref().and_then(Authority::port_u16).unwrap_or_default(),
        http.response.status_code = field::Empty,
        error.type = field::Empty,
        tunnel.bytes.client_to_upstream = field::Empty,
        tunnel.bytes.upstream_to_client = field::Empty,
        acl.decision = field::Empty,
        acl.reason = field::Empty,
        acl.rule = field::Empty,
        otel.status_code = field::Empty,
    );
    let _ = span.set_parent(parent);

    async move {
        let _active = state.metrics.request_started();
        let started = Instant::now();
        let result = match authorize_request(&request, peer, &state) {
            Ok(Some(response)) => Ok(response),
            Ok(None) => prepare_connect(request, &state).await,
            Err(error) => Err(error),
        };
        finish_request(result, "CONNECT", started, &state.metrics)
    }
    .instrument(span)
    .await
}

async fn prepare_connect(
    request: Request<Incoming>,
    state: &ProxyState,
) -> Result<ProxyResponse, ProxyError> {
    let authority = validate_connect_target(request.uri())?;
    let host = authority.host().to_owned();
    let port = authority.port_u16().unwrap_or_default();
    let connect_span = info_span!(
        "proxy.upstream_connect",
        otel.name = "TCP CONNECT",
        otel.kind = "client",
        server.address = %host,
        server.port = port,
        error.type = field::Empty,
        otel.status_code = field::Empty,
    );

    let upstream = async {
        let result = match timeout(
            state.connect_timeout,
            TcpStream::connect(authority.as_str()),
        )
        .await
        {
            Ok(Ok(stream)) => Ok(stream),
            Ok(Err(error)) => Err(ProxyError::new(
                StatusCode::BAD_GATEWAY,
                "upstream_connect",
                "upstream",
                error.to_string(),
            )),
            Err(_) => Err(ProxyError::new(
                StatusCode::GATEWAY_TIMEOUT,
                "upstream_timeout",
                "upstream",
                "upstream connection timed out".to_owned(),
            )),
        };

        if let Err(error) = &result {
            Span::current().record("error.type", error.error_type);
            Span::current().record("otel.status_code", "ERROR");
        }
        result
    }
    .instrument(connect_span)
    .await?;

    let tunnel_span = Span::current();
    let metrics = state.metrics.clone();
    tokio::spawn(run_tunnel(request, upstream, metrics).instrument(tunnel_span));

    Ok(Response::new(empty_body()))
}

async fn run_tunnel(request: Request<Incoming>, mut upstream: TcpStream, metrics: Metrics) {
    let upgraded = match hyper::upgrade::on(request).await {
        Ok(upgraded) => upgraded,
        Err(error) => {
            Span::current().record("error.type", "downstream_upgrade");
            Span::current().record("otel.status_code", "ERROR");
            metrics.error("downstream", "downstream_upgrade");
            error!(error.message = %error, "failed to upgrade CONNECT request");
            return;
        }
    };

    let mut downstream = TokioIo::new(upgraded);
    tunnel::relay(
        &mut downstream,
        &mut upstream,
        &metrics,
        TunnelProtocol::HttpConnect,
    )
    .await;
}

fn finish_request(
    result: Result<ProxyResponse, ProxyError>,
    method: &str,
    started: Instant,
    metrics: &Metrics,
) -> ProxyResponse {
    let mut error_type = None;
    let response = match result {
        Ok(response) => response,
        Err(error) => {
            error_type = Some(error.error_type);
            Span::current().record("error.type", error.error_type);
            Span::current().record("otel.status_code", "ERROR");
            metrics.error(error.phase, error.error_type);
            error!(
                error.type = error.error_type,
                proxy.phase = error.phase,
                error.message = %error.message,
                "proxy request failed"
            );
            error_response(error.status)
        }
    };

    let status = response.status();
    Span::current().record("http.response.status_code", status.as_u16());
    if status.is_server_error() {
        Span::current().record("otel.status_code", "ERROR");
    }
    let duration = started.elapsed();
    metrics.request_finished(method, status.as_u16(), error_type, duration);
    info!(
        http.request.method = method,
        http.response.status_code = status.as_u16(),
        request.duration_ms = duration.as_millis(),
        "proxy request completed"
    );
    response
}

fn validate_http_target(uri: &Uri) -> Result<Authority, ProxyError> {
    if uri.scheme_str() != Some("http") {
        return Err(ProxyError::bad_request(
            "unsupported_scheme",
            "only absolute http:// request targets are supported",
        ));
    }

    let authority = uri.authority().ok_or_else(|| {
        ProxyError::bad_request("invalid_target", "absolute request target has no authority")
    })?;
    if authority.as_str().contains('@') {
        return Err(ProxyError::bad_request(
            "invalid_target",
            "userinfo is not allowed in request targets",
        ));
    }

    Ok(authority.clone())
}

fn validate_connect_target(uri: &Uri) -> Result<Authority, ProxyError> {
    if uri.scheme().is_some() || !uri.path().is_empty() {
        return Err(ProxyError::bad_request(
            "invalid_target",
            "CONNECT target must use authority-form",
        ));
    }

    let authority = uri.authority().ok_or_else(|| {
        ProxyError::bad_request("invalid_target", "CONNECT target has no authority")
    })?;
    if authority.port_u16().is_none() || authority.as_str().contains('@') {
        return Err(ProxyError::bad_request(
            "invalid_target",
            "CONNECT target must contain a valid host and explicit port",
        ));
    }

    Ok(authority.clone())
}

fn error_response(status: StatusCode) -> ProxyResponse {
    let mut response = Response::new(full_body(format!(
        "{}\n",
        status.canonical_reason().unwrap_or("Proxy Error")
    )));
    *response.status_mut() = status;
    response
}

fn box_incoming_body(body: Incoming) -> ProxyBody {
    body.map_err(|error| -> BoxError { Box::new(error) })
        .boxed_unsync()
}

fn full_body(value: impl Into<Bytes>) -> ProxyBody {
    Full::new(value.into())
        .map_err(|never| match never {})
        .boxed_unsync()
}

fn empty_body() -> ProxyBody {
    Empty::new().map_err(|never| match never {}).boxed_unsync()
}

#[derive(Debug)]
struct ProxyError {
    status: StatusCode,
    error_type: &'static str,
    phase: &'static str,
    message: String,
}

impl ProxyError {
    fn new(
        status: StatusCode,
        error_type: &'static str,
        phase: &'static str,
        message: String,
    ) -> Self {
        Self {
            status,
            error_type,
            phase,
            message,
        }
    }

    fn bad_request(error_type: &'static str, message: &'static str) -> Self {
        Self::new(
            StatusCode::BAD_REQUEST,
            error_type,
            "request_validation",
            message.to_owned(),
        )
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        sync::{Arc, Mutex},
        time::Duration,
    };

    use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
    use http::header::PROXY_AUTHORIZATION;
    use hyper::client::conn::http1 as client_http1;
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt},
        task::JoinHandle,
    };

    use super::*;
    use crate::config::{AclConfig, Action, RuleConfig, TelemetryConfig, UserConfig};

    #[test]
    fn validates_absolute_http_targets() {
        let uri = Uri::from_static("http://example.com/path?q=1");
        assert_eq!(validate_http_target(&uri).unwrap(), "example.com");

        assert!(validate_http_target(&Uri::from_static("/path")).is_err());
        assert!(validate_http_target(&Uri::from_static("https://example.com/")).is_err());
        assert!(validate_http_target(&Uri::from_static("http://user@example.com/")).is_err());
    }

    #[test]
    fn validates_connect_authority() {
        let uri = Uri::from_static("example.com:443");
        assert_eq!(validate_connect_target(&uri).unwrap(), "example.com:443");

        assert!(validate_connect_target(&Uri::from_static("example.com")).is_err());
        assert!(validate_connect_target(&Uri::from_static("http://example.com:443")).is_err());
    }

    #[tokio::test]
    async fn forwards_http_and_rewrites_connection_headers() {
        #[derive(Default)]
        struct Observed {
            target: String,
            host: String,
            forwarded_proxy_authorization: bool,
        }

        let observed = Arc::new(Mutex::new(Observed::default()));
        let upstream_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let upstream_addr = upstream_listener.local_addr().unwrap();
        let upstream_observed = observed.clone();
        let upstream_task = tokio::spawn(async move {
            let (stream, _) = upstream_listener.accept().await.unwrap();
            let service = service_fn(move |request: Request<Incoming>| {
                let mut observed = upstream_observed.lock().unwrap();
                observed.target = request.uri().to_string();
                observed.host = request
                    .headers()
                    .get(HOST)
                    .and_then(|value| value.to_str().ok())
                    .unwrap_or_default()
                    .to_owned();
                observed.forwarded_proxy_authorization =
                    request.headers().contains_key("proxy-authorization");
                drop(observed);

                async move {
                    Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"proxied"))))
                }
            });
            http1::Builder::new()
                .serve_connection(TokioIo::new(stream), service)
                .await
                .unwrap();
        });

        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(basic_acl()).await;
        let (mut sender, connection_task) = connect_client(proxy_addr).await;
        let request = Request::builder()
            .method(Method::GET)
            .uri(format!(
                "http://localhost:{}/hello?q=1",
                upstream_addr.port()
            ))
            .header(HOST, "wrong.invalid")
            .header(PROXY_AUTHORIZATION, basic_authorization())
            .body(Full::new(Bytes::new()))
            .unwrap();

        let response = sender.send_request(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = response.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(body, Bytes::from_static(b"proxied"));

        upstream_task.abort();
        let observed = observed.lock().unwrap();
        assert_eq!(observed.target, "/hello?q=1");
        assert_eq!(observed.host, format!("localhost:{}", upstream_addr.port()));
        assert!(!observed.forwarded_proxy_authorization);
        drop(observed);

        proxy_task.abort();
        connection_task.abort();
    }

    #[tokio::test]
    async fn relays_bytes_through_connect_tunnel() {
        let echo_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let echo_addr = echo_listener.local_addr().unwrap();
        let echo_task = tokio::spawn(async move {
            let (mut stream, _) = echo_listener.accept().await.unwrap();
            let mut request = [0_u8; 4];
            stream.read_exact(&mut request).await.unwrap();
            assert_eq!(&request, b"ping");
            stream.write_all(b"pong").await.unwrap();
            stream.shutdown().await.unwrap();
        });

        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(localhost_acl()).await;
        let (mut sender, connection_task) = connect_client(proxy_addr).await;
        let request = Request::builder()
            .method(Method::CONNECT)
            .uri(format!("localhost:{}", echo_addr.port()))
            .body(Full::new(Bytes::new()))
            .unwrap();

        let response = sender.send_request(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);

        let upgraded = hyper::upgrade::on(response).await.unwrap();
        let mut tunnel = TokioIo::new(upgraded);
        tunnel.write_all(b"ping").await.unwrap();
        let mut response = [0_u8; 4];
        tunnel.read_exact(&mut response).await.unwrap();
        assert_eq!(&response, b"pong");

        echo_task.await.unwrap();
        proxy_task.abort();
        connection_task.abort();
    }

    #[tokio::test]
    async fn challenges_unauthenticated_http_and_connect_requests() {
        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(basic_acl()).await;
        let (mut sender, connection_task) = connect_client(proxy_addr).await;

        for request in [
            Request::builder()
                .method(Method::GET)
                .uri("http://192.0.2.1/")
                .body(Full::new(Bytes::new()))
                .unwrap(),
            Request::builder()
                .method(Method::CONNECT)
                .uri("192.0.2.1:443")
                .body(Full::new(Bytes::new()))
                .unwrap(),
        ] {
            let response = sender.send_request(request).await.unwrap();
            assert_eq!(response.status(), StatusCode::PROXY_AUTHENTICATION_REQUIRED);
            assert_eq!(
                response.headers()[PROXY_AUTHENTICATE],
                "Basic realm=\"hermes\", charset=\"UTF-8\""
            );
            response.into_body().collect().await.unwrap();
        }

        proxy_task.abort();
        connection_task.abort();
    }

    #[tokio::test]
    async fn denies_blacklisted_http_domain_before_upstream_connection() {
        let acl = Acl::from_config(AclConfig {
            default_action: Action::Allow,
            users: BTreeMap::new(),
            rules: vec![RuleConfig {
                name: "blocked-domain".to_owned(),
                action: Action::Deny,
                users: vec![],
                networks: vec![],
                domains: vec!["blocked.example".to_owned()],
            }],
        })
        .unwrap();
        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(acl).await;
        let (mut sender, connection_task) = connect_client(proxy_addr).await;
        let request = Request::builder()
            .method(Method::GET)
            .uri("http://blocked.example/")
            .body(Full::new(Bytes::new()))
            .unwrap();

        let response = sender.send_request(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        response.into_body().collect().await.unwrap();

        proxy_task.abort();
        connection_task.abort();
    }

    #[tokio::test]
    async fn domain_whitelist_rejects_ip_literal_connect() {
        let echo_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let echo_addr = echo_listener.local_addr().unwrap();
        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(localhost_acl()).await;
        let (mut sender, connection_task) = connect_client(proxy_addr).await;
        let request = Request::builder()
            .method(Method::CONNECT)
            .uri(echo_addr.to_string())
            .body(Full::new(Bytes::new()))
            .unwrap();

        let response = sender.send_request(request).await.unwrap();
        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        response.into_body().collect().await.unwrap();

        proxy_task.abort();
        connection_task.abort();
    }

    fn basic_authorization() -> String {
        format!(
            "Basic {}",
            BASE64_STANDARD.encode("alice:correct horse battery staple")
        )
    }

    fn basic_acl() -> Acl {
        Acl::from_config(AclConfig {
            default_action: Action::Deny,
            users: BTreeMap::from([(
                "alice".to_owned(),
                UserConfig {
                    password_hash: crate::acl::hash_password(b"correct horse battery staple")
                        .unwrap(),
                },
            )]),
            rules: vec![RuleConfig {
                name: "alice".to_owned(),
                action: Action::Allow,
                users: vec!["alice".to_owned()],
                networks: vec![],
                domains: vec!["localhost".to_owned()],
            }],
        })
        .unwrap()
    }

    fn localhost_acl() -> Acl {
        Acl::from_config(AclConfig {
            default_action: Action::Deny,
            users: BTreeMap::new(),
            rules: vec![RuleConfig {
                name: "localhost".to_owned(),
                action: Action::Allow,
                users: vec![],
                networks: vec!["127.0.0.1/32".to_owned(), "::1/128".to_owned()],
                domains: vec!["localhost".to_owned()],
            }],
        })
        .unwrap()
    }

    async fn spawn_proxy_with_acl(acl: Acl) -> (SocketAddr, JoinHandle<()>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let state = ProxyState::new(&Config {
            listen_addr: address,
            socks5_listen_addr: None,
            connect_timeout: Duration::from_secs(2),
            socks5_handshake_timeout: Duration::from_secs(2),
            upstream_timeout: Duration::from_secs(2),
            telemetry: TelemetryConfig::default(),
            acl: Arc::new(acl),
        });
        let task = tokio::spawn(async move {
            while let Ok((stream, peer)) = listener.accept().await {
                let state = state.clone();
                tokio::spawn(serve_connection(stream, peer, state));
            }
        });
        (address, task)
    }

    async fn connect_client(
        address: SocketAddr,
    ) -> (
        client_http1::SendRequest<Full<Bytes>>,
        JoinHandle<Result<(), hyper::Error>>,
    ) {
        let stream = TcpStream::connect(address).await.unwrap();
        let (sender, connection) = client_http1::handshake(TokioIo::new(stream)).await.unwrap();
        let connection_task = tokio::spawn(connection.with_upgrades());
        (sender, connection_task)
    }
}