use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;
use std::{future::Future, pin::Pin, sync::Arc};
use anyhow::{Context, Result};
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming as IncomingBody;
use hyper::server::conn::http1;
use hyper::Method;
use hyper::{body::Bytes, Response};
use hyper::{service::Service, Request};
use hyper_util::rt::TokioIo;
use serde_json::Value;
use tokio::io::AsyncWriteExt;
use tokio::{net::TcpStream, sync::Mutex};
use tokio_rustls::TlsAcceptor;
use crate::config::Config;
use crate::doslimit::{ConnectionLimits, GlobalLimits};
use crate::errors::rpc_invalid_request;
use crate::query::Query;
use crate::rpc::daemon::dummycom::DummyCom;
use crate::rpc::daemon::ssl_utils::create_tls_acceptor;
use crate::rpc::daemon::to_rpc_error;
use crate::rpc::rpcstats::RpcStats;
use crate::signal::NetworkNotifier;
use crate::thread::{run_with_timeout, spawn_task};
use crate::util::Channel;
use super::connection::Connection;
use super::server::ConnectionId;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;
fn parse_post_request(body: Bytes) -> Result<(String, Value, Vec<Value>)> {
let req_json: Value = serde_json::from_slice(&body).context("invalid JSON format")?;
let empty_params = json!([]);
let dummy_id = json!(0);
let method = req_json
.get("method")
.context("missing method")?
.as_str()
.context("method not a string")?;
let params = req_json.get("params").unwrap_or(&empty_params);
let id = req_json.get("id").unwrap_or(&dummy_id);
let params = match params.as_array() {
Some(p) => p,
None => bail!("Expected 'params' to be an array"),
};
Ok((method.to_string(), id.to_owned(), params.to_vec()))
}
#[derive(Clone)]
struct RestReqService {
connection: Arc<Mutex<Connection>>,
}
impl Service<Request<IncomingBody>> for RestReqService {
type Response = Response<Full<Bytes>>;
type Error = hyper::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn call(&self, req: Request<IncomingBody>) -> Self::Future {
let conn = self.connection.clone();
let error_id = json!(-1);
let to_response = |json_resp: Value| -> hyper::Result<Response<Full<Bytes>>> {
let mut response = Response::new(Full::new(Bytes::from(json_resp.to_string())));
response.headers_mut().insert(
hyper::header::CONTENT_TYPE,
hyper::header::HeaderValue::from_static("application/json"),
);
Ok(response)
};
let res = Box::pin(async move {
if req.method() != Method::POST {
return to_response(to_rpc_error(
rpc_invalid_request("Request method not supported (use POST)".into()),
&error_id,
));
}
let body = req.into_body().collect().await?.to_bytes();
let (method, id, params) = match parse_post_request(body) {
Ok(v) => v,
Err(e) => {
return to_response(to_rpc_error(rpc_invalid_request(e.to_string()), &error_id))
}
};
to_response(conn.lock().await.single_req(&method, ¶ms, &id).await)
});
Box::pin(res)
}
}
async fn run_rest_request(stream: TcpStream, conn: Connection) {
let io = TokioIo::new(stream);
let service = RestReqService {
connection: Arc::new(Mutex::new(conn)),
};
if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
warn!("Failed to serve connection: {:?}", err);
}
}
async fn run_https_request(stream: TcpStream, conn: Connection, ssl_acceptor: TlsAcceptor) {
let service = RestReqService {
connection: Arc::new(Mutex::new(conn)),
};
let tls_stream = match ssl_acceptor.accept(stream).await {
Ok(tls_stream) => tls_stream,
Err(err) => {
warn!("Failed to accept HTTPS connection: {:?}", err);
return;
}
};
let io = TokioIo::new(tls_stream);
if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
warn!("Failed to serve HTTPS connection: {:?}", err);
}
}
#[allow(clippy::too_many_arguments)]
pub async fn start_http_server(
conn_acceptor: Channel<Option<(TcpStream, SocketAddr)>>,
global_limits: Arc<GlobalLimits>,
config: Arc<Config>,
query: Arc<Query>,
stats: Arc<RpcStats>,
connection_limits: ConnectionLimits,
network_notifier: Arc<NetworkNotifier>,
idle_timeout: u64,
subscription_manager: Arc<ScripthashSubscriptions>,
) -> Result<()> {
while let Some(conn) = conn_acceptor.receiver().await.recv().await {
let (mut stream, addr) = match conn {
Some(c) => c,
None => break,
};
let global_limits = global_limits.clone();
let mut connections = match global_limits.inc_connection(&addr.ip()).await {
Err(e) => {
trace!("[{}] dropping http peer - {}", addr, e);
let _ = stream.shutdown().await;
continue;
}
Ok(n) => n,
};
if connections != (0, 0) {
debug!(
"[{} http] connected peer ({:?} out of {:?} connection slots used)",
addr,
connections,
global_limits.connection_limits(),
);
}
let dummycon = DummyCom {};
let conn_id = ConnectionId::new();
let subscription_manager_cpy = subscription_manager.clone();
let conn = Connection::new(
config.clone(),
query.clone(),
addr,
stats.clone(),
connection_limits,
Box::new(dummycon),
network_notifier.clone(),
conn_id,
subscription_manager_cpy,
);
spawn_task("http_peer", async move {
let res = run_with_timeout(
Duration::from_secs(idle_timeout),
run_rest_request(stream, conn),
)
.await;
if res.is_err() {
debug!(
"[{} http] disconnected due to timeout ({}s)",
addr, idle_timeout
);
}
match global_limits.dec_connection(&addr.ip()).await {
Ok(n) => connections = n,
Err(e) => error!("{}", e),
};
debug!(
"[{} http] disconnected peer ({:?} out of {:?} connection slots used)",
addr,
connections,
global_limits.connection_limits(),
);
Ok(())
});
}
info!("HTTP connections no longer accepted");
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn start_https_server(
conn_acceptor: Channel<Option<(TcpStream, SocketAddr)>>,
cert_file: PathBuf,
key_file: PathBuf,
global_limits: Arc<GlobalLimits>,
config: Arc<Config>,
query: Arc<Query>,
stats: Arc<RpcStats>,
connection_limits: ConnectionLimits,
network_notifier: Arc<NetworkNotifier>,
idle_timeout: u64,
subscription_manager: Arc<ScripthashSubscriptions>,
) -> Result<()> {
let ssl_acceptor = create_tls_acceptor(&cert_file, &key_file).await?;
while let Some(conn) = conn_acceptor.receiver().await.recv().await {
let (mut stream, addr) = match conn {
Some(c) => c,
None => break,
};
let global_limits = global_limits.clone();
let mut connections = match global_limits.inc_connection(&addr.ip()).await {
Err(e) => {
trace!("[{}] dropping https peer - {}", addr, e);
let _ = stream.shutdown().await;
continue;
}
Ok(n) => n,
};
if connections != (0, 0) {
debug!(
"[{} https] connected peer ({:?} out of {:?} connection slots used)",
addr,
connections,
global_limits.connection_limits(),
);
}
let dummycon = DummyCom {};
let conn_id = ConnectionId::new();
let subscription_manager_cpy = subscription_manager.clone();
let conn = Connection::new(
config.clone(),
query.clone(),
addr,
stats.clone(),
connection_limits,
Box::new(dummycon),
network_notifier.clone(),
conn_id,
subscription_manager_cpy,
);
let ssl_acceptor = ssl_acceptor.clone();
spawn_task("https_peer", async move {
let res = run_with_timeout(
Duration::from_secs(idle_timeout),
run_https_request(stream, conn, ssl_acceptor),
)
.await;
if res.is_err() {
debug!(
"[{} https] disconnected due to timeout ({}s)",
addr, idle_timeout
);
}
match global_limits.dec_connection(&addr.ip()).await {
Ok(n) => connections = n,
Err(e) => error!("{}", e),
};
debug!(
"[{} https] disconnected peer ({:?} out of {:?} connection slots used)",
addr,
connections,
global_limits.connection_limits(),
);
Ok(())
});
}
info!("HTTPS connections no longer accepted");
Ok(())
}