rama 0.3.0

//! This example is an adapted MITM (http) proxy which is mostly here to demonstrate
//! the HAR Export Layer in action. It can be used for clients, proxies and even servers,
//! to provide HAR export functionality for diagnostic and support purposes.
//!
//! As with most other examples, it is not meant to show a full production-ready setup,
//! but it is purely here to demonstrate a specific feature, HAR Exporting for this example.
//!
//! # Run the example
//!
//! ```sh
//! cargo run --example http_record_har --features=http-full,boring
//! ```
//!
//! ## Expected output
//!
//! The server will start and listen on `:62040`. You can use `curl` to interact with the service:
//!
//! ```sh
//! curl -v -x http://127.0.0.1:62040 --proxy-user 'john:secret' http://www.example.com/
//! curl -k -v -x http://127.0.0.1:62040 --proxy-user 'john:secret' https://www.example.com/
//! ```
//!
//! You can toggle the HAR Recording on and off using:
//!
//! ```sh
//! curl -v -x http://127.0.0.1:62040 --proxy-user 'john:secret' -XPOST http://har.toggle.internal/switch
//! ```
//!
//! This example injects the path of the file that a http request
//! is recorded to. Do not do this in production kids.
//!
//! Once a recording is finished you can import or replay the HAR file in
//! a tool for analysis. For example dev tools of a browser or some special-purpose
//! HAR Analyzer tool.

#![expect(
    clippy::unwrap_used,
    clippy::expect_used,
    reason = "example/test/bench: panic-on-error and print-for-output are the standard patterns for demos and harnesses"
)]

use rama::{
    Layer, Service,
    error::{BoxError, ErrorContext},
    extensions::{Extension, ExtensionsRef},
    http::{
        BodyLimitLayer, HeaderValue, Request, Response, StatusCode,
        client::EasyHttpWebClient,
        layer::{
            compression::CompressionLayer,
            decompression::DecompressionLayer,
            har::{
                self,
                layer::HARExportLayer,
                recorder::{FileRecorder, HarFilePath, Recorder},
            },
            map_response_body::MapResponseBodyLayer,
            proxy_auth::ProxyAuthLayer,
            remove_header::{RemoveRequestHeaderLayer, RemoveResponseHeaderLayer},
            required_header::AddRequiredRequestHeadersLayer,
            trace::TraceLayer,
            upgrade::{DefaultHttpProxyConnectReplyService, UpgradeLayer, Upgraded},
        },
        matcher::{DomainMatcher, MethodMatcher},
        server::HttpServer,
        service::web::{WebService, response::IntoResponse},
    },
    layer::{AddInputExtensionLayer, ConsumeErrLayer, HijackLayer},
    net::user::credentials::basic,
    rt::Executor,
    service::service_fn,
    tcp::server::TcpListener,
    telemetry::tracing::{
        self,
        level_filters::LevelFilter,
        subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt},
    },
    tls::boring::{
        client::{BoringClientConfigExt, EmulateTlsProfileLayer},
        server::TlsAcceptorLayer,
    },
    tls::{
        SecureTransport,
        client::{ServerVerifyMode, TlsClientConfig},
        server::{SelfSignedData, TlsServerConfig},
    },
    ua::{
        layer::emulate::{
            UserAgentEmulateHttpConnectModifierLayer, UserAgentEmulateHttpRequestModifierLayer,
            UserAgentEmulateLayer,
        },
        profile::UserAgentDatabase,
    },
    utils::octets::mib,
};

use std::{
    convert::Infallible,
    sync::{Arc, atomic::AtomicBool},
    time::Duration,
};
use tokio::sync::mpsc;

#[derive(Debug, Clone, Extension)]
struct State {
    mitm_tls_service_config: TlsServerConfig,
    ua_db: Arc<UserAgentDatabase>,
    har_layer: HARExportLayer<FileRecorder, Arc<AtomicBool>>,
    har_toggle_ctl: mpsc::Sender<()>,
    exec: Executor,
}

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    tracing::subscriber::registry()
        .with(fmt::layer())
        .with(
            EnvFilter::builder()
                .with_default_directive(LevelFilter::INFO.into())
                .from_env_lossy(),
        )
        .init();

    let mitm_tls_service_config = new_mitm_tls_service_config();

    let graceful = rama::graceful::Shutdown::default();

    let (har_toggle, har_toggle_ctl) =
        har::toggle::mpsc_toggle(8, graceful.guard_weak().into_cancelled());
    let har_layer = HARExportLayer::new(FileRecorder::default(), har_toggle);

    let state = State {
        mitm_tls_service_config,
        ua_db: Arc::new(UserAgentDatabase::try_embedded()?),
        har_layer,
        har_toggle_ctl,
        exec: Executor::graceful(graceful.guard()),
    };

    graceful.spawn_task_fn(async |guard| {
        let exec = Executor::graceful(guard);

        let tcp_service = TcpListener::build(exec.clone())
            .bind_address("127.0.0.1:62040")
            .await
            .expect("bind tcp proxy to 127.0.0.1:62040");

        let http_mitm_service = new_http_mitm_proxy(&state);
        let http_service = HttpServer::auto(exec.clone()).service(Arc::new(
            (
                TraceLayer::new_for_http(),
                ConsumeErrLayer::default(),
                // See [`ProxyAuthLayer::with_labels`] for more information,
                // e.g. can also be used to extract upstream proxy filters
                ProxyAuthLayer::new(basic!("john", "secret")),
                // used to toggle HAR recording on and off
                // ...
                // NOTE that in a production proxy you would probably
                // put this behind its own local-only socket though,
                // not reachable from the outside, instead of exposing
                // it to the web...
                // ...
                // Remember kids: authentication != security
                HijackLayer::new(
                    DomainMatcher::exact("har.toggle.internal"),
                    Arc::new(WebService::default().with_post("/switch", async |req: Request| {
                        let state = req.extensions().get_ref::<State>().unwrap();
                        if let Err(err) = state.har_toggle_ctl.send(()).await {
                            tracing::error!("failed to toggle HAR Recording: {err}");
                            return StatusCode::INTERNAL_SERVER_ERROR;
                        } else {
                            tracing::debug!(
                                "force a stop recording so the file immediately flushes (DX etc)"
                            );
                            state.har_layer.recorder().stop_record().await;
                        }
                        StatusCode::OK
                    })),
                ),
                UpgradeLayer::new(
                    exec,
                    MethodMatcher::CONNECT,
                    DefaultHttpProxyConnectReplyService::new(),
                    service_fn(http_connect_proxy),
                ),
            )
                .into_layer(http_mitm_service)),
        );

        tcp_service
            .serve(
                (
                    AddInputExtensionLayer::new(state),
                    // protect the http proxy from too large bodies, both from request and response end
                    BodyLimitLayer::symmetric(mib(2)),
                )
                    .into_layer(http_service),
            )
            .await;
    });

    graceful
        .shutdown_with_limit(Duration::from_secs(30))
        .await
        .context("graceful shutdown")?;

    Ok(())
}

