server.rs 28.8 KB
Newer Older
1
// Copyright 2019-2021 Parity Technologies (UK) Ltd.
Niklas Adolfsson's avatar
Niklas Adolfsson committed
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
//
// Permission is hereby granted, free of charge, to any
// person obtaining a copy of this software and associated
// documentation files (the "Software"), to deal in the
// Software without restriction, including without
// limitation the rights to use, copy, modify, merge,
// publish, distribute, sublicense, and/or sell copies of
// the Software, and to permit persons to whom the Software
// is furnished to do so, subject to the following
// conditions:
//
// The above copyright notice and this permission notice
// shall be included in all copies or substantial portions
// of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

Maciej Hirsz's avatar
Maciej Hirsz committed
27
use std::future::Future;
28
use std::net::{SocketAddr, TcpListener as StdTcpListener};
Maciej Hirsz's avatar
Maciej Hirsz committed
29
30
31
use std::pin::Pin;
use std::task::{Context, Poll};

32
use crate::response;
33
use crate::response::{internal_error, malformed};
34
use futures_channel::mpsc;
35
36
use futures_util::future::FutureExt;
use futures_util::stream::{StreamExt, TryStreamExt};
37
use hyper::header::{HeaderMap, HeaderValue};
38
use hyper::server::conn::AddrStream;
Maciej Hirsz's avatar
Maciej Hirsz committed
39
40
use hyper::server::{conn::AddrIncoming, Builder as HyperBuilder};
use hyper::service::{make_service_fn, service_fn};
41
use hyper::{Error as HyperError, Method};
Maciej Hirsz's avatar
Maciej Hirsz committed
42
use jsonrpsee_core::error::{Error, GenericTransportError};
43
use jsonrpsee_core::http_helpers::{self, read_body};
44
use jsonrpsee_core::logger::{self, HttpLogger as Logger};
45
use jsonrpsee_core::server::access_control::AccessControl;
46
47
use jsonrpsee_core::server::helpers::{prepare_error, MethodResponse};
use jsonrpsee_core::server::helpers::{BatchResponse, BatchResponseBuilder};
Maciej Hirsz's avatar
Maciej Hirsz committed
48
use jsonrpsee_core::server::resource_limiting::Resources;
49
use jsonrpsee_core::server::rpc_module::{MethodKind, Methods};
50
use jsonrpsee_core::tracing::{rx_log_from_json, rx_log_from_str, tx_log_from_str, RpcTracing};
51
use jsonrpsee_core::TEN_MB_SIZE_BYTES;
52
use jsonrpsee_types::error::{ErrorCode, ErrorObject, BATCHES_NOT_SUPPORTED_CODE, BATCHES_NOT_SUPPORTED_MSG};
53
use jsonrpsee_types::{Id, Notification, Params, Request};
54
use serde_json::value::RawValue;
55
use tokio::net::{TcpListener, ToSocketAddrs};
56
use tracing_futures::Instrument;
Niklas Adolfsson's avatar
Niklas Adolfsson committed
57

58
59
type Notif<'a> = Notification<'a, Option<&'a RawValue>>;

Niklas Adolfsson's avatar
Niklas Adolfsson committed
60
/// Builder to create JSON-RPC HTTP server.
61
#[derive(Debug)]
62
pub struct Builder<L = ()> {
63
	/// Access control based on HTTP headers.
Niklas Adolfsson's avatar
Niklas Adolfsson committed
64
	access_control: AccessControl,
Maciej Hirsz's avatar
Maciej Hirsz committed
65
	resources: Resources,
Niklas Adolfsson's avatar
Niklas Adolfsson committed
66
	max_request_body_size: u32,
67
	max_response_body_size: u32,
68
	batch_requests_supported: bool,
69
70
	/// Custom tokio runtime to run the server on.
	tokio_runtime: Option<tokio::runtime::Handle>,
71
	logger: L,
72
	max_log_length: u32,
73
	health_api: Option<HealthApi>,
Maciej Hirsz's avatar
Maciej Hirsz committed
74
75
76
77
78
}

impl Default for Builder {
	fn default() -> Self {
		Self {
79
			access_control: AccessControl::default(),
Maciej Hirsz's avatar
Maciej Hirsz committed
80
			max_request_body_size: TEN_MB_SIZE_BYTES,
81
			max_response_body_size: TEN_MB_SIZE_BYTES,
82
			batch_requests_supported: true,
Maciej Hirsz's avatar
Maciej Hirsz committed
83
84
			resources: Resources::default(),
			tokio_runtime: None,
85
			logger: (),
86
			max_log_length: 4096,
87
			health_api: None,
Maciej Hirsz's avatar
Maciej Hirsz committed
88
89
		}
	}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
90
91
}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
92
impl Builder {
Maciej Hirsz's avatar
Maciej Hirsz committed
93
94
95
96
97
98
	/// Create a default server builder.
	pub fn new() -> Self {
		Self::default()
	}
}

