Merge pull request #59 from stakwork/policy-setter

Policy setter

broker/src/init.rs

@@ -1,71 +0,0 @@
-use crate::ChannelRequest;
-use bitcoin::Network;
-use sphinx_key_parser as parser;
-use sphinx_key_parser::MsgDriver;
-use tokio::sync::{mpsc, oneshot};
-use vls_protocol::model::Secret;
-use vls_protocol::{msgs, serde_bolt::WireString};
-use vls_proxy::util::{read_allowlist, read_integration_test_seed};
-
-pub fn blocking_connect(tx: mpsc::Sender<ChannelRequest>, network: Network) {
-    let init_msg_2 = crate::init::make_init_msg(network).expect("couldnt make init msg");
-    let (reply_tx, reply_rx) = oneshot::channel();
-    // Send a request to the MQTT handler to send to signer
-    let request = ChannelRequest {
-        message: init_msg_2,
-        reply_tx,
-    };
-    tx.blocking_send(request).expect("could not blocking send");
-    let res = reply_rx.blocking_recv().expect("couldnt receive");
-    let reply = parser::response_from_bytes(res.reply, 0).expect("couldnt parse init receive");
-    println!("REPLY {:?}", reply);
-}
-
-pub async fn _connect(tx: mpsc::Sender<ChannelRequest>, network: Network) {
-    let init_msg_2 = crate::init::make_init_msg(network).expect("could make init msg");
-    let (reply_tx, reply_rx) = oneshot::channel();
-    // Send a request to the MQTT handler to send to signer
-    let request = ChannelRequest {
-        message: init_msg_2,
-        reply_tx,
-    };
-    let _ = tx.send(request).await;
-    let res = reply_rx.await.expect("couldnt receive");
-    let reply = parser::response_from_bytes(res.reply, 0).expect("could parse init receive");
-    println!("REPLY {:?}", reply);
-}
-
-pub fn make_init_msg(network: Network) -> anyhow::Result<Vec<u8>> {
-    let allowlist = read_allowlist()
-        .into_iter()
-        .map(|s| WireString(s.as_bytes().to_vec()))
-        .collect::<Vec<_>>();
-    let seed = if network == Network::Bitcoin {
-        Some(Secret([
-            0x8c, 0xe8, 0x62, 0xab, 0xd5, 0x6b, 0xb4, 0x6a, 0x61, 0x7f, 0xaf, 0x13, 0x50, 0xc1,
-            0xca, 0xf5, 0xb1, 0xee, 0x02, 0x97, 0xbf, 0xf3, 0xb8, 0xc9, 0x56, 0x63, 0x58, 0x9f,
-            0xec, 0x8c, 0x45, 0x79,
-        ]))
-    } else {
-        read_integration_test_seed()
-            .map(|s| Secret(s))
-            .or(Some(Secret([1; 32])))
-    };
-    // FIXME remove this
-    log::info!("allowlist {:?} seed {:?}", allowlist, seed);
-    let init = msgs::HsmdInit2 {
-        derivation_style: 0,
-        network_name: WireString(network.to_string().as_bytes().to_vec()),
-        dev_seed: seed,
-        dev_allowlist: allowlist,
-    };
-    let sequence = 0;
-    let mut md = MsgDriver::new_empty();
-    msgs::write_serial_request_header(&mut md, sequence, 0)?;
-    msgs::write(&mut md, init)?;
-    Ok(md.bytes())
-    // msgs::read_serial_response_header(&mut serial, sequence)?;
-    // let init_reply: msgs::HsmdInit2Reply = msgs::read_message(&mut serial)?;
-    // log::info!("init reply {:?}", init_reply);
-    // Ok(())
-}

broker/src/main.rs

@@ -1,6 +1,5 @@
 #![feature(once_cell)]
 mod chain_tracker;
-mod init;
 mod mqtt;
 mod run_test;
 mod unix_fd;

parser/Cargo.toml

@@ -6,7 +6,10 @@ edition = "2021"
 [dependencies]
 vls-protocol = { git = "https://gitlab.com/lightning-signer/validating-lightning-signer.git" }
 serde = { version = "1.0", default-features = false }
+rmp-serde = "1.1.0"
 serde_bolt = { version = "0.2", default-features = false }
+sphinx-auther = "0.1.9"
+anyhow = "1"
 
 [features]
 default = ["std"]