async fn http_connect_proxy(upgraded: Upgraded) -> Result<(), Infallible> {
    let state = upgraded.extensions().get_ref::<State>().unwrap();
    let http_service = Arc::new(new_http_mitm_proxy(state));

    let executor = state.exec.clone();

    let mut http_tp = HttpServer::auto(executor);
    http_tp.h2_mut().set_enable_connect_protocol();

    let http_transport_service = http_tp.service(http_service);

    let https_service = TlsAcceptorLayer::new(state.mitm_tls_service_config.clone())
        .with_store_client_hello(true)
        .into_layer(http_transport_service);

    https_service.serve(upgraded).await.expect("infallible");

    Ok(())
}

fn new_http_mitm_proxy(
    state: &State,
) -> impl Service<Request, Output = Response, Error = Infallible> {
    (
        MapResponseBodyLayer::new_boxed_streaming_body(),
        TraceLayer::new_for_http(),
        ConsumeErrLayer::default(),
        UserAgentEmulateLayer::new(state.ua_db.clone())
            .with_try_auto_detect_user_agent(true)
            .with_is_optional(true),
        CompressionLayer::new(),
        AddRequiredRequestHeadersLayer::new(),
        EmulateTlsProfileLayer::new(),
    )
        .into_layer(service_fn(http_mitm_proxy))
}

async fn http_mitm_proxy(req: Request) -> Result<Response, Infallible> {
    // This function will receive all requests going through this proxy,
    // be it sent via HTTP or HTTPS, both are equally visible. Hence... MITM

    // NOTE: use a custom connector (layers) in case you wish to add custom features,
    // such as upstream proxies or other configurations
    let tls_config = req
        .extensions()
        .get_ref::<SecureTransport>()
        .and_then(|st| st.client_hello())
        .map(TlsClientConfig::new_from_client_hello)
        .unwrap_or_else(TlsClientConfig::default_http)
        .with_server_verify(ServerVerifyMode::Disable);

    let state = req.extensions().get_ref::<State>().unwrap();

    // NOTE: in a production proxy you most likely
    // wouldn't want to build this each invocation,
    // but instead have a pre-built one as a struct local
    let client = EasyHttpWebClient::connector_builder()
        .with_default_transport_connector()
        .with_default_dns_connector()
        .with_tls_proxy_support_using_boringssl()
        .with_proxy_support()
        .with_tls_support_using_boringssl(tls_config)
        .with_custom_connector(UserAgentEmulateHttpConnectModifierLayer::default())
        .with_default_http_connector(state.exec.clone())
        .build_client()
        .with_jit_layer(UserAgentEmulateHttpRequestModifierLayer::default());

    // these are not desired for WS MITM flow, but they are for regular HTTP flow
    let client = (
        RemoveResponseHeaderLayer::hop_by_hop(),
        RemoveRequestHeaderLayer::hop_by_hop(),
        MapResponseBodyLayer::new_boxed_streaming_body(),
        DecompressionLayer::new(),
        state.har_layer.clone(),
    )
        .into_layer(client);

    match client.serve(req).await {
        Ok(mut resp) => {
            if let Some(har_fp) = resp
                .extensions()
                .get_ref::<HarFilePath>()
                .map(|fp| fp.display().to_string())
                .map(|fp| HeaderValue::try_from(fp).unwrap())
            {
                resp.headers_mut().insert("x-rama-har-file-path", har_fp);
            }

            Ok(resp)
        }
        Err(err) => {
            tracing::error!("error in client request: {err:?}");
            Ok(StatusCode::INTERNAL_SERVER_ERROR.into_response())
        }
    }
}

// NOTE: for a production service you ideally use
// an issued TLS cert (if possible via ACME). Or at the very least
// load it in from memory/file, so that your clients can install the certificate for trust.
fn new_mitm_tls_service_config() -> TlsServerConfig {
    TlsServerConfig::new()
        .try_with_self_signed(SelfSignedData {
            organisation_name: Some("Example Server Acceptor".to_owned()),
            ..Default::default()
        })
        .expect("self-signed")
        .with_alpn_http_auto()
}