99
100
impl<L> Builder<L> {
	/// Add a logger to the builder [`Logger`](../jsonrpsee_core/logger/trait.Logger.html).
Maciej Hirsz's avatar
Maciej Hirsz committed
101
102
	///
	/// ```
103
	/// use std::{time::Instant, net::SocketAddr};
104
	/// use hyper::Request;
Maciej Hirsz's avatar
Maciej Hirsz committed
105
	///
106
	/// use jsonrpsee_core::logger::{HttpLogger, Headers, MethodKind, Params};
Maciej Hirsz's avatar
Maciej Hirsz committed
107
108
	/// use jsonrpsee_http_server::HttpServerBuilder;
	///
Maciej Hirsz's avatar
Maciej Hirsz committed
109
	/// #[derive(Clone)]
110
	/// struct MyLogger;
Maciej Hirsz's avatar
Maciej Hirsz committed
111
	///
112
	/// impl HttpLogger for MyLogger {
Maciej Hirsz's avatar
Maciej Hirsz committed
113
114
	///     type Instant = Instant;
	///
115
116
	///     // Called once the HTTP request is received, it may be a single JSON-RPC call
	///     // or batch.
117
	///     fn on_request(&self, _remote_addr: SocketAddr, _request: &Request<hyper::Body>) -> Instant {
Maciej Hirsz's avatar
Maciej Hirsz committed
118
119
	///         Instant::now()
	///     }
Niklas Adolfsson's avatar
Niklas Adolfsson committed
120
	///
121
122
	///     // Called once a single JSON-RPC method call is processed, it may be called multiple times
	///     // on batches.
123
124
	///     fn on_call(&self, method_name: &str, params: Params, kind: MethodKind) {
	///         println!("Call to method: '{}' params: {:?}, kind: {}", method_name, params, kind);
125
126
127
128
129
130
131
	///     }
	///
	///     // Called once a single JSON-RPC call is completed, it may be called multiple times
	///     // on batches.
	///     fn on_result(&self, method_name: &str, success: bool, started_at: Instant) {
	///         println!("Call to '{}' took {:?}", method_name, started_at.elapsed());
	///     }
Maciej Hirsz's avatar
Maciej Hirsz committed
132
	///
Niklas Adolfsson's avatar
Niklas Adolfsson committed
133
	///     // Called the entire JSON-RPC is completed, called on once for both single calls or batches.
134
135
	///     fn on_response(&self, result: &str, started_at: Instant) {
	///         println!("complete JSON-RPC response: {}, took: {:?}", result, started_at.elapsed());
Maciej Hirsz's avatar
Maciej Hirsz committed
136
137
138
	///     }
	/// }
	///
139
	/// let builder = HttpServerBuilder::new().set_logger(MyLogger);
Maciej Hirsz's avatar
Maciej Hirsz committed
140
	/// ```
141
	pub fn set_logger<T: Logger>(self, logger: T) -> Builder<T> {
Maciej Hirsz's avatar
Maciej Hirsz committed
142
		Builder {
143
			access_control: self.access_control,
Maciej Hirsz's avatar
Maciej Hirsz committed
144
			max_request_body_size: self.max_request_body_size,
145
			max_response_body_size: self.max_response_body_size,
146
			batch_requests_supported: self.batch_requests_supported,
Maciej Hirsz's avatar
Maciej Hirsz committed
147
148
			resources: self.resources,
			tokio_runtime: self.tokio_runtime,
149
			logger,
150
			max_log_length: self.max_log_length,
151
			health_api: self.health_api,
Maciej Hirsz's avatar
Maciej Hirsz committed
152
153
154
		}
	}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
155
156
157
158
159
	/// Sets the maximum size of a request body in bytes (default is 10 MiB).
	pub fn max_request_body_size(mut self, size: u32) -> Self {
		self.max_request_body_size = size;
		self
	}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
160

161
162
163
164
165
166
	/// Sets the maximum size of a response body in bytes (default is 10 MiB).
	pub fn max_response_body_size(mut self, size: u32) -> Self {
		self.max_response_body_size = size;
		self
	}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
167
168
169
170
171
	/// Sets access control settings.
	pub fn set_access_control(mut self, acl: AccessControl) -> Self {
		self.access_control = acl;
		self
	}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
172

173
174
175
176
177
178
179
	/// Enables or disables support of [batch requests](https://www.jsonrpc.org/specification#batch).
	/// By default, support is enabled.
	pub fn batch_requests_supported(mut self, supported: bool) -> Self {
		self.batch_requests_supported = supported;
		self
	}

180
181
182
	/// Register a new resource kind. Errors if `label` is already registered, or if the number of
	/// registered resources on this server instance would exceed 8.
	///
183
	/// See the module documentation for [`resource_limiting`](../jsonrpsee_utils/server/resource_limiting/index.html#resource-limiting)
184
	/// for details.
Maciej Hirsz's avatar
Maciej Hirsz committed
185
186
187
188
189
190
	pub fn register_resource(mut self, label: &'static str, capacity: u16, default: u16) -> Result<Self, Error> {
		self.resources.register(label, capacity, default)?;

		Ok(self)
	}

191
192
193
	/// Configure a custom [`tokio::runtime::Handle`] to run the server on.
	///
	/// Default: [`tokio::spawn`]
194
	pub fn custom_tokio_runtime(mut self, rt: tokio::runtime::Handle) -> Self {
195
		self.tokio_runtime = Some(rt);
196
		self
197
198
	}

199
	/// Enable health endpoint.
200
201
202
203
204
205
206
207
	/// Allows you to expose one of the methods under GET /<path> The method will be invoked with no parameters.
	/// Error returned from the method will be converted to status 500 response.
	/// Expects a tuple with (</path>, <rpc-method-name>).
	///
	/// Fails if the path is missing `/`.
	pub fn health_api(mut self, path: impl Into<String>, method: impl Into<String>) -> Result<Self, Error> {
		let path = path.into();

Niklas Adolfsson's avatar
Niklas Adolfsson committed
208
		if !path.starts_with('/') {
209
210
211
			return Err(Error::Custom(format!("Health endpoint path must start with `/` to work, got: {}", path)));
		}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
212
		self.health_api = Some(HealthApi { path, method: method.into() });
213
		Ok(self)
214
215
	}

216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
	/// Finalizes the configuration of the server with customized TCP settings on the socket and on hyper.
	///
	/// ```rust
	/// use jsonrpsee_http_server::HttpServerBuilder;
	/// use socket2::{Domain, Socket, Type};
	/// use std::net::TcpListener;
	///
	/// #[tokio::main]
	/// async fn main() {
	///   let addr = "127.0.0.1:0".parse().unwrap();
	///   let domain = Domain::for_address(addr);
	///   let socket = Socket::new(domain, Type::STREAM, None).unwrap();
	///   socket.set_nonblocking(true).unwrap();
	///
	///   let address = addr.into();
	///   socket.bind(&address).unwrap();
	///   socket.listen(4096).unwrap();
	///
	///   let listener: TcpListener = socket.into();
	///   let local_addr = listener.local_addr().ok();
	///
	///   // hyper does some settings on the provided socket, ensure that nothing breaks our "expected settings".
	///
	///   let listener = hyper::Server::from_tcp(listener)
	///     .unwrap()
	///     .tcp_sleep_on_accept_errors(true)
	///     .tcp_keepalive(None)
	///     .tcp_nodelay(true);
	///
	///   let server = HttpServerBuilder::new().build_from_hyper(listener, addr).unwrap();
	/// }
	/// ```
	pub fn build_from_hyper(
		self,
		listener: hyper::server::Builder<AddrIncoming>,
		local_addr: SocketAddr,
252
	) -> Result<Server<L>, Error> {
253
		Ok(Server {
254
			access_control: self.access_control,
255
256
257
258
			listener,
			local_addr: Some(local_addr),
			max_request_body_size: self.max_request_body_size,
			max_response_body_size: self.max_response_body_size,
259
			batch_requests_supported: self.batch_requests_supported,
260
261
			resources: self.resources,
			tokio_runtime: self.tokio_runtime,
262
			logger: self.logger,
263
			max_log_length: self.max_log_length,
264
			health_api: self.health_api,
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
		})
	}

