pairing: Attach signer to the pairing process

Signer now receives requests from the scheduler for further processing.
The signer processes a ApprovePairingRequest and responds to the
scheduler.

Signed-off-by: Peter Neuroth <[email protected]>

libs/gl-client-py/tests/test_pairing.py

@@ -1,13 +1,18 @@
+import time
 from fixtures import *
 from glclient.pairing import NewDeviceClient, AttestationDeviceClient
+from glclient import Credentials
 
 @pytest.fixture
 def attestation_device(clients):
     c = clients.new()
     c.register()
     yield c
 
-def test_pairing_session(attestation_device, creds):
+def test_pairing_session(sclient, signer, creds):
+    # Run the signer in the background
+    signer.run_in_thread()
+
     name = "new_device"
     desc = "my_description"
     restrs = "method^list"
@@ -19,13 +24,14 @@ def test_pairing_session(attestation_device, creds):
     m = next(session_iter)
     assert(m.data)
 
-    # # register an "attestation device"
-    # res = sclient.register(signer)
-    # ac = AttestationDevicePairingClient(auth=res.auth)
+    # register attestation device.
+    res = sclient.register(signer)
+    sclient.schedule()
+    creds = Credentials.as_device().from_bytes(res.creds).build()
+    ac = AttestationDeviceClient(creds=creds)
 
     # check for pairing data.
     session_id = m.data.split(':')[1]
-    ac = AttestationDeviceClient(creds=attestation_device.creds())
     m = ac.get_pairing_data(session_id)
     assert(m.session_id)
     assert(m.csr)
@@ -38,7 +44,7 @@ def test_pairing_session(attestation_device, creds):
     # and with our rune.
     ac.approve_pairing(
         m.session_id,
-        attestation_device.node_id,
+        signer.node_id(),
         m.device_name,
         m.restrs
     )
@@ -51,6 +57,9 @@ def test_pairing_session(attestation_device, creds):
     # assert(m.rune) fixme: enable once we pass back a rune during the tests.
     assert(m.creds)
 
