feat(transport-http): JWT auth layer

Cargo.toml

@@ -139,6 +139,7 @@ thiserror-no-std = "2.0.2"
 url = "2.5"
 derive_more = { version = "1.0.0", default-features = false }
 http = "1.1.0"
+jsonwebtoken = "9.3.0"
 
 ## serde
 serde = { version = "1.0", default-features = false, features = [

crates/rpc-types-engine/Cargo.toml

@@ -41,7 +41,7 @@ jsonrpsee-types = { version = "0.24", optional = true }
 
 # jwt
 rand = { workspace = true, optional = true }
-jsonwebtoken = { version = "9.3.0", optional = true }
+jsonwebtoken = { workspace = true, optional = true }
 
 [features]
 default = ["jwt", "std", "serde"]

crates/rpc-types-engine/src/jwt.rs

@@ -258,6 +258,11 @@ impl JwtSecret {
         let algo = jsonwebtoken::Header::new(Algorithm::HS256);
         jsonwebtoken::encode(&algo, claims, &key)
     }
+
+    /// Returns the secret key as a byte slice.
+    pub const fn as_bytes(&self) -> &[u8] {
+        &self.0
+    }
 }
 
 impl core::fmt::Debug for JwtSecret {

crates/transport-http/Cargo.toml

@@ -34,6 +34,10 @@ http-body-util = { workspace = true, optional = true }
 hyper = { workspace = true, default-features = false, optional = true }
 hyper-util = { workspace = true, features = ["full"], optional = true }
 
+# auth layer
+alloy-rpc-types-engine = { workspace = true, optional = true }
+jsonwebtoken = { workspace = true, optional = true }
+
 [features]
 default = ["reqwest", "reqwest-default-tls"]
 reqwest = [
@@ -52,6 +56,13 @@ hyper = [
     "dep:tower",
     "dep:tracing",
 ]
+jwt-auth = [
+    "hyper",
+    "dep:alloy-rpc-types-engine",
+    "alloy-rpc-types-engine/jwt",
+    "alloy-rpc-types-engine/serde",
+    "dep:jsonwebtoken",
+]
 reqwest-default-tls = ["reqwest?/default-tls"]
 reqwest-native-tls = ["reqwest?/native-tls"]
 reqwest-rustls-tls = ["reqwest?/rustls-tls"]

crates/transport-http/src/hyper_transport.rs

@@ -69,12 +69,15 @@ impl<B, S> HyperClient<B, S> {
     }
 }
 
-impl<B, S> Http<HyperClient<B, S>>
+impl<B, S, ResBody> Http<HyperClient<B, S>>
 where
-    S: Service<Request<B>, Response = HyperResponse> + Clone + Send + Sync + 'static,
+    S: Service<Request<B>, Response = Response<ResBody>> + Clone + Send + Sync + 'static,
     S::Future: Send,
     S::Error: std::error::Error + Send + Sync + 'static,
     B: From<Vec<u8>> + Send + 'static + Clone,
+    ResBody: BodyExt + Send + 'static,
+    ResBody::Error: std::error::Error + Send + Sync + 'static,
+    ResBody::Data: Send,
 {
     /// Make a request to the server using the given service.
     fn request_hyper(&self, req: RequestPacket) -> TransportFut<'static> {

crates/transport-http/src/layers/auth.rs

@@ -0,0 +1,141 @@
+use crate::hyper::{
+    header::{HeaderMap, AUTHORIZATION},
+    Request, Response,
+};
+use alloy_rpc_types_engine::{Claims, JwtSecret};
+use alloy_transport::{TransportError, TransportErrorKind};
+use hyper::header::HeaderValue;
+use jsonwebtoken::{decode, get_current_timestamp, DecodingKey, Validation};
+use std::{
+    future::Future,
+    pin::Pin,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use tower::{Layer, Service};
+
+/// The [`AuthLayer`] uses the provided [`JwtSecret`] to generate and validate the jwt token
+/// in the requests.
+///
+/// The generated token is inserted into the [`AUTHORIZATION`] header of the request.
+#[derive(Clone, Debug)]
+pub struct AuthLayer {
+    secret: JwtSecret,
+    latency_buffer: u64,
+}
+
+impl AuthLayer {
+    /// Create a new [`AuthLayer`].
+    pub const fn new(secret: JwtSecret) -> Self {
+        Self { secret, latency_buffer: 5000 }
+    }
+
+    /// We use this buffer to perfom an extra check on the `iat` field to prevent sending any
+    /// requests with tokens that are valid now but may not be upon reaching the server.
+    ///
+    /// In milliseconds. Default is 5s.
+    pub const fn with_latency_buffer(self, latency_buffer: u64) -> Self {
+        Self { latency_buffer, ..self }
+    }
+}
+
+impl<S> Layer<S> for AuthLayer {
+    type Service = AuthService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        AuthService::new(inner, self.secret, self.latency_buffer)
+    }
+}
+
+/// A service that generates and validates the jwt token in the requests using the provided secret.
+#[derive(Clone, Debug)]
+pub struct AuthService<S> {
+    inner: S,
+    secret: JwtSecret,
+    /// In milliseconds.
+    latency_buffer: u64,
+    most_recent_claim: Option<Claims>,
+}
+
+impl<S> AuthService<S> {
+    /// Create a new [`AuthService`] with the given inner service.
+    pub const fn new(inner: S, secret: JwtSecret, latency_buffer: u64) -> Self {
+        Self { inner, secret, latency_buffer, most_recent_claim: None }
+    }
+
+    /// Validate the token in the request headers.
+    ///
+    /// Returns `true` if the token is valid and `iat` is beyond the grace buffer.
+    pub fn validate(&self) -> bool {
+        if let Some(claim) = self.most_recent_claim.as_ref() {
+            let curr_secs = get_current_timestamp();
+            if claim.iat.abs_diff(curr_secs) * 1000 <= self.latency_buffer {
+                return false;
+            }
+        }
+
+        true
+    }
+}
+
+impl<S, B, ResBody> Service<Request<B>> for AuthService<S>
+where
+    S: Service<hyper::Request<B>, Response = Response<ResBody>, Error = TransportError>
+        + Clone
+        + Send
+        + Sync
+        + 'static,
+    S::Future: Send,
+    S::Error: std::error::Error + Send + Sync + 'static,
+    B: From<Vec<u8>> + Send + 'static + Clone + Sync,
+    ResBody: hyper::body::Body + Send + 'static,
+    ResBody::Error: std::error::Error + Send + Sync + 'static,
+    ResBody::Data: Send,
+{
+    type Response = Response<ResBody>;
+    type Error = TransportError;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Response<ResBody>, TransportError>> + Send + 'static>>;
+
+    fn poll_ready(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request<B>) -> Self::Future {
+        if self.validate() {
+            Box::pin(self.inner.call(req))
+        } else {
+            // Generate a fresh token from the secret and insert it into the request.
+            match create_token_from_secret(&self.secret) {
+                Ok(token) => {
+                    // Decode the token to get the claims.
+                    let validation = Validation::new(jsonwebtoken::Algorithm::HS256);
+                    let decoding_key = DecodingKey::from_secret(self.secret.as_bytes());
+                    let token_data = decode::<Claims>(&token, &decoding_key, &validation);
+                    self.most_recent_claim = token_data.ok().map(|data| data.claims);
+
+                    let mut req = req;
+                    req.headers_mut().insert(AUTHORIZATION, HeaderValue::from_str(&token).unwrap());
+                    Box::pin(self.inner.call(req))
+                }
+                Err(e) => {
+                    let e = TransportErrorKind::custom(e);
+
+                    Box::pin(async move { Err(e) })
+                }
+            }
+        }
+    }
+}
+
+fn create_token_from_secret(secret: &JwtSecret) -> Result<String, jsonwebtoken::errors::Error> {
+    let token = secret.encode(&Claims {
+        iat: (SystemTime::now().duration_since(UNIX_EPOCH).unwrap() + Duration::from_secs(60))
+            .as_secs(),
+        exp: None,
+    })?;
+
+    Ok(format!("Bearer {}", token))
+}

crates/transport-http/src/layers/mod.rs

@@ -0,0 +1,7 @@
+//! tower http like layer implementations that work over the http::Request type.
+#![cfg(all(not(target_arch = "wasm32"), feature = "hyper"))]
+
+#[cfg(feature = "jwt-auth")]
+mod auth;
+#[cfg(feature = "jwt-auth")]
+pub use auth::{AuthLayer, AuthService};

crates/transport-http/src/lib.rs

@@ -20,6 +20,11 @@ pub use hyper;
 #[cfg(all(not(target_arch = "wasm32"), feature = "hyper"))]
 pub use hyper_util;
 
+#[cfg(all(not(target_arch = "wasm32"), feature = "hyper", feature = "jwt-auth"))]
+mod layers;
+#[cfg(all(not(target_arch = "wasm32"), feature = "hyper", feature = "jwt-auth"))]
+pub use layers::{AuthLayer, AuthService};
+
 #[cfg(all(not(target_arch = "wasm32"), feature = "hyper"))]
 mod hyper_transport;
 #[cfg(all(not(target_arch = "wasm32"), feature = "hyper"))]