	/// Finalizes the configuration of the server with customized TCP settings on the socket.
	/// Note, that [`hyper`] might overwrite some of the TCP settings on the socket
	/// if you want full-control of socket settings use [`Builder::build_from_hyper`] instead.
	///
	/// ```rust
	/// use jsonrpsee_http_server::HttpServerBuilder;
	/// use socket2::{Domain, Socket, Type};
	/// use std::time::Duration;
	///
	/// #[tokio::main]
	/// async fn main() {
	///   let addr = "127.0.0.1:0".parse().unwrap();
	///   let domain = Domain::for_address(addr);
	///   let socket = Socket::new(domain, Type::STREAM, None).unwrap();
	///   socket.set_nonblocking(true).unwrap();
	///
	///   let address = addr.into();
	///   socket.bind(&address).unwrap();
	///
	///   socket.listen(4096).unwrap();
	///
	///   let server = HttpServerBuilder::new().build_from_tcp(socket).unwrap();
	/// }
	/// ```
292
	pub fn build_from_tcp(self, listener: impl Into<StdTcpListener>) -> Result<Server<L>, Error> {
293
294
295
296
297
298
299
300
301
302
303
		let listener = listener.into();
		let local_addr = listener.local_addr().ok();

		let listener = hyper::Server::from_tcp(listener)?;

		Ok(Server {
			listener,
			local_addr,
			access_control: self.access_control,
			max_request_body_size: self.max_request_body_size,
			max_response_body_size: self.max_response_body_size,
304
			batch_requests_supported: self.batch_requests_supported,
305
306
			resources: self.resources,
			tokio_runtime: self.tokio_runtime,
307
			logger: self.logger,
308
			max_log_length: self.max_log_length,
309
			health_api: self.health_api,
310
311
312
		})
	}

313
	/// Finalizes the configuration of the server.
314
315
316
317
318
319
320
321
322
323
	///
	/// ```rust
	/// #[tokio::main]
	/// async fn main() {
	///   let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	///   let occupied_addr = listener.local_addr().unwrap();
	///   let addrs: &[std::net::SocketAddr] = &[
	///       occupied_addr,
	///       "127.0.0.1:0".parse().unwrap(),
	///   ];
324
325
	///   assert!(jsonrpsee_http_server::HttpServerBuilder::default().build(occupied_addr).await.is_err());
	///   assert!(jsonrpsee_http_server::HttpServerBuilder::default().build(addrs).await.is_ok());
326
327
	/// }
	/// ```
328
	pub async fn build(self, addrs: impl ToSocketAddrs) -> Result<Server<L>, Error> {
329
		let listener = TcpListener::bind(addrs).await?.into_std()?;
Niklas Adolfsson's avatar
Niklas Adolfsson committed
330
331

		let local_addr = listener.local_addr().ok();
332
333
334
335
336
337
338
339
		let listener = hyper::Server::from_tcp(listener)?.tcp_nodelay(true);

		Ok(Server {
			listener,
			local_addr,
			access_control: self.access_control,
			max_request_body_size: self.max_request_body_size,
			max_response_body_size: self.max_response_body_size,
340
			batch_requests_supported: self.batch_requests_supported,
341
342
			resources: self.resources,
			tokio_runtime: self.tokio_runtime,
343
			logger: self.logger,
344
			max_log_length: self.max_log_length,
345
			health_api: self.health_api,
346
		})
Niklas Adolfsson's avatar
Niklas Adolfsson committed
347
	}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
348
}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
349

