gl-client 0.4.0

use crate::credentials::{RuneProvider, TlsConfigProvider};
use crate::metrics::{savings_percent, signer_state_response_wire_bytes};
use crate::pb::scheduler::{scheduler_client::SchedulerClient, NodeInfoRequest, UpgradeRequest};
use crate::pb::scheduler::{
    signer_request, signer_response, ApprovePairingRequest, ApprovePairingResponse, SignerResponse,
};
/// 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.
use crate::pb::{node_client::NodeClient, Empty, HsmRequest, HsmRequestContext, HsmResponse};
use crate::pb::{PendingRequest, SignerStateEntry};
use crate::runes;
use crate::signer::resolve::Resolver;
use crate::tls::TlsConfig;
use crate::{node, node::Client};
use anyhow::{anyhow, Result};
use base64::engine::general_purpose;
use base64::Engine;
use bytes::BufMut;
use http::uri::InvalidUri;
use lightning_signer::bitcoin::hashes::Hash;
use lightning_signer::bitcoin::secp256k1::{
    ecdsa::Signature as SecpSignature, Message as SecpMessage, PublicKey, Secp256k1, SecretKey,
};
use lightning_signer::bitcoin::Network;
use lightning_signer::node::NodeServices;
use lightning_signer::policy::filter::FilterRule;
use lightning_signer::util::crypto_utils;
use log::{debug, error, info, trace, warn};
use ring::digest::{digest, SHA256};
use ring::signature::{UnparsedPublicKey, ECDSA_P256_SHA256_FIXED};
use runeauth::{Condition, Restriction, Rune, RuneError};
use std::convert::{TryFrom, TryInto};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::SystemTime;
use tokio::sync::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};
use vls_protocol::serde_bolt::Octets;
use vls_protocol_signer::approver::{Approve, MemoApprover};
use vls_protocol_signer::handler;
use vls_protocol_signer::handler::Handler;

mod approver;
mod auth;
pub mod model;
mod report;
mod resolve;

const VERSION: &str = "v25.12";
const GITHASH: &str = env!("GIT_HASH");
const RUNE_VERSION: &str = "gl0";
// This is the same derivation key that is used by core lightning itself.
const RUNE_DERIVATION_SECRET: &str = "gl-commando";
const STATE_DERIVATION_SECRET: &str = "greenlight/state-signing/v1";
const STATE_SIGNING_DOMAIN: &[u8] = b"greenlight/state-signing/v1\0";
const STATE_SIGNATURE_OVERRIDE_ACK: &str = "I_ACCEPT_OPERATOR_ASSISTED_STATE_OVERRIDE";
const COMPACT_SIGNATURE_LEN: usize = 64;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StateSignatureMode {
    Off,
    Soft,
    Hard,
}

