Merge pull request #282 from kinode-dao/dr/add-ws-routing-flag-and-enforce

fix: enforce ws-routing port

Cargo.lock

@@ -2572,7 +2572,7 @@ dependencies = [
 
 [[package]]
 name = "kinode"
-version = "0.6.1"
+version = "0.6.2"
 dependencies = [
  "aes-gcm",
  "alloy-primitives",
@@ -2645,7 +2645,7 @@ dependencies = [
 
 [[package]]
 name = "kinode_lib"
-version = "0.6.1"
+version = "0.6.2"
 dependencies = [
  "lib",
 ]
@@ -2738,7 +2738,7 @@ checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67"
 
 [[package]]
 name = "lib"
-version = "0.6.1"
+version = "0.6.2"
 dependencies = [
  "alloy-rpc-types",
  "lazy_static",

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "kinode_lib"
 authors = ["KinodeDAO"]
-version = "0.6.1"
+version = "0.6.2"
 edition = "2021"
 description = "A general-purpose sovereign cloud computing platform"
 homepage = "https://kinode.org"

kinode/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "kinode"
 authors = ["KinodeDAO"]
-version = "0.6.1"
+version = "0.6.2"
 edition = "2021"
 description = "A general-purpose sovereign cloud computing platform"
 homepage = "https://kinode.org"

kinode/src/eth/mod.rs

@@ -251,7 +251,7 @@ async fn handle_network_error(
     // if we hold active subscriptions for the remote node that this error refers to,
     // close them here -- they will need to resubscribe
     // TODO is this necessary?
-    if let Some(sub_map) = active_subscriptions.get(&wrapped_error.error.target) {
+    if let Some((_who, sub_map)) = active_subscriptions.remove(&wrapped_error.error.target) {
         for (_sub_id, sub) in sub_map.iter() {
             if let ActiveSub::Local(handle) = sub {
                 verbose_print(

kinode/src/http/utils.rs

@@ -100,18 +100,18 @@ pub fn deserialize_headers(hashmap: HashMap<String, String>) -> HeaderMap {
     header_map
 }
 
-pub async fn find_open_port(start_at: u16, end_at: u16) -> Option<u16> {
+pub async fn find_open_port(start_at: u16, end_at: u16) -> Option<TcpListener> {
     for port in start_at..end_at {
         let bind_addr = format!("0.0.0.0:{}", port);
-        if is_port_available(&bind_addr).await {
+        if let Some(bound) = is_port_available(&bind_addr).await {
-            return Some(port);
+            return Some(bound);
         }
     }
     None
 }
 
-pub async fn is_port_available(bind_addr: &str) -> bool {
+pub async fn is_port_available(bind_addr: &str) -> Option<TcpListener> {
-    TcpListener::bind(bind_addr).await.is_ok()
+    TcpListener::bind(bind_addr).await.ok()
 }
 
 pub fn _binary_encoded_string_to_bytes(s: &str) -> Vec<u8> {

kinode/src/main.rs

@@ -44,6 +44,7 @@ const DEFAULT_PROVIDERS_MAINNET: &str = include_str!("eth/default_providers_main
 async fn serve_register_fe(
     home_directory_path: &str,
     our_ip: String,
+    ws_networking: (tokio::net::TcpListener, bool),
     http_server_port: u16,
     testnet: bool,
 ) -> (Identity, Vec<u8>, Keyfile) {
@@ -66,7 +67,7 @@ async fn serve_register_fe(
 
     let (tx, mut rx) = mpsc::channel::<(Identity, Keyfile, Vec<u8>)>(1);
     let (our, decoded_keyfile, encoded_keyfile) = tokio::select! {
-        _ = register::register(tx, kill_rx, our_ip, http_server_port, disk_keyfile, testnet) => {
+        _ = register::register(tx, kill_rx, our_ip, ws_networking, http_server_port, disk_keyfile, testnet) => {
             panic!("registration failed")
         }
         Some((our, decoded_keyfile, encoded_keyfile)) = rx.recv() => {
@@ -97,6 +98,11 @@ async fn main() {
             arg!(--port <PORT> "Port to bind [default: first unbound at or above 8080]")
                 .value_parser(value_parser!(u16)),
         )
+        .arg(
+            arg!(--"ws-port" <PORT> "Kinode internal WebSockets protocol port [default: first unbound at or above 9000]")
+                .alias("network-router-port")
+                .value_parser(value_parser!(u16)),
+        )
         .arg(
             arg!(--testnet "If set, use Sepolia testnet")
                 .default_value("false")
@@ -118,11 +124,6 @@ async fn main() {
     let app = app
         .arg(arg!(--password <PASSWORD> "Networking password"))
         .arg(arg!(--"fake-node-name" <NAME> "Name of fake node to boot"))
-        .arg(
-            arg!(--"network-router-port" <PORT> "Network router port")
-                .default_value("9001")
-                .value_parser(value_parser!(u16)),
-        )
         .arg(
             arg!(--detached <IS_DETACHED> "Run in detached mode (don't accept input)")
                 .action(clap::ArgAction::SetTrue),
@@ -131,21 +132,16 @@ async fn main() {
     let matches = app.get_matches();
 
     let home_directory_path = matches.get_one::<String>("home").unwrap();
-    let (port, port_flag_used) = match matches.get_one::<u16>("port") {
-        Some(port) => (*port, true),
-        None => (8080, false),
-    };
     let on_testnet = *matches.get_one::<bool>("testnet").unwrap();
 
+    let http_port = matches.get_one::<u16>("port");
+    let ws_networking_port = matches.get_one::<u16>("ws-port");
+
     #[cfg(not(feature = "simulation-mode"))]
     let is_detached = false;
     #[cfg(feature = "simulation-mode")]
-    let (password, network_router_port, fake_node_name, is_detached) = (
+    let (password, fake_node_name, is_detached) = (
         matches.get_one::<String>("password"),
-        matches
-            .get_one::<u16>("network-router-port")
-            .unwrap()
-            .clone(),
         matches.get_one::<String>("fake-node-name"),
         *matches.get_one::<bool>("detached").unwrap(),
     );
@@ -262,35 +258,59 @@ async fn main() {
         }
     };
 
-    let http_server_port = if port_flag_used {
+    let http_server_port = if let Some(port) = http_port {
-        match http::utils::find_open_port(port, port + 1).await {
+        match http::utils::find_open_port(*port, port + 1).await {
-            Some(port) => port,
+            Some(bound) => bound.local_addr().unwrap().port(),
             None => {
                 println!(
                     "error: couldn't bind {}; first available port found was {}. \
                     Set an available port with `--port` and try again.",
                     port,
-                    http::utils::find_open_port(port, port + 1000)
+                    http::utils::find_open_port(*port, port + 1000)
                         .await
-                        .expect("no ports found in range"),
+                        .expect("no ports found in range")
+                        .local_addr()
+                        .unwrap()
+                        .port(),
                 );
                 panic!();
             }
         }
     } else {
-        match http::utils::find_open_port(port, port + 1000).await {
+        match http::utils::find_open_port(8080, 8999).await {
-            Some(port) => port,
+            Some(bound) => bound.local_addr().unwrap().port(),
             None => {
                 println!(
-                    "error: couldn't bind any ports between {port} and {}. \
+                    "error: couldn't bind any ports between 8080 and 8999. \
-                    Set an available port with `--port` and try again.",
+                    Set an available port with `--port` and try again."
-                    port + 1000,
                 );
                 panic!();
             }
         }
     };
 
+    // if the --ws-port flag is used, bind to that port right away.
+    // if the flag is not used, find the first available port between 9000 and 65535.
+    // NOTE: if the node has a different port specified in its onchain (direct) id,
+    // booting will fail if the flag was used to select a different port.
+    // if the flag was not used, the bound port will be dropped in favor of the onchain port.
+
+    let (ws_tcp_handle, flag_used) = if let Some(port) = ws_networking_port {
+        (
+            http::utils::find_open_port(*port, port + 1)
+                .await
+                .expect("ws-port selected with flag could not be bound"),
+            true,
+        )
+    } else {
+        (
+            http::utils::find_open_port(9000, 65535)
+                .await
+                .expect("no ports found in range 9000-65535 for websocket server"),
+            false,
+        )
+    };
+
     println!(
         "login or register at http://localhost:{}\r",
         http_server_port
@@ -300,6 +320,7 @@ async fn main() {
     let (our, encoded_keyfile, decoded_keyfile) = serve_register_fe(
         home_directory_path,
         our_ip.to_string(),
+        (ws_tcp_handle, flag_used),
         http_server_port,
         on_testnet, // true if testnet mode
     )
@@ -313,7 +334,8 @@ async fn main() {
                     serve_register_fe(
                         &home_directory_path,
                         our_ip.to_string(),
-                        http_server_port.clone(),
+                        (ws_tcp_handle, flag_used),
+                        http_server_port,
                         on_testnet, // true if testnet mode
                     )
                     .await
@@ -514,7 +536,7 @@ async fn main() {
     ));
     #[cfg(feature = "simulation-mode")]
     tasks.spawn(net::mock_client(
-        network_router_port,
+        *ws_networking_port.unwrap_or(&9000),
         our.name.clone(),
         kernel_message_sender.clone(),
         net_message_receiver,

kinode/src/register.rs

@@ -48,6 +48,8 @@ sol! {
     ) public view virtual returns (bool authed);
 
     function nodes(bytes32) external view returns (address, uint96);
+
+    function ip(bytes32) external view returns (uint128, uint16, uint16, uint16, uint16);
 }
 
 pub fn _ip_to_number(ip: &str) -> Result<u32, &'static str> {
@@ -115,7 +117,8 @@ pub async fn register(
     tx: RegistrationSender,
     kill_rx: oneshot::Receiver<bool>,
     ip: String,
-    port: u16,
+    ws_networking: (tokio::net::TcpListener, bool),
+    http_port: u16,
     keyfile: Option<Vec<u8>>,
     testnet: bool,
 ) {
@@ -124,14 +127,12 @@ pub async fn register(
     let net_keypair = Arc::new(serialized_networking_keypair.as_ref().to_vec());
     let tx = Arc::new(tx);
 
-    // TODO: if IP is localhost, don't allow registration as direct
+    let ws_port = ws_networking.0.local_addr().unwrap().port();
-    let ws_port = crate::http::utils::find_open_port(9000, 65535)
+    let ws_flag_used = ws_networking.1;
-        .await
-        .expect(
-            "Unable to find free port between 9000 and 65535 for a new websocket, are you kidding?",
-        );
 
-    // This is a temporary identity, passed to the UI. If it is confirmed through a /boot or /confirm-change-network-keys, then it will be used to replace the current identity
+    // This is a **temporary** identity, passed to the UI.
+    // If it is confirmed through a /boot or /confirm-change-network-keys,
+    // then it will be used to replace the current identity.
     let our_temp_id = Arc::new(Identity {
         networking_key: format!("0x{}", public_key),
         name: "".to_string(),
@@ -167,6 +168,7 @@ pub async fn register(
     let net_keypair = warp::any().map(move || net_keypair.clone());
     let tx = warp::any().map(move || tx.clone());
     let ip = warp::any().map(move || ip.clone());
+    let ws_port = warp::any().map(move || if ws_flag_used { Some(ws_port) } else { None });
 
     let static_files = warp::path("static").and(static_dir!("src/register-ui/build/static/"));
 
@@ -197,7 +199,29 @@ pub async fn register(
             } else {
                 warp::reply::json(&"0xa")
             }
-        }));
+        }))
+        .or(warp::path("our").and(warp::get()).and(keyfile.clone()).map(
+            move |keyfile: Option<Vec<u8>>| {
+                if let Some(keyfile) = keyfile {
+                    if let Ok((username, _, _, _, _, _)) = bincode::deserialize::<(
+                        String,
+                        Vec<String>,
+                        Vec<u8>,
+                        Vec<u8>,
+                        Vec<u8>,
+                        Vec<u8>,
+                    )>(keyfile.as_ref())
+                    {
+                        return warp::reply::html(username);
+                    }
+                }
+                warp::reply::html(String::new())
+            },
+        ));
+
+    let boot_provider = provider.clone();
+    let login_provider = provider.clone();
+    let import_provider = provider.clone();
 
     let api = warp::path("info")
         .and(
@@ -225,7 +249,7 @@ pub async fn register(
                 .and(our_temp_id.clone())
                 .and(net_keypair.clone())
                 .and_then(move |boot_info, tx, our_temp_id, net_keypair| {
-                    let provider = provider.clone();
+                    let boot_provider = boot_provider.clone();
                     handle_boot(
                         boot_info,
                         tx,
@@ -233,7 +257,7 @@ pub async fn register(
                         net_keypair,
                         testnet,
                         kns_address,
-                        provider,
+                        boot_provider,
                     )
                 }),
         ))
@@ -242,17 +266,33 @@ pub async fn register(
                 .and(warp::body::content_length_limit(1024 * 16))
                 .and(warp::body::json())
                 .and(ip.clone())
+                .and(ws_port.clone())
                 .and(tx.clone())
-                .and_then(handle_import_keyfile),
+                .and_then(move |boot_info, ip, ws_port, tx| {
+                    let import_provider = import_provider.clone();
+                    handle_import_keyfile(boot_info, ip, ws_port, tx, kns_address, import_provider)
+                }),
         ))
         .or(warp::path("login").and(
             warp::post()
                 .and(warp::body::content_length_limit(1024 * 16))
                 .and(warp::body::json())
                 .and(ip)
+                .and(ws_port.clone())
                 .and(tx.clone())
                 .and(keyfile.clone())
-                .and_then(handle_login),
+                .and_then(move |boot_info, ip, ws_port, tx, keyfile| {
+                    let login_provider = login_provider.clone();
+                    handle_login(
+                        boot_info,
+                        ip,
+                        ws_port,
+                        tx,
+                        keyfile,
+                        kns_address,
+                        login_provider,
+                    )
+                }),
         ))
         .or(warp::path("confirm-change-network-keys").and(
             warp::post()
@@ -276,9 +316,9 @@ pub async fn register(
         .or(api)
         .with(warp::reply::with::headers(headers));
 
-    let _ = open::that(format!("http://localhost:{}/", port));
+    let _ = open::that(format!("http://localhost:{}/", http_port));
     warp::serve(routes)
-        .bind_with_graceful_shutdown(([0, 0, 0, 0], port), async {
+        .bind_with_graceful_shutdown(([0, 0, 0, 0], http_port), async {
             kill_rx.await.ok();
         })
         .1
@@ -388,7 +428,25 @@ async fn handle_boot(
         ..Default::default()
     };
 
-    let Ok(tld) = provider.call(tx, None).await else {
+    // this call can fail if the indexer has not caught up to the transaction
+    // that just got confirmed on our frontend. for this reason, we retry
+    // the call a few times before giving up.
+
+    let mut attempts = 0;
+    let mut tld_result = Err(());
+    while attempts < 5 {
+        match provider.call(tx.clone(), None).await {
+            Ok(tld) => {
+                tld_result = Ok(tld);
+                break;
+            }
+            Err(_) => {
+                attempts += 1;
+                tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
+            }
+        }
+    }
+    let Ok(tld) = tld_result else {
         return Ok(warp::reply::with_status(
             warp::reply::json(&"Failed to fetch TLD contract for username"),
             StatusCode::INTERNAL_SERVER_ERROR,
@@ -419,13 +477,13 @@ async fn handle_boot(
 
     let Ok(authed) = provider.call(tx, None).await else {
         return Ok(warp::reply::with_status(
-            warp::reply::json(&"Failed to fetch TLD contract for username"),
+            warp::reply::json(&"Failed to fetch associated address for username"),
             StatusCode::INTERNAL_SERVER_ERROR,
         )
         .into_response());
     };
 
-    let is_ok = bool::abi_decode(&authed, false).map_err(|_| warp::reject())?;
+    let is_ok = bool::abi_decode(&authed, false).unwrap_or(false);
 
     if !is_ok {
         return Ok(warp::reply::with_status(
@@ -482,7 +540,10 @@ async fn handle_boot(
 async fn handle_import_keyfile(
     info: ImportKeyfileInfo,
     ip: String,
+    ws_networking_port: Option<u16>,
     sender: Arc<RegistrationSender>,
+    kns_address: EthAddress,
+    provider: Arc<Provider<PubSubFrontend>>,
 ) -> Result<impl Reply, Rejection> {
     // if keyfile was not present in node and is present from user upload
     let encoded_keyfile = match base64::decode(info.keyfile.clone()) {
@@ -496,58 +557,52 @@ async fn handle_import_keyfile(
         }
     };
 
-    let Some(ws_port) = crate::http::utils::find_open_port(9000, 9999).await else {
+    let (decoded_keyfile, mut our) =
+        match keygen::decode_keyfile(&encoded_keyfile, &info.password_hash) {
+            Ok(k) => {
+                let our = Identity {
+                    name: k.username.clone(),
+                    networking_key: format!(
+                        "0x{}",
+                        hex::encode(k.networking_keypair.public_key().as_ref())
+                    ),
+                    ws_routing: if k.routers.is_empty() {
+                        Some((ip, 9000))
+                    } else {
+                        None
+                    },
+                    allowed_routers: k.routers.clone(),
+                };
+
+                (k, our)
+            }
+            Err(_) => {
+                return Ok(warp::reply::with_status(
+                    warp::reply::json(&"Incorrect password_hash".to_string()),
+                    StatusCode::UNAUTHORIZED,
+                )
+                .into_response())
+            }
+        };
+
+    if let Err(e) = assign_ws_routing(&mut our, kns_address, provider, ws_networking_port).await {
         return Ok(warp::reply::with_status(
-            warp::reply::json(&"Unable to find free port"),
+            warp::reply::json(&e.to_string()),
             StatusCode::INTERNAL_SERVER_ERROR,
         )
         .into_response());
-    };
+    }
-
-    let (decoded_keyfile, our) = match keygen::decode_keyfile(&encoded_keyfile, &info.password_hash)
-    {
-        Ok(k) => {
-            let our = Identity {
-                name: k.username.clone(),
-                networking_key: format!(
-                    "0x{}",
-                    hex::encode(k.networking_keypair.public_key().as_ref())
-                ),
-                ws_routing: if k.routers.is_empty() {
-                    Some((ip, ws_port))
-                } else {
-                    None
-                },
-                allowed_routers: k.routers.clone(),
-            };
-
-            (k, our)
-        }
-        Err(_) => {
-            return Ok(warp::reply::with_status(
-                warp::reply::json(&"Incorrect password_hash".to_string()),
-                StatusCode::UNAUTHORIZED,
-            )
-            .into_response())
-        }
-    };
-
-    // if !networking_info_valid(rpc_url, ip, ws_port, &our).await {
-    //     return Ok(warp::reply::with_status(
-    //         warp::reply::json(&"Networking info invalid".to_string()),
-    //         StatusCode::UNAUTHORIZED,
-    //     )
-    //     .into_response());
-    // }
-
     success_response(sender, our, decoded_keyfile, encoded_keyfile).await
 }
 
 async fn handle_login(
     info: LoginInfo,
     ip: String,
+    ws_networking_port: Option<u16>,
     sender: Arc<RegistrationSender>,
     encoded_keyfile: Option<Vec<u8>>,
+    kns_address: EthAddress,
+    provider: Arc<Provider<PubSubFrontend>>,
 ) -> Result<impl Reply, Rejection> {
     if encoded_keyfile.is_none() {
         return Ok(warp::reply::with_status(
@@ -558,42 +613,41 @@ async fn handle_login(
     }
     let encoded_keyfile = encoded_keyfile.unwrap();
 
-    let Some(ws_port) = crate::http::utils::find_open_port(9000, 65535).await else {
+    let (decoded_keyfile, mut our) =
+        match keygen::decode_keyfile(&encoded_keyfile, &info.password_hash) {
+            Ok(k) => {
+                let our = Identity {
+                    name: k.username.clone(),
+                    networking_key: format!(
+                        "0x{}",
+                        hex::encode(k.networking_keypair.public_key().as_ref())
+                    ),
+                    ws_routing: if k.routers.is_empty() {
+                        Some((ip, 9000))
+                    } else {
+                        None
+                    },
+                    allowed_routers: k.routers.clone(),
+                };
+
+                (k, our)
+            }
+            Err(_) => {
+                return Ok(warp::reply::with_status(
+                    warp::reply::json(&"Incorrect password_hash"),
+                    StatusCode::UNAUTHORIZED,
+                )
+                .into_response())
+            }
+        };
+
+    if let Err(e) = assign_ws_routing(&mut our, kns_address, provider, ws_networking_port).await {
         return Ok(warp::reply::with_status(
-            warp::reply::json(&"Unable to find free port"),
+            warp::reply::json(&e.to_string()),
             StatusCode::INTERNAL_SERVER_ERROR,
         )
         .into_response());
-    };
+    }
-
-    let (decoded_keyfile, our) = match keygen::decode_keyfile(&encoded_keyfile, &info.password_hash)
-    {
-        Ok(k) => {
-            let our = Identity {
-                name: k.username.clone(),
-                networking_key: format!(
-                    "0x{}",
-                    hex::encode(k.networking_keypair.public_key().as_ref())
-                ),
-                ws_routing: if k.routers.is_empty() {
-                    Some((ip, ws_port))
-                } else {
-                    None
-                },
-                allowed_routers: k.routers.clone(),
-            };
-
-            (k, our)
-        }
-        Err(_) => {
-            return Ok(warp::reply::with_status(
-                warp::reply::json(&"Incorrect password_hash"),
-                StatusCode::UNAUTHORIZED,
-            )
-            .into_response())
-        }
-    };
-
     success_response(sender, our, decoded_keyfile, encoded_keyfile).await
 }
 
@@ -657,6 +711,54 @@ async fn confirm_change_network_keys(
     success_response(sender, our.clone(), decoded_keyfile, encoded_keyfile).await
 }
 
+async fn assign_ws_routing(
+    our: &mut Identity,
+    kns_address: EthAddress,
+    provider: Arc<Provider<PubSubFrontend>>,
+    ws_networking_port: Option<u16>,
+) -> anyhow::Result<()> {
+    let namehash = FixedBytes::<32>::from_slice(&keygen::namehash(&our.name));
+    let ip_call = ipCall { _0: namehash }.abi_encode();
+    let tx_input = TransactionInput::new(Bytes::from(ip_call));
+    let tx = TransactionRequest {
+        to: Some(kns_address),
+        input: tx_input,
+        ..Default::default()
+    };
+
+    let Ok(ip_data) = provider.call(tx, None).await else {
+        return Err(anyhow::anyhow!("Failed to fetch node IP data from PKI"));
+    };
+
+    let Ok((ip, ws, _wt, _tcp, _udp)) = <(u128, u16, u16, u16, u16)>::abi_decode(&ip_data, false)
+    else {
+        return Err(anyhow::anyhow!("Failed to decode node IP data from PKI"));
+    };
+
+    let node_ip = format!(
+        "{}.{}.{}.{}",
+        (ip >> 24) & 0xFF,
+        (ip >> 16) & 0xFF,
+        (ip >> 8) & 0xFF,
+        ip & 0xFF
+    );
+
+    if node_ip != *"0.0.0.0" || ws != 0 {
+        // direct node
+        if let Some(chosen_port) = ws_networking_port {
+            if chosen_port != ws {
+                return Err(anyhow::anyhow!(
+                    "Binary used --ws-port flag to set port to {}, but node is using port {} onchain.",
+                    chosen_port,
+                    ws
+                ));
+            }
+        }
+        our.ws_routing = Some((node_ip, ws));
+    }
+    Ok(())
+}
+
 async fn success_response(
     sender: Arc<RegistrationSender>,
     our: Identity,

lib/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lib"
 authors = ["KinodeDAO"]
-version = "0.6.1"
+version = "0.6.2"
 edition = "2021"
 description = "A general-purpose sovereign cloud computing platform"
 homepage = "https://kinode.org"