350
351
352
353
354
355
#[derive(Debug, Clone)]
struct HealthApi {
	path: String,
	method: String,
}

356
/// Handle used to run or stop the server.
357
#[derive(Debug)]
358
pub struct ServerHandle {
359
	stop_sender: mpsc::Sender<()>,
360
	pub(crate) handle: Option<tokio::task::JoinHandle<()>>,
361
362
}

363
impl ServerHandle {
364
	/// Requests server to stop. Returns an error if server was already stopped.
365
	pub fn stop(mut self) -> Result<tokio::task::JoinHandle<()>, Error> {
366
		let stop = self.stop_sender.try_send(()).map(|_| self.handle.take());
367
368
369
370
		match stop {
			Ok(Some(handle)) => Ok(handle),
			_ => Err(Error::AlreadyStopped),
		}
371
372
373
	}
}

374
375
376
377
378
379
380
381
382
383
384
385
386
impl Future for ServerHandle {
	type Output = ();

	fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
		let handle = match &mut self.handle {
			Some(handle) => handle,
			None => return Poll::Ready(()),
		};

		handle.poll_unpin(cx).map(|_| ())
	}
}

387
388
/// An HTTP JSON RPC server.
#[derive(Debug)]
389
pub struct Server<L = ()> {
Niklas Adolfsson's avatar
Niklas Adolfsson committed
390
391
392
393
394
395
	/// Hyper server.
	listener: HyperBuilder<AddrIncoming>,
	/// Local address
	local_addr: Option<SocketAddr>,
	/// Max request body size.
	max_request_body_size: u32,
396
397
	/// Max response body size.
	max_response_body_size: u32,
398
399
400
401
	/// Max length for logging for request and response
	///
	/// Logs bigger than this limit will be truncated.
	max_log_length: u32,
402
403
	/// Whether batch requests are supported by this server or not.
	batch_requests_supported: bool,
404
	/// Access control.
Niklas Adolfsson's avatar
Niklas Adolfsson committed
405
	access_control: AccessControl,
406
	/// Tracker for currently used resources on the server.
Maciej Hirsz's avatar
Maciej Hirsz committed
407
	resources: Resources,
408
409
	/// Custom tokio runtime to run the server on.
	tokio_runtime: Option<tokio::runtime::Handle>,
410
	logger: L,
411
	health_api: Option<HealthApi>,
Niklas Adolfsson's avatar
Niklas Adolfsson committed
412
}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
413