impl Default for StateSignatureMode {
    fn default() -> Self {
        Self::Soft
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StateSignatureOverrideConfig {
    pub ack: String,
    pub note: Option<String>,
}

#[derive(Clone, Debug, Default)]
pub struct SignerConfig {
    pub state_signature_mode: StateSignatureMode,
    pub state_signature_override: Option<StateSignatureOverrideConfig>,
}

#[derive(Debug, Default)]
struct OverrideSignatureUsage {
    missing_keys: Vec<String>,
    invalid_keys: Vec<String>,
}

impl OverrideSignatureUsage {
    fn is_used(&self) -> bool {
        !self.missing_keys.is_empty() || !self.invalid_keys.is_empty()
    }
}

#[derive(Clone)]
pub struct Signer {
    secret: [u8; 32],
    state_signing_secret: SecretKey,
    state_signing_pubkey: PublicKey,
    state_signature_mode: StateSignatureMode,
    state_signature_override_enabled: bool,
    state_signature_override_note: Option<String>,
    master_rune: Rune,
    services: NodeServices,
    tls: TlsConfig,
    id: Vec<u8>,

    /// Cached version of the init response
    init: Vec<u8>,

    network: Network,
    state: Arc<Mutex<crate::persist::State>>,
}

#[derive(thiserror::Error, Debug)]
pub enum Error {
    #[error("could not connect to scheduler: ")]
    SchedulerConnection(),

    #[error("scheduler returned an error: {0}")]
    Scheduler(tonic::Status),

    #[error("could not connect to node: {0}")]
    NodeConnection(#[from] tonic::transport::Error),

    #[error("connection to node lost: {0}")]
    NodeDisconnect(#[from] tonic::Status),

    #[error("authentication error: {0}")]
    Auth(crate::Error),

    #[error("scheduler returned faulty URI: {0}")]
    InvalidUri(#[from] InvalidUri),

    #[error("resolver error: request {0:?}, context: {1:?}")]
    Resolver(Vec<u8>, Vec<crate::signer::model::Request>),

    #[error("error asking node to be upgraded: {0}")]
    Upgrade(tonic::Status),

    #[error("protocol error: {0}")]
    Protocol(#[from] vls_protocol::Error),

    #[error("other: {0}")]
    Other(anyhow::Error),

    #[error("could not approve pairing request: {0}")]
    ApprovePairingRequestError(String),
}

impl Signer {
    pub fn new<T>(secret: Vec<u8>, network: Network, creds: T) -> Result<Signer, anyhow::Error>
    where
        T: TlsConfigProvider,
    {
        Self::new_with_config(secret, network, creds, SignerConfig::default())
    }

    pub fn new_with_config<T>(
        secret: Vec<u8>,
        network: Network,
        creds: T,
        config: SignerConfig,
    ) -> Result<Signer, anyhow::Error>
    where
        T: TlsConfigProvider,
    {
        use lightning_signer::policy::{
            filter::PolicyFilter, simple_validator::SimpleValidatorFactory,
        };
        use lightning_signer::signer::ClockStartingTimeFactory;
        use lightning_signer::util::clock::StandardClock;

        info!("Initializing signer for {VERSION} ({GITHASH}) (VLS)");
        let state_signature_mode = config.state_signature_mode;
        let (state_signature_override_enabled, state_signature_override_note) =
            match config.state_signature_override {
                Some(override_config) => {
                    if state_signature_mode == StateSignatureMode::Off {
                        return Err(anyhow!(
                            "state signature override is incompatible with state signature mode off"
                        ));
                    }

                    if override_config.ack != STATE_SIGNATURE_OVERRIDE_ACK {
                        return Err(anyhow!(
                            "invalid state signature override ack, expected {}",
                            STATE_SIGNATURE_OVERRIDE_ACK
                        ));
                    }

                    let note = override_config
                        .note
                        .and_then(|n| {
                            let trimmed = n.trim().to_string();
                            if trimmed.is_empty() {
                                None
                            } else {
                                Some(trimmed)
                            }
                        });
                    (true, note)
                }
                None => (false, None),
            };
        let mut sec: [u8; 32] = [0; 32];
        sec.copy_from_slice(&secret[0..32]);

        // The persister takes care of persisting metadata across
        // restarts
        let persister = Arc::new(crate::persist::MemoryPersister::new());
        let mut policy =
            lightning_signer::policy::simple_validator::make_default_simple_policy(network);

        policy.filter = PolicyFilter::default();
        policy.filter.merge(PolicyFilter {
            // TODO: Remove once we have fully switched over to zero-fee anchors
            rules: vec![
                FilterRule::new_warn("policy-channel-safe-type-anchors"),
                FilterRule::new_warn("policy-routing-balanced"),
                FilterRule::new_warn("policy-commitment-retry-same"),
            ],
        });

        // TODO Remove this once VLS has implemented the fee budget
        // per payment, rather than the fee budget per HTLC.
        // Ref: https://github.com/Blockstream/greenlight/issues/538
        {
            policy.max_feerate_per_kw = 150_000;
            policy.filter.merge(PolicyFilter {
                rules: vec![
                    FilterRule::new_warn("policy-commitment-fee-range"),
                    FilterRule::new_warn("policy-mutual-fee-range"),
                ],
            });
        }

        policy.filter.merge(PolicyFilter {
            // TODO: Remove once we have implemented zero invoice support
            rules: vec![
                FilterRule::new_warn("policy-routing-balanced"),
                FilterRule::new_warn("policy-htlc-fee-range"),
            ],
        });

        // TODO: Remove once we found the desync issue
        policy.filter.merge(PolicyFilter {
            rules: vec![
                // "policy failure: get_per_commitment_secret: cannot
                // revoke commitment_number 312 when
                // next_holder_commit_num is 313"
                FilterRule::new_warn("policy-revoke-new-commitment-signed"),
                // "policy failure: get_current_holder_commitment_info:
                // invalid next holder commitment number: 3974 != 3973"
                // Triggered during force-close signing when tower
                // committed the state advance but CLN didn't.
                FilterRule::new_warn("policy-other"),
            ],
        });

        // Increase the invoices limit. Results in a larger state, but
        // bumping into this is rather annoying.
        policy.max_invoices = 10_000usize;

        // Relaxed max_routing_fee since we no longer have the
        // presplitter which was causing the HTLCs to be smaller.
        policy.max_routing_fee_msat = 1_000_000;

        let validator_factory = Arc::new(SimpleValidatorFactory::new_with_policy(policy));
        let starting_time_factory = ClockStartingTimeFactory::new();
        let clock = Arc::new(StandardClock());

        let services = NodeServices {
            validator_factory,
            starting_time_factory,
            persister: persister.clone(),
            trusted_oracle_pubkeys: vec![],
            clock,
        };

        let mut handler = handler::HandlerBuilder::new(network, 0 as u64, services.clone(), sec)
            .build()
            .map_err(|e| anyhow!("building root_handler: {:?}", e))?;

        // Calling init on the `InitHandler` from above puts it into a
        // state that it can be upgraded into the `RootHandler` that
        // we need for the rest of the run.
        let init = Signer::initmsg(&mut handler)?;

        let init = HsmdInitReplyV4::from_vec(init)
            .map_err(|e| anyhow!("Failed to parse init message as HsmdInitReplyV4: {:?}", e))?;

        let id = init.node_id.0.to_vec();
        use vls_protocol::msgs::SerBolt;
        let init = init.as_vec();

        // Init master rune. We create the rune seed from the nodes
        // seed by deriving a hardened key tagged with "rune secret".
        let rune_secret = crypto_utils::hkdf_sha256(&sec, RUNE_DERIVATION_SECRET.as_bytes(), &[]);
        let mr = Rune::new_master_rune(&rune_secret, vec![], None, Some(RUNE_VERSION.to_string()))?;
        let state_signing_secret = Self::derive_state_signing_secret(&sec)?;
        let state_signing_pubkey =
            PublicKey::from_secret_key(&Secp256k1::signing_only(), &state_signing_secret);

        trace!("Initialized signer for node_id={}", hex::encode(&id));
        Ok(Signer {
            secret: sec,
            state_signing_secret,
            state_signing_pubkey,
            state_signature_mode,
            state_signature_override_enabled,
            state_signature_override_note,
            master_rune: mr,
            services,
            tls: creds.tls_config(),
            id,
            init,
            network,
            state: persister.state(),
        })
    }

    fn init_handler(&self) -> Result<handler::InitHandler, anyhow::Error> {
        let h = handler::HandlerBuilder::new(
            self.network,
            0 as u64,
            self.services.clone(),
            self.secret,
        )
        .build()
        .map_err(|e| anyhow!("building root_handler: {:?}", e))?;

        Ok(h)
    }

    fn handler(&self) -> Result<handler::RootHandler, anyhow::Error> {
        let mut h = self.init_handler()?;
        h.handle(Signer::initreq())
            .map_err(|e| anyhow!("Failed to handle hsmd_init message: {:?}", e))?;
        Ok(h.into())
    }

    fn handler_with_approver(
        &self,
        approver: Arc<dyn Approve>,
    ) -> Result<handler::RootHandler, Error> {
        let mut h = handler::HandlerBuilder::new(
            self.network,
            0 as u64,
            self.services.clone(),
            self.secret,
        )
        .approver(approver)
        .build()
        .map_err(|e| crate::signer::Error::Other(anyhow!("Could not create handler: {:?}", e)))?;
        h.handle(Signer::initreq())
            .map_err(|e| Error::Other(anyhow!("Failed to handle hsmd_init message: {:?}", e)))?;
        Ok(h.into())
    }

    /// Create an `init` request that we can pass to the signer.
    fn initreq() -> vls_protocol::msgs::Message {
        vls_protocol::msgs::Message::HsmdInit(vls_protocol::msgs::HsmdInit {
            key_version: vls_protocol::model::Bip32KeyVersion {
                pubkey_version: 0,
                privkey_version: 0,
            },
            chain_params: lightning_signer::bitcoin::BlockHash::all_zeros(),
            encryption_key: None,
            dev_privkey: None,
            dev_bip32_seed: None,
            dev_channel_secrets: None,
            dev_channel_secrets_shaseed: None,
            hsm_wire_min_version: 4,
            hsm_wire_max_version: 6,
        })
    }

    fn initmsg(handler: &mut vls_protocol_signer::handler::InitHandler) -> Result<Vec<u8>, Error> {
        let (_req, response) = handler
            .handle(Signer::initreq())
            .map_err(|e| Error::Other(anyhow!("Failed to handle init request: {:?}", e)))?;
        Ok(response.map(|a| a.as_vec()).unwrap_or_default())
    }

    fn derive_state_signing_secret(secret: &[u8; 32]) -> Result<SecretKey, anyhow::Error> {
        let key = crypto_utils::hkdf_sha256(secret, STATE_DERIVATION_SECRET.as_bytes(), &[]);
        SecretKey::from_slice(&key).map_err(|e| anyhow!("failed to derive state signing key: {e}"))
    }

    fn state_signature_digest(key: &str, version: u64, value: &[u8]) -> [u8; 32] {
        let mut payload =
            Vec::with_capacity(STATE_SIGNING_DOMAIN.len() + 4 + key.len() + 8 + 4 + value.len());
        payload.extend_from_slice(STATE_SIGNING_DOMAIN);
        payload.extend_from_slice(&(key.len() as u32).to_be_bytes());
        payload.extend_from_slice(key.as_bytes());
        payload.extend_from_slice(&version.to_be_bytes());
        payload.extend_from_slice(&(value.len() as u32).to_be_bytes());
        payload.extend_from_slice(value);
        let hash = digest(&SHA256, &payload);
        let mut out = [0u8; 32];
        out.copy_from_slice(hash.as_ref());
        out
    }

    fn sign_state_payload(
        &self,
        key: &str,
        version: u64,
        value: &[u8],
    ) -> Result<Vec<u8>, anyhow::Error> {
        let digest = Self::state_signature_digest(key, version, &value);
        let msg = SecpMessage::from_digest_slice(&digest).map_err(|e| {
            anyhow!(
                "failed to build state signature digest for key {}: {}",
                key,
                e
            )
        })?;
        let sig = Secp256k1::signing_only().sign_ecdsa(&msg, &self.state_signing_secret);
        Ok(sig.serialize_compact().to_vec())
    }

    fn verify_state_entry_signature(&self, entry: &SignerStateEntry) -> Result<(), anyhow::Error> {
        if entry.signature.len() != COMPACT_SIGNATURE_LEN {
            return Err(anyhow!(
                "expected {} signature bytes, got {}",
                COMPACT_SIGNATURE_LEN,
                entry.signature.len()
            ));
        }
        let digest = Self::state_signature_digest(&entry.key, entry.version, &entry.value);
        let msg = SecpMessage::from_digest_slice(&digest)
            .map_err(|e| anyhow!("failed to build digest message: {}", e))?;
        let sig = SecpSignature::from_compact(&entry.signature)
            .map_err(|e| anyhow!("invalid compact signature: {}", e))?;
        Secp256k1::verification_only()
            .verify_ecdsa(&msg, &sig, &self.state_signing_pubkey)
            .map_err(|e| anyhow!("signature verification failed: {}", e))?;
        Ok(())
    }

    fn state_signature_mode_label(&self) -> &'static str {
        match self.state_signature_mode {
            StateSignatureMode::Off => "off",
            StateSignatureMode::Soft => "soft",
            StateSignatureMode::Hard => "hard",
        }
    }

    fn inspect_incoming_state_signatures(
        &self,
        entries: &[SignerStateEntry],
    ) -> Result<OverrideSignatureUsage, Error> {
        if self.state_signature_mode == StateSignatureMode::Off {
            return Ok(OverrideSignatureUsage::default());
        }

        let mut usage = OverrideSignatureUsage::default();
        let mut first_error: Option<anyhow::Error> = None;

        for entry in entries.iter() {
            if entry.signature.is_empty() {
                if self.state_signature_mode == StateSignatureMode::Hard {
                    usage.missing_keys.push(entry.key.clone());
                    if first_error.is_none() {
                        first_error = Some(anyhow!("missing state signature for key {}", entry.key));
                    }
                }
                continue;
            }

            if let Err(e) = self.verify_state_entry_signature(entry) {
                usage.invalid_keys.push(entry.key.clone());
                if first_error.is_none() {
                    first_error = Some(anyhow!("invalid state signature for key {}: {}", entry.key, e));
                }
            }
        }

        if usage.is_used() && !self.state_signature_override_enabled {
            return Err(Error::Other(
                first_error
                    .unwrap_or_else(|| anyhow!("state signature verification failed unexpectedly")),
            ));
        }

        Ok(usage)
    }

    async fn report_state_signature_override_enabled(&self) {
        if !self.state_signature_override_enabled {
            return;
        }

        let message = report::build_state_signature_override_enabled_message(
            self.state_signature_mode_label(),
            &self.id,
            self.state_signature_override_note.as_deref(),
        );
        warn!("{}", message);
        report::Reporter::report(crate::pb::scheduler::SignerRejection {
            msg: message,
            request: None,
            git_version: GITHASH.to_string(),
            node_id: self.node_id(),
        })
        .await;
    }

    async fn report_state_signature_override_usage(
        &self,
        req: &HsmRequest,
        usage: &OverrideSignatureUsage,
    ) {
        if !usage.is_used() || !self.state_signature_override_enabled {
            return;
        }

        let message = report::build_state_signature_override_used_message(
            self.state_signature_mode_label(),
            req.request_id as u64,
            &usage.missing_keys,
            &usage.invalid_keys,
            self.state_signature_override_note.as_deref(),
        );
        warn!("{}", message);
        report::Reporter::report(crate::pb::scheduler::SignerRejection {
            msg: message,
            request: Some(req.clone()),
            git_version: GITHASH.to_string(),
            node_id: self.node_id(),
        })
        .await;
    }

    /// Filter out any request that is not signed, such that the
    /// remainder is the minimal set to reconcile state changes
    /// against.
    ///
    /// Returns an error if a signature failed verification or if the
    /// rune verification failed.
    fn check_request_auth(
        &self,
        requests: Vec<crate::pb::PendingRequest>,
    ) -> Vec<Result<crate::pb::PendingRequest, anyhow::Error>> {
        // Filter out requests lacking a required field. They are unverifiable anyway.
        // Todo: partition results to provide more detailed errors.
        requests
            .into_iter()
            .filter(|r| !r.pubkey.is_empty() && !r.signature.is_empty() && !r.rune.is_empty())
            .map(|r| {
                let pk = UnparsedPublicKey::new(&ECDSA_P256_SHA256_FIXED, &r.pubkey);
                let mut data = r.request.clone();

                // If we have a timestamp associated we must add it to
                // the payload being checked. Same thing happens on
                // the client too.
                if r.timestamp != 0 {
                    data.put_u64(r.timestamp);
                }

                pk.verify(&data, &r.signature)
                    .map_err(|e| anyhow!("signature verification failed: {}", e))?;

                self.verify_rune(r.clone())
                    .map(|_| r)
                    .map_err(|e| anyhow!("rune verification failed: {}", e))
            })
            .collect()
    }

    /// Verifies that the public key of the request and the signers rune version
    /// match the corresponding restrictions of the rune.
    fn verify_rune(&self, request: crate::pb::PendingRequest) -> Result<(), anyhow::Error> {
        let rune64 = general_purpose::URL_SAFE.encode(request.rune);
        let rune = Rune::from_base64(&rune64)?;

        // A valid gl-rune must contain a pubkey field as this  is bound to the
        // signer. Against the rules of runes we do not accept a rune that has
        // no restriction on a public key.
        if !rune.to_string().contains("pubkey=") {
            return Err(anyhow!("rune is missing pubkey field"));
        }

        // Currently we only use a 0 unique_id and a pubkey field to allow
        // for delegation in the future but we could also set the public
        // key as the unique_id in the future and add a method that allows
        // to create new empty runes.
        let unique_id = rune.get_id();
        let ver_id = match unique_id {
            Some(id) => format!("{}-{}", id, RUNE_VERSION),
            None => String::default(),
        };

        // Check that the request points to `cln.Node`.
        let mut parts = request.uri.split('/');
        parts.next();
        match parts.next() {
            Some(service) => {
                if service != "cln.Node" && service != "greenlight.Node" {
                    debug!("request from unknown service {}.", service);
                    return Err(anyhow!("service {} is not valid", service));
                }
            }
            None => {
                debug!("could not extract service from the uri while verifying rune.");
                return Err(anyhow!("can not extract service from uri"));
            }
        };

        // Extract the method from the request uri: eg. `/cln.Node/CreateInvoice`
        // becomes `createinvoice`.
        let method = match parts.next() {
            Some(m) => m.to_lowercase(),
            None => {
                debug!("could not extract method from uri while verifying rune.");
                return Err(anyhow!("can not extract uri form request"));
            }
        };

        let ctx = runes::Context {
            method,
            pubkey: hex::encode(request.pubkey),
            time: SystemTime::now(),
            unique_id: ver_id,
        };

        match self.master_rune.check_with_reason(&rune64, ctx) {
            Ok(_) => Ok(()),
            Err(e) => Err(e.into()),
        }
    }

    /// Given the URI of the running node, connect to it and stream
    /// requests from it. The requests are then verified and processed
    /// using the `Hsmd`.
    pub async fn run_once(&self, node_uri: Uri) -> Result<(), Error> {
        info!("Connecting to node at {}", node_uri);

        let tls_config = if node_uri.host().unwrap_or_default().contains("blckstrm") {
            self.tls.inner.clone()
        } else {
            self.tls.inner.clone().domain_name("localhost")
        };

        let c = Endpoint::from_shared(node_uri.to_string())?
            .tls_config(tls_config)?
            .tcp_keepalive(Some(crate::TCP_KEEPALIVE))
            .http2_keep_alive_interval(crate::TCP_KEEPALIVE)
            .keep_alive_timeout(crate::TCP_KEEPALIVE_TIMEOUT)
            .keep_alive_while_idle(true)
            .connect_lazy();

        let mut client = NodeClient::new(c);

        let mut stream = client
            .stream_hsm_requests(Request::new(Empty::default()))
            .await?
            .into_inner();

        info!("Starting to stream signer requests");
        loop {
            let req = match stream
                .message()
                .await
                .map_err(|e| Error::NodeDisconnect(e))?
            {
                Some(r) => r,
                None => {
                    warn!("Signer request stream ended, the node shouldn't do this.");
                    return Ok(());
                }
            };
            let request_id = req.request_id;
            let hex_req = hex::encode(&req.raw);
            let signer_state = req.signer_state.clone();
            trace!("Received request {}", hex_req);

            match self.process_request(req.clone()).await {
                Ok(response) => {
                    trace!("Sending response {}", hex::encode(&response.raw));
                    client
                        .respond_hsm_request(response)
                        .await
                        .map_err(|e| Error::NodeDisconnect(e))?;
                }
                Err(e) => {
                    // Report the rejection to the server for debugging
                    // Note: process_request already reports some errors, but we ensure
                    // all errors are reported here as a safety net
                    report::Reporter::report(crate::pb::scheduler::SignerRejection {
                        msg: format!("Error in run_once loop: {:?}", e),
                        request: Some(req.clone()),
                        git_version: GITHASH.to_string(),
                        node_id: self.node_id(),
                    })
                    .await;

                    let response = HsmResponse {
                        raw: vec![],
                        request_id,
                        error: format!("{:?}", e),
                        signer_state: vec![],
                    };
                    client
                        .respond_hsm_request(response)
                        .await
                        .map_err(|e| Error::NodeDisconnect(e))?;
                    warn!(
                        "Rejected request {} with error: {}. State: {:?}",
                        hex_req, e, signer_state,
                    )
                }
            };
        }
    }

    fn authenticate_request(
        &self,
        msg: &vls_protocol::msgs::Message,
        reqs: &Vec<model::Request>,
    ) -> Result<(), Error> {
        log::trace!(
            "Resolving signature request against pending grpc commands: {:?}",
            reqs
        );

        // Quick path out of here: we can't find a resolution for a
        // request, then abort!
        Resolver::try_resolve(msg, &reqs)?;

        Ok(())
    }

    async fn process_request(&self, req: HsmRequest) -> Result<HsmResponse, Error> {
        debug!("Processing request {:?}", req);
        let req = req;

        let signature_usage = self.inspect_incoming_state_signatures(&req.signer_state)?;
        if signature_usage.is_used() {
            self.report_state_signature_override_usage(&req, &signature_usage)
                .await;
        }

        let incoming_state = crate::persist::State::try_from(req.signer_state.as_slice())
            .map_err(|e| Error::Other(anyhow!("Failed to decode signer state: {e}")))?;


        // Create sketch from incoming state (nodelet's view) so we can
        // send back any entries the nodelet doesn't know about yet,
        // including the initial VLS state created during Signer::new().
        let prestate_sketch = incoming_state.sketch();

        let prestate_log = {
            debug!("Updating local signer state with state from node");
            let mut state = self.state.lock().map_err(|e| {
                Error::Other(anyhow!("Failed to acquire state lock: {:?}", e))
            })?;
            let merge_res = state.merge(&incoming_state).map_err(|e| {
                Error::Other(anyhow!("Failed to merge signer state: {:?}", e))
            })?;
            if merge_res.has_conflicts() {
                debug!(
                    "State merge ignored stale versions (count={})",
                    merge_res.conflict_count
                );
            }
            trace!("Processing request {}", hex::encode(&req.raw));
            serde_json::to_string(&*state).map_err(|e| {
                Error::Other(anyhow!("Failed to serialize signer state for logging: {:?}", e))
            })?
        };

        // The first two bytes represent the message type. Check that
        // it is not a `sign-message` request (type 23).
        if let &[h, l, ..] = req.raw.as_slice() {
            let typ = ((h as u16) << 8) | (l as u16);
            if typ == 23 {
                warn!("Refusing to process sign-message request");
                return Err(Error::Other(anyhow!(
                    "Cannot process sign-message requests from node."
                )));
            }
        }

        let ctxrequests: Vec<model::Request> = self
            .check_request_auth(req.requests.clone())
            .into_iter()
            .filter_map(|r| r.ok())
            .map(|r| decode_request(r))
            .filter_map(|r| match r {
                Ok(r) => Some(r),
                Err(e) => {
                    log::error!("Unable to decode request in context: {}", e);
                    None
                }
            })
            .collect::<Vec<model::Request>>();

        let msg = vls_protocol::msgs::from_vec(req.raw.clone()).map_err(|e| Error::Protocol(e))?;
        log::debug!("Handling message {:?}", msg);
        log::trace!("Signer state {}", prestate_log);

        if let Err(e) = self.authenticate_request(&msg, &ctxrequests) {
            report::Reporter::report(crate::pb::scheduler::SignerRejection {
                msg: e.to_string(),
                request: Some(req.clone()),
                git_version: GITHASH.to_string(),
                node_id: self.node_id(),
            })
            .await;
            #[cfg(not(feature = "permissive"))]
            return Err(Error::Resolver(req.raw, ctxrequests));
        };

        // If present, add the close_to_addr to the allowlist
        for parsed_request in ctxrequests.iter() {
            match parsed_request {
                model::Request::GlConfig(gl_config) => {
                    let pubkey = PublicKey::from_slice(&self.id);
                    match pubkey {
                        Ok(p) => {
                            let _ = self
                                .services
                                .persister
                                .update_node_allowlist(&p, vec![gl_config.close_to_addr.clone()]);
                        }
                        Err(e) => debug!("Could not parse public key {:?}: {:?}", self.id, e),
                    }
                }
                _ => {}
            }
        }

        use auth::Authorizer;
        let auth = auth::GreenlightAuthorizer {};
        let approvals = auth.authorize(&ctxrequests).map_err(|e| Error::Auth(e))?;
        debug!("Current approvals: {:?}", approvals);

        let approver = Arc::new(MemoApprover::new(approver::ReportingApprover::new(
            #[cfg(feature = "permissive")]
            vls_protocol_signer::approver::PositiveApprover(),
            #[cfg(not(feature = "permissive"))]
            vls_protocol_signer::approver::NegativeApprover(),
        )));
        approver.approve(approvals);
        let root_handler = self.handler_with_approver(approver)?;

        log::trace!("Updating state from context");
        if let Err(e) = update_state_from_context(&ctxrequests, &root_handler) {
            error!("Failed to update state from context: {:?}", e);
            report::Reporter::report(crate::pb::scheduler::SignerRejection {
                msg: format!("Failed to update state from context: {:?}", e),
                request: Some(req.clone()),
                git_version: GITHASH.to_string(),
                node_id: self.node_id(),
            })
            .await;
            return Err(Error::Other(anyhow!(
                "Failed to update state from context: {:?}",
                e
            )));
        }
        log::trace!("State updated");

        // Match over root and client handler.
        let response = match req.context.clone() {
            Some(HsmRequestContext { dbid: 0, .. }) | None => {
                // This is the main daemon talking to us.
                root_handler.handle(msg)
            }
            Some(c) => {
                let node_id_len = c.node_id.len();
                let pk: [u8; 33] = c.node_id.try_into().map_err(|_| {
                    Error::Other(anyhow!(
                        "Invalid node_id length in context: expected 33 bytes, got {}",
                        node_id_len
                    ))
                })?;
                let pk = vls_protocol::model::PubKey(pk);
                root_handler
                    .for_new_client(1 as u64, pk, c.dbid)
                    .handle(msg)
            }
        };

        let response = match response {
            Ok(r) => r,
            Err(e) => {
                report::Reporter::report(crate::pb::scheduler::SignerRejection {
                    msg: format!("{:?}", e),
                    request: Some(req.clone()),
                    git_version: GITHASH.to_string(),
                    node_id: self.node_id(),
                })
                .await;
                return Err(Error::Other(anyhow!("processing request: {e:?}")));
            }
        };

        let signer_state: Vec<crate::pb::SignerStateEntry> = {
            debug!("Serializing state changes to report to node");
            let mut state = self.state.lock().map_err(|e| {
                Error::Other(anyhow!(
                    "Failed to acquire state lock for serialization: {:?}",
                    e
                ))
            })?;
            state
                .resign_signatures(|key, version, value| {
                    self.sign_state_payload(key, version, value)
                })
                .map_err(|e| Error::Other(anyhow!("Failed to sign signer state entries: {e}")))?;
            let full_wire_bytes = {
                let full_entries: Vec<crate::pb::SignerStateEntry> = state.clone().into();
                signer_state_response_wire_bytes(&full_entries)
            };
            let diff_state = prestate_sketch.diff_state(&state);
            let diff_entries: Vec<crate::pb::SignerStateEntry> = diff_state.into();
            let diff_wire_bytes = signer_state_response_wire_bytes(&diff_entries);
            let saved_percent = savings_percent(full_wire_bytes, diff_wire_bytes);
            trace!(
                "Signer state diff entries={}, wire_bytes={}, full_wire_bytes={}, saved {}% bandwidth syncing the state",
                diff_entries.len(),
                diff_wire_bytes,
                full_wire_bytes,
                saved_percent
            );
            diff_entries
        };
        Ok(HsmResponse {
            raw: response.as_vec(),
            request_id: req.request_id,
            signer_state,
            error: "".to_owned(),
        })
    }

    pub fn node_id(&self) -> Vec<u8> {
        self.id.clone()
    }

    pub fn get_init(&self) -> Vec<u8> {
        self.init.clone()
    }

    /// Retrieve the messages we know `lightningd` will ask when
    /// starting. Since we can't be attached during startup, or on
    /// background sync runs, we need to stash them at the `scheduler`
    /// so we can start without a signer present.
    pub fn get_startup_messages(&self) -> Vec<StartupMessage> {
        let mut init_handler = self.init_handler().unwrap();

        let init = StartupMessage {
            request: Signer::initreq().inner().as_vec(),
            response: init_handler
                .handle(Signer::initreq())
                .unwrap()
                .1
                .map(|a| a.as_vec())
                .unwrap_or_default(),
        };

        let requests = vec![
            // v22.11 introduced an addiotiona startup message, the
            // bolt12 key generation
            vls_protocol::msgs::Message::DeriveSecret(vls_protocol::msgs::DeriveSecret {
                info: Octets("bolt12-invoice-base".as_bytes().to_vec()),
            }),
            // SCB needs a secret derived too
            vls_protocol::msgs::Message::DeriveSecret(vls_protocol::msgs::DeriveSecret {
                info: Octets("scb secret".as_bytes().to_vec()),
            }),
            // Commando needs a secret for its runes
            vls_protocol::msgs::Message::DeriveSecret(vls_protocol::msgs::DeriveSecret {
                info: Octets("commando".as_bytes().to_vec()),
            }),
            // The node alias key
            vls_protocol::msgs::Message::DeriveSecret(vls_protocol::msgs::DeriveSecret {
                info: Octets("node-alias-base".as_bytes().to_vec()),
            }),
            vls_protocol::msgs::Message::DeriveSecret(vls_protocol::msgs::DeriveSecret {
                info: Octets("offer-blinded-path".as_bytes().to_vec()),
            }),
        ];

        let serialized: Vec<Vec<u8>> = requests.iter().map(|m| m.inner().as_vec()).collect();
        let responses: Vec<Vec<u8>> = requests
            .into_iter()
            .map(|r| self.handler().unwrap().handle(r).unwrap().as_vec())
            .collect();

        let mut msgs: Vec<StartupMessage> = serialized
            .into_iter()
            .zip(responses)
            .map(|r| {
                log::debug!("Storing canned request-response: {:?} -> {:?}", r.0, r.1);

                StartupMessage {
                    request: r.0,
                    response: r.1,
                }
            })
            .collect();

        msgs.insert(0, init);

        msgs
    }

    pub fn bip32_ext_key(&self) -> Vec<u8> {
        use vls_protocol::{msgs, msgs::Message};
        // This should never fail since we validated it during construction
        let initmsg = msgs::from_vec(self.init.clone())
            .expect("init message should be valid (validated during Signer::new)");

        match initmsg {
            Message::HsmdInit2Reply(m) => m.bip32.0.to_vec(),
            Message::HsmdInitReplyV4(m) => m.bip32.0.to_vec(),
            Message::HsmdInitReplyV2(m) => m.bip32.0.to_vec(),
            m => {
                // This should never happen, but log and return empty instead of panicking
                error!("Unknown initmsg type {:?}, cannot extract bip32 key", m);
                vec![]
            }
        }
    }

    pub fn legacy_bip32_ext_key(&self) -> Vec<u8> {
        let mut handler = match self.init_handler() {
            Ok(h) => h,
            Err(e) => {
                error!("Failed to retrieve handler for legacy bip32 key: {:?}", e);
                return vec![];
            }
        };
        let req = vls_protocol::msgs::Message::HsmdInit(vls_protocol::msgs::HsmdInit {
            key_version: vls_protocol::model::Bip32KeyVersion {
                pubkey_version: 0,
                privkey_version: 0,
            },
            chain_params: lightning_signer::bitcoin::BlockHash::all_zeros(),
            encryption_key: None,
            dev_privkey: None,
            dev_bip32_seed: None,
            dev_channel_secrets: None,
            dev_channel_secrets_shaseed: None,
            hsm_wire_min_version: 1,
            hsm_wire_max_version: 2,
        });

        let initmsg = match handler.handle(req) {
            Ok((_req, response)) => response.map(|a| a.as_vec()).unwrap_or_default(),
            Err(e) => {
                error!("Failed to handle legacy init message: {:?}", e);
                return vec![];
            }
        };

        if initmsg.len() <= 35 {
            error!(
                "Legacy init message too short: expected >35 bytes, got {}",
                initmsg.len()
            );
            return vec![];
        }
        initmsg[35..].to_vec()
    }

    /// Connect to the scheduler given by the environment variable
    /// `GL_SCHEDULER_GRPC_URI` (of the default URI) and wait for the
    /// node to be scheduled. Once scheduled, connect to the node
    /// directly and start streaming and processing requests.
    pub async fn run_forever(&self, shutdown: mpsc::Receiver<()>) -> Result<(), anyhow::Error> {
        let scheduler_uri = crate::utils::scheduler_uri();
        debug!("Starting signer run loop");
        self.report_state_signature_override_enabled().await;
        let res = Self::run_forever_with_uri(&self, shutdown, scheduler_uri).await;
        debug!("Exited signer run loop");
        res
    }

    /// Create and, if necessary, upgrade the scheduler
    async fn init_scheduler(
        &self,
        scheduler_uri: String,
    ) -> Result<SchedulerClient<tonic::transport::channel::Channel>> {
        info!("Connecting to scheduler at {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)
            .keep_alive_timeout(crate::TCP_KEEPALIVE_TIMEOUT)
            .keep_alive_while_idle(true)
            .connect_lazy();
        let mut scheduler = SchedulerClient::new(channel);

        // Upgrade node if necessary.
        // If it fails due to connection error, sleep and retry. Re-throw all other errors.
        loop {
            let call_start = tokio::time::Instant::now();
            debug!("Sending maybe_upgrade to {}", self.version());
            #[allow(deprecated)]
            let res = scheduler
                .maybe_upgrade(UpgradeRequest {
                    initmsg: self.init.clone(),
                    signer_version: self.version().to_owned(),
                    startupmsgs: self
                        .get_startup_messages()
                        .into_iter()
                        .map(|s| s.into())
                        .collect(),
                })
                .await;
            debug!(
                "Server returned {:?} after {}s",
                res,
                call_start.elapsed().as_secs()
            );

            match res {
                Err(e) => match e.code() {
                    Code::Unavailable => {
                        debug!("Cannot connect to scheduler, sleeping and retrying");
                        sleep(Duration::from_secs(3)).await;
                        continue;
                    }
                    _ => Err(Error::Upgrade(e))?,
                },
                Ok(r) => {
                    debug!("Server reports version {}", r.into_inner().old_version)
                }
            }
            break;
        }
        Ok(scheduler)
    }

    /// The core signer loop. Connects to the signer and keeps the connection alive.
    ///
    /// Used as inner loop for `run_forever_with_uri`.
    async fn run_forever_inner(
        &self,
        mut scheduler: SchedulerClient<tonic::transport::channel::Channel>,
    ) -> Result<(), anyhow::Error> {
        loop {
            info!("Calling scheduler.get_node_info");
            let node_info_res = scheduler
                .get_node_info(NodeInfoRequest {
                    node_id: self.id.clone(),

                    // This `wait` parameter means that the scheduler will
                    // not automatically schedule the node. Rather we are
                    // telling the scheduler we want to be told as soon as
                    // the node is being scheduled so we can re-attach to
                    // that.
                    wait: true,
                })
                .await;

            let node_info = match node_info_res.map(|v| v.into_inner()) {
                Ok(v) => {
                    info!("Got node_info from scheduler: {:?}", v);
                    v
                }
                Err(e) => {
                    trace!("Got an error from the scheduler: {e}. Sleeping before retrying");
                    sleep(Duration::from_millis(1000)).await;
                    continue;
                }
            };

            if node_info.grpc_uri.is_empty() {
                trace!("Got an empty GRPC URI, node is not scheduled, sleeping and retrying");
                sleep(Duration::from_millis(1000)).await;
                continue;
            }

            if let Err(e) = self
                .run_once(Uri::from_maybe_shared(node_info.grpc_uri)?)
                .await
            {
                warn!("Error running against node: {e}");
            }
        }
    }

    pub async fn run_forever_with_uri(
        &self,
        mut shutdown: mpsc::Receiver<()>,
        scheduler_uri: String,
    ) -> Result<(), anyhow::Error> {
        let scheduler = self.init_scheduler(scheduler_uri).await?;
        tokio::select! {
            run_forever_inner_res = self.run_forever_inner(scheduler.clone()) => {
                error!("Inner signer loop exited unexpectedly: {run_forever_inner_res:?}");
            },
            run_forever_scheduler_res = self.run_forever_scheduler(scheduler) => {
                error!("Scheduler signer loop exited unexpectedly: {run_forever_scheduler_res:?}")
            }
            _ = shutdown.recv() => debug!("Received the signal to exit the signer loop")
        };

        info!("Exiting the signer loop");
        Ok(())
    }

    async fn run_forever_scheduler(
        &self,
        scheduler: SchedulerClient<tonic::transport::Channel>,
    ) -> Result<(), anyhow::Error> {
        loop {
            if let Err(e) = self.run_once_scheduler(scheduler.clone()).await {
                warn!("Error running schduler, trying again: {e}");
            }
        }
    }

    async fn run_once_scheduler(
        &self,
        mut scheduler: SchedulerClient<tonic::transport::Channel>,
    ) -> Result<(), anyhow::Error> {
        let (sender, rx) = mpsc::channel(1);
        let outbound = ReceiverStream::new(rx);
        // let inbound_future = scheduler.signer_requests_stream(outbound);

        let mut stream = scheduler
            .signer_requests_stream(outbound)
            .await?
            .into_inner();

        trace!("Starting to stream signer requests from scheduler");

        loop {
            match stream.message().await {
                Ok(Some(msg)) => {
                    let req_id = msg.request_id;
                    trace!("Processing scheduler request {}", req_id);
                    match msg.request {
                        Some(signer_request::Request::ApprovePairing(req)) => {
                            if let Err(e) = self
                                .process_pairing_approval(req_id, req, sender.clone())
                                .await
                            {
                                debug!("Could not process pairing approval: {:?}", e);
                            }
                        }
                        None => {
                            debug!("Received an empty signing request");
                        }
                    };
                }
                Ok(None) => {
                    debug!("End of stream, this should not happen by the server");
                    return Err(anyhow!("Scheduler closed the stream"));
                }
                Err(e) => {
                    debug!("Got an error from the scheduler {}", e);
                    return Err(anyhow!("Scheduler stream error {e:?}"));
                }
            };
        }
    }

    async fn process_pairing_approval(
        &self,
        req_id: u32,
        req: ApprovePairingRequest,
        stream: mpsc::Sender<SignerResponse>,
    ) -> Result<()> {
        // Can only sign for it's own id!
        let node_id = self.node_id();

        let mut data = vec![];
        data.put(req.device_id.as_bytes());
        data.put_u64(req.timestamp);
        data.put(&node_id[..]);
        data.put(req.device_name.as_bytes());
        data.put(req.restrictions.as_bytes());

        // Check that the signature matches
        let pk = UnparsedPublicKey::new(&ECDSA_P256_SHA256_FIXED, req.pubkey.clone());
        pk.verify(&data, &req.sig)
            .map_err(|e| Error::ApprovePairingRequestError(e.to_string()))?;

        // Decode rune as we expect it to be in byte format in the pending request.
        let rune = general_purpose::URL_SAFE
            .decode(req.rune.clone())
            .map_err(|e| Error::ApprovePairingRequestError(e.to_string()))?;

        // Check that the rune matches
        self.verify_rune(PendingRequest {
            request: vec![],
            uri: "/cln.Node/ApprovePairing".to_string(),
            signature: req.sig,
            pubkey: req.pubkey,
            timestamp: req.timestamp,
            rune,
        })?;

        let restrs: Vec<Vec<&str>> = req
            .restrictions
            .split('&')
            .map(|s| s.split('|').collect::<Vec<&str>>())
            .collect();

        // Create the rune that approves pairing
        let rune = self.create_rune(None, restrs)?;

        let _ = stream
            .send(SignerResponse {
                request_id: req_id,
                response: Some(signer_response::Response::ApprovePairing(
                    ApprovePairingResponse {
                        device_id: req.device_id,
                        node_id,
                        rune,
                    },
                )),
            })
            .await?;

        Ok(())
    }

    // TODO See comment on `sign_device_key`.
    pub fn sign_challenge(&self, challenge: Vec<u8>) -> Result<Vec<u8>, anyhow::Error> {
        if challenge.len() != 32 {
            return Err(anyhow!("challenge is not 32 bytes long"));
        }
        let (sig, _) = self.sign_message(challenge)?;
        Ok(sig)
    }

    /// Signs the devices public key. This signature is meant to be appended
    /// to any payload signed by the device so that the signer can verify that
    /// it knows the device.
    ///
    // TODO Use a lower-level API that bypasses the LN message
    // prefix. This will allow us to re-expose the sign-message API to
    // node users.
    pub fn sign_device_key(&self, key: &[u8]) -> Result<Vec<u8>, anyhow::Error> {
        if key.len() != 65 {
            return Err(anyhow!("key is not 65 bytes long"));
        }
        let (sig, _) = self.sign_message(key.to_vec())?;
        Ok(sig)
    }

    /// Signs a message with the hsmd client. Returns a tuple with the signature
    /// and the unmodified recovery id.
    pub fn sign_message(&self, msg: Vec<u8>) -> Result<(Vec<u8>, u8), anyhow::Error> {
        if msg.len() > u16::MAX as usize {
            return Err(anyhow!("Message exceeds max len of {}", u16::MAX));
        }

        let len = u16::to_be_bytes(msg.len() as u16);
        if len.len() != 2 {
            return Err(anyhow!(
                "Message to be signed has unexpected len {}",
                len.len()
            ));
        }

        let req = vls_protocol::msgs::SignMessage {
            message: Octets(msg),
        };
        let response = self
            .handler()?
            .handle(vls_protocol::msgs::Message::SignMessage(req))
            .unwrap();

        // The signature returned by VLS consists of the signature with the
        // recovery id appended.
        let complete_sig = response.as_vec();
        let sig = complete_sig[2..66].to_vec();
        let recovery_id = complete_sig[66];
        Ok((sig, recovery_id))
    }

    /// Signs an invoice.
    pub fn sign_invoice(&self, msg: Vec<u8>) -> Result<Vec<u8>, anyhow::Error> {
        if msg.len() > u16::MAX as usize {
            return Err(anyhow!("Message exceeds max len of {}", u16::MAX));
        }

        let sig = self
            .handler()?
            .handle(vls_protocol::msgs::from_vec(msg.clone())?)
            .map_err(|_| anyhow!("Sign invoice failed"))?;
        Ok(sig.as_vec()[2..67].to_vec())
    }

    /// Create a Node stub from this instance of the signer, configured to
    /// talk to the corresponding node.
    pub async fn node<Creds>(&self, creds: Creds) -> Result<Client, anyhow::Error>
    where
        Creds: TlsConfigProvider + RuneProvider,
    {
        node::Node::new(self.node_id(), creds)?.schedule().await
    }

    pub fn version(&self) -> &'static str {
        VERSION
    }

    /// Creates a base64 string called a rune which is used to authorize
    /// commands on the node and to issue signatures from the signer. Each new
    /// rune must contain a `pubkey` field that equals the public key that is
    /// used to sign-off signature requests. Nobody can remove restrictions from
    /// a rune.
    ///
    /// If a `rune` is supplied the restrictions are added to this rune. This
    /// way one can invoke a rune that only allows for a subset of commands.
    ///
    /// `restrictions` is a vector of restrictions where each restriction itself
    /// is a vector of one ore more alternatives.
    ///
    /// - =: passes if equal ie. identical. e.g. method=withdraw
    /// - /: not equals, e.g. method/withdraw
    /// - ^: starts with, e.g. id^024b9a1fa8e006f1e3937f
    /// - $: ends with, e.g. id$381df1cc449605.
    /// - ~: contains, e.g. id~006f1e3937f65f66c40.
    /// - <: is a decimal integer, and is less than. e.g. time<1656759180
    /// - \>: is a decimal integer, and is greater than. e.g. time>1656759180
    /// - {: preceeds in alphabetical order (or matches but is shorter), e.g. id{02ff.
    /// - }: follows in alphabetical order (or matches but is longer), e.g. id}02ff.
    /// - #: a comment, ignored, e.g. dumb example#.
    /// - !: only passes if the name does not exist. e.g. something!.  Every other operator except # fails if name does not exist!
    ///
    /// # Examples
    /// This creates a fresh rune that is only restricted to a pubkey:
    ///
    /// `create_rune(None, vec![vec!["pubkey=000000"]])`
    ///
    /// "wjEjvKoFJToMLBv4QVbJpSbMoGFlnYVxs8yy40PIBgs9MC1nbDAmcHVia2V5PTAwMDAwMA"
    ///
    /// This adds a restriction to the rune, in this case a restriction that only
    /// allows to call methods that start with "list" or "get", basically a
    /// read-only rune:
    ///
    /// `create_rune("wjEjvKoFJToMLBv4QVbJpSbMoGFlnYVxs8yy40PIBgs9MC1nbDAmcHVia2V5PTAwMDAwMA", vec![vec!["method^list", "method^get"]])`
    ///
    pub fn create_rune(
        &self,
        rune: Option<&str>,
        restrictions: Vec<Vec<&str>>,
    ) -> Result<String, anyhow::Error> {
        if let Some(rune) = rune {
            // We got a rune, add restrictions to it!
            let mut rune: Rune = Rune::from_base64(rune)?;
            restrictions.into_iter().for_each(|alts| {
                let joined = alts.join("|");
                _ = rune.add_restriction(joined.as_str())
            });
            return Ok(rune.to_base64());
        } else {
            let res: Vec<Restriction> = restrictions
                .into_iter()
                .map(|alts| {
                    let joined = alts.join("|");
                    Restriction::try_from(joined.as_str())
                })
                .collect::<Result<Vec<Restriction>, RuneError>>()?;

            // New rune, we need a unique id.
            // FIXME: Add a counter that persists in SSS.
            let unique_id = 0;

            // Check that at least one restriction has a `pubkey` field set.
            let has_pubkey_field = res.iter().any(|r: &Restriction| {
                r.alternatives
                    .iter()
                    .any(|a| a.get_field() == *"pubkey" && a.get_condition() == Condition::Equal)
            });
            if !has_pubkey_field {
                return Err(anyhow!("Missing a restriction on the pubkey"));
            }

            let rune = Rune::new(
                self.master_rune.authcode(),
                res,
                Some(unique_id.to_string()),
                Some(RUNE_VERSION.to_string()),
            )?;
            Ok(rune.to_base64())
        }
    }
}

/// Look through the context requests and update the state
/// accordingly. This is useful to modify allowlists and invoice lists
/// extracted from the authenticated requests.
fn update_state_from_context(
    requests: &Vec<model::Request>,
    handler: &handler::RootHandler,
) -> Result<(), Error> {
    log::debug!("Updating state from {} context request", requests.len());
    let node = handler.node();

    requests.iter().for_each(|r| {
        if let Err(e) = update_state_from_request(r, &node) {
            log::warn!("Failed to update state from request: {:?}", e);
        }
    });
    Ok(())
}

fn update_state_from_request(
    request: &model::Request,
    node: &lightning_signer::node::Node,
) -> Result<(), Error> {
    use lightning_signer::invoice::Invoice;
    use std::str::FromStr;
    match request {
        model::Request::SendPay(model::cln::SendpayRequest {
            bolt11: Some(inv), ..
        }) => match Invoice::from_str(inv) {
            Ok(invoice) => {
                log::debug!(
                    "Adding invoice {:?} as side-effect of this sendpay {:?}",
                    invoice,
                    request
                );
                if let Err(e) = node.add_invoice(invoice) {
                    log::warn!("Failed to add invoice to node state: {:?}", e);
                }
            }
            Err(e) => {
                log::warn!("Failed to parse invoice from sendpay request: {:?}", e);
            }
        },
        _ => {}
    }

    Ok(())
}

/// Used to decode incoming requests into their corresponding protobuf
/// message. This is used by the E2E verification to verify that
/// incoming requests match up with the user intent. User intent here
/// refers to there being a user-signed command that matches the
/// effects we are being asked to sign off.
fn decode_request(r: crate::pb::PendingRequest) -> Result<model::Request, anyhow::Error> {
    // Strip the compressions flag (1 byte) and the length prefix (4
    // bytes big endian) and we're left with just the payload. See
    // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
    // for technical details.
    //
    // Notice that we assume that the compression flag is off.
    assert_eq!(r.request[0], 0u8);
    let payload = &r.request[5..];

    crate::signer::model::cln::decode_request(&r.uri, payload)
        .or_else(|_| crate::signer::model::greenlight::decode_request(&r.uri, payload))
}

/// A `(request, response)`-tuple passed to the scheduler to allow
/// signerless startups.
pub struct StartupMessage {
    request: Vec<u8>,
    response: Vec<u8>,
}

impl From<StartupMessage> for crate::pb::scheduler::StartupMessage {
    fn from(r: StartupMessage) -> Self {
        Self {
            request: r.request,
            response: r.response,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::credentials;
    use crate::pb;
    use serde_json::json;
    use vls_protocol::msgs::SerBolt;

    fn test_override_config(note: Option<&str>) -> StateSignatureOverrideConfig {
        StateSignatureOverrideConfig {
            ack: STATE_SIGNATURE_OVERRIDE_ACK.to_string(),
            note: note.map(str::to_string),
        }
    }

    fn mk_signer(mode: StateSignatureMode) -> Signer {
        Signer::new_with_config(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
            SignerConfig {
                state_signature_mode: mode,
                state_signature_override: None,
            },
        )
        .unwrap()
    }

    fn mk_signer_with_override(mode: StateSignatureMode, note: Option<&str>) -> Signer {
        Signer::new_with_config(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
            SignerConfig {
                state_signature_mode: mode,
                state_signature_override: Some(test_override_config(note)),
            },
        )
        .unwrap()
    }

    fn heartbeat_raw() -> Vec<u8> {
        vls_protocol::msgs::GetHeartbeat {}.as_vec()
    }

    fn mk_state_entry(key: &str, version: u64, value: serde_json::Value) -> SignerStateEntry {
        SignerStateEntry {
            version,
            key: key.to_string(),
            value: serde_json::to_vec(&value).unwrap(),
            signature: vec![],
        }
    }

    /// We should not sign messages that we get from the node, since
    /// we're using the sign_message RPC message to create TLS
    /// certificate attestations. We can remove this limitation once
    /// we switch over to a salted hash when signing those
    /// attestations.
    #[tokio::test]
    async fn test_sign_message_rejection() {
        let signer = Signer::new(
            vec![0 as u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
        )
        .unwrap();

        let msg = hex::decode("0017000B48656c6c6f20776f726c64").unwrap();
        assert!(signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: msg,
                signer_state: vec![],
                requests: Vec::new(),
            },)
            .await
            .is_err());
    }

    /// We should reject a signing request with an empty message.
    #[tokio::test]
    async fn test_empty_message() {
        let signer = Signer::new(
            vec![0 as u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
        )
        .unwrap();

        assert_eq!(
            signer
                .process_request(HsmRequest {
                    request_id: 0,
                    context: None,
                    raw: vec![],
                    signer_state: vec![],
                    requests: Vec::new(),
                },)
                .await
                .unwrap_err()
                .to_string(),
            *"protocol error: ShortRead"
        )
    }

    #[test]
    fn test_state_signature_roundtrip() {
        let signer = mk_signer(StateSignatureMode::Soft);
        let mut entry = mk_state_entry("state/test", 1, json!({"v": 1}));
        entry.signature = signer
            .sign_state_payload(&entry.key, entry.version, &entry.value)
            .unwrap();
        assert!(signer.verify_state_entry_signature(&entry).is_ok());
    }

    #[tokio::test]
    async fn test_soft_mode_accepts_missing_signature_and_repairs() {
        let signer = mk_signer(StateSignatureMode::Soft);
        let key = "state/test".to_string();
        let req = HsmRequest {
            request_id: 42,
            context: None,
            raw: heartbeat_raw(),
            signer_state: vec![mk_state_entry(&key, 1, json!({"v": 1}))],
            requests: vec![],
        };
        let response = signer.process_request(req).await.unwrap();
        let repaired = response
            .signer_state
            .iter()
            .find(|e| e.key == key)
            .expect("expected repaired entry in diff");
        assert_eq!(repaired.signature.len(), COMPACT_SIGNATURE_LEN);
    }

    #[tokio::test]
    async fn test_soft_mode_rejects_invalid_signature() {
        let signer = mk_signer(StateSignatureMode::Soft);
        let mut entry = mk_state_entry("state/test", 1, json!({"v": 1}));
        entry.signature = vec![1u8; COMPACT_SIGNATURE_LEN];
        let err = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![entry],
                requests: vec![],
            })
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("invalid state signature"));
    }

    #[tokio::test]
    async fn test_hard_mode_rejects_missing_signature() {
        let signer = mk_signer(StateSignatureMode::Hard);
        let err = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![mk_state_entry("state/test", 1, json!({"v": 1}))],
                requests: vec![],
            })
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("missing state signature"));
    }

    #[tokio::test]
    async fn test_hard_mode_rejects_invalid_signature() {
        let signer = mk_signer(StateSignatureMode::Hard);
        let mut entry = mk_state_entry("state/test", 1, json!({"v": 1}));
        entry.signature = vec![2u8; COMPACT_SIGNATURE_LEN];
        let err = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![entry],
                requests: vec![],
            })
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("invalid state signature"));
    }

    #[tokio::test]
    async fn test_hard_mode_accepts_valid_signature() {
        let signer = mk_signer(StateSignatureMode::Hard);
        let mut entry = mk_state_entry("state/test", 1, json!({"v": 1}));
        entry.signature = signer
            .sign_state_payload(&entry.key, entry.version, &entry.value)
            .unwrap();

        let res = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![entry],
                requests: vec![],
            })
            .await;
        assert!(res.is_ok());
    }

    #[tokio::test]
    async fn test_off_mode_accepts_invalid_and_missing_signatures() {
        let signer = mk_signer(StateSignatureMode::Off);
        let mut invalid = mk_state_entry("state/invalid", 1, json!({"v": 1}));
        invalid.signature = vec![3u8; COMPACT_SIGNATURE_LEN];
        let missing = mk_state_entry("state/missing", 1, json!({"v": 2}));
        let res = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![invalid, missing],
                requests: vec![],
            })
            .await;
        assert!(res.is_ok());
    }

    #[tokio::test]
    async fn test_soft_mode_override_accepts_invalid_across_requests() {
        let signer = mk_signer_with_override(StateSignatureMode::Soft, Some("test override"));

        let mut invalid1 = mk_state_entry("state/invalid1", 1, json!({"v": 1}));
        invalid1.signature = vec![4u8; COMPACT_SIGNATURE_LEN];
        signer
            .process_request(HsmRequest {
                request_id: 1,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![invalid1],
            requests: vec![],
            })
            .await
            .unwrap();
        let snapshot1: Vec<SignerStateEntry> = {
            let state_guard = signer.state.lock().unwrap();
            state_guard.clone().into()
        };
        let persisted1 = snapshot1.iter().find(|e| e.key == "state/invalid1").unwrap();
        assert_eq!(persisted1.signature, vec![4u8; COMPACT_SIGNATURE_LEN]);

        let mut invalid2 = mk_state_entry("state/invalid2", 1, json!({"v": 2}));
        invalid2.signature = vec![5u8; COMPACT_SIGNATURE_LEN];
        let res2 = signer
            .process_request(HsmRequest {
                request_id: 2,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![invalid2],
                requests: vec![],
            })
            .await;
        assert!(res2.is_ok());

        let snapshot2: Vec<SignerStateEntry> = {
            let state_guard = signer.state.lock().unwrap();
            state_guard.clone().into()
        };
        let persisted2 = snapshot2.iter().find(|e| e.key == "state/invalid2").unwrap();
        assert_eq!(persisted2.signature, vec![5u8; COMPACT_SIGNATURE_LEN]);
    }

    #[tokio::test]
    async fn test_hard_mode_override_accepts_missing_across_requests() {
        let signer = mk_signer_with_override(StateSignatureMode::Hard, Some("test override"));

        for (key, request_id) in [("state/missing1", 10u32), ("state/missing2", 11u32)] {
            signer
                .process_request(HsmRequest {
                    request_id,
                    context: None,
                    raw: heartbeat_raw(),
                    signer_state: vec![mk_state_entry(key, 1, json!({"v": request_id}))],
                    requests: vec![],
                })
                .await
                .unwrap();
            let snapshot: Vec<SignerStateEntry> = {
                let state_guard = signer.state.lock().unwrap();
                state_guard.clone().into()
            };
            let persisted = snapshot.iter().find(|entry| entry.key == key).unwrap();
            assert_eq!(persisted.signature.len(), COMPACT_SIGNATURE_LEN);
        }
    }

    #[tokio::test]
    async fn test_hard_mode_override_accepts_invalid_signature() {
        let signer = mk_signer_with_override(StateSignatureMode::Hard, Some("test override"));
        let mut invalid = mk_state_entry("state/invalid-hard", 1, json!({"v": 3}));
        invalid.signature = vec![6u8; COMPACT_SIGNATURE_LEN];

        signer
            .process_request(HsmRequest {
                request_id: 12,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![invalid],
                requests: vec![],
            })
            .await
            .unwrap();

        let snapshot: Vec<SignerStateEntry> = {
            let state_guard = signer.state.lock().unwrap();
            state_guard.clone().into()
        };
        let persisted = snapshot
            .iter()
            .find(|entry| entry.key == "state/invalid-hard")
            .unwrap();
        assert_eq!(persisted.signature, vec![6u8; COMPACT_SIGNATURE_LEN]);
    }

    #[tokio::test]
    async fn test_override_preserves_invalid_signature_without_touching_valid_signature() {
        let signer = mk_signer_with_override(StateSignatureMode::Soft, Some("repair"));

        let mut valid = mk_state_entry("state/valid", 1, json!({"v": 1}));
        valid.signature = signer
            .sign_state_payload(&valid.key, valid.version, &valid.value)
            .unwrap();
        let mut invalid = mk_state_entry("state/invalid", 1, json!({"v": 2}));
        invalid.signature = vec![7u8; COMPACT_SIGNATURE_LEN];

        signer
            .process_request(HsmRequest {
                request_id: 13,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![valid.clone(), invalid],
                requests: vec![],
            })
            .await
            .unwrap();

        let snapshot: Vec<SignerStateEntry> = {
            let state_guard = signer.state.lock().unwrap();
            state_guard.clone().into()
        };

        let persisted_valid = snapshot.iter().find(|entry| entry.key == "state/valid").unwrap();
        let preserved_invalid = snapshot
            .iter()
            .find(|entry| entry.key == "state/invalid")
            .unwrap();

        assert_eq!(persisted_valid.signature, valid.signature);
        assert_eq!(preserved_invalid.signature, vec![7u8; COMPACT_SIGNATURE_LEN]);
    }

    #[test]
    fn test_override_rejected_when_mode_off() {
        let signer = Signer::new_with_config(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
            SignerConfig {
                state_signature_mode: StateSignatureMode::Off,
                state_signature_override: Some(test_override_config(Some("test"))),
            },
        );
        let err = signer.err().unwrap().to_string();
        assert!(err.contains("incompatible with state signature mode off"));
    }

    #[test]
    fn test_override_rejected_for_invalid_ack() {
        let signer = Signer::new_with_config(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
            SignerConfig {
                state_signature_mode: StateSignatureMode::Soft,
                state_signature_override: Some(StateSignatureOverrideConfig {
                    ack: "WRONG".to_string(),
                    note: None,
                }),
            },
        );
        let err = signer.err().unwrap().to_string();
        assert!(err.contains("invalid state signature override ack"));
    }

    #[tokio::test]
    async fn test_malformed_state_value_returns_error() {
        let signer = mk_signer(StateSignatureMode::Soft);
        let entry = SignerStateEntry {
            version: 1,
            key: "nodes/bad".to_string(),
            value: b"{".to_vec(),
            signature: vec![],
        };
        let err = signer
            .process_request(HsmRequest {
                request_id: 0,
                context: None,
                raw: heartbeat_raw(),
                signer_state: vec![entry],
                requests: vec![],
            })
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("Failed to decode signer state"));
    }

    #[test]
    fn test_sign_message_max_size() {
        let signer = Signer::new(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
        )
        .unwrap();

        // We test if we reject a message that is too long.
        let msg = [0u8; u16::MAX as usize + 1];
        assert_eq!(
            signer.sign_message(msg.to_vec()).unwrap_err().to_string(),
            format!("Message exceeds max len of {}", u16::MAX)
        );
    }

    /// Some users were relying on the broken behavior of
    /// `bip32_ext_key`. We need to ensure that the behavior remains
    /// stable for the time being until we have ensured no users of it
    /// remain.
    #[test]
    fn test_legacy_bip32_key() {
        let signer = Signer::new(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
        )
        .unwrap();

        let bip32 = signer.legacy_bip32_ext_key();
        let expected: Vec<u8> = vec![
            4, 136, 178, 30, 2, 175, 86, 45, 251, 0, 0, 0, 0, 119, 232, 160, 181, 114, 16, 182, 23,
            70, 246, 204, 254, 122, 233, 131, 242, 174, 134, 193, 120, 104, 70, 176, 202, 168, 243,
            142, 127, 239, 60, 157, 212, 3, 162, 85, 18, 86, 240, 176, 177, 84, 94, 241, 92, 64,
            175, 69, 165, 146, 101, 79, 180, 195, 27, 117, 8, 66, 110, 100, 36, 246, 115, 48, 193,
            189, 3, 247, 195, 58, 236, 143, 230, 177, 91, 217, 66, 67, 19, 204, 22, 96, 65, 140,
            86, 195, 109, 50, 228, 94, 193, 173, 103, 252, 196, 192, 173, 243, 223,
        ];

        assert_eq!(bip32, expected);
    }

    /// We want to ensure that we can not generate a rune that is unrestricted
    /// 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,
            credentials::Nobody::default(),
        )
        .unwrap();

        let alt = "pubkey=112233";
        let wrong_alt = "pubkey^112233";

        // Check empty restrictions.
        assert!(signer.create_rune(None, vec![]).is_err());

        // Check wrong restriction.
        assert!(signer.create_rune(None, vec![vec![wrong_alt]]).is_err());

        // Check good restriction.
        assert!(signer.create_rune(None, vec![vec![alt]]).is_ok());

        // Check at least one alternative in one restriction.
        assert!(signer
            .create_rune(None, vec![vec![wrong_alt], vec![wrong_alt, alt]])
            .is_ok());
    }

    #[test]
    fn test_rune_expansion() {
        let signer = Signer::new(
            vec![0u8; 32],
            Network::Bitcoin,
            credentials::Nobody::default(),
        )
        .unwrap();
        let rune = "wjEjvKoFJToMLBv4QVbJpSbMoGFlnYVxs8yy40PIBgs9MC1nbDAmcHVia2V5PTAwMDAwMA==";

        let new_rune = signer
            .create_rune(Some(rune), vec![vec!["method^get"]])
            .unwrap();
        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,
            credentials::Nobody::default(),
        )
        .unwrap();

        // This is just a placeholder public key, could also be a different one;
        let pubkey = signer.node_id();
        let pubkey_rest = format!("pubkey={}", hex::encode(&pubkey));

        // Create a rune that allows methods that start with `create`.
        let rune = signer
            .create_rune(None, vec![vec![&pubkey_rest], vec!["method^create"]])
            .unwrap();

        // A method/uri that starts with `create` is ok.
        let uri = "/cln.Node/CreateInvoice".to_string();
        let r = pb::PendingRequest {
            request: vec![],
            uri,
            signature: vec![],
            pubkey: pubkey.clone(),
            timestamp: 0,
            rune: general_purpose::URL_SAFE.decode(&rune).unwrap(),
        };
        assert!(signer.verify_rune(r).is_ok());

        // method/uri `Pay` is not allowed by the rune.
        let uri = "/cln.Node/Pay".to_string();
        let r = pb::PendingRequest {
            request: vec![],
            uri,
            signature: vec![],
            pubkey: pubkey.clone(),
            timestamp: 0,
            rune: general_purpose::URL_SAFE.decode(&rune).unwrap(),
        };
        assert!(signer.verify_rune(r).is_err());

        // The `greenlight.Node` service also needs to be accepted for
        // setting the `close_to_addr`.
        let uri = "/greenlight.Node/CreateInvoice".to_string();
        let r = pb::PendingRequest {
            request: vec![],
            uri,
            signature: vec![],
            pubkey: pubkey.clone(),
            timestamp: 0,
            rune: general_purpose::URL_SAFE.decode(&rune).unwrap(),
        };
        assert!(signer.verify_rune(r).is_ok());

        // A service other than `cln.Node` and `greenlight.Node` is
        // not allowed.
        let uri = "/wrong.Service/CreateInvoice".to_string();
        let r = pb::PendingRequest {
            request: vec![],
            uri,
            signature: vec![],
            pubkey: pubkey.clone(),
            timestamp: 0,
            rune: general_purpose::URL_SAFE.decode(&rune).unwrap(),
        };
        assert!(signer.verify_rune(r).is_err());
    }

    #[test]
    fn test_empty_rune_is_valid() {
        let creds = credentials::Nobody::default();
        let signer = Signer::new(vec![0u8; 32], Network::Bitcoin, creds).unwrap();

        // This is just a placeholder public key, could also be a different one;
        let pubkey = signer.node_id();
        let pubkey_rest = format!("pubkey={}", hex::encode(&pubkey));

        let rune = signer.create_rune(None, vec![vec![&pubkey_rest]]).unwrap();
        let uri = "/cln.Node/Pay".to_string();
        assert!(signer
            .verify_rune(crate::pb::PendingRequest {
                request: vec![],
                uri,
                signature: vec![],
                pubkey,
                timestamp: 0,
                rune: general_purpose::URL_SAFE.decode(rune).unwrap(),
            })
            .is_ok());
    }

    #[test]
    fn test_empty_rune_checks_pubkey() {
        let creds = credentials::Nobody::default();
        let signer = Signer::new(vec![0u8; 32], Network::Bitcoin, creds).unwrap();

        // This is just a placeholder public key, could also be a different one;
        let pubkey = signer.node_id();
        let pubkey_rest = format!("pubkey={}", hex::encode(&pubkey));

        let rune = signer.create_rune(None, vec![vec![&pubkey_rest]]).unwrap();
        let uri = "/cln.Node/Pay".to_string();
        assert!(signer
            .verify_rune(crate::pb::PendingRequest {
                request: vec![],
                uri,
                signature: vec![],
                pubkey: hex::decode("33aabb").unwrap(),
                timestamp: 0,
                rune: general_purpose::URL_SAFE.decode(rune).unwrap(),
            })
            .is_err());
    }
}