jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(deps): bump http, hyper etc. to 1.0; jsonrpsee 0.23 (#7018)

Browse Source 
Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>
This commit is contained in:
DaniPopes
2024-06-11 11:37:49 +02:00
committed by GitHub
parent e9d8cdab49
commit 55317eb004
26 changed files with 554 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

797
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

58
Cargo.toml
View File

@ -119,7 +119,7 @@ members = [
"examples/rpc-db/",
"examples/txpool-tracing/",
"testing/ef-tests/",
"testing/testing-utils"
"testing/testing-utils",
]
default-members = ["bin/reth"]
@ -330,9 +330,15 @@ reth-trie-parallel = { path = "crates/trie/parallel" }
reth-trie-types = { path = "crates/trie/types" }
# revm
revm = { version = "9.0.0", features = [ "std", "secp256k1", "blst", ], default-features = false }
revm-primitives = { version = "4.0.0", features = [ "std", ], default-features = false }
revm-inspectors = { git = "https://github.com/paradigmxyz/evm-inspectors", rev = "53aa2b2" }
revm = { version = "9.0.0", features = [
"std",
"secp256k1",
"blst",
], default-features = false }
revm-primitives = { version = "4.0.0", features = [
"std",
], default-features = false }
revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "5e3058a" }
# eth
alloy-chains = "0.1.15"
@ -341,21 +347,21 @@ alloy-dyn-abi = "0.7.2"
alloy-sol-types = "0.7.2"
alloy-rlp = "0.3.4"
alloy-trie = "0.4"
alloy-rpc-types = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-rpc-types-anvil = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-rpc-types-trace = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-rpc-types-beacon = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-genesis = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-node-bindings = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-provider = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93", default-features = false, features = [
alloy-rpc-types = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-rpc-types-anvil = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-rpc-types-trace = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-rpc-types-beacon = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-genesis = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-node-bindings = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-provider = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d", default-features = false, features = [
"reqwest",
] }
alloy-eips = { git = "https://github.com/alloy-rs/alloy", default-features = false, rev = "cc68b93" }
alloy-signer = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-signer-wallet = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-network = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "cc68b93" }
alloy-eips = { git = "https://github.com/alloy-rs/alloy", default-features = false, rev = "14ed25d" }
alloy-signer = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-signer-wallet = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-network = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "14ed25d" }
# misc
auto_impl = "1"
@ -415,21 +421,25 @@ async-trait = "0.1.68"
futures = "0.3.26"
pin-project = "1.0.12"
futures-util = "0.3.25"
hyper = "0.14.25"
hyper = "1.3"
hyper-util = "0.1.5"
reqwest = { version = "0.12", default-features = false }
tower = "0.4"
tower-http = "0.4"
http = "0.2.8"
http-body = "0.4.5"
tower-http = "0.5"
# p2p
discv5 = "0.6.0"
igd-next = "0.14.3"
# rpc
jsonrpsee = "0.22"
jsonrpsee-core = "0.22"
jsonrpsee-types = "0.22"
jsonrpsee = "0.23"
jsonrpsee-core = "0.23"
jsonrpsee-types = "0.23"
jsonrpsee-http-client = "0.23"
# http
http = "1.0"
http-body = "1.0"
# crypto
secp256k1 = { version = "0.28", default-features = false, features = [

3
crates/consensus/debug-client/src/providers/rpc.rs
View File

@ -3,6 +3,7 @@ use alloy_eips::BlockNumberOrTag;
use alloy_provider::{Provider, ProviderBuilder};
use futures::StreamExt;
use reth_node_core::rpc::types::RichBlock;
use reth_rpc_types::BlockTransactionsKind;
use tokio::sync::mpsc::Sender;
/// Block provider that fetches new blocks from an RPC endpoint using a websocket connection.
@ -32,7 +33,7 @@ impl BlockProvider for RpcBlockProvider {
while let Some(block) = stream.next().await {
let full_block = ws_provider
.get_block_by_hash(block.header.hash.unwrap(), true)
.get_block_by_hash(block.header.hash.unwrap(), BlockTransactionsKind::Full)
.await
.expect("failed to get block")
.expect("block not found");

4
crates/node-core/Cargo.toml
View File

@ -69,7 +69,9 @@ serde.workspace = true
serde_json.workspace = true
# http/rpc
hyper.workspace = true
http.workspace = true
jsonrpsee.workspace = true
tower.workspace = true
# tracing
tracing.workspace = true

52
crates/node-core/src/metrics/prometheus_exporter.rs
View File

@ -2,10 +2,7 @@
use crate::metrics::version_metrics::register_version_metrics;
use eyre::WrapErr;
use hyper::{
service::{make_service_fn, service_fn},
Body, Request, Response, Server,
};
use http::Response;
use metrics::describe_gauge;
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
use metrics_util::layers::{PrefixLayer, Stack};
@ -64,29 +61,36 @@ async fn start_endpoint<F: Hook + 'static>(
hook: Arc<F>,
task_executor: TaskExecutor,
) -> eyre::Result<()> {
let make_svc = make_service_fn(move |_| {
let handle = handle.clone();
let hook = Arc::clone(&hook);
async move {
Ok::<_, Infallible>(service_fn(move |_: Request<Body>| {
let listener =
tokio::net::TcpListener::bind(listen_addr).await.wrap_err("Could not bind to address")?;
task_executor.spawn_with_graceful_shutdown_signal(|signal| async move {
let mut shutdown = signal.ignore_guard();
loop {
let io = tokio::select! {
res = listener.accept() => match res {
Ok((stream, _remote_addr)) => stream,
Err(err) => {
tracing::error!(%err, "failed to accept connection");
continue;
}
},
_ = &mut shutdown => break,
};
let handle = handle.clone();
let hook = hook.clone();
let service = tower::service_fn(move |_| {
(hook)();
let metrics = handle.render();
async move { Ok::<_, Infallible>(Response::new(Body::from(metrics))) }
}))
}
});
async move { Ok::<_, Infallible>(Response::new(metrics)) }
});
let server =
Server::try_bind(&listen_addr).wrap_err("Could not bind to address")?.serve(make_svc);
task_executor.spawn_with_graceful_shutdown_signal(move |signal| async move {
if let Err(error) = server
.with_graceful_shutdown(async move {
let _ = signal.await;
})
.await
{
tracing::error!(%error, "metrics endpoint crashed")
if let Err(error) =
jsonrpsee::server::serve_with_graceful_shutdown(io, service, &mut shutdown).await
{
tracing::error!(%error, "metrics endpoint crashed")
}
}
});

5
crates/optimism/node/Cargo.toml
View File

@ -33,10 +33,7 @@ revm-primitives.workspace = true
# async
async-trait.workspace = true
hyper.workspace = true
reqwest = { workspace = true, default-features = false, features = [
"rustls-tls-native-roots",
] }
reqwest = { workspace = true, features = ["rustls-tls-native-roots"] }
tracing.workspace = true
# misc

3
crates/optimism/node/src/rpc.rs
View File

@ -12,9 +12,6 @@ use std::sync::{atomic::AtomicUsize, Arc};
/// Error type when interacting with the Sequencer
#[derive(Debug, thiserror::Error)]
pub enum SequencerRpcError {
/// Wrapper around a [`hyper::Error`].
#[error(transparent)]
HyperError(#[from] hyper::Error),
/// Wrapper around an [`reqwest::Error`].
#[error(transparent)]
HttpError(#[from] reqwest::Error),

22
crates/rpc/ipc/src/server/mod.rs
View File

@ -91,7 +91,7 @@ where
/// async fn run_server() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
/// let server = Builder::default().build("/tmp/my-uds");
/// let mut module = RpcModule::new(());
/// module.register_method("say_hello", |_, _| "lo")?;
/// module.register_method("say_hello", |_, _, _| "lo")?;
/// let handle = server.start(module).await?;
///
/// // In this example we don't care about doing shutdown so let's it run forever.
@ -390,7 +390,7 @@ where
let rpc_service = self.rpc_middleware.service(RpcService::new(
self.inner.methods.clone(),
max_response_body_size,
self.inner.conn_id as usize,
self.inner.conn_id.into(),
cfg,
));
// an ipc connection needs to handle read+write concurrently
@ -896,7 +896,7 @@ mod tests {
let endpoint = dummy_endpoint();
let server = Builder::default().max_response_body_size(100).build(&endpoint);
let mut module = RpcModule::new(());
module.register_method("anything", |_, _| "a".repeat(101)).unwrap();
module.register_method("anything", |_, _, _| "a".repeat(101)).unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -911,7 +911,7 @@ mod tests {
let endpoint = dummy_endpoint();
let server = Builder::default().max_request_body_size(100).build(&endpoint);
let mut module = RpcModule::new(());
module.register_method("anything", |_, _| "succeed").unwrap();
module.register_method("anything", |_, _, _| "succeed").unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -939,7 +939,7 @@ mod tests {
let endpoint = dummy_endpoint();
let server = Builder::default().max_connections(2).build(&endpoint);
let mut module = RpcModule::new(());
module.register_method("anything", |_, _| "succeed").unwrap();
module.register_method("anything", |_, _, _| "succeed").unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -973,7 +973,7 @@ mod tests {
let server = Builder::default().build(&endpoint);
let mut module = RpcModule::new(());
let msg = r#"{"jsonrpc":"2.0","id":83,"result":"0x7a69"}"#;
module.register_method("eth_chainId", move |_, _| msg).unwrap();
module.register_method("eth_chainId", move |_, _, _| msg).unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -987,7 +987,7 @@ mod tests {
let endpoint = dummy_endpoint();
let server = Builder::default().build(&endpoint);
let mut module = RpcModule::new(());
module.register_method("anything", |_, _| "ok").unwrap();
module.register_method("anything", |_, _, _| "ok").unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -1013,7 +1013,7 @@ mod tests {
let server = Builder::default().build(&endpoint);
let mut module = RpcModule::new(());
let msg = r#"{"admin":"1.0","debug":"1.0","engine":"1.0","eth":"1.0","ethash":"1.0","miner":"1.0","net":"1.0","rpc":"1.0","txpool":"1.0","web3":"1.0"}"#;
module.register_method("rpc_modules", move |_, _| msg).unwrap();
module.register_method("rpc_modules", move |_, _, _| msg).unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());
@ -1036,7 +1036,7 @@ mod tests {
"subscribe_hello",
"s_hello",
"unsubscribe_hello",
|_, pending, tx| async move {
|_, pending, tx, _| async move {
let rx = tx.subscribe();
let stream = BroadcastStream::new(rx);
pipe_from_stream_with_bounded_buffer(pending, stream).await?;
@ -1088,8 +1088,8 @@ mod tests {
let mut module = RpcModule::new(());
let goodbye_msg = r#"{"jsonrpc":"2.0","id":1,"result":"goodbye"}"#;
let hello_msg = r#"{"jsonrpc":"2.0","id":2,"result":"hello"}"#;
module.register_method("say_hello", move |_, _| hello_msg).unwrap();
module.register_method("say_goodbye", move |_, _| goodbye_msg).unwrap();
module.register_method("say_hello", move |_, _, _| hello_msg).unwrap();
module.register_method("say_goodbye", move |_, _, _| goodbye_msg).unwrap();
let handle = server.start(module).await.unwrap();
tokio::spawn(handle.stopped());

37
crates/rpc/ipc/src/server/rpc_service.rs
View File

@ -6,15 +6,15 @@ use jsonrpsee::{
IdProvider,
},
types::{error::reject_too_many_subscriptions, ErrorCode, ErrorObject, Request},
BoundedSubscriptions, ConnectionDetails, MethodCallback, MethodResponse, MethodSink, Methods,
SubscriptionState,
BoundedSubscriptions, ConnectionId, Extensions, MethodCallback, MethodResponse, MethodSink,
Methods, SubscriptionState,
};
use std::sync::Arc;
/// JSON-RPC service middleware.
#[derive(Clone, Debug)]
pub struct RpcService {
conn_id: usize,
conn_id: ConnectionId,
methods: Methods,
max_response_body_size: usize,
cfg: RpcServiceCfg,
@ -39,7 +39,7 @@ impl RpcService {
pub(crate) const fn new(
methods: Methods,
max_response_body_size: usize,
conn_id: usize,
conn_id: ConnectionId,
cfg: RpcServiceCfg,
) -> Self {
Self { methods, max_response_body_size, conn_id, cfg }
@ -58,6 +58,7 @@ impl<'a> RpcServiceT<'a> for RpcService {
let params = req.params();
let name = req.method_name();
let id = req.id().clone();
let extensions = Extensions::new();
match self.methods.method_with_name(name) {
None => {
@ -65,31 +66,17 @@ impl<'a> RpcServiceT<'a> for RpcService {
ResponseFuture::ready(rp)
}
Some((_name, method)) => match method {
MethodCallback::Sync(callback) => {
let rp = (callback)(id, params, max_response_body_size, extensions);
ResponseFuture::ready(rp)
}
MethodCallback::Async(callback) => {
let params = params.into_owned();
let id = id.into_owned();
let fut = (callback)(id, params, conn_id, max_response_body_size);
let fut = (callback)(id, params, conn_id, max_response_body_size, extensions);
ResponseFuture::future(fut)
}
MethodCallback::AsyncWithDetails(callback) => {
let params = params.into_owned();
let id = id.into_owned();
// Note: Add the `Request::extensions` to the connection details when available
// here.
let fut = (callback)(
id,
params,
ConnectionDetails::_new(conn_id),
max_response_body_size,
);
ResponseFuture::future(fut)
}
MethodCallback::Sync(callback) => {
let rp = (callback)(id, params, max_response_body_size);
ResponseFuture::ready(rp)
}
MethodCallback::Subscription(callback) => {
let RpcServiceCfg::CallsAndSubscriptions {
bounded_subscriptions,
@ -110,7 +97,7 @@ impl<'a> RpcServiceT<'a> for RpcService {
subscription_permit: p,
};
let fut = callback(id.clone(), params, sink, conn_state);
let fut = callback(id.clone(), params, sink, conn_state, extensions);
ResponseFuture::future(fut)
} else {
let max = bounded_subscriptions.max();
@ -129,7 +116,7 @@ impl<'a> RpcServiceT<'a> for RpcService {
return ResponseFuture::ready(rp);
};
let rp = callback(id, params, conn_id, max_response_body_size);
let rp = callback(id, params, conn_id, max_response_body_size, extensions);
ResponseFuture::ready(rp)
}
},

2
crates/rpc/rpc-builder/Cargo.toml
View File

@ -30,7 +30,7 @@ reth-engine-primitives.workspace = true
jsonrpsee = { workspace = true, features = ["server"] }
tower-http = { workspace = true, features = ["full"] }
tower = { workspace = true, features = ["full"] }
hyper.workspace = true
http.workspace = true
pin-project.workspace = true
# metrics

7
crates/rpc/rpc-builder/src/auth.rs
View File

@ -1,6 +1,5 @@
use crate::error::{RpcError, ServerKind};
use hyper::header::AUTHORIZATION;
pub use jsonrpsee::server::ServerBuilder;
use http::header::AUTHORIZATION;
use jsonrpsee::{
core::RegisterMethodError,
http_client::{transport::HttpBackend, HeaderMap},
@ -8,7 +7,6 @@ use jsonrpsee::{
Methods,
};
use reth_engine_primitives::EngineTypes;
pub use reth_ipc::server::Builder as IpcServerBuilder;
use reth_rpc::EthSubscriptionIdProvider;
use reth_rpc_api::servers::*;
use reth_rpc_layer::{
@ -19,6 +17,9 @@ use reth_rpc_server_types::constants;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use tower::layer::util::Identity;
pub use jsonrpsee::server::ServerBuilder;
pub use reth_ipc::server::Builder as IpcServerBuilder;
/// Server configuration for the auth server.
#[derive(Debug)]
pub struct AuthServerConfig {

2
crates/rpc/rpc-builder/src/cors.rs
View File

@ -1,4 +1,4 @@
use hyper::{http::HeaderValue, Method};
use http::{HeaderValue, Method};
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
/// Error thrown when parsing cors domains went wrong

16
crates/rpc/rpc-builder/src/lib.rs
View File

@ -160,8 +160,7 @@ use crate::{
metrics::RpcRequestMetrics,
};
use error::{ConflictingModules, RpcError, ServerKind};
use hyper::{header::AUTHORIZATION, HeaderMap};
pub use jsonrpsee::server::ServerBuilder;
use http::{header::AUTHORIZATION, HeaderMap};
use jsonrpsee::{
core::RegisterMethodError,
server::{AlreadyStoppedError, IdProvider, RpcServiceBuilder, Server, ServerHandle},
@ -170,9 +169,6 @@ use jsonrpsee::{
use reth_engine_primitives::EngineTypes;
use reth_evm::ConfigureEvm;
use reth_ipc::server::IpcServer;
pub use reth_ipc::server::{
Builder as IpcServerBuilder, RpcServiceBuilder as IpcRpcServiceBuilder,
};
use reth_network_api::{noop::NoopNetwork, NetworkInfo, Peers};
use reth_provider::{
AccountReader, BlockReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider,
@ -204,14 +200,16 @@ use std::{
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
pub use tower::layer::util::{Identity, Stack};
use tower_http::cors::CorsLayer;
use tracing::{instrument, trace};
pub use crate::eth::{EthConfig, EthHandlers};
// re-export for convenience
pub use jsonrpsee::server::ServerBuilder;
pub use reth_ipc::server::{
Builder as IpcServerBuilder, RpcServiceBuilder as IpcRpcServiceBuilder,
};
pub use reth_rpc_server_types::{constants, RethRpcModule, RpcModuleSelection};
pub use tower::layer::util::{Identity, Stack};
/// Auth server utilities.
pub mod auth;
@ -227,6 +225,7 @@ pub mod error;
/// Eth utils
mod eth;
pub use eth::{EthConfig, EthHandlers};
// Rpc server metrics
mod metrics;
@ -1959,6 +1958,7 @@ impl RpcServerHandle {
client.expect("failed to create http client").into()
}
/// Returns a ws client connected to the server.
pub async fn ws_client(&self) -> Option<jsonrpsee::ws_client::WsClient> {
let url = self.ws_url()?;

6
crates/rpc/rpc-engine-api/Cargo.toml
View File

@ -34,11 +34,11 @@ metrics.workspace = true
# misc
async-trait.workspace = true
thiserror.workspace = true
jsonrpsee-types.workspace = true
jsonrpsee-core.workspace = true
tracing.workspace = true
jsonrpsee-types.workspace = true
serde.workspace = true
thiserror.workspace = true
tracing.workspace = true
[dev-dependencies]
reth-ethereum-engine-primitives.workspace = true

1
crates/rpc/rpc-engine-api/src/engine_api.rs
View File

@ -804,6 +804,7 @@ where
self.inner.metrics.latency.exchange_transition_configuration.record(start.elapsed());
Ok(res?)
}
/// Handler for `engine_getClientVersionV1`
///
/// See also <https://github.com/ethereum/execution-apis/blob/03911ffc053b8b806123f1fc237184b0092a485a/src/engine/identification.md>

7
crates/rpc/rpc-layer/Cargo.toml
View File

@ -14,15 +14,14 @@ workspace = true
alloy-rpc-types-engine.workspace = true
http.workspace = true
hyper.workspace = true
tower.workspace = true
http-body.workspace = true
jsonrpsee-http-client.workspace = true
pin-project.workspace = true
tower.workspace = true
tracing.workspace = true
[dev-dependencies]
hyper = { workspace = true, features = ["client", "tcp"] }
reqwest.workspace = true
assert_matches.workspace = true
tokio = { workspace = true, features = ["macros"] }
tempfile.workspace = true

11
crates/rpc/rpc-layer/src/auth_client_layer.rs
View File

@ -1,11 +1,10 @@
use crate::{Claims, JwtSecret};
use http::HeaderValue;
use hyper::{header::AUTHORIZATION, service::Service};
use http::{header::AUTHORIZATION, HeaderValue};
use std::{
task::{Context, Poll},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tower::Layer;
use tower::{Layer, Service};
/// A layer that adds a new JWT token to every request using `AuthClientService`.
#[derive(Debug)]
@ -41,9 +40,9 @@ impl<S> AuthClientService<S> {
}
}
impl<S, B> Service<hyper::Request<B>> for AuthClientService<S>
impl<S, B> Service<http::Request<B>> for AuthClientService<S>
where
S: Service<hyper::Request<B>>,
S: Service<http::Request<B>>,
{
type Response = S::Response;
type Error = S::Error;
@ -53,7 +52,7 @@ where
self.inner.poll_ready(cx)
}
fn call(&mut self, mut request: hyper::Request<B>) -> Self::Future {
fn call(&mut self, mut request: http::Request<B>) -> Self::Future {
request.headers_mut().insert(AUTHORIZATION, secret_to_bearer_header(&self.secret));
self.inner.call(request)
}

70
crates/rpc/rpc-layer/src/auth_layer.rs
View File

@ -1,6 +1,5 @@
use super::AuthValidator;
use http::{Request, Response};
use http_body::Body;
use jsonrpsee_http_client::{HttpRequest, HttpResponse};
use pin_project::pin_project;
use std::{
future::Future,
@ -65,7 +64,7 @@ where
/// This type is the actual implementation of the middleware. It follows the [`Service`]
/// specification to correctly proxy Http requests to its inner service after headers validation.
#[derive(Debug)]
#[derive(Clone, Debug)]
pub struct AuthService<S, V> {
/// Performs auth validation logics
validator: V,
@ -73,16 +72,15 @@ pub struct AuthService<S, V> {
inner: S,
}
impl<ReqBody, ResBody, S, V> Service<Request<ReqBody>> for AuthService<S, V>
impl<S, V> Service<HttpRequest> for AuthService<S, V>
where
S: Service<Request<ReqBody>, Response = Response<ResBody>>,
V: AuthValidator<ResponseBody = ResBody>,
ReqBody: Body,
ResBody: Body,
S: Service<HttpRequest, Response = HttpResponse>,
V: AuthValidator,
Self: Clone,
{
type Response = Response<ResBody>;
type Response = HttpResponse;
type Error = S::Error;
type Future = ResponseFuture<S::Future, ResBody>;
type Future = ResponseFuture<S::Future>;
/// If we get polled it means that we dispatched an authorized Http request to the inner layer.
/// So we just poll the inner layer ourselves.
@ -96,7 +94,7 @@ where
/// Returns a future that wraps either:
/// - The inner service future for authorized requests
/// - An error Http response in case of authorization errors
fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
fn call(&mut self, req: HttpRequest) -> Self::Future {
match self.validator.validate(req.headers()) {
Ok(_) => ResponseFuture::future(self.inner.call(req)),
Err(res) => ResponseFuture::invalid_auth(res),
@ -106,39 +104,35 @@ where
#[pin_project]
#[allow(missing_debug_implementations)]
pub struct ResponseFuture<F, B> {
pub struct ResponseFuture<F> {
#[pin]
kind: Kind<F, B>,
kind: Kind<F>,
}
impl<F, B> ResponseFuture<F, B>
where
B: Body,
{
impl<F> ResponseFuture<F> {
const fn future(future: F) -> Self {
Self { kind: Kind::Future { future } }
}
const fn invalid_auth(err_res: Response<B>) -> Self {
const fn invalid_auth(err_res: HttpResponse) -> Self {
Self { kind: Kind::Error { response: Some(err_res) } }
}
}
#[pin_project(project = KindProj)]
enum Kind<F, B> {
enum Kind<F> {
Future {
#[pin]
future: F,
},
Error {
response: Option<Response<B>>,
response: Option<HttpResponse>,
},
}
impl<F, B, E> Future for ResponseFuture<F, B>
impl<F, E> Future for ResponseFuture<F>
where
F: Future<Output = Result<Response<B>, E>>,
B: Body,
F: Future<Output = Result<HttpResponse, E>>,
{
type Output = F::Output;
@ -158,12 +152,11 @@ mod tests {
use super::*;
use crate::JwtAuthValidator;
use alloy_rpc_types_engine::{Claims, JwtError, JwtSecret};
use http::{header, Method, Request, StatusCode};
use hyper::{body, Body};
use jsonrpsee::{
server::{RandomStringIdProvider, ServerBuilder, ServerHandle},
RpcModule,
};
use reqwest::{header, StatusCode};
use std::{
net::SocketAddr,
time::{SystemTime, UNIX_EPOCH},
@ -234,25 +227,20 @@ mod tests {
async fn send_request(jwt: Option<String>) -> (StatusCode, String) {
let server = spawn_server().await;
let client = hyper::Client::new();
let client =
reqwest::Client::builder().timeout(std::time::Duration::from_secs(1)).build().unwrap();
let jwt = jwt.unwrap_or_default();
let address = format!("http://{AUTH_ADDR}:{AUTH_PORT}");
let bearer = format!("Bearer {jwt}");
let body = r#"{"jsonrpc": "2.0", "method": "greet_melkor", "params": [], "id": 1}"#;
let req = Request::builder()
.method(Method::POST)
.header(header::AUTHORIZATION, bearer)
let response = client
.post(&format!("http://{AUTH_ADDR}:{AUTH_PORT}"))
.bearer_auth(jwt.unwrap_or_default())
.body(body)
.header(header::CONTENT_TYPE, "application/json")
.uri(address)
.body(Body::from(body))
.send()
.await
.unwrap();
let res = client.request(req).await.unwrap();
let status = res.status();
let body_bytes = body::to_bytes(res.into_body()).await.unwrap();
let body = String::from_utf8(body_bytes.to_vec()).expect("response was not valid utf-8");
let status = response.status();
let body = response.text().await.unwrap();
server.stop().unwrap();
server.stopped().await;
@ -278,7 +266,7 @@ mod tests {
// Create a mock rpc module
let mut module = RpcModule::new(());
module.register_method("greet_melkor", |_, _| "You are the dark lord").unwrap();
module.register_method("greet_melkor", |_, _, _| "You are the dark lord").unwrap();
server.start(module)
}

15
crates/rpc/rpc-layer/src/jwt_validator.rs
View File

@ -1,7 +1,7 @@
use http::{header, HeaderMap, Response, StatusCode};
use tracing::error;
use crate::{AuthValidator, JwtError, JwtSecret};
use http::{header, HeaderMap, Response, StatusCode};
use jsonrpsee_http_client::{HttpBody, HttpResponse};
use tracing::error;
/// Implements JWT validation logics and integrates
/// to an Http [`AuthLayer`][crate::AuthLayer]
@ -22,9 +22,7 @@ impl JwtAuthValidator {
}
impl AuthValidator for JwtAuthValidator {
type ResponseBody = hyper::Body;
fn validate(&self, headers: &HeaderMap) -> Result<(), Response<Self::ResponseBody>> {
fn validate(&self, headers: &HeaderMap) -> Result<(), HttpResponse> {
match get_bearer(headers) {
Some(jwt) => match self.secret.validate(&jwt) {
Ok(_) => Ok(()),
@ -55,14 +53,13 @@ fn get_bearer(headers: &HeaderMap) -> Option<String> {
Some(token.into())
}
fn err_response(err: JwtError) -> Response<hyper::Body> {
let body = hyper::Body::from(err.to_string());
fn err_response(err: JwtError) -> HttpResponse {
// We build a response from an error message.
// We don't cope with headers or other structured fields.
// Then we are safe to "expect" on the result.
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.body(body)
.body(HttpBody::new(err.to_string()))
.expect("This should never happen")
}

8
crates/rpc/rpc-layer/src/lib.rs
View File

@ -8,7 +8,8 @@
#![cfg_attr(not(test), warn(unused_crate_dependencies))]
#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
use http::{HeaderMap, Response};
use http::HeaderMap;
use jsonrpsee_http_client::HttpResponse;
mod auth_client_layer;
mod auth_layer;
@ -24,10 +25,7 @@ pub use jwt_validator::JwtAuthValidator;
/// General purpose trait to validate Http Authorization headers. It's supposed to be integrated as
/// a validator trait into an [`AuthLayer`].
pub trait AuthValidator {
/// Body type of the error response
type ResponseBody;
/// This function is invoked by the [`AuthLayer`] to perform validation on Http headers.
/// The result conveys validation errors in the form of an Http response.
fn validate(&self, headers: &HeaderMap) -> Result<(), Response<Self::ResponseBody>>;
fn validate(&self, headers: &HeaderMap) -> Result<(), HttpResponse>;
}

2
crates/rpc/rpc/src/eth/api/transactions.rs
View File

@ -1734,7 +1734,7 @@ pub(crate) fn build_transaction_receipt_with_block_receipts(
}
let rpc_receipt = reth_rpc_types::Receipt {
status: receipt.success,
status: receipt.success.into(),
cumulative_gas_used: receipt.cumulative_gas_used as u128,
logs,
};

7
crates/rpc/rpc/src/otterscan.rs
View File

@ -158,7 +158,12 @@ where
.drain(page_start..page_end)
.map(|receipt| {
let receipt = receipt.inner.map_inner(|receipt| OtsReceipt {
status: receipt.inner.receipt.status,
status: receipt
.inner
.receipt
.status
.as_eip658()
.expect("ETH API returned pre-EIP-658 status"),
cumulative_gas_used: receipt.inner.receipt.cumulative_gas_used as u64,
logs: None,
logs_bloom: None,

9
crates/tasks/src/shutdown.rs
View File

@ -13,8 +13,6 @@ use std::{
use tokio::sync::oneshot;
/// A Future that resolves when the shutdown event has been fired.
///
/// The [`TaskManager`](crate)
#[derive(Debug)]
pub struct GracefulShutdown {
shutdown: Shutdown,
@ -25,6 +23,13 @@ impl GracefulShutdown {
pub(crate) fn new(shutdown: Shutdown, guard: GracefulShutdownGuard) -> Self {
Self { shutdown, guard: Some(guard) }
}
/// Returns a new shutdown future that is ignores the returned [`GracefulShutdownGuard`].
///
/// This just maps the return value of the future to `()`, it does not drop the guard.
pub fn ignore_guard(self) -> impl Future<Output = ()> + Send + Sync + Unpin + 'static {
self.map(drop)
}
}
impl Future for GracefulShutdown {

5
deny.toml
View File

@ -89,10 +89,9 @@ unknown-registry = "warn"
# in the allow list is encountered
unknown-git = "deny"
allow-git = [
# TODO: remove, see ./Cargo.toml
# TODO: Please avoid adding new entries to this list.
"https://github.com/alloy-rs/alloy",
"https://github.com/foundry-rs/block-explorers",
"https://github.com/bluealloy/revm",
"https://github.com/paradigmxyz/evm-inspectors",
"https://github.com/sigp/discv5",
"https://github.com/paradigmxyz/revm-inspectors",
]

8
examples/beacon-api-sidecar-fetcher/Cargo.toml
View File

@ -11,10 +11,10 @@ reth-node-ethereum.workspace = true
alloy-rpc-types-beacon.workspace = true
serde.workspace = true
serde_json.workspace = true
clap.workspace = true
futures-util.workspace = true
eyre.workspace = true
thiserror.workspace = true
futures-util.workspace = true
reqwest.workspace = true
serde_json.workspace = true
serde.workspace = true
thiserror.workspace = true

5
examples/beacon-api-sse/Cargo.toml
View File

@ -8,10 +8,11 @@ license.workspace = true
[dependencies]
reth.workspace = true
reth-node-ethereum.workspace = true
alloy-rpc-types-beacon.workspace = true
clap.workspace = true
tracing.workspace = true
futures-util.workspace = true
mev-share-sse = { version = "0.3.0", default-features = false }
tokio = { workspace = true, features = ["time"] }
mev-share-sse = { version = "0.2.0" , default-features = false }
tracing.workspace = true