414
impl<L: Logger> Server<L> {
Niklas Adolfsson's avatar
Niklas Adolfsson committed
415
	/// Returns socket address to which the server is bound.
416
417
	pub fn local_addr(&self) -> Result<SocketAddr, Error> {
		self.local_addr.ok_or_else(|| Error::Custom("Local address not found".into()))
Niklas Adolfsson's avatar
Niklas Adolfsson committed
418
419
	}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
420
	/// Start the server.
421
	pub fn start(mut self, methods: impl Into<Methods>) -> Result<ServerHandle, Error> {
Niklas Adolfsson's avatar
Niklas Adolfsson committed
422
		let max_request_body_size = self.max_request_body_size;
423
		let max_response_body_size = self.max_response_body_size;
424
		let max_log_length = self.max_log_length;
425
		let acl = self.access_control;
426
427
		let (tx, mut rx) = mpsc::channel(1);
		let listener = self.listener;
Maciej Hirsz's avatar
Maciej Hirsz committed
428
		let resources = self.resources;
429
		let logger = self.logger;
430
		let batch_requests_supported = self.batch_requests_supported;
Maciej Hirsz's avatar
Maciej Hirsz committed
431
		let methods = methods.into().initialize_resources(&resources)?;
432
		let health_api = self.health_api;
Niklas Adolfsson's avatar
Niklas Adolfsson committed
433

434
435
		let make_service = make_service_fn(move |conn: &AddrStream| {
			let remote_addr = conn.remote_addr();
Niklas Adolfsson's avatar
Niklas Adolfsson committed
436
			let methods = methods.clone();
437
			let acl = acl.clone();
Maciej Hirsz's avatar
Maciej Hirsz committed
438
			let resources = resources.clone();
439
			let logger = logger.clone();
440
			let health_api = health_api.clone();
Niklas Adolfsson's avatar
Niklas Adolfsson committed
441
442
443

			async move {
				Ok::<_, HyperError>(service_fn(move |request| {
444
					let request_start = logger.on_request(remote_addr, &request);
445

Niklas Adolfsson's avatar
Niklas Adolfsson committed
446
					let methods = methods.clone();
447
					let acl = acl.clone();
Maciej Hirsz's avatar
Maciej Hirsz committed
448
					let resources = resources.clone();
449
					let logger = logger.clone();
450
					let health_api = health_api.clone();
David's avatar
David committed
451
452
453

					// Run some validation on the http request, then read the body and try to deserialize it into one of
					// two cases: a single RPC request or a batch of RPC requests.
Niklas Adolfsson's avatar
Niklas Adolfsson committed
454
					async move {
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
						let keys = request.headers().keys().map(|k| k.as_str());
						let cors_request_headers = http_helpers::get_cors_request_headers(request.headers());

						let host = match http_helpers::read_header_value(request.headers(), "host") {
							Some(origin) => origin,
							None => return Ok(malformed()),
						};
						let maybe_origin = http_helpers::read_header_value(request.headers(), "origin");

						if let Err(e) = acl.verify_host(host) {
							tracing::warn!("Denied request: {:?}", e);
							return Ok(response::host_not_allowed());
						}

						if let Err(e) = acl.verify_origin(maybe_origin, host) {
							tracing::warn!("Denied request: {:?}", e);
							return Ok(response::invalid_allow_origin());
						}

						if let Err(e) = acl.verify_headers(keys, cors_request_headers) {
							tracing::warn!("Denied request: {:?}", e);
							return Ok(response::invalid_allow_headers());
Niklas Adolfsson's avatar
Niklas Adolfsson committed
477
						}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
478

479
480
481
482
483
						// Only `POST` and `OPTIONS` methods are allowed.
						match *request.method() {
							// An OPTIONS request is a CORS preflight request. We've done our access check
							// above so we just need to tell the browser that the request is OK.
							Method::OPTIONS => {
484
								let origin = match maybe_origin {
485
486
487
									Some(origin) => origin,
									None => return Ok(malformed()),
								};
488
489

								let allowed_headers = acl.allowed_headers().to_cors_header_value();
490
491
492
493
494
495
496
497
498
499
500
501
502
								let allowed_header_bytes = allowed_headers.as_bytes();

								let res = hyper::Response::builder()
									.header("access-control-allow-origin", origin)
									.header("access-control-allow-methods", "POST")
									.header("access-control-allow-headers", allowed_header_bytes)
									.body(hyper::Body::empty())
									.unwrap_or_else(|e| {
										tracing::error!("Error forming preflight response: {}", e);
										internal_error()
									});

								Ok(res)
Niklas Adolfsson's avatar
Niklas Adolfsson committed
503
							}
504
505
506
507
508
							// The actual request. If it's a CORS request we need to remember to add
							// the access-control-allow-origin header (despite preflight) to allow it
							// to be read in a browser.
							Method::POST if content_type_is_json(&request) => {
								let origin = return_origin_if_different_from_host(request.headers()).cloned();
509
								let mut res = process_validated_request(ProcessValidatedRequest {
510
									request,
511
									logger,
512
513
514
									methods,
									resources,
									max_request_body_size,
515
									max_response_body_size,
516
									max_log_length,
517
									batch_requests_supported,
518
519
									request_start,
								})
520
521
522
523
								.await?;

								if let Some(origin) = origin {
									res.headers_mut().insert("access-control-allow-origin", origin);
524
								}
525
								Ok(res)
Niklas Adolfsson's avatar
Niklas Adolfsson committed
526
							}
527
528
							Method::GET => match health_api.as_ref() {
								Some(health) if health.path.as_str() == request.uri().path() => {
529
530
									process_health_request(
										health,
531
										logger,
532
533
										methods,
										max_response_body_size,
534
										request_start,
535
536
537
										max_log_length,
									)
									.await
538
539
540
								}
								_ => Ok(response::method_not_allowed()),
							},
541
542
543
							// Error scenarios:
							Method::POST => Ok(response::unsupported_content_type()),
							_ => Ok(response::method_not_allowed()),
David's avatar
David committed
544
						}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
545
546
547
548
					}
				}))
			}
		});
Niklas Adolfsson's avatar
Niklas Adolfsson committed
549

550
551
552
553
554
555
556
557
558
559
		let rt = match self.tokio_runtime.take() {
			Some(rt) => rt,
			None => tokio::runtime::Handle::current(),
		};

		let handle = rt.spawn(async move {
			let server = listener.serve(make_service);
			let _ = server.with_graceful_shutdown(async move { rx.next().await.map_or((), |_| ()) }).await;
		});

560
		Ok(ServerHandle { handle: Some(handle), stop_sender: tx })
Niklas Adolfsson's avatar
Niklas Adolfsson committed
561
562
563
	}
}

