Merge pull request dmnd-pool#135 from Fi3/AddPoolSignature

Add pool signature

Author: jbesraa

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "demand-cli"
-version = "0.2.1"
+version = "0.2.2"
 edition = "2021"
 
 [dependencies]

Cargo.toml

@@ -7,7 +7,7 @@ use std::{
     path::PathBuf,
     time::Duration,
 };
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info};
 
 use crate::{
     shared::error::Error, HashUnit, DEFAULT_SV1_HASHPOWER, PRODUCTION_URL, STAGING_URL,
@@ -50,6 +50,8 @@ struct Args {
     monitor: bool,
     #[clap(long, short = 'u')]
     auto_update: bool,
+    #[clap(long)]
+    signature: Option<String>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -109,6 +111,7 @@ pub struct Configuration {
     api_server_port: String,
     monitor: bool,
     auto_update: bool,
+    signature: String,
 }
 impl Configuration {
     pub fn token() -> Option<String> {
@@ -213,6 +216,10 @@ impl Configuration {
         CONFIG.auto_update
     }
 
+    pub fn signature() -> String {
+        CONFIG.signature.clone()
+    }
+
     // Loads config from CLI, file, or env vars with precedence: CLI > file > env.
     fn load_config() -> Self {
         let args = Args::parse();
@@ -226,7 +233,23 @@ impl Configuration {
             .token
             .or(config.token)
             .or_else(|| std::env::var("TOKEN").ok());
-        debug!("User Token: {:?}", token);
+        println!("User Token: {:?}", token);
+
+        let signature = match args.signature {
+            Some(s) => {
+                if s.len() == 2 {
+                    println!("Signature provided: DDx{}", s);
+                    format!("DDx{}", s)
+                } else {
+                    println!("Invalid signature provided, using DDxDD");
+                    "DDxDD".to_string()
+                }
+            }
+            None => {
+                println!("Signature not provided, using DDxDD");
+                "DDxDD".to_string()
+            }
+        };
 
         let tp_address = args
             .tp_address
@@ -261,13 +284,13 @@ impl Configuration {
         let downstream_hashrate;
         if let Some(hashpower) = expected_hashrate {
             downstream_hashrate = hashpower;
-            info!(
+            println!(
                 "Using downstream hashrate: {}h/s",
                 HashUnit::format_value(hashpower)
             );
         } else {
             downstream_hashrate = DEFAULT_SV1_HASHPOWER;
-            warn!(
+            println!(
                 "No downstream hashrate provided, using default value: {}h/s",
                 HashUnit::format_value(DEFAULT_SV1_HASHPOWER)
             );
@@ -332,6 +355,7 @@ impl Configuration {
             api_server_port,
             monitor,
             auto_update,
+            signature,
         }
     }
 }

src/config.rs

@@ -31,7 +31,7 @@ mod translator;
 const TRANSLATOR_BUFFER_SIZE: usize = 32;
 const MIN_EXTRANONCE_SIZE: u16 = 6;
 const MIN_EXTRANONCE2_SIZE: u16 = 5;
-const UPSTREAM_EXTRANONCE1_SIZE: usize = 15;
+const UPSTREAM_EXTRANONCE1_SIZE: usize = 20;
 const DEFAULT_SV1_HASHPOWER: f32 = 100_000_000_000_000.0;
 const SHARE_PER_MIN: f32 = 10.0;
 const CHANNEL_DIFF_UPDTATE_INTERVAL: u32 = 10;
@@ -124,7 +124,13 @@ async fn main() {
     let mut router = router::Router::new(pool_addresses, auth_pub_k, None, None);
     let epsilon = Duration::from_millis(30_000);
     let best_upstream = router.select_pool_connect().await;
-    initialize_proxy(&mut router, best_upstream, epsilon).await;
+    initialize_proxy(
+        &mut router,
+        best_upstream,
+        epsilon,
+        Configuration::signature(),
+    )
+    .await;
     info!("exiting");
     tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
 }
@@ -133,6 +139,7 @@ async fn initialize_proxy(
     router: &mut Router,
     mut pool_addr: Option<std::net::SocketAddr>,
     epsilon: Duration,
+    signature: String,
 ) {
     loop {
         let stats_sender = api::stats::StatsSender::new();
@@ -155,17 +162,23 @@ async fn initialize_proxy(
         let sv1_ingress_abortable = ingress::sv1_ingress::start_listen_for_downstream(downs_sv1_tx);
 
         let (translator_up_tx, mut translator_up_rx) = channel(10);
-        let translator_abortable =
-            match translator::start(downs_sv1_rx, translator_up_tx, stats_sender.clone()).await {
-                Ok(abortable) => abortable,
-                Err(e) => {
-                    error!("Impossible to initialize translator: {e}");
-                    // Impossible to start the proxy so we restart proxy
-                    ProxyState::update_translator_state(TranslatorState::Down);
-                    ProxyState::update_tp_state(TpState::Down);
-                    return;
-                }
-            };
+        let translator_abortable = match translator::start(
+            downs_sv1_rx,
+            translator_up_tx,
+            stats_sender.clone(),
+            signature.clone(),
+        )
+        .await
+        {
+            Ok(abortable) => abortable,
+            Err(e) => {
+                error!("Impossible to initialize translator: {e}");
+                // Impossible to start the proxy so we restart proxy
+                ProxyState::update_translator_state(TranslatorState::Down);
+                ProxyState::update_tp_state(TpState::Down);
+                return;
+            }
+        };
 
         let (from_jdc_to_share_accounter_send, from_jdc_to_share_accounter_recv) = channel(10);
         let (from_share_accounter_to_jdc_send, from_share_accounter_to_jdc_recv) = channel(10);

src/main.rs

@@ -35,6 +35,7 @@ pub async fn start(
         Option<Address>,
     )>,
     stats_sender: crate::api::stats::StatsSender,
+    signature: String,
 ) -> Result<AbortOnDrop, Error<'static>> {
     let task_manager = TaskManager::initialize(pool_connection.clone());
     let abortable = task_manager
@@ -102,6 +103,7 @@ pub async fn start(
         target.clone(),
         diff_config.clone(),
         send_to_up,
+        signature,
     )
     .await?;
 

src/translator/mod.rs

@@ -86,6 +86,7 @@ pub struct Upstream {
     // than the configured percentage
     pub(super) difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
     pub sender: TSender<Mining<'static>>,
+    signature: String,
 }
 
 impl PartialEq for Upstream {
@@ -109,6 +110,7 @@ impl Upstream {
         target: Arc<Mutex<Vec<u8>>>,
         difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
         sender: TSender<Mining<'static>>,
+        signature: String,
     ) -> ProxyResult<'static, Arc<Mutex<Self>>> {
         Ok(Arc::new(Mutex::new(Self {
             extranonce_prefix: None,
@@ -123,6 +125,7 @@ impl Upstream {
             target,
             difficulty_config,
             sender,
+            signature,
         })))
     }
 
@@ -383,8 +386,18 @@ impl Upstream {
                             return;
                         }
                     };
+                    let signature = match self_.safe_lock(|s| s.signature.clone()) {
+                        Ok(s) => s,
+                        Err(e) => {
+                            error!("Translator upstream mutex corrupted: {e}");
+                            return;
+                        }
+                    };
 
                     sv2_submit.channel_id = channel_id;
+                    let mut extranonce = signature.as_bytes().to_vec();
+                    extranonce.extend_from_slice(&sv2_submit.extranonce.to_vec());
+                    sv2_submit.extranonce = extranonce.try_into().unwrap();
 
                     let message =
                         roles_logic_sv2::parsers::Mining::SubmitSharesExtended(sv2_submit);
@@ -506,12 +519,16 @@ impl ParseUpstreamMiningMessages<Downstream, NullDownstreamMiningSelector, NoRou
     /// role in a SV1 `mining.subscribe` message response.
     fn handle_open_extended_mining_channel_success(
         &mut self,
-        m: roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess,
+        mut m: roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess,
     ) -> Result<SendTo<Downstream>, RolesLogicError> {
         info!(
             "Handling OpenExtendedMiningChannelSuccess message from Pool for Channel Id: {}",
             m.channel_id
         );
+        let mut prefix = m.extranonce_prefix.to_vec();
+        prefix.extend_from_slice(self.signature.as_bytes());
+        m.extranonce_prefix = prefix.try_into().unwrap();
+        m.extranonce_size -= self.signature.len() as u16;
         let tproxy_e1_len =
             proxy_extranonce1_len(m.extranonce_size as usize, self.min_extranonce_size.into())
                 as u16;