done

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",

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"] }

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<String>,
+    pub http_port: u16,
+}
+
+#[derive(Clone, Debug)]
+pub enum InitInfoState {
+    Waiting(Receiver<InitInfo>),
+    Resolved(Arc<InitInfo>),
+}
+
+#[derive(Clone, Debug)]
+pub struct InitInfoArc {
+    state: Arc<RwLock<InitInfoState>>,
+}
+
+impl InitInfoArc {
+    pub fn new(r: Receiver<InitInfo>) -> Self {
+        InitInfoArc {
+            state: Arc::new(RwLock::new(InitInfoState::Waiting(r))),
+        }
+    }
+    pub async fn wait_for_info(self) -> Arc<InitInfo> {
+        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(_) => (),
+                },
+            }
+        }
+    }
+}

src/lightning_socket.rs

@@ -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<PathBuf>),
-    Resolved(Arc<PathBuf>),
-}
-
-#[derive(Clone, Debug)]
-pub struct LightningSocketArc {
-    state: Arc<RwLock<LightningSocketState>>,
-}
-
-impl LightningSocketArc {
-    pub fn new(r: Receiver<PathBuf>) -> Self {
-        LightningSocketArc {
-            state: Arc::new(RwLock::new(LightningSocketState::Waiting(r))),
-        }
-    }
-    pub fn wait_for_path(self) -> BoxFuture<'static, Arc<PathBuf>> {
-        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()
-    }
-}

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<Output = Result<UnixStream, E>> + Unpin + 'static,
     E: std::error::Error + Send + Sync + 'static,
 >(
     pool: Pool<UnixStream, F, U, E>,
+    init_info_fut: InitInfoArc,
     req: Request<Body>,
 ) -> Result<Response<Body>, 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<UnixStream, F, U, E>,
+    init_info_fut: InitInfoArc,
     req: Request<Body>,
 ) -> Result<Response<Body>, 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 {

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<Option<T>, D::Error>
 where
     T: serde::Deserialize<'de>,
@@ -110,10 +112,21 @@ pub struct RpcError {
     pub data: Option<Value>,
 }
 
-pub fn handle_stdio_rpc(send_side: Sender<PathBuf>) {
+pub struct StdErrWrapper(std::io::Stdin);
+impl std::io::Read for StdErrWrapper {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        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<InitInfo>) {
     // 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<PathBuf>) {
                         }),
                     },
                     "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<PathBuf>) {
         };
         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<PathBuf>, conf: RpcParams) -> Result<(), failure::Error> {
+fn init(send_side: Sender<InitInfo>, 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<LightningInit> 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<u16, D::Error> {
+    #[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: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
-        #[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,
-        })
-    }
-}