564
565
566
567
568
569
570
571
572
573
574
575
576
577
// Checks the origin and host headers. If they both exist, return the origin if it does not match the host.
// If one of them doesn't exist (origin most probably), or they are identical, return None.
fn return_origin_if_different_from_host(headers: &HeaderMap) -> Option<&HeaderValue> {
	if let (Some(origin), Some(host)) = (headers.get("origin"), headers.get("host")) {
		if origin != host {
			Some(origin)
		} else {
			None
		}
	} else {
		None
	}
}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
578
/// Checks that content type of received request is valid for JSON-RPC.
579
580
fn content_type_is_json(request: &hyper::Request<hyper::Body>) -> bool {
	is_json(request.headers().get("content-type"))
Niklas Adolfsson's avatar
Niklas Adolfsson committed
581
582
}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
583
584
585
/// Returns true if the `content_type` header indicates a valid JSON message.
fn is_json(content_type: Option<&hyper::header::HeaderValue>) -> bool {
	match content_type.and_then(|val| val.to_str().ok()) {
586
		Some(content)
Niklas Adolfsson's avatar
Niklas Adolfsson committed
587
588
589
590
591
			if content.eq_ignore_ascii_case("application/json")
				|| content.eq_ignore_ascii_case("application/json; charset=utf-8")
				|| content.eq_ignore_ascii_case("application/json;charset=utf-8") =>
		{
			true
Niklas Adolfsson's avatar
Niklas Adolfsson committed
592
		}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
593
		_ => false,
Niklas Adolfsson's avatar
Niklas Adolfsson committed
594
595
	}
}
596

597
struct ProcessValidatedRequest<L: Logger> {
598
	request: hyper::Request<hyper::Body>,
599
	logger: L,
600
601
602
	methods: Methods,
	resources: Resources,
	max_request_body_size: u32,
603
	max_response_body_size: u32,
604
	max_log_length: u32,
605
	batch_requests_supported: bool,
606
	request_start: L::Instant,
607
608
609
}

