[COMPLETE] MTP-MIGRATION [Some errors to be found]

src/server/api.rs

@@ -1,6 +1,6 @@
 use crate::load_keyring;
 use crate::sql::sql;
-use crate::sql::sql::{get_by_user_id, get_omikron_by_id};
+use crate::sql::sql::{get_by_user_id, get_iota_by_id, get_omikron_by_id};
 use crate::sql::user_online_tracker::get_iota_primary_omikron_connection;
 use crate::transport::omikron_manager::get_random_omikron;
 use crate::util::file_util::get_directory;
@@ -64,12 +64,13 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
         ["api", "get", "omikron"] => {
             if let Ok(omikron_conn) = get_random_omikron().await {
                 if let Some(id) = omikron_conn.get_omikron_id().await {
-                    if let Ok((public_key, ip_address)) = sql::get_omikron_by_id(id).await {
+                    if let Ok((public_key, ip_address, port)) = sql::get_omikron_by_id(id).await {
                         let mut res = JsonValue::new_object();
                         res["status"] = "success".into();
                         res["id"] = id.into();
                         res["public_key"] = public_key.to_base64().into();
                         res["ip_address"] = ip_address.into();
+                        res["port"] = port.into();
 
                         (StatusCode::OK, res.dump())
                     } else {
@@ -99,20 +100,22 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
                 let mut res = JsonValue::new_object();
                 res["status"] = "error_bad_request".into();
                 (StatusCode::BAD_REQUEST, res.dump())
-            } else if let Ok((public_key, ip_address)) = get_omikron_by_id(id).await {
+            } else if let Ok((public_key, ip_address, port)) = get_omikron_by_id(id).await {
                 let mut res = JsonValue::new_object();
                 res["status"] = "success".into();
                 res["id"] = id.into();
                 res["public_key"] = public_key.to_base64().into();
                 res["ip_address"] = ip_address.into();
+                res["port"] = port.into();
                 (StatusCode::OK, res.dump())
             } else if let Some(omikron_id) = get_iota_primary_omikron_connection(id) {
-                if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
+                if let Ok((public_key, ip_address, port)) = get_omikron_by_id(omikron_id).await {
                     let mut res = JsonValue::new_object();
                     res["status"] = "success".into();
                     res["id"] = omikron_id.into();
                     res["public_key"] = public_key.to_base64().into();
                     res["ip_address"] = ip_address.into();
+                    res["port"] = port.into();
                     (StatusCode::OK, res.dump())
                 } else {
                     let mut res = JsonValue::new_object();
@@ -122,12 +125,14 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
             } else if let Ok((_, iota_id, _, _, _, _, _, _, _, _, _, _)) = get_by_user_id(id).await
             {
                 if let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) {
-                    if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
+                    if let Ok((public_key, ip_address, port)) = get_omikron_by_id(omikron_id).await
+                    {
                         let mut res = JsonValue::new_object();
                         res["status"] = "success".into();
                         res["id"] = omikron_id.into();
                         res["public_key"] = public_key.to_base64().into();
                         res["ip_address"] = ip_address.into();
+                        res["port"] = port.into();
                         (StatusCode::OK, res.dump())
                     } else {
                         let mut res = JsonValue::new_object();
@@ -146,6 +151,29 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
             }
         }
 
+        // ==================================================
+        // GET IOTA BY ID
+        // ==================================================
+        ["api", "get", "iota", id] => {
+            let id: i64 = id.parse().unwrap_or(0);
+
+            if id == 0 {
+                let mut res = JsonValue::new_object();
+                res["status"] = "error_bad_request".into();
+                (StatusCode::BAD_REQUEST, res.dump())
+            } else if let Ok((id, public_key)) = get_iota_by_id(id).await {
+                let mut res = JsonValue::new_object();
+                res["status"] = "success".into();
+                res["iota_id"] = id.into();
+                res["public_key"] = public_key.to_base64().into();
+                (StatusCode::OK, res.dump())
+            } else {
+                let mut res = JsonValue::new_object();
+                res["status"] = "error_not_found".into();
+                (StatusCode::NOT_FOUND, res.dump())
+            }
+        }
+
         // ==================================================
         // GET ID BY USERNAME
         // ==================================================

src/server/server.rs

@@ -28,7 +28,7 @@ pub async fn start(port: u16) -> anyhow::Result<()> {
         .with_no_client_auth()
         .with_single_cert(cert_chain, key)?;
 
-    config.alpn_protocols = vec![b"h2".to_vec(), b"hmtp/1.1".to_vec()];
+    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
 
     let bind_addr = std::env::var("BIND_ADDRESS").unwrap_or_else(|_| "0.0.0.0".to_string());
     let addr = format!("{}:{}", bind_addr, port);

src/sql/sql.rs

@@ -85,11 +85,19 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
         id BIGINT NOT NULL PRIMARY KEY,
         public_key BLOB NOT NULL,
         location VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
-        ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin
+        ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
+        port INT(11) NOT NULL DEFAULT 959
         )",
     )
     .execute(&pool)
     .await;
+    // Retrofits `port` onto omikrons tables created before this column existed;
+    // `CREATE TABLE IF NOT EXISTS` above is a no-op against an already-existing table.
+    let _ = sqlx::query(
+        "ALTER TABLE omikrons ADD COLUMN IF NOT EXISTS port INT(11) NOT NULL DEFAULT 959",
+    )
+    .execute(&pool)
+    .await;
     let _ = sqlx::query(
         "CREATE TABLE IF NOT EXISTS
         notifications (
@@ -684,7 +692,7 @@ pub async fn delete_iota(id: i64) -> Result<(), sqlx::Error> {
 //                                         OMIKRONS
 // ==========================================================================================
 
-pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String), sqlx::Error> {
+pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String, u16), sqlx::Error> {
     let pool = {
         let db_lock = SQL_DB.read().await;
         db_lock
@@ -693,18 +701,22 @@ pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String), sql
             .expect("Database pool not initialized")
     };
 
-    let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>)>(
+    let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>, i32)>(
-        "SELECT public_key, ip_address FROM omikrons WHERE id = ?",
+        "SELECT public_key, ip_address, port FROM omikrons WHERE id = ?",
     )
     .bind(id)
     .fetch_optional(&pool)
     .await?;
 
     match row {
-        Some((public_key, ip_address)) => {
+        Some((public_key, ip_address, port)) => {
             let bundle = PublicKeyBundle::from_bytes(&public_key)
                 .map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
-            Ok((bundle, String::from_utf8_lossy(&ip_address).to_string()))
+            Ok((
+                bundle,
+                String::from_utf8_lossy(&ip_address).to_string(),
+                port as u16,
+            ))
         }
         _ => Err(sqlx::Error::RowNotFound),
     }

src/transport/omikron_connection.rs

@@ -1051,7 +1051,7 @@ pub async fn get_by_omikron_id(
     sql::get_omikron_by_id(omikron_id as i64)
         .await
         .ok()
-        .map(|(bundle, _ip_address)| bundle)
+        .map(|(bundle, _ip_address, _port)| bundle)
 }
 pub async fn complete_register(_pub_key: PublicKeyBundle, _description: Option<String>) -> u64 {
     0
@@ -1094,11 +1094,23 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
     let mut host: Host = Host::new(host_config).await?;
     log!("OmikronServer listening on port {}", port);
 
-    while let Ok(Some(mut connection)) = host.accept().await {
+    loop {
+        let mut conn = match host.accept().await {
+            Ok(Some(conn)) => conn,
+            Ok(None) => break,
+            Err(e) => {
+                // A single omikron's failed/aborted handshake (bad auth, a
+                // probe, a mid-handshake disconnect) must not take down the
+                // whole listener - only that connection attempt is lost.
+                log_err!(0, PrintType::Omega, "Rejected omikron connection: {}", e);
+                continue;
+            }
+        };
+
+        let omikron_connection = OmikronConnection::new(conn.sender, conn.client_id);
         tokio::spawn(async move {
-            let conn = OmikronConnection::new(connection.sender, connection.client_id);
+            omikron_manager::add_omikron(omikron_connection.clone()).await;
-            omikron_manager::add_omikron(conn.clone()).await;
+            omikron_connection.handle(&mut conn.receiver).await;
-            conn.handle(&mut connection.receiver).await;
         });
     }