+    signer.shutdown()
+    # FIXME: add a blocking shutdown call that waits for the signer to shutdown.
+    time.sleep(2)
 
 def test_paring_data_validation(attestation_device, creds):
     """A simple test to ensure that data validation works as intended.

libs/gl-client/Cargo.toml

@@ -2,10 +2,7 @@
 name = "gl-client"
 version = "0.1.9"
 edition = "2018"
-authors = [
-        "Christian Decker",
-        "The Greenlight Team"
-]
+authors = ["Christian Decker", "The Greenlight Team"]
 description = "Client library for Greenlight, and basis for language bindings."
 repository = "https://github.com/Blockstream/greenlight"
 license = "MIT"
@@ -32,15 +29,19 @@ picky-asn1-der = "0.4"
 pin-project = "1.1.3"
 prost = "0.11"
 prost-derive = "0.11"
-reqwest = {version="^0.11", features=["json", "rustls-tls-native-roots"], default-features = false}
+reqwest = { version = "^0.11", features = [
+        "json",
+        "rustls-tls-native-roots",
+], default-features = false }
 ring = "~0.16.20"
 runeauth = "0.1"
 rustls-pemfile = "1.0.3"
 sha256 = "1.1.4"
 tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1"
 tonic = { version = "^0.8", features = ["tls", "transport"] }
 tower = { version = "0.4" }
-rcgen = { version = "0.10.0", features = ["pem", "x509-parser"]}
+rcgen = { version = "0.10.0", features = ["pem", "x509-parser"] }
 tempfile = "3.3.0"
 url = "2.4.0"
 serde = { version = "1", features = [ "derive" ] }
@@ -60,9 +61,9 @@ futures = "0.3.28"
 async-trait = "0.1.72"
 
 rand = "0.8.5"
-uuid = {version = "1.4.0", features=["serde"]}
+uuid = { version = "1.4.0", features = ["serde"] }
 time = { version = "0.3", features = ["macros"] }
 
 [build-dependencies]
 tonic-build = "^0.8"
-serde = { version = "1", features = [ "derive" ] }
+serde = { version = "1", features = ["derive"] }

libs/gl-client/src/pairing/attestation_device.rs

@@ -14,7 +14,7 @@ use picky::{pem::Pem, x509::Csr};
 use picky_asn1_x509::{PublicKey, SubjectPublicKeyInfo};
 use ring::{
     rand,
-    signature::{self, EcdsaKeyPair},
+    signature::{self, EcdsaKeyPair, KeyPair},
 };
 
 use rustls_pemfile as pemfile;
@@ -139,6 +139,7 @@ impl Client<Connected> {
                 restrs: restrs.to_string(),
                 sig: sig,
                 rune: self.rune.clone(),
+                pubkey: kp.public_key().as_ref().to_vec(),
             })
             .await?
             .into_inner())

libs/gl-client/src/signer/mod.rs

@@ -1,5 +1,9 @@
 use crate::credentials::Credentials;
 use crate::pb::scheduler::{scheduler_client::SchedulerClient, NodeInfoRequest, UpgradeRequest};
+use crate::pb::scheduler::{
+    signer_request, signer_response, ApprovePairingRequest, ApprovePairingResponse, SignerResponse,
+};
+use crate::pb::PendingRequest;
 /// The core signer system. It runs in a dedicated thread or using the
 /// caller thread, streaming incoming requests, verifying them,
 /// signing if ok, and then shipping the response to the node.
@@ -19,13 +23,15 @@ use lightning_signer::node::NodeServices;
 use lightning_signer::policy::filter::FilterRule;
 use lightning_signer::util::crypto_utils;
 use log::{debug, info, trace, warn};
+use ring::signature::{UnparsedPublicKey, ECDSA_P256_SHA256_FIXED};
 use runeauth::{Condition, MapChecker, Restriction, Rune, RuneError};
 use std::collections::HashMap;
 use std::convert::{TryFrom, TryInto};
 use std::sync::Arc;
 use std::sync::Mutex;
 use tokio::sync::{broadcast, mpsc};
 use tokio::time::{sleep, Duration};
+use tokio_stream::wrappers::ReceiverStream;
 use tonic::transport::{Endpoint, Uri};
 use tonic::{Code, Request};
 use vls_protocol::msgs::{DeBolt, HsmdInitReplyV4};
@@ -246,7 +252,6 @@ impl Signer {
         requests: Vec<crate::pb::PendingRequest>,
     ) -> Vec<Result<crate::pb::PendingRequest, anyhow::Error>> {
         // Filter out requests lacking a required field. They are unverifiable anyway.
-        use ring::signature::{UnparsedPublicKey, ECDSA_P256_SHA256_FIXED};
         // Todo: partition results to provide more detailed errors.
         requests
             .into_iter()
@@ -623,7 +628,7 @@ impl Signer {
             &scheduler_uri
         );
 
-        let channel = Endpoint::from_shared(scheduler_uri)?
+        let channel = Endpoint::from_shared(scheduler_uri.clone())?
             .tls_config(self.tls.inner.clone())?
             .tcp_keepalive(Some(crate::TCP_KEEPALIVE))
             .http2_keep_alive_interval(crate::TCP_KEEPALIVE)
@@ -668,9 +673,10 @@ impl Signer {
 
         let shutdown = self.shutdown_connector(shutdown, tx.clone());
         let node_runner = self.run_forever_node(tx.subscribe(), scheduler.clone());
+        let scheduler_runner = self.run_forever_scheduler(tx.subscribe(), scheduler.clone());
 
-        let _ = tokio::join!(node_runner, shutdown);
-        todo!()
+        let _ = tokio::join!(node_runner, scheduler_runner, shutdown);
+        Ok(())
     }
 
     async fn shutdown_connector(
@@ -742,6 +748,150 @@ impl Signer {
         Ok(())
     }
 
+    async fn run_forever_scheduler(
+        &self,
+        mut shutdown: broadcast::Receiver<()>,
+        mut scheduler: SchedulerClient<tonic::transport::Channel>,
+    ) -> Result<(), anyhow::Error> {
+        loop {
+            let (sender, rx) = mpsc::channel(1);
+            let outbound = ReceiverStream::new(rx);
+            let inbound_future = scheduler.signer_requests_stream(outbound);
+
+            let mut stream = tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    debug!("Received the signal to exit the signer loop");
+                    return Ok(());
+                }
+                stream = inbound_future => match stream {
+                    Ok(s) => s.into_inner(),
+                    Err(e) => {
+                        debug!("Failed to start stream: {}", e);
+                        sleep(Duration::from_secs(5)).await;
+                        continue;
+                    },
+                },
+            };
+
+            debug!("Starting to stream signer requests from scheduler");
+
+            loop {
+                tokio::select! {
+                    biased;
+                    _ = shutdown.recv() => {
+                        debug!("Received the signal to exit the signer loop");
+                        return Ok(());
+                    },
+                    msg = stream.message() => match msg {
+                        Ok(Some(msg)) => {
+                            let req_id = msg.request_id;
+                            debug!("Processing scheduler request {}", req_id);
+                            match msg.request {
+                                Some(signer_request::Request::ApprovePairing(req)) => self.process_pairing_approval(req_id, req, sender.clone()).await,
+                                None => {
+                                    debug!("Received an empty signing request");
+                                }
+                            };
+                        },
+                        Ok(None) => {
+                            debug!("End of stream, this should not happen by the server");
+                            break;
+                        },
+                        Err(e) => {
+                            debug!("Got an error from the scheduler {}", e);
+                            break;
+                        },
+                    },
+                };
+            }
+        }
+    }
+
+    async fn process_pairing_approval(
+        &self,
+        req_id: u32,
+        req: ApprovePairingRequest,
+        stream: mpsc::Sender<SignerResponse>,
+    ) -> () {
+        let mut data = vec![];
+        data.put(req.session_id.as_bytes());
+        data.put_u64(req.timestamp);
+        data.put(&req.node_id[..]);
+        data.put(req.device_name.as_bytes());
+        data.put(req.restrs.as_bytes());
+
+        // Check that the signature matches
+        let pk = UnparsedPublicKey::new(&ECDSA_P256_SHA256_FIXED, req.pubkey.clone());
+        if pk.verify(&data, &req.sig).is_err() {
+            debug!("Got an invalid signature processing pairing approval");
+            return;
+        }
+
+        // Decode rune as we expect it to be bytes in the pending request.
+        let rune = match general_purpose::URL_SAFE.decode(req.rune.clone()) {
+            Ok(r) => r,
+            Err(e) => {
+                debug!("Could not decode rune processing pairing approval {}", e);
+                return;
+            }
+        };
+
+        // Check that the rune matches
+        match self.verify_rune(PendingRequest {
+            request: vec![],
+            uri: "/cln.Node/ApprovePairing".to_string(),
+            signature: req.sig,
+            pubkey: req.pubkey,
+            timestamp: req.timestamp,
+            rune,
+        }) {
+            Ok(_) => (),
+            Err(e) => {
+                debug!(
+                    "Got an invalid rune {} processing pairing approval {}",
+                    req.rune, e
+                );
+                return;
+            }
+        };
+
+        let restrs: Vec<Vec<&str>> = req
+            .restrs
+            .split('&')
+            .map(|s| s.split('|').collect::<Vec<&str>>())
+            .collect();
+
+        // Create the rune that approves pairing
+        let rune = match self.create_rune(None, restrs) {
+            Ok(r) => r,
+            Err(e) => {
+                debug!("Could not create rune during pairing approval {}", e);
+                return;
+            }
+        };
+
+        match stream
+            .send(SignerResponse {
+                request_id: req_id,
+                response: Some(signer_response::Response::ApprovePairing(
+                    ApprovePairingResponse {
+                        session_id: req.session_id,
+                        node_id: req.node_id,
+                        rune: rune,
+                    },
+                )),
+            })
+            .await
+        {
+            Ok(_) => (),
+            Err(e) => debug!(
+                "Could not respond to stream during pairing approval {:?}",
+                e
+            ),
+        };
+    }
+
     // TODO See comment on `sign_device_key`.
     pub fn sign_challenge(&self, challenge: Vec<u8>) -> Result<Vec<u8>, anyhow::Error> {
         if challenge.len() != 32 {
@@ -983,8 +1133,8 @@ impl From<StartupMessage> for crate::pb::scheduler::StartupMessage {
 
 #[cfg(test)]
 mod tests {
-    use crate::tls;
     use crate::pb;
+    use crate::tls;
 
     use super::*;
 
@@ -1080,8 +1230,12 @@ mod tests {
     /// on the public key "pubkey=<public-key-of-devices-tls-cert>".
     #[test]
     fn test_rune_expects_pubkey() {
-        let signer =
-            Signer::new(vec![0u8; 32], Network::Bitcoin, tls::TlsConfig::new().unwrap()).unwrap();
+        let signer = Signer::new(
+            vec![0u8; 32],
+            Network::Bitcoin,
+            tls::TlsConfig::new().unwrap(),
+        )
+        .unwrap();
 
         let alt = "pubkey=112233";
         let wrong_alt = "pubkey^112233";
@@ -1103,21 +1257,29 @@ mod tests {
 
     #[test]
     fn test_rune_expansion() {
-        let signer =
-            Signer::new(vec![0u8; 32], Network::Bitcoin,  tls::TlsConfig::new().unwrap()).unwrap();
+        let signer = Signer::new(
+            vec![0u8; 32],
+            Network::Bitcoin,
+            tls::TlsConfig::new().unwrap(),
+        )
+        .unwrap();
         let rune = "wjEjvKoFJToMLBv4QVbJpSbMoGFlnYVxs8yy40PIBgs9MC1nbDAmcHVia2V5PTAwMDAwMA==";
 
         let new_rune = signer
             .create_rune(Some(rune), vec![vec!["method^get"]])
             .unwrap();
-        let rs = Rune::from_base64(&new_rune).unwrap().to_string();
-        assert!(rs.contains("0-gl0&pubkey=000000&method^get"))
+        let stream = Rune::from_base64(&new_rune).unwrap().to_string();
+        assert!(stream.contains("0-gl0&pubkey=000000&method^get"))
     }
 
     #[test]
     fn test_rune_checks_method() {
-        let signer =
-            Signer::new(vec![0u8; 32], Network::Bitcoin,  tls::TlsConfig::new().unwrap()).unwrap();
+        let signer = Signer::new(
+            vec![0u8; 32],
+            Network::Bitcoin,
+            tls::TlsConfig::new().unwrap(),
+        )
+        .unwrap();
 
         // This is just a placeholder public key, could also be a different one;
         let pubkey = signer.node_id();