/// Process a verified request, it implies a POST request with content type JSON.
610
611
async fn process_validated_request<L: Logger>(
	input: ProcessValidatedRequest<L>,
612
) -> Result<hyper::Response<hyper::Body>, HyperError> {
613
614
	let ProcessValidatedRequest {
		request,
615
		logger,
616
617
618
619
620
621
622
623
624
		methods,
		resources,
		max_request_body_size,
		max_response_body_size,
		max_log_length,
		batch_requests_supported,
		request_start,
	} = input;

625
626
	let (parts, body) = request.into_parts();

627
	let (body, is_single) = match read_body(&parts.headers, body, max_request_body_size).await {
628
		Ok(r) => r,
629
		Err(GenericTransportError::TooLarge) => return Ok(response::too_large(max_request_body_size)),
630
631
632
633
634
635
636
637
638
		Err(GenericTransportError::Malformed) => return Ok(response::malformed()),
		Err(GenericTransportError::Inner(e)) => {
			tracing::error!("Internal error reading request body: {}", e);
			return Ok(response::internal_error());
		}
	};

	// Single request or notification
	if is_single {
639
640
		let call = CallData {
			conn_id: 0,
641
			logger: &logger,
642
643
644
645
646
647
648
			methods: &methods,
			max_response_body_size,
			max_log_length,
			resources: &resources,
			request_start,
		};
		let response = process_single_request(body, call).await;
649
		logger.on_response(&response.result, request_start);
650
651
652
653
654
655
656
657
		Ok(response::ok_response(response.result))
	}
	// Batch of requests or notifications
	else if !batch_requests_supported {
		let err = MethodResponse::error(
			Id::Null,
			ErrorObject::borrowed(BATCHES_NOT_SUPPORTED_CODE, &BATCHES_NOT_SUPPORTED_MSG, None),
		);
658
		logger.on_response(&err.result, request_start);
659
660
661
662
663
664
665
666
		Ok(response::ok_response(err.result))
	}
	// Batch of requests or notifications
	else {
		let response = process_batch_request(Batch {
			data: body,
			call: CallData {
				conn_id: 0,
667
				logger: &logger,
668
669
670
671
672
673
674
675
				methods: &methods,
				max_response_body_size,
				max_log_length,
				resources: &resources,
				request_start,
			},
		})
		.await;
676
		logger.on_response(&response.result, request_start);
677
678
679
		Ok(response::ok_response(response.result))
	}
}
680

681
async fn process_health_request<L: Logger>(
682
	health_api: &HealthApi,
683
	logger: L,
684
685
	methods: Methods,
	max_response_body_size: u32,
686
	request_start: L::Instant,
687
688
689
	max_log_length: u32,
) -> Result<hyper::Response<hyper::Body>, HyperError> {
	let trace = RpcTracing::method_call(&health_api.method);
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
	async {
		tx_log_from_str("HTTP health API", max_log_length);
		let response = match methods.method_with_name(&health_api.method) {
			None => MethodResponse::error(Id::Null, ErrorObject::from(ErrorCode::MethodNotFound)),
			Some((_name, method_callback)) => match method_callback.inner() {
				MethodKind::Sync(callback) => {
					(callback)(Id::Number(0), Params::new(None), max_response_body_size as usize)
				}
				MethodKind::Async(callback) => {
					(callback)(Id::Number(0), Params::new(None), 0, max_response_body_size as usize, None).await
				}
				MethodKind::Subscription(_) | MethodKind::Unsubscription(_) => {
					MethodResponse::error(Id::Null, ErrorObject::from(ErrorCode::InternalError))
				}
			},
		};
706

707
		rx_log_from_str(&response.result, max_log_length);
708
709
		logger.on_result(&health_api.method, response.success, request_start);
		logger.on_response(&response.result, request_start);
710

711
712
713
714
715
716
		if response.success {
			#[derive(serde::Deserialize)]
			struct RpcPayload<'a> {
				#[serde(borrow)]
				result: &'a serde_json::value::RawValue,
			}
717

718
719
720
721
722
			let payload: RpcPayload = serde_json::from_str(&response.result)
				.expect("valid JSON-RPC response must have a result field and be valid JSON; qed");
			Ok(response::ok_response(payload.result.to_string()))
		} else {
			Ok(response::internal_error())
723
		}
724
	}
725
726
	.instrument(trace.into_span())
	.await
727
728
729
}

#[derive(Debug, Clone)]
730
struct Batch<'a, L: Logger> {
731
	data: Vec<u8>,
732
	call: CallData<'a, L>,
733
734
735
}

#[derive(Debug, Clone)]
736
struct CallData<'a, L: Logger> {
737
	conn_id: usize,
738
	logger: &'a L,
739
740
741
742
	methods: &'a Methods,
	max_response_body_size: u32,
	max_log_length: u32,
	resources: &'a Resources,
743
	request_start: L::Instant,
744
745
746
}

#[derive(Debug, Clone)]
747
struct Call<'a, L: Logger> {
748
749
	params: Params<'a>,
	name: &'a str,
750
	call: CallData<'a, L>,
751
752
753
754
755
756
	id: Id<'a>,
}

