diff --git a/Cargo.lock b/Cargo.lock index abd2ae0..6a0285a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -34,6 +34,12 @@ dependencies = [ "libc", ] +[[package]] +name = "base64" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b41b7ea54a0c9d92199de89e20e58d49f02f8e699814ef3fdf266f6f748d15c7" + [[package]] name = "bitflags" version = "1.2.1" @@ -50,6 +56,7 @@ checksum = "130aac562c0dd69c56b3b1cc8ffd2e17be31d0b6c25b61c96b76231aa23e39e1" name = "c-lightning-http-plugin" version = "0.1.0" dependencies = [ + "base64", "crossbeam-channel", "failure", "futures", @@ -96,9 +103,9 @@ dependencies = [ [[package]] name = "failure" -version = "0.1.6" +version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8273f13c977665c5db7eb2b99ae520952fe5ac831ae4cd09d80c4c7042b5ed9" +checksum = "b8529c2421efa3066a5cbd8063d2244603824daccb6936b079010bb2aa89464b" dependencies = [ "backtrace", "failure_derive", @@ -106,9 +113,9 @@ dependencies = [ [[package]] name = "failure_derive" -version = "0.1.6" +version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0bc225b78e0391e4b8683440bf2e63c2deeeb2ce5189eab46e2b68c6d3725d08" +checksum = "030a733c8287d6213886dd487564ff5c8f6aae10278b3588ed177f9d18f8d231" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 6477dbb..2ba5e0c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,10 +7,11 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +base64 = "0.11.0" crossbeam-channel = "0.4.2" -failure = "0.1.6" +failure = "0.1.7" futures = { version = "0.3.4", features = ["async-await"] } -hyper = { version = "0.13.2", features = ["stream"] } +hyper = { version = "0.13.3", features = ["stream"] } lazy_async_pool = "0.3.0" lazy_static = "1.4.0" serde = { version = "1.0", features = ["derive"] } diff --git a/src/init_info.rs b/src/init_info.rs new file mode 100644 index 0000000..504ec2e --- /dev/null +++ b/src/init_info.rs @@ -0,0 +1,49 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use crossbeam_channel::Receiver; +use tokio::sync::RwLock; + +#[derive(Clone, Debug)] +pub struct InitInfo { + pub socket_path: PathBuf, + pub auth_header: Option, + pub http_port: u16, +} + +#[derive(Clone, Debug)] +pub enum InitInfoState { + Waiting(Receiver), + Resolved(Arc), +} + +#[derive(Clone, Debug)] +pub struct InitInfoArc { + state: Arc>, +} + +impl InitInfoArc { + pub fn new(r: Receiver) -> Self { + InitInfoArc { + state: Arc::new(RwLock::new(InitInfoState::Waiting(r))), + } + } + pub async fn wait_for_info(self) -> Arc { + loop { + let guard = self.state.read().await; + match &*guard { + InitInfoState::Resolved(ref path) => return path.clone(), + InitInfoState::Waiting(receiver) => match receiver.try_recv() { + Ok(ii) => { + let arc_ii = Arc::new(ii); + drop(guard); // turns out this is important + let mut guard = self.state.write().await; + *guard = InitInfoState::Resolved(arc_ii.clone()); + return arc_ii; + } + Err(_) => (), + }, + } + } + } +} diff --git a/src/lightning_socket.rs b/src/lightning_socket.rs deleted file mode 100644 index 1d4e11d..0000000 --- a/src/lightning_socket.rs +++ /dev/null @@ -1,45 +0,0 @@ -use std::path::PathBuf; -use std::sync::Arc; - -use crossbeam_channel::Receiver; -use futures::future::BoxFuture; -use futures::future::FutureExt; -use tokio::sync::RwLock; - -#[derive(Clone, Debug)] -pub enum LightningSocketState { - Waiting(Receiver), - Resolved(Arc), -} - -#[derive(Clone, Debug)] -pub struct LightningSocketArc { - state: Arc>, -} - -impl LightningSocketArc { - pub fn new(r: Receiver) -> Self { - LightningSocketArc { - state: Arc::new(RwLock::new(LightningSocketState::Waiting(r))), - } - } - pub fn wait_for_path(self) -> BoxFuture<'static, Arc> { - async move { - let guard = self.state.read().await; - match &*guard { - LightningSocketState::Resolved(ref path) => path.clone(), - LightningSocketState::Waiting(receiver) => match receiver.try_recv() { - Ok(pb) => { - let arc_pb = Arc::new(pb); - drop(guard); - let mut guard = self.state.write().await; - *guard = LightningSocketState::Resolved(arc_pb.clone()); - arc_pb - } - Err(_) => self.clone().wait_for_path().await, - }, - } - } - .boxed() - } -} diff --git a/src/main.rs b/src/main.rs index b1241dd..815384f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,10 +13,10 @@ use tokio::net::UnixStream; use crate::async_io::RpcResponseStream; use crate::async_io::TokioCompatAsyncRead; -use crate::lightning_socket::LightningSocketArc; +use crate::init_info::InitInfoArc; mod async_io; -mod lightning_socket; +mod init_info; mod rpc; type BoxedByteStream = Box< @@ -26,16 +26,36 @@ type BoxedByteStream = Box< + Send, >; +async fn handle_auth( + init_info_fut: InitInfoArc, + auth: Option<&hyper::header::HeaderValue>, +) -> bool { + if let (Some(received), Some(expected)) = + (auth, &init_info_fut.wait_for_info().await.auth_header) + { + received == expected + } else { + false + } +} + async fn handle_inner< F: Fn() -> U + Send + Sync + 'static, U: Future> + Unpin + 'static, E: std::error::Error + Send + Sync + 'static, >( pool: Pool, + init_info_fut: InitInfoArc, req: Request, ) -> Result, Error> { match req.method() { &Method::POST => { + if !handle_auth(init_info_fut, req.headers().get("Authorization")).await { + return Response::builder().header("Content-Type", "application/json") + .status(hyper::StatusCode::UNAUTHORIZED) + .body(Body::from("{\"id\":null,\"jsonrpc\":\"2.0\",\"error\":{\"code\":5,\"message\":\"Unauthorized\"}}")) + .map_err(Error::from); + } let mut ustream = pool.get().await?; let body = req.into_body(); tokio::io::copy( @@ -71,9 +91,10 @@ async fn handle< E: std::error::Error + Send + Sync + 'static, >( pool: Pool, + init_info_fut: InitInfoArc, req: Request, ) -> Result, Error> { - match handle_inner(pool, req).await { + match handle_inner(pool, init_info_fut, req).await { Err(e) => Response::builder() .header("Content-Type", "application/json") .status(hyper::StatusCode::INTERNAL_SERVER_ERROR) @@ -81,7 +102,7 @@ async fn handle< id: crate::rpc::JsonRpcV2Id::Null, jsonrpc: Default::default(), result: crate::rpc::RpcResult::Error(crate::rpc::RpcError { - code: serde_json::Number::from(5), + code: serde_json::Number::from(0), message: "internal server error", data: Some(serde_json::Value::String(format!("{}", e))), }), @@ -94,31 +115,35 @@ async fn handle< #[tokio::main] async fn main() { // Construct our SocketAddr to listen on... - let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); - let (send_side, recv_side) = crossbeam_channel::bounded(1); - // fork thread for stdio - let lightning_socket_fut = LightningSocketArc::new(recv_side); + std::thread::spawn(move || { + crate::rpc::handle_stdio_rpc(send_side); + }); + // fork thread for stdio + let init_info_fut = InitInfoArc::new(recv_side); + + let init_info_fut_cap = init_info_fut.clone(); let pool = Pool::new(0, move || { - let lightning_socket_fut = lightning_socket_fut.clone(); - async move { UnixStream::connect(&*lightning_socket_fut.wait_for_path().await).await } + let init_info_fut = init_info_fut_cap.clone(); + async move { UnixStream::connect(&*init_info_fut.wait_for_info().await.socket_path).await } .boxed() }); // And a MakeService to handle each connection... - let handler = move |req| handle((&pool).clone(), req); + let init_info_fut_cap = init_info_fut.clone(); + let handler = move |req| handle((&pool).clone(), init_info_fut_cap.clone(), req); let make_service = make_service_fn(|_conn| { let handler = handler.clone(); futures::future::ok::<_, Error>(service_fn(handler)) }); - // Then bind and serve... - let server = Server::bind(&addr).serve(make_service); + let port = init_info_fut.wait_for_info().await.http_port; - std::thread::spawn(move || { - crate::rpc::handle_stdio_rpc(send_side); - }); + // Then bind and serve... + let server = Server::bind(&SocketAddr::from(([127, 0, 0, 1], port))).serve(make_service); + + eprintln!("Serving RPC on port {}", port); // And run forever... if let Err(e) = server.await { diff --git a/src/rpc.rs b/src/rpc.rs index cdb32c9..3a7b398 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -4,6 +4,8 @@ use crossbeam_channel::Sender; use serde_json::StreamDeserializer; use serde_json::Value; +use crate::init_info::InitInfo; + fn deserialize_some<'de, T, D>(deserializer: D) -> Result, D::Error> where T: serde::Deserialize<'de>, @@ -110,10 +112,21 @@ pub struct RpcError { pub data: Option, } -pub fn handle_stdio_rpc(send_side: Sender) { +pub struct StdErrWrapper(std::io::Stdin); +impl std::io::Read for StdErrWrapper { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + use std::io::Write; + let n = self.0.read(buf)?; + std::io::stderr().write_all(&buf[..n])?; + std::io::stderr().flush()?; + Ok(n) + } +} + +pub fn handle_stdio_rpc(send_side: Sender) { // create serde stream let req_stream: StreamDeserializer<_, RpcReq> = - StreamDeserializer::new(serde_json::de::IoRead::new(std::io::stdin())); + StreamDeserializer::new(serde_json::de::IoRead::new(StdErrWrapper(std::io::stdin()))); // for request in stream for e_req in req_stream { let rpc_result = match e_req { @@ -135,7 +148,26 @@ pub fn handle_stdio_rpc(send_side: Sender) { }), }, "getmanifest" => RpcResult::Result(serde_json::json!({ - "options": [], + "options": [ + { + "name": "http-user", + "type": "string", + "default": "lightning", + "description": "Basic-Auth user header for http authentication" + }, + { + "name": "http-pass", + "type": "string", + "default": "", + "description": "Basic-Auth password header for http authentication, not setting this will result in requests being rejected" + }, + { + "name": "http-port", + "type": "int", + "default": 8080, + "description": "Http port for web server listening" + } + ], "rpcmethods": [], "subscriptions": [], "hooks": [], @@ -168,49 +200,100 @@ pub fn handle_stdio_rpc(send_side: Sender) { }; serde_json::to_writer(std::io::stdout(), &rpc_result) .unwrap_or_else(|e| eprintln!("error writing rpc response: {}", e)); + match rpc_result.result { + RpcResult::Error(e) => eprintln!("{:?}", e), + _ => (), + }; print!("\n\n"); } } -fn init(send_side: Sender, conf: RpcParams) -> Result<(), failure::Error> { +fn init(send_side: Sender, conf: RpcParams) -> Result<(), failure::Error> { let arg0 = match conf { RpcParams::ByPosition(mut a) => a .pop() .ok_or(failure::format_err!("no arguments supplied"))?, RpcParams::ByName(a) => serde_json::Value::Object(a), }; - let conf: LightningConfig = serde_json::from_value(arg0)?; + let conf: LightningInit = serde_json::from_value(arg0)?; send_side - .send(conf.lightning_dir.join(conf.rpc_file)) - .unwrap_or_default(); // ignore send error: means the reciever has already received and been dropped + .send(conf.into()) + .unwrap_or_else(|e| eprintln!("SEND ERROR: {}", e)); // ignore send error: means the reciever has already received and been dropped Ok(()) } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct LightningInit { + options: LightningOptions, + configuration: LightningConfig, +} + +impl From for InitInfo { + fn from(li: LightningInit) -> Self { + InitInfo { + socket_path: li + .configuration + .lightning_dir + .join(li.configuration.rpc_file), + auth_header: if li.options.http_pass.is_empty() { + None + } else { + Some(format!( + "Basic {}", + base64::encode(&format!( + "{}:{}", + li.options.http_user, li.options.http_pass + )) + )) + }, + http_port: li.options.http_port, + } + } +} + +fn default_user() -> String { + "lightning".to_owned() +} + +fn default_pass() -> String { + "".to_owned() +} + +fn default_port() -> u16 { + 8080 +} + +#[derive(Clone, Debug, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct LightningOptions { + #[serde(default = "default_user")] + http_user: String, + #[serde(default = "default_pass")] + http_pass: String, + #[serde(default = "default_port")] + #[serde(deserialize_with = "deser_str_num")] + http_port: u16, +} + +fn deser_str_num<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result { + #[derive(serde::Deserialize)] + #[serde(untagged)] + enum StrNum { + Str(String), + Num(u16), + } + let sn: StrNum = serde::Deserialize::deserialize(deserializer)?; + Ok(match sn { + StrNum::Str(s) => s.parse().map_err(|e| serde::de::Error::custom(e))?, + StrNum::Num(n) => n, + }) +} + +#[derive(Clone, Debug, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] pub struct LightningConfig { lightning_dir: PathBuf, rpc_file: String, startup: bool, } - -impl<'de> serde::Deserialize<'de> for LightningConfig { - fn deserialize>(d: D) -> Result { - #[derive(serde::Deserialize)] - struct Complete { - configuration: LightningConfigDefault, - } - #[derive(serde::Deserialize)] - #[serde(rename_all = "kebab-case")] - pub struct LightningConfigDefault { - lightning_dir: PathBuf, - rpc_file: String, - startup: bool, - } - let complete = Complete::deserialize(d)?; - Ok(LightningConfig { - lightning_dir: complete.configuration.lightning_dir, - rpc_file: complete.configuration.rpc_file, - startup: complete.configuration.startup, - }) - } -}