Dedicated HTTP server/client crate

Also have better names for the HTTP routes

Cargo.lock

@@ -1614,7 +1614,9 @@ checksum = "d87c48c02e0dc5e3b849a2041db3029fd066650f8f717c07bf8ed78ccb895cac"
 dependencies = [
  "http",
  "hyper",
+ "log",
  "rustls 0.20.2",
+ "rustls-native-certs 0.6.1",
  "tokio",
  "tokio-rustls 0.23.2",
 ]
@@ -1863,12 +1865,12 @@ dependencies = [
  "mas-config",
  "mas-email",
  "mas-handlers",
+ "mas-http",
  "mas-storage",
  "mas-tasks",
  "mas-templates",
  "mas-warp-utils",
  "opentelemetry",
- "opentelemetry-http",
  "opentelemetry-jaeger",
  "opentelemetry-otlp",
  "opentelemetry-semantic-conventions",
@@ -1878,7 +1880,6 @@ dependencies = [
  "serde_yaml",
  "tokio",
  "tower",
- "tower-http",
  "tracing",
  "tracing-appender",
  "tracing-opentelemetry",
@@ -1987,6 +1988,22 @@ dependencies = [
  "warp",
 ]
 
+[[package]]
+name = "mas-http"
+version = "0.1.0"
+dependencies = [
+ "bytes 1.1.0",
+ "http",
+ "http-body",
+ "hyper",
+ "hyper-rustls 0.23.0",
+ "opentelemetry",
+ "opentelemetry-http",
+ "tower",
+ "tower-http",
+ "tracing",
+]
+
 [[package]]
 name = "mas-iana"
 version = "0.1.0"
@@ -4084,7 +4101,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "03650267ad175b51c47d02ed9547fc7d4ba2c7e5cb76df0bed67edd1825ae297"
 dependencies = [
  "async-compression",
- "base64 0.13.0",
  "bitflags",
  "bytes 1.1.0",
  "futures-core",
@@ -4092,11 +4108,7 @@ dependencies = [
  "http",
  "http-body",
  "http-range-header",
- "httpdate",
  "iri-string",
- "mime",
- "mime_guess",
- "percent-encoding",
  "pin-project-lite",
  "tokio",
  "tokio-util",

crates/cli/Cargo.toml

@@ -13,7 +13,6 @@ clap = { version = "3.0.14", features = ["derive"] }
 dotenv = "0.15.0"
 schemars = { version = "0.8.8", features = ["url", "chrono"] }
 tower = { version = "0.4.11", features = ["full"] }
-tower-http = { version = "0.2.1", features = ["full"] }
 hyper = { version = "0.14.16", features = ["full"] }
 serde_yaml = "0.8.23"
 warp = "0.3.2"
@@ -28,15 +27,15 @@ tracing-appender = "0.2.0"
 tracing-subscriber = { version = "0.3.7", features = ["env-filter"] }
 tracing-opentelemetry = "0.17.0"
 opentelemetry = { version = "0.17.0", features = ["trace", "metrics", "rt-tokio"] }
-opentelemetry-http = "0.6.0"
 opentelemetry-semantic-conventions = "0.9.0"
 opentelemetry-jaeger = { version = "0.16.0", features = ["rt-tokio", "reqwest_collector_client"], optional = true }
 opentelemetry-otlp = { version = "0.10.0", features = ["trace", "metrics"], optional = true }
 opentelemetry-zipkin = { version = "0.15.0", features = ["reqwest-client", "reqwest-rustls"], default-features = false, optional = true }
 
 mas-config = { path = "../config" }
-mas-handlers = { path = "../handlers" }
 mas-email = { path = "../email" }
+mas-handlers = { path = "../handlers" }
+mas-http = { path = "../http" }
 mas-storage = { path = "../storage" }
 mas-tasks = { path = "../tasks" }
 mas-templates = { path = "../templates" }

crates/cli/src/commands/server.rs

@@ -21,20 +21,15 @@ use std::{
 use anyhow::Context;
 use clap::Parser;
 use futures::{future::TryFutureExt, stream::TryStreamExt};
-use hyper::{header, Server};
+use hyper::Server;
 use mas_config::RootConfig;
 use mas_email::{MailTransport, Mailer};
 use mas_storage::MIGRATOR;
 use mas_tasks::TaskQueue;
 use mas_templates::Templates;
-use tower::{make::Shared, ServiceBuilder};
-use tower_http::{
-    compression::CompressionLayer, sensitive_headers::SetSensitiveHeadersLayer, trace::TraceLayer,
-};
+use tower::make::Shared;
 use tracing::{error, info};
 
-use crate::telemetry::{OtelMakeSpan, OtelOnResponse};
-
 #[derive(Parser, Debug, Default)]
 pub(super) struct Options {
     /// Automatically apply pending migrations
@@ -216,23 +211,7 @@ impl Options {
 
         let warp_service = warp::service(root);
 
-        let service = ServiceBuilder::new()
-            // Add high level tracing/logging to all requests
-            .layer(
-                TraceLayer::new_for_http()
-                    .make_span_with(OtelMakeSpan)
-                    .on_response(OtelOnResponse),
-            )
-            // Set a timeout
-            .timeout(Duration::from_secs(10))
-            // Compress responses
-            .layer(CompressionLayer::new())
-            // Mark the `Authorization` and `Cookie` headers as sensitive so it doesn't show in logs
-            .layer(SetSensitiveHeadersLayer::new(vec![
-                header::AUTHORIZATION,
-                header::COOKIE,
-            ]))
-            .service(warp_service);
+        let service = mas_http::server(warp_service);
 
         info!("Listening on http://{}", listener.local_addr().unwrap());
 

crates/cli/src/telemetry.rs

@@ -16,7 +16,6 @@ use std::{net::SocketAddr, time::Duration};
 
 use anyhow::bail;
 use futures::stream::{Stream, StreamExt};
-use hyper::{header, Version};
 use mas_config::{MetricsExporterConfig, Propagator, TelemetryConfig, TracingExporterConfig};
 use opentelemetry::{
     global,
@@ -27,16 +26,12 @@ use opentelemetry::{
         trace::Tracer,
         Resource,
     },
-    trace::TraceContextExt,
 };
-use opentelemetry_http::HeaderExtractor;
 #[cfg(feature = "jaeger")]
 use opentelemetry_jaeger::Propagator as JaegerPropagator;
 use opentelemetry_semantic_conventions as semcov;
 #[cfg(feature = "zipkin")]
 use opentelemetry_zipkin::{B3Encoding, Propagator as ZipkinPropagator};
-use tower_http::trace::{MakeSpan, OnResponse};
-use tracing::field;
 use url::Url;
 
 pub fn setup(config: &TelemetryConfig) -> anyhow::Result<Option<Tracer>> {
@@ -240,75 +235,3 @@ fn resource() -> Resource {
 
     resource.merge(&detected)
 }
-
-#[derive(Debug, Clone, Default)]
-pub struct OtelMakeSpan;
-
-impl<B> MakeSpan<B> for OtelMakeSpan {
-    fn make_span(&mut self, request: &hyper::Request<B>) -> tracing::Span {
-        // Extract the context from the headers
-        let headers = request.headers();
-        let extractor = HeaderExtractor(headers);
-
-        let cx = opentelemetry::global::get_text_map_propagator(|propagator| {
-            propagator.extract(&extractor)
-        });
-
-        let cx = if cx.span().span_context().is_remote() {
-            cx
-        } else {
-            opentelemetry::Context::new()
-        };
-
-        // Attach the context so when the request span is created it gets properly
-        // parented
-        let _guard = cx.attach();
-
-        let version = match request.version() {
-            Version::HTTP_09 => "0.9",
-            Version::HTTP_10 => "1.0",
-            Version::HTTP_11 => "1.1",
-            Version::HTTP_2 => "2.0",
-            Version::HTTP_3 => "3.0",
-            _ => "",
-        };
-
-        let span = tracing::info_span!(
-            "request",
-            http.method = %request.method(),
-            http.target = %request.uri(),
-            http.flavor = version,
-            http.status_code = field::Empty,
-            http.user_agent = field::Empty,
-            otel.kind = "server",
-            otel.status_code = field::Empty,
-        );
-
-        if let Some(user_agent) = headers
-            .get(header::USER_AGENT)
-            .and_then(|s| s.to_str().ok())
-        {
-            span.record("http.user_agent", &user_agent);
-        }
-
-        span
-    }
-}
-
-#[derive(Debug, Clone, Default)]
-pub struct OtelOnResponse;
-
-impl<B> OnResponse<B> for OtelOnResponse {
-    fn on_response(self, response: &hyper::Response<B>, _latency: Duration, span: &tracing::Span) {
-        let s = response.status();
-        let status = if s.is_success() {
-            "ok"
-        } else if s.is_client_error() || s.is_server_error() {
-            "error"
-        } else {
-            "unset"
-        };
-        span.record("otel.status_code", &status);
-        span.record("http.status_code", &s.as_u16());
-    }
-}

crates/email/Cargo.toml

@@ -9,13 +9,13 @@ license = "Apache-2.0"
 anyhow = "1.0.53"
 async-trait = "0.1.52"
 tokio = { version = "1.16.1", features = ["macros"] }
-
-mas-templates = { path = "../templates" }
-mas-config = { path = "../config" }
 tracing = "0.1.30"
 aws-sdk-sesv2 = "0.6.0"
 aws-config = "0.6.0"
 
+mas-templates = { path = "../templates" }
+mas-config = { path = "../config" }
+
 [dependencies.lettre]
 version = "0.10.0-rc.4"
 default-features = false

crates/email/src/transport/aws_ses.rs

@@ -35,7 +35,8 @@ impl Transport {
     /// Constructs a [`Transport`] from a given AWS shared config
     #[must_use]
     pub fn new(config: &aws_config::Config) -> Self {
-        let client = Client::new(config);
+        let config = aws_sdk_sesv2::Config::from(config);
+        let client = Client::from_conf(config);
         Self { client }
     }
 }

crates/handlers/src/health.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use hyper::header::CONTENT_TYPE;
-use mas_warp_utils::{errors::WrapError, filters::database::connection};
+use mas_warp_utils::{
+    errors::WrapError,
+    filters::{self, database::connection},
+};
 use mime::TEXT_PLAIN;
 use sqlx::{pool::PoolConnection, PgPool, Postgres};
 use tracing::{info_span, Instrument};
@@ -21,6 +24,7 @@ use warp::{filters::BoxedFilter, reply::with_header, Filter, Rejection, Reply};
 
 pub fn filter(pool: &PgPool) -> BoxedFilter<(Box<dyn Reply>,)> {
     warp::path!("health")
+        .and(filters::trace::name("GET /health"))
         .and(warp::get())
         .and(connection(pool))
         .and_then(get)

crates/handlers/src/lib.rs

@@ -26,6 +26,7 @@ use mas_email::Mailer;
 use mas_jose::StaticKeystore;
 use mas_static_files::filter as static_files;
 use mas_templates::Templates;
+use mas_warp_utils::filters;
 use sqlx::PgPool;
 use warp::{filters::BoxedFilter, Filter, Reply};
 
@@ -61,7 +62,8 @@ pub fn root(
         &config.http,
         &config.csrf,
     );
-    let static_files = static_files(config.http.web_root.clone());
+    let static_files =
+        static_files(config.http.web_root.clone()).and(filters::trace::name("GET static file"));
 
     let filter = health.or(views).unify().or(static_files).unify().or(oauth2);
 

crates/handlers/src/oauth2/authorization.rs

@@ -40,6 +40,7 @@ use mas_templates::{FormPostContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         database::transaction,
         session::{optional_session, session},
         with_templates,
@@ -222,6 +223,7 @@ pub fn filter(
     let clients_config_2 = clients_config.clone();
 
     let authorize = warp::path!("oauth2" / "authorize")
+        .and(filters::trace::name("GET /oauth2/authorize"))
         .and(warp::get())
         .map(move || clients_config.clone())
         .and(warp::query())
@@ -230,6 +232,7 @@ pub fn filter(
         .and_then(get);
 
     let step = warp::path!("oauth2" / "authorize" / "step")
+        .and(filters::trace::name("GET /oauth2/authorize/step"))
         .and(warp::get())
         .and(warp::query())
         .and(session(pool, encrypter))

crates/handlers/src/oauth2/discovery.rs

@@ -23,7 +23,7 @@ use mas_iana::{
     },
 };
 use mas_jose::SigningKeystore;
-use mas_warp_utils::filters::url_builder::UrlBuilder;
+use mas_warp_utils::filters::{self, url_builder::UrlBuilder};
 use oauth2_types::{
     oidc::{ClaimType, Metadata, SubjectType},
     requests::{Display, GrantType, ResponseMode},
@@ -184,6 +184,7 @@ pub(super) fn filter(
     };
 
     warp::path!(".well-known" / "openid-configuration")
+        .and(filters::trace::name("GET /.well-known/configuration"))
         .and(warp::get())
         .map(move || {
             let ret: Box<dyn Reply> = Box::new(warp::reply::json(&metadata));

crates/handlers/src/oauth2/introspection.rs

@@ -20,7 +20,7 @@ use mas_storage::oauth2::{
 };
 use mas_warp_utils::{
     errors::WrapError,
-    filters::{client::client_authentication, database::connection, url_builder::UrlBuilder},
+    filters::{self, client::client_authentication, database::connection, url_builder::UrlBuilder},
 };
 use oauth2_types::requests::{IntrospectionRequest, IntrospectionResponse};
 use sqlx::{pool::PoolConnection, PgPool, Postgres};
@@ -37,6 +37,7 @@ pub fn filter(
         .to_string();
 
     warp::path!("oauth2" / "introspect")
+        .and(filters::trace::name("POST /oauth2/introspect"))
         .and(
             warp::post()
                 .and(connection(pool))

crates/handlers/src/oauth2/keys.rs

@@ -15,12 +15,13 @@
 use std::sync::Arc;
 
 use mas_jose::{ExportJwks, StaticKeystore};
-use mas_warp_utils::errors::WrapError;
+use mas_warp_utils::{errors::WrapError, filters};
 use warp::{filters::BoxedFilter, Filter, Rejection, Reply};
 
 pub(super) fn filter(key_store: &Arc<StaticKeystore>) -> BoxedFilter<(Box<dyn Reply>,)> {
     let key_store = key_store.clone();
     warp::path!("oauth2" / "keys.json")
+        .and(filters::trace::name("GET /oauth2/keys.json"))
         .and(warp::get().map(move || key_store.clone()).and_then(get))
         .boxed()
 }

crates/handlers/src/oauth2/token.rs

@@ -37,7 +37,7 @@ use mas_storage::{
 };
 use mas_warp_utils::{
     errors::WrapError,
-    filters::{client::client_authentication, database::connection, url_builder::UrlBuilder},
+    filters::{self, client::client_authentication, database::connection, url_builder::UrlBuilder},
     reply::with_typed_header,
 };
 use oauth2_types::{
@@ -108,6 +108,7 @@ pub fn filter(
     let issuer = builder.oidc_issuer();
 
     warp::path!("oauth2" / "token")
+        .and(filters::trace::name("POST /oauth2/token"))
         .and(
             warp::post()
                 .and(client_authentication(clients_config, audience))

crates/handlers/src/oauth2/userinfo.rs

@@ -14,7 +14,10 @@
 
 use mas_data_model::{AccessToken, Session};
 use mas_storage::PostgresqlBackend;
-use mas_warp_utils::filters::authenticate::{authentication, recover_unauthorized};
+use mas_warp_utils::filters::{
+    self,
+    authenticate::{authentication, recover_unauthorized},
+};
 use serde::Serialize;
 use sqlx::PgPool;
 use warp::{filters::BoxedFilter, Filter, Rejection, Reply};
@@ -27,6 +30,7 @@ struct UserInfo {
 
 pub(super) fn filter(pool: &PgPool) -> BoxedFilter<(Box<dyn Reply>,)> {
     warp::path!("oauth2" / "userinfo")
+        .and(filters::trace::name("GET /oauth2/userinfo"))
         .and(
             warp::get()
                 .or(warp::post())

crates/handlers/src/views/account/emails.rs

@@ -27,6 +27,7 @@ use mas_templates::{AccountEmailsContext, EmailVerificationContext, TemplateCont
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::{protected_form, updated_csrf_token},
         database::{connection, transaction},
@@ -52,6 +53,7 @@ pub(super) fn filter(
     let mailer = mailer.clone();
 
     let get = with_templates(templates)
+        .and(filters::trace::name("GET /account/emails"))
         .and(encrypted_cookie_saver(encrypter))
         .and(updated_csrf_token(encrypter, csrf_config))
         .and(session(pool, encrypter))
@@ -59,6 +61,7 @@ pub(super) fn filter(
         .and_then(get);
 
     let post = with_templates(templates)
+        .and(filters::trace::name("POST /account/emails"))
         .and(warp::any().map(move || mailer.clone()))
         .and(url_builder(http_config))
         .and(encrypted_cookie_saver(encrypter))

crates/handlers/src/views/account/mod.rs

@@ -26,6 +26,7 @@ use mas_templates::{AccountContext, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::updated_csrf_token,
         database::connection,
@@ -47,6 +48,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     let get = warp::get()
+        .and(filters::trace::name("GET /account"))
         .and(with_templates(templates))
         .and(encrypted_cookie_saver(encrypter))
         .and(updated_csrf_token(encrypter, csrf_config))

crates/handlers/src/views/account/password.rs

@@ -23,6 +23,7 @@ use mas_templates::{EmptyContext, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::{protected_form, updated_csrf_token},
         database::transaction,
@@ -54,8 +55,12 @@ pub(super) fn filter(
         .and(protected_form(encrypter))
         .and_then(post);
 
-    let get = warp::get().and(get);
-    let post = warp::post().and(post);
+    let get = warp::get()
+        .and(get)
+        .and(filters::trace::name("GET /account/passwords"));
+    let post = warp::post()
+        .and(post)
+        .and(filters::trace::name("POST /account/passwords"));
     let filter = get.or(post).unify();
 
     warp::path!("password").and(filter).boxed()

crates/handlers/src/views/index.rs

@@ -17,6 +17,7 @@ use mas_data_model::BrowserSession;
 use mas_storage::PostgresqlBackend;
 use mas_templates::{IndexContext, TemplateContext, Templates};
 use mas_warp_utils::filters::{
+    self,
     cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
     csrf::updated_csrf_token,
     session::optional_session,
@@ -34,6 +35,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     warp::path::end()
+        .and(filters::trace::name("GET /"))
         .and(warp::get())
         .and(url_builder(http_config))
         .and(with_templates(templates))

crates/handlers/src/views/login.rs

@@ -22,6 +22,7 @@ use mas_templates::{LoginContext, LoginFormField, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::{protected_form, updated_csrf_token},
         database::connection,
@@ -90,6 +91,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     let get = warp::get()
+        .and(filters::trace::name("GET /login"))
         .and(with_templates(templates))
         .and(connection(pool))
         .and(encrypted_cookie_saver(encrypter))
@@ -99,6 +101,7 @@ pub(super) fn filter(
         .and_then(get);
 
     let post = warp::post()
+        .and(filters::trace::name("POST /login"))
         .and(with_templates(templates))
         .and(connection(pool))
         .and(encrypted_cookie_saver(encrypter))

crates/handlers/src/views/logout.rs

@@ -17,13 +17,14 @@ use mas_data_model::BrowserSession;
 use mas_storage::{user::end_session, PostgresqlBackend};
 use mas_warp_utils::{
     errors::WrapError,
-    filters::{csrf::protected_form, database::transaction, session::session},
+    filters::{self, csrf::protected_form, database::transaction, session::session},
 };
 use sqlx::{PgPool, Postgres, Transaction};
 use warp::{filters::BoxedFilter, hyper::Uri, Filter, Rejection, Reply};
 
 pub(super) fn filter(pool: &PgPool, encrypter: &Encrypter) -> BoxedFilter<(Box<dyn Reply>,)> {
     warp::path!("logout")
+        .and(filters::trace::name("POST /logout"))
         .and(warp::post())
         .and(session(pool, encrypter))
         .and(transaction(pool))

crates/handlers/src/views/reauth.rs

@@ -20,6 +20,7 @@ use mas_templates::{ReauthContext, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::{protected_form, updated_csrf_token},
         database::{connection, transaction},
@@ -87,6 +88,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     let get = warp::get()
+        .and(filters::trace::name("GET /reauth"))
         .and(with_templates(templates))
         .and(connection(pool))
         .and(encrypted_cookie_saver(encrypter))
@@ -96,6 +98,7 @@ pub(super) fn filter(
         .and_then(get);
 
     let post = warp::post()
+        .and(filters::trace::name("POST /reauth"))
         .and(session(pool, encrypter))
         .and(transaction(pool))
         .and(protected_form(encrypter))

crates/handlers/src/views/register.rs

@@ -26,6 +26,7 @@ use mas_templates::{RegisterContext, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::{protected_form, updated_csrf_token},
         database::{connection, transaction},
@@ -96,6 +97,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     let get = warp::get()
+        .and(filters::trace::name("GET /register"))
         .and(with_templates(templates))
         .and(connection(pool))
         .and(encrypted_cookie_saver(encrypter))
@@ -105,6 +107,7 @@ pub(super) fn filter(
         .and_then(get);
 
     let post = warp::post()
+        .and(filters::trace::name("POST /register"))
         .and(transaction(pool))
         .and(encrypted_cookie_saver(encrypter))
         .and(protected_form(encrypter))

crates/handlers/src/views/verify.rs

@@ -26,6 +26,7 @@ use mas_templates::{EmptyContext, TemplateContext, Templates};
 use mas_warp_utils::{
     errors::WrapError,
     filters::{
+        self,
         cookies::{encrypted_cookie_saver, EncryptedCookieSaver},
         csrf::updated_csrf_token,
         database::transaction,
@@ -43,6 +44,7 @@ pub(super) fn filter(
     csrf_config: &CsrfConfig,
 ) -> BoxedFilter<(Box<dyn Reply>,)> {
     warp::path!("verify" / String)
+        .and(filters::trace::name("GET /verify"))
         .and(warp::get())
         .and(with_templates(templates))
         .and(encrypted_cookie_saver(encrypter))

crates/http/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "mas-http"
+version = "0.1.0"
+authors = ["Quentin Gliech <quenting@element.io>"]
+edition = "2021"
+license = "Apache-2.0"
+
+[dependencies]
+bytes = "1.1.0"
+http = "0.2.6"
+http-body = "0.4.4"
+hyper = "0.14.16"
+hyper-rustls = { version = "0.23.0", features = ["http1", "http2"] }
+opentelemetry = "0.17.0"
+opentelemetry-http = "0.6.0"
+tower = { version = "0.4.11", features = ["timeout", "limit"] }
+tower-http = { version = "0.2.1", features = ["follow-redirect", "decompression-full", "set-header", "trace"] }
+tracing = "0.1.30"

crates/http/src/lib.rs

@@ -0,0 +1,202 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::time::Duration;
+
+use http::{header::USER_AGENT, HeaderValue, Request, Response, Version};
+use http_body::combinators::BoxBody;
+use hyper::{client::HttpConnector, Client};
+use hyper_rustls::HttpsConnectorBuilder;
+use opentelemetry::trace::TraceContextExt;
+use opentelemetry_http::HeaderExtractor;
+use tower::{
+    limit::ConcurrencyLimitLayer,
+    timeout::TimeoutLayer,
+    util::{BoxCloneService, BoxService},
+    BoxError, Service, ServiceBuilder, ServiceExt,
+};
+use tower_http::{
+    follow_redirect::FollowRedirectLayer,
+    set_header::SetRequestHeaderLayer,
+    trace::{MakeSpan, OnResponse, TraceLayer},
+};
+use tracing::field;
+
+static MAS_USER_AGENT: HeaderValue =
+    HeaderValue::from_static("matrix-authentication-service/0.0.1");
+
+type Body = BoxBody<bytes::Bytes, BoxError>;
+
+pub fn client(
+    operation: &'static str,
+) -> BoxService<
+    Request<Body>,
+    Response<impl http_body::Body<Data = bytes::Bytes, Error = hyper::Error>>,
+    BoxError,
+> {
+    let mut http = HttpConnector::new();
+    http.enforce_http(false);
+
+    let https = HttpsConnectorBuilder::new()
+        .with_native_roots()
+        .https_or_http()
+        .enable_http1()
+        .enable_http2()
+        .wrap_connector(http);
+
+    let client = Client::builder().build(https);
+
+    ServiceBuilder::new()
+        .layer(
+            TraceLayer::new_for_http()
+                .make_span_with(MakeOtelSpan::client(operation))
+                .on_response(OtelOnResponse),
+        )
+        .layer(TimeoutLayer::new(Duration::from_secs(10)))
+        .layer(FollowRedirectLayer::new())
+        .layer(ConcurrencyLimitLayer::new(10))
+        .layer(SetRequestHeaderLayer::overriding(
+            USER_AGENT,
+            MAS_USER_AGENT.clone(),
+        ))
+        .service(client)
+        .boxed()
+}
+
+#[allow(clippy::type_complexity)]
+pub fn server<ReqBody, ResBody, S>(
+    service: S,
+) -> BoxCloneService<Request<ReqBody>, Response<BoxBody<ResBody::Data, ResBody::Error>>, BoxError>
+where
+    S: Service<Request<ReqBody>, Response = Response<ResBody>> + Clone + Send + 'static,
+    ReqBody: http_body::Body + 'static,
+    ResBody: http_body::Body + Sync + Send + 'static,
+    ResBody::Error: std::fmt::Display + 'static,
+    S::Future: Send + 'static,
+    S::Error: Into<BoxError> + 'static,
+{
+    ServiceBuilder::new()
+        .map_response(|r: Response<_>| r.map(BoxBody::new))
+        .layer(
+            TraceLayer::new_for_http()
+                .make_span_with(MakeOtelSpan::server())
+                .on_response(OtelOnResponse),
+        )
+        .layer(TimeoutLayer::new(Duration::from_secs(10)))
+        .service(service)
+        .boxed_clone()
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct MakeOtelSpan {
+    operation: Option<&'static str>,
+    kind: &'static str,
+    extract: bool,
+}
+
+impl MakeOtelSpan {
+    fn client(operation: &'static str) -> Self {
+        Self {
+            operation: Some(operation),
+            extract: false,
+            kind: "client",
+        }
+    }
+
+    fn server() -> Self {
+        Self {
+            operation: None,
+            extract: true,
+            kind: "server",
+        }
+    }
+}
+
+impl<B> MakeSpan<B> for MakeOtelSpan {
+    fn make_span(&mut self, request: &Request<B>) -> tracing::Span {
+        let cx = if self.extract {
+            // Extract the context from the headers
+            let headers = request.headers();
+            let extractor = HeaderExtractor(headers);
+
+            let cx = opentelemetry::global::get_text_map_propagator(|propagator| {
+                propagator.extract(&extractor)
+            });
+
+            if cx.span().span_context().is_remote() {
+                cx
+            } else {
+                opentelemetry::Context::new()
+            }
+        } else {
+            opentelemetry::Context::current()
+        };
+
+        // Attach the context so when the request span is created it gets properly
+        // parented
+        let _guard = cx.attach();
+
+        // Extract the context from the headers
+        let headers = request.headers();
+
+        let version = match request.version() {
+            Version::HTTP_09 => "0.9",
+            Version::HTTP_10 => "1.0",
+            Version::HTTP_11 => "1.1",
+            Version::HTTP_2 => "2.0",
+            Version::HTTP_3 => "3.0",
+            _ => "",
+        };
+
+        let span = tracing::info_span!(
+            "request",
+            otel.name = field::Empty,
+            otel.kind = self.kind,
+            otel.status_code = field::Empty,
+            http.method = %request.method(),
+            http.target = %request.uri(),
+            http.flavor = version,
+            http.status_code = field::Empty,
+            http.user_agent = field::Empty,
+        );
+
+        if let Some(operation) = &self.operation {
+            span.record("otel.name", operation);
+        }
+
+        if let Some(user_agent) = headers.get(USER_AGENT).and_then(|s| s.to_str().ok()) {
+            span.record("http.user_agent", &user_agent);
+        }
+
+        span
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct OtelOnResponse;
+
+impl<B> OnResponse<B> for OtelOnResponse {
+    fn on_response(self, response: &hyper::Response<B>, _latency: Duration, span: &tracing::Span) {
+        let s = response.status();
+        let status = if s.is_success() {
+            "ok"
+        } else if s.is_client_error() || s.is_server_error() {
+            "error"
+        } else {
+            "unset"
+        };
+        span.record("otel.status_code", &status);
+        span.record("http.status_code", &s.as_u16());
+    }
+}

crates/warp-utils/src/filters/mod.rs

@@ -1,4 +1,4 @@
-// Copyright 2021 The Matrix.org Foundation C.I.C.
+// Copyright 2021, 2022 The Matrix.org Foundation C.I.C.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ pub mod csrf;
 pub mod database;
 pub mod headers;
 pub mod session;
+pub mod trace;
 pub mod url_builder;
 
 use std::convert::Infallible;

crates/warp-utils/src/filters/trace.rs

@@ -0,0 +1,33 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Route tracing utility
+
+use std::convert::Infallible;
+
+use tracing::Span;
+use warp::Filter;
+
+/// Set the name of that route
+#[must_use]
+pub fn name(
+    name: &'static str,
+) -> impl Filter<Extract = (), Error = Infallible> + Clone + Send + Sync + 'static {
+    warp::any()
+        .map(move || {
+            let span = Span::current();
+            span.record("otel.name", &name);
+        })
+        .untuple_one()
+}