diff --git a/Cargo.lock b/Cargo.lock index faaeaa4c86..206bdcb00e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2066,6 +2066,8 @@ name = "execution_layer" version = "0.1.0" dependencies = [ "async-trait", + "bytes 1.1.0", + "environment", "eth1", "eth2_serde_utils 0.1.0", "futures", @@ -2076,6 +2078,8 @@ dependencies = [ "slog", "tokio", "types", + "warp", + "warp_utils", ] [[package]] diff --git a/beacon_node/execution_layer/Cargo.toml b/beacon_node/execution_layer/Cargo.toml index 7883d02caa..7680c9e831 100644 --- a/beacon_node/execution_layer/Cargo.toml +++ b/beacon_node/execution_layer/Cargo.toml @@ -17,3 +17,7 @@ eth2_serde_utils = { path = "../../consensus/serde_utils" } serde_json = "1.0.58" serde = { version = "1.0.116", features = ["derive"] } eth1 = { path = "../eth1" } +warp = { git = "https://github.com/paulhauner/warp ", branch = "cors-wildcard" } +warp_utils = { path = "../../common/warp_utils" } +environment = { path = "../../lighthouse/environment" } +bytes = "1.1.0" diff --git a/beacon_node/execution_layer/src/engine_api/http.rs b/beacon_node/execution_layer/src/engine_api/http.rs index ff292b0435..eb638b6a5f 100644 --- a/beacon_node/execution_layer/src/engine_api/http.rs +++ b/beacon_node/execution_layer/src/engine_api/http.rs @@ -53,11 +53,9 @@ impl HttpJsonRpc { id: 1, }; - let url: &str = self.url.as_ref(); - let body: JsonResponseBody = self .client - .post(url) + .post(self.url.full.clone()) .timeout(timeout) .header(CONTENT_TYPE, "application/json") .json(&body) @@ -297,3 +295,67 @@ struct JsonForkChoiceUpdatedRequest { head_block_hash: Hash256, finalized_block_hash: Hash256, } + +#[cfg(test)] +mod test { + use super::*; + use crate::test_utils::MockServer; + use std::future::Future; + use std::sync::Arc; + use types::MainnetEthSpec; + + struct Tester { + server: MockServer, + echo_client: Arc, + } + + impl Tester { + pub fn new() -> Self { + let server = MockServer::unit_testing::(); + let echo_url = SensitiveUrl::parse(&format!("{}/echo", server.url())).unwrap(); + let echo_client = Arc::new(HttpJsonRpc::new(echo_url).unwrap()); + + Self { + server, + echo_client, + } + } + + pub async fn assert_request_equals( + self, + request_func: R, + expected_json: serde_json::Value, + ) -> Self + where + R: Fn(Arc) -> F, + F: Future, + { + request_func(self.echo_client.clone()).await; + let request_bytes = self.server.last_echo_request().await; + let request_json: serde_json::Value = + serde_json::from_slice(&request_bytes).expect("request was not valid json"); + if request_json != expected_json { + panic!( + "json mismatch!\n\nobserved: {}\n\nexpected: {}\n\n", + request_json.to_string(), + expected_json.to_string() + ) + } + self + } + } + + #[tokio::test] + async fn forkchoice_updated_request() { + Tester::new() + .assert_request_equals( + |client| async move { + let _ = client + .forkchoice_updated(Hash256::repeat_byte(0), Hash256::repeat_byte(1)) + .await; + }, + json!("meow"), + ) + .await; + } +} diff --git a/beacon_node/execution_layer/src/lib.rs b/beacon_node/execution_layer/src/lib.rs index 589568a51d..465d994866 100644 --- a/beacon_node/execution_layer/src/lib.rs +++ b/beacon_node/execution_layer/src/lib.rs @@ -5,6 +5,7 @@ use slog::Logger; mod engine_api; mod engines; +pub mod test_utils; #[derive(Debug)] pub enum Error { diff --git a/beacon_node/execution_layer/src/test_utils/mod.rs b/beacon_node/execution_layer/src/test_utils/mod.rs new file mode 100644 index 0000000000..1b28d1199a --- /dev/null +++ b/beacon_node/execution_layer/src/test_utils/mod.rs @@ -0,0 +1,175 @@ +use bytes::Bytes; +use environment::null_logger; +use serde::{Deserialize, Serialize}; +use slog::{info, Logger}; +use std::future::Future; +use std::marker::PhantomData; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; +use std::sync::Arc; +use tokio::sync::{oneshot, RwLock}; +use types::EthSpec; +use warp::Filter; + +pub struct MockServer { + _shutdown_tx: oneshot::Sender<()>, + listen_socket_addr: SocketAddr, + last_echo_request: Arc>>, +} + +impl MockServer { + pub fn unit_testing() -> Self { + let last_echo_request = Arc::new(RwLock::new(None)); + + let ctx: Arc> = Arc::new(Context { + config: <_>::default(), + log: null_logger().unwrap(), + last_echo_request: last_echo_request.clone(), + _phantom: PhantomData, + }); + + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + + let shutdown_future = async { + // Ignore the result from the channel, shut down regardless. + let _ = shutdown_rx.await; + }; + + let (listen_socket_addr, server_future) = serve(ctx, shutdown_future).unwrap(); + + tokio::spawn(server_future); + + Self { + _shutdown_tx: shutdown_tx, + listen_socket_addr, + last_echo_request, + } + } + + pub fn url(&self) -> String { + format!( + "http://{}:{}", + self.listen_socket_addr.ip(), + self.listen_socket_addr.port() + ) + } + + pub async fn last_echo_request(&self) -> Bytes { + self.last_echo_request + .write() + .await + .take() + .expect("last echo request is none") + } +} + +#[derive(Debug)] +pub enum Error { + Warp(warp::Error), + Other(String), +} + +impl From for Error { + fn from(e: warp::Error) -> Self { + Error::Warp(e) + } +} + +impl From for Error { + fn from(e: String) -> Self { + Error::Other(e) + } +} + +/// A wrapper around all the items required to spawn the HTTP server. +/// +/// The server will gracefully handle the case where any fields are `None`. +pub struct Context { + pub config: Config, + pub log: Logger, + pub last_echo_request: Arc>>, + pub _phantom: PhantomData, +} + +/// Configuration for the HTTP server. +#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)] +pub struct Config { + pub listen_addr: Ipv4Addr, + pub listen_port: u16, + pub allow_origin: Option, +} + +impl Default for Config { + fn default() -> Self { + Self { + listen_addr: Ipv4Addr::new(127, 0, 0, 1), + listen_port: 0, + allow_origin: None, + } + } +} + +/// Creates a server that will serve requests using information from `ctx`. +/// +/// The server will shut down gracefully when the `shutdown` future resolves. +/// +/// ## Returns +/// +/// This function will bind the server to the provided address and then return a tuple of: +/// +/// - `SocketAddr`: the address that the HTTP server will listen on. +/// - `Future`: the actual server future that will need to be awaited. +/// +/// ## Errors +/// +/// Returns an error if the server is unable to bind or there is another error during +/// configuration. +pub fn serve( + ctx: Arc>, + shutdown: impl Future + Send + Sync + 'static, +) -> Result<(SocketAddr, impl Future), Error> { + let config = &ctx.config; + let log = ctx.log.clone(); + + // Configure CORS. + let cors_builder = { + let builder = warp::cors() + .allow_method("GET") + .allow_headers(vec!["Content-Type"]); + + warp_utils::cors::set_builder_origins( + builder, + config.allow_origin.as_deref(), + (config.listen_addr, config.listen_port), + )? + }; + + let inner_ctx = ctx.clone(); + let routes = warp::post() + .and(warp::path("echo")) + .and(warp::body::bytes()) + .and(warp::any().map(move || inner_ctx.clone())) + .and_then(|bytes: Bytes, ctx: Arc>| async move { + *ctx.last_echo_request.write().await = Some(bytes.clone()); + Ok::<_, warp::reject::Rejection>( + warp::http::Response::builder().status(200).body(bytes), + ) + }) + // Add a `Server` header. + .map(|reply| warp::reply::with_header(reply, "Server", "lighthouse-mock-execution-client")) + .with(cors_builder.build()); + + let (listening_socket, server) = warp::serve(routes).try_bind_with_graceful_shutdown( + SocketAddrV4::new(config.listen_addr, config.listen_port), + async { + shutdown.await; + }, + )?; + + info!( + log, + "Metrics HTTP server started"; + "listen_address" => listening_socket.to_string(), + ); + + Ok((listening_socket, server)) +}