use crate::{
Layer, Service,
cli::ForwardKind,
combinators::{Either, Either3},
error::{BoxError, BoxErrorExt, ErrorContext},
extensions::ExtensionsRef,
http::{
BodyLimitLayer, Request, Response, Version,
body::util::BodyExt,
convert::curl,
core::h2::frame::EarlyFrameCapture,
fingerprint::{AkamaiH2, Ja4H},
header::USER_AGENT,
headers::exotic::XClacksOverhead,
layer::set_header::SetResponseHeaderLayer,
layer::{required_header::AddRequiredResponseHeadersLayer, trace::TraceLayer},
proto::h2::PseudoHeaderOrder,
server::HttpServer,
service::web::{extract::Json, response::IntoResponse},
ws::handshake::{
matcher::WebSocketMatcher,
server::{WebSocketAcceptor, WebSocketEchoService},
},
},
layer::limit::policy::UnlimitedPolicy,
layer::{ConsumeErrLayer, LimitLayer, TimeoutLayer, limit::policy::ConcurrentPolicy},
net::address::ip::geo::IpGeoDb,
net::forwarded::Forwarded,
net::stream::SocketInfo,
net::{AuthorityInputExt, Protocol, ProtocolInputExt},
proxy::haproxy::server::HaProxyLayer,
rt::Executor,
tcp::TcpStream,
telemetry::tracing,
ua::{UserAgent, layer::classifier::UserAgentClassifierLayer, profile::UserAgentDatabase},
utils::octets::mib,
};
use rama_core::error::ErrorExt as _;
use rama_http::layer::upgrade::UpgradeLayer;
use serde::Serialize;
use serde_json::json;
use std::{convert::Infallible, sync::Arc, time::Duration};
core::cfg_select! {
feature = "boring" => {
use crate::tls::boring::server::TlsAcceptorLayer;
}
feature = "rustls" => {
use crate::tls::rustls::server::TlsAcceptorLayer;
}
_ => {}
}
#[cfg(any(feature = "rustls", feature = "boring"))]
use crate::{
tls::fingerprint::{Ja3, Ja4, PeetPrint},
tls::{
SecureTransport,
client::ClientHelloExtension,
client::{ECHClientHello, NegotiatedTlsParameters},
server::TlsServerConfig,
},
};
#[derive(Debug, Clone)]
pub struct EchoServiceBuilder<H> {
concurrent_limit: usize,
body_limit: usize,
timeout: Duration,
forward: Option<ForwardKind>,
#[cfg(any(feature = "rustls", feature = "boring"))]
tls_server_config: Option<TlsServerConfig>,
http_version: Option<Version>,
ws_support: bool,
http_service_builder: H,
uadb: Option<std::sync::Arc<UserAgentDatabase>>,
geo_db: Option<std::sync::Arc<IpGeoDb>>,
}
impl Default for EchoServiceBuilder<()> {
fn default() -> Self {
Self {
concurrent_limit: 0,
body_limit: mib(1),
timeout: Duration::ZERO,
forward: None,
#[cfg(any(feature = "rustls", feature = "boring"))]
tls_server_config: None,
http_version: None,
ws_support: false,
http_service_builder: (),
uadb: None,
geo_db: None,
}
}
}
impl EchoServiceBuilder<()> {
#[must_use]
pub fn new() -> Self {
Self::default()
}
}
impl<H> EchoServiceBuilder<H> {
crate::utils::macros::generate_set_and_with! {
pub fn concurrent(mut self, limit: usize) -> Self {
self.concurrent_limit = limit;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn body_limit(mut self, limit: usize) -> Self {
self.body_limit = limit;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn forward(mut self, kind: Option<ForwardKind>) -> Self {
self.forward = kind;
self
}
}
crate::utils::macros::generate_set_and_with! {
#[cfg(any(feature = "rustls", feature = "boring"))]
pub fn tls_server_config(mut self, cfg: Option<TlsServerConfig>) -> Self {
self.tls_server_config = cfg;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn http_version(mut self, version: Option<Version>) -> Self {
self.http_version = version;
self
}
}
pub fn with_http_layer<H2>(self, layer: H2) -> EchoServiceBuilder<(H, H2)> {
EchoServiceBuilder {
concurrent_limit: self.concurrent_limit,
body_limit: self.body_limit,
timeout: self.timeout,
forward: self.forward,
#[cfg(any(feature = "rustls", feature = "boring"))]
tls_server_config: self.tls_server_config,
http_version: self.http_version,
ws_support: self.ws_support,
http_service_builder: (self.http_service_builder, layer),
uadb: self.uadb,
geo_db: self.geo_db,
}
}
crate::utils::macros::generate_set_and_with! {
pub fn user_agent_database(
mut self,
db: Option<std::sync::Arc<UserAgentDatabase>>,
) -> Self {
self.uadb = db;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn geo_db(mut self, db: Option<std::sync::Arc<IpGeoDb>>) -> Self {
self.geo_db = db;
self
}
}
crate::utils::macros::generate_set_and_with! {
pub fn ws_support(
mut self,
support: bool,
) -> Self {
self.ws_support = support;
self
}
}
}
impl<H> EchoServiceBuilder<H>
where
H: Layer<EchoService, Service: Service<Request, Output = Response, Error = BoxError>>,
{
#[expect(unused_mut)]
pub fn build(
mut self,
exec: Executor,
) -> Result<impl Service<TcpStream, Output = (), Error = Infallible>, BoxError> {
let tcp_forwarded_layer = match &self.forward {
Some(ForwardKind::HaProxy) => Some(HaProxyLayer::default()),
_ => None,
};
let http_service = Arc::new(self.build_http(exec.clone()));
let tcp_service_builder = (
ConsumeErrLayer::trace_as(tracing::Level::DEBUG),
LimitLayer::new(if self.concurrent_limit > 0 {
Either::A(ConcurrentPolicy::max(self.concurrent_limit))
} else {
Either::B(UnlimitedPolicy::new())
}),
if !self.timeout.is_zero() {
TimeoutLayer::new(self.timeout)
} else {
TimeoutLayer::never()
},
tcp_forwarded_layer,
BodyLimitLayer::request_only(self.body_limit),
#[cfg(any(feature = "rustls", feature = "boring"))]
self.tls_server_config
.map(|cfg| TlsAcceptorLayer::new(cfg).with_store_client_hello(true)),
);
let http_transport_service = match self.http_version {
Some(Version::HTTP_2) => Either3::A({
let mut http = HttpServer::new_h2(exec);
if self.ws_support {
http.h2_mut().set_enable_connect_protocol();
}
http.service(http_service)
}),
Some(Version::HTTP_11 | Version::HTTP_10 | Version::HTTP_09) => {
Either3::B(HttpServer::new_http1(exec).service(http_service))
}
Some(version) => {
return Err(BoxError::from_static_str("unsupported http version")
.context_debug_field("version", version));
}
None => Either3::C({
let mut http = HttpServer::auto(exec);
if self.ws_support {
http.h2_mut().set_enable_connect_protocol();
}
http.service(http_service)
}),
};
Ok(tcp_service_builder.into_layer(http_transport_service))
}
pub fn build_http(
&self,
exec: Executor,
) -> impl Service<Request, Output: IntoResponse, Error = Infallible> + use<H> {
let http_forwarded_layer = super::http_forwarded_layer(self.forward.as_ref());
let geo_attribution = self.geo_db.as_ref().and_then(|db| {
let notices: Vec<_> = db.attributions().collect();
(!notices.is_empty()).then(|| crate::cli::service::geo::geo_attribution_layer(notices))
});
(
TraceLayer::new_for_http(),
SetResponseHeaderLayer::<XClacksOverhead>::if_not_present_default_typed(),
AddRequiredResponseHeadersLayer::default(),
geo_attribution,
UserAgentClassifierLayer::new(),
ConsumeErrLayer::default(),
http_forwarded_layer,
self.ws_support.then(|| {
UpgradeLayer::new(
exec,
WebSocketMatcher::default(),
{
let acceptor = WebSocketAcceptor::default()
.with_protocols_flex(true)
.with_echo_protocols();
#[cfg(feature = "compression")]
{
acceptor.with_per_message_deflate_overwrite_extensions()
}
#[cfg(not(feature = "compression"))]
{
acceptor
}
},
ConsumeErrLayer::trace_as(tracing::Level::DEBUG)
.into_layer(WebSocketEchoService::default()),
)
}),
)
.into_layer(self.http_service_builder.layer(EchoService {
uadb: self.uadb.clone(),
geo_db: self.geo_db.clone(),
}))
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct EchoService {
uadb: Option<std::sync::Arc<UserAgentDatabase>>,
geo_db: Option<std::sync::Arc<IpGeoDb>>,
}
impl Service<Request> for EchoService {
type Output = Response;
type Error = BoxError;
async fn serve(&self, req: Request) -> Result<Self::Output, Self::Error> {
let user_agent_info = req
.extensions()
.get_ref()
.map(|ua: &UserAgent| {
json!({
"user_agent": ua.header_str().to_owned(),
"kind": ua.info().map(|info| info.kind.to_string()),
"version": ua.info().and_then(|info| info.version),
"platform": ua.platform().map(|v| v.to_string()),
})
})
.unwrap_or_default();
let authority = req
.authority()
.context("echo: resolve request authority")?
.to_string();
let scheme = req.protocol().unwrap_or(&Protocol::HTTP).to_string();
let ua_str = req
.headers()
.get(USER_AGENT)
.and_then(|h| h.to_str().ok())
.map(ToOwned::to_owned);
tracing::debug!(
user_agent.original = ua_str,
"echo request received from ua with ua header",
);
#[derive(Debug, Serialize)]
struct FingerprintProfileData {
hash: String,
verbose: String,
matched: bool,
}
let ja4h = Ja4H::compute(&req)
.inspect_err(|err| tracing::error!("ja4h compute failure: {err:?}"))
.ok()
.map(|ja4h| {
let mut profile_ja4h: Option<FingerprintProfileData> = None;
if let Some(uadb) = self.uadb.as_deref()
&& let Some(profile) =
ua_str.as_deref().and_then(|s| uadb.get_exact_header_str(s))
{
let matched_ja4h = match req.version() {
Version::HTTP_10 | Version::HTTP_11 => profile
.http
.ja4h_h1_navigate(Some(req.method().clone()))
.inspect_err(|err| {
tracing::trace!(
"ja4h computation of matched profile for incoming h1 req: {err:?}"
)
})
.ok(),
Version::HTTP_2 => profile
.http
.ja4h_h2_navigate(Some(req.method().clone()))
.inspect_err(|err| {
tracing::trace!(
"ja4h computation of matched profile for incoming h2 req: {err:?}"
)
})
.ok(),
_ => None,
};
if let Some(tgt) = matched_ja4h {
let hash = format!("{tgt}");
let matched = format!("{ja4h}") == hash;
profile_ja4h = Some(FingerprintProfileData {
hash,
verbose: format!("{tgt:?}"),
matched,
});
}
}
json!({
"hash": format!("{ja4h}"),
"verbose": format!("{ja4h:?}"),
"profile": profile_ja4h,
})
});
let (parts, body) = req.into_parts();
let body = body
.collect()
.await
.context("collect request body for echo purposes")?
.to_bytes();
let curl_request = curl::cmd_string_for_request_parts_and_payload(&parts, &body);
let headers: Vec<_> = parts
.headers
.into_ordered_iter()
.map(|(name, value)| {
(
name,
std::str::from_utf8(value.as_bytes())
.map(|s| s.to_owned())
.unwrap_or_else(|_| format!("0x{:x?}", value.as_bytes())),
)
})
.collect();
let body = hex::encode(body.as_ref());
#[cfg(any(feature = "rustls", feature = "boring"))]
let tls_info = parts
.extensions
.get_ref::<SecureTransport>()
.and_then(|st| st.client_hello())
.map(|hello| {
let ja4 = Ja4::compute(parts.extensions.extensions())
.inspect_err(|err| tracing::trace!("ja4 computation: {err:?}"))
.ok();
let mut profile_ja4: Option<FingerprintProfileData> = None;
if let Some(uadb) = self.uadb.as_deref()
&& let Some(profile) =
ua_str.as_deref().and_then(|s| uadb.get_exact_header_str(s))
{
let matched_ja4 = profile
.tls
.compute_ja4(
parts
.extensions
.get_ref::<NegotiatedTlsParameters>()
.map(|param| param.protocol_version),
)
.inspect_err(|err| {
tracing::trace!("ja4 computation of matched profile: {err:?}")
})
.ok();
if let (Some(src), Some(tgt)) = (ja4.as_ref(), matched_ja4) {
let hash = format!("{tgt}");
let matched = format!("{src}") == hash;
profile_ja4 = Some(FingerprintProfileData {
hash,
verbose: format!("{tgt:?}"),
matched,
});
}
}
let ja4 = ja4.map(|ja4| {
json!({
"hash": format!("{ja4}"),
"verbose": format!("{ja4:?}"),
"profile": profile_ja4,
})
});
let ja3 = Ja3::compute(parts.extensions.extensions())
.inspect_err(|err| tracing::trace!("ja3 computation: {err:?}"))
.ok();
let mut profile_ja3: Option<FingerprintProfileData> = None;
if let Some(uadb) = self.uadb.as_deref()
&& let Some(profile) =
ua_str.as_deref().and_then(|s| uadb.get_exact_header_str(s))
{
let matched_ja3 = profile
.tls
.compute_ja3(
parts
.extensions
.get_ref::<NegotiatedTlsParameters>()
.map(|param| param.protocol_version),
)
.inspect_err(|err| {
tracing::trace!("ja3 computation of matched profile: {err:?}")
})
.ok();
if let (Some(src), Some(tgt)) = (ja3.as_ref(), matched_ja3) {
let hash = format!("{tgt:x}");
let matched = format!("{src:x}") == hash;
profile_ja3 = Some(FingerprintProfileData {
hash,
verbose: format!("{tgt}"),
matched,
});
}
}
let ja3 = ja3.map(|ja3| {
json!({
"hash": format!("{ja3:x}"),
"verbose": format!("{ja3}"),
"profile": profile_ja3,
})
});
let peet = PeetPrint::compute(parts.extensions.extensions())
.inspect_err(|err| tracing::trace!("peet computation: {err:?}"))
.ok();
let mut profile_peet: Option<FingerprintProfileData> = None;
if let Some(uadb) = self.uadb.as_deref()
&& let Some(profile) =
ua_str.as_deref().and_then(|s| uadb.get_exact_header_str(s))
{
let matched_peet = profile
.tls
.compute_peet()
.inspect_err(|err| {
tracing::trace!("peetprint computation of matched profile: {err:?}")
})
.ok();
if let (Some(src), Some(tgt)) = (peet.as_ref(), matched_peet) {
let hash = format!("{tgt}");
let matched = format!("{src}") == hash;
profile_peet = Some(FingerprintProfileData {
hash,
verbose: format!("{tgt:?}"),
matched,
});
}
}
let peet = peet.map(|peet| {
json!({
"hash": format!("{peet}"),
"verbose": format!("{peet:?}"),
"profile": profile_peet,
})
});
json!({
"header": {
"version": hello.protocol_version().to_string(),
"cipher_suites": hello
.cipher_suites().iter().map(|s| s.to_string()).collect::<Vec<_>>(),
"compression_algorithms": hello
.compression_algorithms().iter().map(|s| s.to_string()).collect::<Vec<_>>(),
},
"extensions": hello.extensions().iter().map(|extension| match extension {
ClientHelloExtension::ServerName(domain) => json!({
"id": extension.id().to_string(),
"data": domain,
}),
ClientHelloExtension::SignatureAlgorithms(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::SupportedVersions(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::ApplicationLayerProtocolNegotiation(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::ApplicationSettings{ protocols, .. } => json!({
"id": extension.id().to_string(),
"data": protocols.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::SupportedGroups(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::ECPointFormats(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::CertificateCompression(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::DelegatedCredentials(v) => json!({
"id": extension.id().to_string(),
"data": v.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
}),
ClientHelloExtension::RecordSizeLimit(v) => json!({
"id": extension.id().to_string(),
"data": v.to_string(),
}),
ClientHelloExtension::EncryptedClientHello(ech) => match ech {
ECHClientHello::Outer(ech) => json!({
"id": extension.id().to_string(),
"data": {
"type": "outer",
"cipher_suite": {
"aead_id": ech.cipher_suite.aead_id.to_string(),
"kdf_id": ech.cipher_suite.kdf_id.to_string(),
},
"config_id": ech.config_id,
"enc": format!("0x{}", hex::encode(&ech.enc)),
"payload": format!("0x{}", hex::encode(&ech.payload)),
},
}),
ECHClientHello::Inner => json!({
"id": extension.id().to_string(),
"data": {
"type": "inner",
},
})
}
ClientHelloExtension::Opaque { id, data } => if data.is_empty() {
json!({
"id": id.to_string()
})
} else {
json!({
"id": id.to_string(),
"data": format!("0x{}", hex::encode(data))
})
},
}).collect::<Vec<_>>(),
"ja3": ja3,
"ja4": ja4,
"peet": peet
})
});
#[cfg(not(any(feature = "rustls", feature = "boring")))]
let tls_info: Option<()> = None;
let mut h2 = None;
if parts.version == Version::HTTP_2 {
let early_frames = parts.extensions.get_ref::<EarlyFrameCapture>();
let pseudo_headers = parts.extensions.get_ref::<PseudoHeaderOrder>();
let akamai_h2 = AkamaiH2::compute(&parts.extensions)
.inspect_err(|err| tracing::trace!("akamai h2 compute failure: {err:?}"))
.ok()
.map(|akamai| {
json!({
"hash": format!("{akamai}"),
"verbose": format!("{akamai:?}"),
})
});
h2 = Some(json!({
"early_frames": early_frames,
"pseudo_headers": pseudo_headers,
"akamai_h2": akamai_h2,
}));
}
let geo = self
.geo_db
.as_ref()
.and_then(|db| {
parts
.extensions
.get_ref::<Forwarded>()
.and_then(|f| f.client_ip())
.or_else(|| {
parts
.extensions
.get_ref::<SocketInfo>()
.map(|s| s.peer_addr().ip_addr)
})
.and_then(|ip| db.resolve(ip))
})
.map(|info| serde_json::to_value(&info).unwrap_or_default())
.unwrap_or(serde_json::Value::Null);
Ok(Json(json!({
"ua": user_agent_info,
"geo": geo,
"http": {
"version": format!("{:?}", parts.version),
"scheme": scheme,
"method": format!("{:?}", parts.method),
"authority": authority,
"path": parts.uri.path_or_root().into_owned(),
"query": parts.uri.query().map(|q| q.as_encoded_str().into_owned()),
"h2": h2,
"headers": headers,
"payload": body,
"ja4h": ja4h,
"curl": curl_request,
},
"tls": tls_info,
"socket_addr": parts.extensions.get_ref::<Forwarded>()
.and_then(|f|
f.client_socket_addr().map(|addr| addr.to_string())
.or_else(|| f.client_ip().map(|ip| ip.to_string()))
).or_else(|| parts.extensions.get_ref::<SocketInfo>().map(|v| v.peer_addr().to_string())),
}))
.into_response())
}
}