// Batch responses must be sent back as a single message so we read the results from each
// request in the batch and read the results off of a new channel, `rx_batch`, and then send the
// complete batch response back to the client over `tx`.
757
async fn process_batch_request<L>(b: Batch<'_, L>) -> BatchResponse
758
where
759
	L: Logger,
760
761
762
763
764
765
766
767
768
{
	let Batch { data, call } = b;

	if let Ok(batch) = serde_json::from_slice::<Vec<Request>>(&data) {
		let max_response_size = call.max_response_body_size;
		let batch = batch.into_iter().map(|req| Ok((req, call.clone())));

		let batch_stream = futures_util::stream::iter(batch);

769
		let trace = RpcTracing::batch();
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
		return async {
			let batch_response = batch_stream
				.try_fold(
					BatchResponseBuilder::new_with_limit(max_response_size as usize),
					|batch_response, (req, call)| async move {
						let params = Params::new(req.params.map(|params| params.get()));
						let response = execute_call(Call { name: &req.method, params, id: req.id, call }).await;
						batch_response.append(&response)
					},
				)
				.await;

			match batch_response {
				Ok(batch) => batch.finish(),
				Err(batch_err) => batch_err,
			}
		}
		.instrument(trace.into_span())
		.await;
789
790
791
792
793
	}

	if let Ok(batch) = serde_json::from_slice::<Vec<Notif>>(&data) {
		return if !batch.is_empty() {
			BatchResponse { result: "".to_string(), success: true }
794
		} else {
795
796
797
			BatchResponse::error(Id::Null, ErrorObject::from(ErrorCode::InvalidRequest))
		};
	}
798

799
800
801
802
803
804
805
	// "If the batch rpc call itself fails to be recognized as an valid JSON or as an
	// Array with at least one value, the response from the Server MUST be a single
	// Response object." – The Spec.
	let (id, code) = prepare_error(&data);
	BatchResponse::error(id, ErrorObject::from(code))
}

806
async fn process_single_request<L: Logger>(data: Vec<u8>, call: CallData<'_, L>) -> MethodResponse {
807
808
	if let Ok(req) = serde_json::from_slice::<Request>(&data) {
		let trace = RpcTracing::method_call(&req.method);
809
810
811
812
813
814
815
816
817
		async {
			rx_log_from_json(&req, call.max_log_length);
			let params = Params::new(req.params.map(|params| params.get()));
			let name = &req.method;
			let id = req.id;
			execute_call(Call { name, params, id, call }).await
		}
		.instrument(trace.into_span())
		.await
818
819
	} else if let Ok(req) = serde_json::from_slice::<Notif>(&data) {
		let trace = RpcTracing::notification(&req.method);
820
821
		let span = trace.into_span();
		let _enter = span.enter();
822
		rx_log_from_json(&req, call.max_log_length);
823

824
825
826
827
828
		MethodResponse { result: String::new(), success: true }
	} else {
		let (id, code) = prepare_error(&data);
		MethodResponse::error(id, ErrorObject::from(code))
	}
829
}
830

831
async fn execute_call<L: Logger>(c: Call<'_, L>) -> MethodResponse {
832
	let Call { name, id, params, call } = c;
833
	let CallData { resources, methods, logger, max_response_body_size, max_log_length, conn_id, request_start } = call;
834
835

	let response = match methods.method_with_name(name) {
836
		None => {
837
			logger.on_call(name, params.clone(), logger::MethodKind::Unknown);
838
839
			MethodResponse::error(id, ErrorObject::from(ErrorCode::MethodNotFound))
		}
840
		Some((name, method)) => match &method.inner() {
841
			MethodKind::Sync(callback) => {
842
				logger.on_call(name, params.clone(), logger::MethodKind::MethodCall);
843
844
845
846
847
848
849
850
851
852
853

				match method.claim(name, resources) {
					Ok(guard) => {
						let r = (callback)(id, params, max_response_body_size as usize);
						drop(guard);
						r
					}
					Err(err) => {
						tracing::error!("[Methods::execute_with_resources] failed to lock resources: {:?}", err);
						MethodResponse::error(id, ErrorObject::from(ErrorCode::ServerIsBusy))
					}
854
				}
855
856
			}
			MethodKind::Async(callback) => {
857
				logger.on_call(name, params.clone(), logger::MethodKind::MethodCall);
858
859
860
861
				match method.claim(name, resources) {
					Ok(guard) => {
						let id = id.into_owned();
						let params = params.into_owned();
862

863
864
865
866
867
868
						(callback)(id, params, conn_id, max_response_body_size as usize, Some(guard)).await
					}
					Err(err) => {
						tracing::error!("[Methods::execute_with_resources] failed to lock resources: {:?}", err);
						MethodResponse::error(id, ErrorObject::from(ErrorCode::ServerIsBusy))
					}
869
				}
870
			}
871
			MethodKind::Subscription(_) | MethodKind::Unsubscription(_) => {
872
				logger.on_call(name, params.clone(), logger::MethodKind::Unknown);
873
874
				tracing::error!("Subscriptions not supported on HTTP");
				MethodResponse::error(id, ErrorObject::from(ErrorCode::InternalError))
875
876
877
878
			}
		},
	};

879
	tx_log_from_str(&response.result, max_log_length);
880
	logger.on_result(name, response.success, request_start);
881
	response
882
}