parser/src/control.rs

@@ -0,0 +1,57 @@
+use serde::{Deserialize, Serialize};
+use sphinx_auther::nonce;
+use sphinx_auther::secp256k1::{PublicKey, SecretKey};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum ControlMessage {
+    Nonce,
+    QueryPolicy,
+    UpdatePolicy(Policy),
+    Ota(OtaParams),
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum ControlResponse {
+    Nonce(u64),
+    PolicyCurrent(Policy),
+    PolicyUpdated(Policy),
+    OtaConfirm(OtaParams),
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Policy {
+    pub sats_per_day: u64,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct OtaParams {
+    pub version: u64,
+    pub url: String,
+}
+
+// u64 is the nonce. Each signature must have a higher nonce
+pub struct Controller(SecretKey, PublicKey, u64);
+
+impl Controller {
+    pub fn new(sk: SecretKey, pk: PublicKey, nonce: u64) -> Self {
+        Self(sk, pk, nonce)
+    }
+    pub fn build_msg(&mut self, msg: ControlMessage) -> anyhow::Result<Vec<u8>> {
+        let data = rmp_serde::to_vec(&msg)?;
+        let ret = nonce::build_msg(&data, &self.0, self.2)?;
+        self.2 = self.2 + 1;
+        Ok(ret)
+    }
+    pub fn build_response(&self, msg: ControlResponse) -> anyhow::Result<Vec<u8>> {
+        Ok(rmp_serde::to_vec(&msg)?)
+    }
+    pub fn parse_msg(&mut self, input: &[u8]) -> anyhow::Result<ControlMessage> {
+        let msg = nonce::parse_msg(input, &self.1, self.2)?;
+        let ret = rmp_serde::from_slice(&msg)?;
+        self.2 = self.2 + 1;
+        Ok(ret)
+    }
+    pub fn parse_response(&self, input: &[u8]) -> anyhow::Result<ControlResponse> {
+        Ok(rmp_serde::from_slice(input)?)
+    }
+}

parser/src/lib.rs

@@ -1,63 +1,11 @@
+pub mod control;
+
 use serde::ser;
 use std::cmp::min;
 use std::io;
 use vls_protocol::msgs::{self, DeBolt, Message};
 use vls_protocol::serde_bolt::{Error, Read, Result, Write};
 
-pub fn raw_request_from_bytes(
-    message: Vec<u8>,
-    sequence: u16,
-    dbid: u64,
-) -> vls_protocol::Result<Vec<u8>> {
-    let mut md = MsgDriver::new_empty();
-    msgs::write_serial_request_header(&mut md, sequence, dbid)?;
-    msgs::write_vec(&mut md, message)?;
-    Ok(md.bytes())
-}
-
-pub fn request_from_msg<T: ser::Serialize + DeBolt>(
-    msg: T,
-    sequence: u16,
-    dbid: u64,
-) -> vls_protocol::Result<Vec<u8>> {
-    let mut md = MsgDriver::new_empty();
-    msgs::write_serial_request_header(&mut md, sequence, dbid)?;
-    msgs::write(&mut md, msg)?;
-    Ok(md.bytes())
-}
-
-pub fn raw_response_from_msg<T: ser::Serialize + DeBolt>(
-    msg: T,
-    sequence: u16,
-) -> vls_protocol::Result<Vec<u8>> {
-    let mut m = MsgDriver::new_empty();
-    msgs::write_serial_response_header(&mut m, sequence)?;
-    msgs::write(&mut m, msg)?;
-    Ok(m.bytes())
-}
-
-pub fn request_from_bytes<T: DeBolt>(msg: Vec<u8>) -> vls_protocol::Result<(T, u16, u64)> {
-    let mut m = MsgDriver::new(msg);
-    let (sequence, dbid) = msgs::read_serial_request_header(&mut m)?;
-    let reply: T = msgs::read_message(&mut m)?;
-    Ok((reply, sequence, dbid))
-}
-
-pub fn raw_response_from_bytes(
-    res: Vec<u8>,
-    expected_sequence: u16,
-) -> vls_protocol::Result<Vec<u8>> {
-    let mut m = MsgDriver::new(res);
-    msgs::read_serial_response_header(&mut m, expected_sequence)?;
-    Ok(msgs::read_raw(&mut m)?)
-}
-
-pub fn response_from_bytes(res: Vec<u8>, expected_sequence: u16) -> vls_protocol::Result<Message> {
-    let mut m = MsgDriver::new(res);
-    msgs::read_serial_response_header(&mut m, expected_sequence)?;
-    Ok(msgs::read(&mut m)?)
-}
-
 pub struct MsgDriver(Vec<u8>);
 
 impl MsgDriver {
@@ -112,6 +60,60 @@ impl Write for MsgDriver {
     }
 }
 
+pub fn raw_request_from_bytes(
+    message: Vec<u8>,
+    sequence: u16,
+    dbid: u64,
+) -> vls_protocol::Result<Vec<u8>> {
+    let mut md = MsgDriver::new_empty();
+    msgs::write_serial_request_header(&mut md, sequence, dbid)?;
+    msgs::write_vec(&mut md, message)?;
+    Ok(md.bytes())
+}
+
+pub fn request_from_msg<T: ser::Serialize + DeBolt>(
+    msg: T,
+    sequence: u16,
+    dbid: u64,
+) -> vls_protocol::Result<Vec<u8>> {
+    let mut md = MsgDriver::new_empty();
+    msgs::write_serial_request_header(&mut md, sequence, dbid)?;
+    msgs::write(&mut md, msg)?;
+    Ok(md.bytes())
+}
+
+pub fn raw_response_from_msg<T: ser::Serialize + DeBolt>(
+    msg: T,
+    sequence: u16,
+) -> vls_protocol::Result<Vec<u8>> {
+    let mut m = MsgDriver::new_empty();
+    msgs::write_serial_response_header(&mut m, sequence)?;
+    msgs::write(&mut m, msg)?;
+    Ok(m.bytes())
+}
+
+pub fn request_from_bytes<T: DeBolt>(msg: Vec<u8>) -> vls_protocol::Result<(T, u16, u64)> {
+    let mut m = MsgDriver::new(msg);
+    let (sequence, dbid) = msgs::read_serial_request_header(&mut m)?;
+    let reply: T = msgs::read_message(&mut m)?;
+    Ok((reply, sequence, dbid))
+}
+
+pub fn raw_response_from_bytes(
+    res: Vec<u8>,
+    expected_sequence: u16,
+) -> vls_protocol::Result<Vec<u8>> {
+    let mut m = MsgDriver::new(res);
+    msgs::read_serial_response_header(&mut m, expected_sequence)?;
+    Ok(msgs::read_raw(&mut m)?)
+}
+
+pub fn response_from_bytes(res: Vec<u8>, expected_sequence: u16) -> vls_protocol::Result<Message> {
+    let mut m = MsgDriver::new(res);
+    msgs::read_serial_response_header(&mut m, expected_sequence)?;
+    Ok(msgs::read(&mut m)?)
+}
+
 #[cfg(test)]
 mod tests {
     use crate::MsgDriver;

signer/src/derive.rs

@@ -0,0 +1,13 @@
+use vls_core::{
+    bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey},
+    bitcoin::Network,
+    signer::derive::{key_derive, KeyDerivationStyle},
+};
+use vls_protocol_signer::lightning_signer as vls_core;
+
+pub fn node_keys(network: &Network, seed: &[u8]) -> (PublicKey, SecretKey) {
+    let style = KeyDerivationStyle::Native;
+    let deriver = key_derive(style, network.clone());
+    let ctx = Secp256k1::new();
+    deriver.node_keys(seed, &ctx)
+}

signer/src/lib.rs

@@ -1,3 +1,4 @@
+mod derive;
 mod randomstartingtime;
 
 use lightning_signer::bitcoin::blockdata::constants::ChainHash;
@@ -8,20 +9,19 @@ use lightning_signer::policy::simple_validator::{make_simple_policy, SimpleValid
 use lightning_signer::util::clock::StandardClock;
 use lightning_signer::util::velocity::{VelocityControlIntervalType, VelocityControlSpec};
 use randomstartingtime::RandomStartingTimeFactory;
-pub use vls_protocol_signer::lightning_signer;
-pub use vls_protocol_signer::vls_protocol;
-// use lightning_signer::persist::DummyPersister;
 use std::sync::Arc;
 use vls_protocol::model::PubKey;
 use vls_protocol::msgs::{self, read_serial_request_header, write_serial_response_header, Message};
 use vls_protocol::serde_bolt::WireString;
 use vls_protocol_signer::handler::{Handler, RootHandler};
+pub use vls_protocol_signer::lightning_signer;
 use vls_protocol_signer::lightning_signer::bitcoin::Network;
 use vls_protocol_signer::lightning_signer::wallet::Wallet;
+pub use vls_protocol_signer::vls_protocol;
 
-pub use sphinx_key_parser::MsgDriver;
+pub use derive::node_keys as derive_node_keys;
+pub use sphinx_key_parser::{control, MsgDriver};
 pub use sphinx_key_persister::FsPersister;
-
 pub struct InitResponse {
     pub root_handler: RootHandler,
     pub init_reply: Vec<u8>,
@@ -29,6 +29,23 @@ pub struct InitResponse {
 
 pub const ROOT_STORE: &str = "/sdcard/store";
 
+pub fn set_policies(
+    root_handler: &RootHandler,
+    network: Network,
+    sats_per_day: u64,
+) -> anyhow::Result<()> {
+    let mut policy = make_simple_policy(network);
+    policy.filter = PolicyFilter::new_permissive();
+    let velocity_spec = VelocityControlSpec {
+        limit: sats_per_day,
+        interval_type: VelocityControlIntervalType::Daily,
+    };
+    policy.global_velocity_control = velocity_spec;
+    let validator_factory = Arc::new(SimpleValidatorFactory::new_with_policy(policy));
+    root_handler.node.set_validator_factory(validator_factory);
+    Ok(())
+}
+
 pub fn init(bytes: Vec<u8>, network: Network) -> anyhow::Result<InitResponse> {
     // let persister: Arc<dyn Persist> = Arc::new(DummyPersister);
     let mut md = MsgDriver::new(bytes);
@@ -48,8 +65,8 @@ pub fn init(bytes: Vec<u8>, network: Network) -> anyhow::Result<InitResponse> {
     let mut policy = make_simple_policy(network);
     policy.filter = PolicyFilter::new_permissive();
     let velocity_spec = VelocityControlSpec {
-        limit: 10,
-        interval_type: VelocityControlIntervalType::Hourly,
+        limit: 1_000_000, // default a million sats per day
+        interval_type: VelocityControlIntervalType::Daily,
     };
     policy.global_velocity_control = velocity_spec;
     let validator_factory = Arc::new(SimpleValidatorFactory::new_with_policy(policy));

sphinx-key/Cargo.lock

@@ -1969,6 +1969,19 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "sphinx-auther"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07ba95c8bd0600a9853ed6320701423362bfeac8d69034ed9585cb289d849701"
+dependencies = [
+ "anyhow",
+ "base64",
+ "hex",
+ "log",
+ "secp256k1",
+]
+
 [[package]]
 name = "sphinx-crypter"
 version = "0.1.0"
@@ -2007,8 +2020,11 @@ dependencies = [
 name = "sphinx-key-parser"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
+ "rmp-serde",
  "serde",
  "serde_bolt",
+ "sphinx-auther",
  "vls-protocol",
 ]
 

sphinx-key/src/conn/mqtt.rs

@@ -14,7 +14,6 @@ use std::thread;
 
 pub const VLS_TOPIC: &str = "sphinx";
 pub const CONTROL_TOPIC: &str = "sphinx-control";
-pub const OTA_TOPIC: &str = "sphinx-ota";
 pub const RETURN_TOPIC: &str = "sphinx-return";
 pub const USERNAME: &str = "sphinx-key";
 pub const PASSWORD: &str = "sphinx-key-pass";
@@ -90,9 +89,6 @@ pub fn start_listening(
                                     CONTROL_TOPIC => tx
                                         .send(CoreEvent::Control(msg.data().to_vec()))
                                         .expect("couldnt send Event::Control"),
-                                    OTA_TOPIC => tx
-                                        .send(CoreEvent::Ota(msg.data().to_vec()))
-                                        .expect("couldnt send Event::Ota"),
                                     _ => log::warn!("unrecognized topic {}", topic),
                                 };
                             } else {

sphinx-key/src/core/events.rs

@@ -1,7 +1,8 @@
-use crate::conn::mqtt::{CONTROL_TOPIC, OTA_TOPIC, QOS, RETURN_TOPIC, VLS_TOPIC};
+use crate::conn::mqtt::{CONTROL_TOPIC, QOS, RETURN_TOPIC, VLS_TOPIC};
 use crate::core::config::Config;
 use crate::core::init::make_init_msg;
 
+use sphinx_key_signer::control::Controller;
 use sphinx_key_signer::lightning_signer::bitcoin::Network;
 use sphinx_key_signer::vls_protocol::model::PubKey;
 use sphinx_key_signer::{self, InitResponse};
@@ -19,7 +20,6 @@ pub enum Event {
     Connected,
     Disconnected,
     VlsMessage(Vec<u8>),
-    Ota(Vec<u8>),
     Control(Vec<u8>),
 }
 
@@ -36,6 +36,12 @@ pub enum Status {
     Signing,
 }
 
+// the controller validates Control messages
+pub fn controller_from_seed(network: &Network, seed: &[u8]) -> Controller {
+    let (pk, sk) = sphinx_key_signer::derive_node_keys(network, seed);
+    Controller::new(sk, pk, 0)
+}
+
 // the main event loop
 #[cfg(not(feature = "pingpong"))]
 pub fn make_event_loop(
@@ -56,8 +62,6 @@ pub fn make_event_loop(
                     .expect("could not MQTT subscribe");
                 mqtt.subscribe(CONTROL_TOPIC, QOS)
                     .expect("could not MQTT subscribe");
-                mqtt.subscribe(OTA_TOPIC, QOS)
-                    .expect("could not MQTT subscribe");
                 led_tx.send(Status::Connected).unwrap();
                 break;
             }
@@ -71,6 +75,10 @@ pub fn make_event_loop(
         root_handler,
         init_reply: _,
     } = sphinx_key_signer::init(init_msg, network).expect("failed to init signer");
+
+    // make the controller to validate Control messages
+    let mut ctrlr = controller_from_seed(&network, &config.seed[..]);
+
     // signing loop
     let dummy_peer = PubKey([0; 33]);
     while let Ok(event) = rx.recv() {
@@ -81,8 +89,6 @@ pub fn make_event_loop(
                     .expect("could not MQTT subscribe");
                 mqtt.subscribe(CONTROL_TOPIC, QOS)
                     .expect("could not MQTT subscribe");
-                mqtt.subscribe(OTA_TOPIC, QOS)
-                    .expect("could not MQTT subscribe");
                 led_tx.send(Status::Connected).unwrap();
             }
             Event::Disconnected => {
@@ -107,8 +113,15 @@ pub fn make_event_loop(
                     }
                 };
             }
-            Event::Control(_) => (),
-            Event::Ota(_) => (),
+            Event::Control(ref msg_bytes) => {
+                match ctrlr.parse_msg(msg_bytes) {
+                    Ok(msg) => {
+                        log::info!("CONTROL MSG {:?}", msg);
+                        // create a response and mqtt pub here
+                    }
+                    Err(e) => log::warn!("error parsing ctrl msg {:?}", e),
+                };
+            }
         }
     }
 
@@ -147,7 +160,6 @@ pub fn make_event_loop(
                 log::info!("GOT A Event::Disconnected msg!");
             }
             Event::Control(_) => (),
-            Event::Ota(_) => (),
         }
     }
 

tester/src/main.rs

@@ -3,6 +3,7 @@ use sphinx_key_signer::lightning_signer::bitcoin::Network;
 
 use clap::{App, AppSettings, Arg};
 use rumqttc::{self, AsyncClient, Event, MqttOptions, Packet, QoS};
+use sphinx_key_signer::control::Controller;
 use sphinx_key_signer::vls_protocol::{model::PubKey, msgs};
 use sphinx_key_signer::{self, InitResponse};
 use std::env;
@@ -11,9 +12,11 @@ use std::str::FromStr;
 use std::time::Duration;
 
 const SUB_TOPIC: &str = "sphinx";
+const CONTROL_TOPIC: &str = "sphinx-control";
 const PUB_TOPIC: &str = "sphinx-return";
 const USERNAME: &str = "sphinx-key";
 const PASSWORD: &str = "sphinx-key-pass";
+const DEV_SEED: [u8; 32] = [0; 32];
 
 #[tokio::main(worker_threads = 1)]
 async fn main() -> Result<(), Box<dyn Error>> {
@@ -57,6 +60,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
             .subscribe(SUB_TOPIC, QoS::AtMostOnce)
             .await
             .expect("could not mqtt subscribe");
+        client
+            .subscribe(CONTROL_TOPIC, QoS::AtMostOnce)
+            .await
+            .expect("could not mqtt subscribe");
+
+        // make the controller to validate Control messages
+        let mut ctrlr = controller_from_seed(&Network::Regtest, &DEV_SEED[..]);
 
         if is_test {
             // test handler loop
@@ -64,24 +74,39 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 match eventloop.poll().await {
                     Ok(event) => {
                         // println!("{:?}", event);
-                        if let Some(ping_bytes) = incoming_bytes(event) {
-                            let (ping, sequence, dbid): (msgs::Ping, u16, u64) =
-                                parser::request_from_bytes(ping_bytes).expect("read ping header");
-                            if is_log {
-                                println!("sequence {}", sequence);
-                                println!("dbid {}", dbid);
-                                println!("INCOMING: {:?}", ping);
+                        if let Some((topic, msg_bytes)) = incoming_bytes(event) {
+                            match topic.as_str() {
+                                SUB_TOPIC => {
+                                    let (ping, sequence, dbid): (msgs::Ping, u16, u64) =
+                                        parser::request_from_bytes(msg_bytes)
+                                            .expect("read ping header");
+                                    if is_log {
+                                        println!("sequence {}", sequence);
+                                        println!("dbid {}", dbid);
+                                        println!("INCOMING: {:?}", ping);
+                                    }
+                                    let pong = msgs::Pong {
+                                        id: ping.id,
+                                        message: ping.message,
+                                    };
+                                    let bytes = parser::raw_response_from_msg(pong, sequence)
+                                        .expect("couldnt parse raw response");
+                                    client
+                                        .publish(PUB_TOPIC, QoS::AtMostOnce, false, bytes)
+                                        .await
+                                        .expect("could not mqtt publish");
+                                }
+                                CONTROL_TOPIC => {
+                                    match ctrlr.parse_msg(&msg_bytes) {
+                                        Ok(msg) => {
+                                            log::info!("CONTROL MSG {:?}", msg);
+                                            // create a response and mqtt pub here
+                                        }
+                                        Err(e) => log::warn!("error parsing ctrl msg {:?}", e),
+                                    };
+                                }
+                                _ => log::info!("invalid topic"),
                             }
-                            let pong = msgs::Pong {
-                                id: ping.id,
-                                message: ping.message,
-                            };
-                            let bytes = parser::raw_response_from_msg(pong, sequence)
-                                .expect("couldnt parse raw response");
-                            client
-                                .publish(PUB_TOPIC, QoS::AtMostOnce, false, bytes)
-                                .await
-                                .expect("could not mqtt publish");
                         }
                     }
                     Err(e) => {
@@ -97,7 +122,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 if let Ok(init_event) = eventloop.poll().await {
                     // this may be another kind of message like MQTT ConnAck
                     // loop around again and wait for the init
-                    if let Some(init_msg_bytes) = incoming_bytes(init_event) {
+                    if let Some((_topic, init_msg_bytes)) = incoming_bytes(init_event) {
                         let InitResponse {
                             root_handler,
                             init_reply,
@@ -122,7 +147,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                     match eventloop.poll().await {
                         Ok(event) => {
                             let dummy_peer = PubKey([0; 33]);
-                            if let Some(msg_bytes) = incoming_bytes(event) {
+                            if let Some((_topic, msg_bytes)) = incoming_bytes(event) {
                                 match sphinx_key_signer::handle(
                                     rh,
                                     msg_bytes,
@@ -151,10 +176,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 }
 
-fn incoming_bytes(event: Event) -> Option<Vec<u8>> {
+fn incoming_bytes(event: Event) -> Option<(String, Vec<u8>)> {
     if let Event::Incoming(packet) = event {
         if let Packet::Publish(p) = packet {
-            return Some(p.payload.to_vec());
+            return Some((p.topic, p.payload.to_vec()));
         }
     }
     None
@@ -200,3 +225,8 @@ pub fn setup_logging(who: &str, level_arg: &str) {
         .apply()
         .expect("log config");
 }
+
+pub fn controller_from_seed(network: &Network, seed: &[u8]) -> Controller {
+    let (pk, sk) = sphinx_key_signer::derive_node_keys(network, seed);
+    Controller::new(sk, pk, 0)
+}