fix(web): support SSH host key prompts

crates/dbx-web/src/main.rs

@@ -2,6 +2,7 @@ mod auth;
 mod error;
 mod routes;
 mod sse;
+mod ssh_prompt;
 mod state;
 
 use std::collections::{HashMap, HashSet};
@@ -239,8 +240,11 @@ async fn main() {
         nacos_imports: RwLock::new(HashMap::new()),
         login_rate_limit: tokio::sync::Mutex::new(state::LoginRateLimit { fail_count: 0, locked_until: None }),
         export_files: RwLock::new(HashMap::new()),
+        ssh_prompts: Arc::new(ssh_prompt::SshPromptHub::new()),
     });
 
+    ssh_prompt::install_web_ssh_prompt_bridge(web_state.ssh_prompts.clone());
+
     // API routes
     let api = Router::new()
         // Auth
@@ -283,6 +287,8 @@ async fn main() {
         // System
         .route("/system/fonts", get(routes::jdbc::list_system_fonts))
         .route("/ssh/config-hosts", get(routes::ssh_config::list_ssh_config_hosts))
+        .route("/ssh/prompts", get(routes::ssh_prompt::stream_ssh_prompts))
+        .route("/ssh/prompts/resolve", post(routes::ssh_prompt::resolve_ssh_prompt))
         // Tunnel profiles
         .route("/tunnel-profiles/list", get(routes::tunnel_profiles::load_tunnel_profiles))
         .route("/tunnel-profiles/save", post(routes::tunnel_profiles::save_tunnel_profiles))

crates/dbx-web/src/routes/connection.rs

@@ -500,6 +500,7 @@ mod tests {
             nacos_imports: RwLock::new(HashMap::new()),
             login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
             export_files: RwLock::new(HashMap::new()),
+            ssh_prompts: Arc::new(crate::ssh_prompt::SshPromptHub::new()),
         });
         (state, dir)
     }

crates/dbx-web/src/routes/mod.rs

@@ -28,6 +28,7 @@ pub mod schema_cache;
 pub mod schema_diff;
 pub mod sql_file;
 pub mod ssh_config;
+pub mod ssh_prompt;
 pub mod tab_runtime_cache;
 pub mod table_export;
 pub mod table_import;

crates/dbx-web/src/routes/mongo.rs

@@ -840,6 +840,7 @@ mod tests {
             nacos_imports: RwLock::new(HashMap::new()),
             login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
             export_files: RwLock::new(HashMap::new()),
+            ssh_prompts: Arc::new(crate::ssh_prompt::SshPromptHub::new()),
         });
         (state, dir)
     }

crates/dbx-web/src/routes/mq.rs

@@ -1546,6 +1546,7 @@ mod tests {
             nacos_imports: RwLock::new(HashMap::new()),
             login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
             export_files: RwLock::new(HashMap::new()),
+            ssh_prompts: Arc::new(crate::ssh_prompt::SshPromptHub::new()),
         });
         (state, dir)
     }

crates/dbx-web/src/routes/ssh_prompt.rs

@@ -0,0 +1,59 @@
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use axum::Json;
+use dbx_core::db::ssh_prompt::SshPromptAnswer;
+use futures::Stream;
+use serde::Deserialize;
+use std::convert::Infallible;
+use std::sync::Arc;
+
+use crate::error::AppError;
+use crate::ssh_prompt::SshPromptEvent;
+use crate::state::WebState;
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ResolveSshPromptRequest {
+    pub id: String,
+    pub action: String,
+    pub remember: Option<bool>,
+    pub secret: Option<String>,
+}
+
+pub async fn stream_ssh_prompts(
+    State(state): State<Arc<WebState>>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let (replay, mut receiver) = state.ssh_prompts.subscribe();
+    let stream = async_stream::stream! {
+        for event in replay {
+            yield Ok(ssh_prompt_event(event));
+        }
+        loop {
+            match receiver.recv().await {
+                Ok(event) => yield Ok(ssh_prompt_event(event)),
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => break,
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    };
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
+pub async fn resolve_ssh_prompt(
+    State(state): State<Arc<WebState>>,
+    Json(request): Json<ResolveSshPromptRequest>,
+) -> Result<Json<()>, AppError> {
+    let answer = match request.action.as_str() {
+        "accept" => SshPromptAnswer::Accept { remember: request.remember.unwrap_or(false) },
+        "reject" => SshPromptAnswer::Reject,
+        "secret" => SshPromptAnswer::Secret(request.secret.unwrap_or_default()),
+        other => return Err(AppError::bad_request(format!("Unknown SSH prompt action: {other}"))),
+    };
+
+    state.ssh_prompts.resolve(&request.id, answer).map_err(AppError::bad_request)?;
+    Ok(Json(()))
+}
+
+fn ssh_prompt_event(event: SshPromptEvent) -> Event {
+    Event::default().data(serde_json::to_string(&event).expect("SSH prompt events must serialize"))
+}

crates/dbx-web/src/ssh_prompt.rs

@@ -0,0 +1,162 @@
+use dbx_core::db::ssh_prompt::{SshHostKeyNotice, SshPromptAnswer, SshPromptEnvelope, SshPromptRequest};
+use serde::Serialize;
+use std::sync::{Arc, Mutex};
+use tokio::sync::{broadcast, mpsc};
+
+const EVENT_BUFFER_SIZE: usize = 32;
+const PROMPT_CHANNEL_SIZE: usize = 16;
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum SshPromptEvent {
+    Sync { pending_ids: Vec<String> },
+    Prompt { request: SshPromptRequest },
+    Notice { notice: SshHostKeyNotice },
+    Dismiss { id: String },
+}
+
+struct PendingPrompt {
+    responder: tokio::sync::oneshot::Sender<SshPromptAnswer>,
+    request: SshPromptRequest,
+}
+
+pub struct SshPromptHub {
+    pending: Mutex<Vec<PendingPrompt>>,
+    events: broadcast::Sender<SshPromptEvent>,
+}
+
+impl SshPromptHub {
+    pub fn new() -> Self {
+        let (events, _) = broadcast::channel(EVENT_BUFFER_SIZE);
+        Self { pending: Mutex::new(Vec::new()), events }
+    }
+
+    pub fn register(&self, envelope: SshPromptEnvelope) {
+        let request = envelope.request;
+        self.pending.lock().unwrap().push(PendingPrompt { responder: envelope.responder, request: request.clone() });
+        let _ = self.events.send(SshPromptEvent::Prompt { request });
+    }
+
+    pub fn subscribe(&self) -> (Vec<SshPromptEvent>, broadcast::Receiver<SshPromptEvent>) {
+        let receiver = self.events.subscribe();
+        let pending = self.pending.lock().unwrap();
+        let mut replay = vec![SshPromptEvent::Sync {
+            pending_ids: pending.iter().map(|prompt| prompt.request.id.clone()).collect(),
+        }];
+        replay.extend(pending.iter().map(|prompt| SshPromptEvent::Prompt { request: prompt.request.clone() }));
+        (replay, receiver)
+    }
+
+    pub fn resolve(&self, id: &str, answer: SshPromptAnswer) -> Result<(), String> {
+        let pending = {
+            let mut pending = self.pending.lock().unwrap();
+            let index = pending
+                .iter()
+                .position(|prompt| prompt.request.id == id)
+                .ok_or_else(|| format!("No pending SSH prompt with id {id}"))?;
+            pending.remove(index)
+        };
+        let result = pending.responder.send(answer);
+        let _ = self.events.send(SshPromptEvent::Dismiss { id: id.to_string() });
+        result.map_err(|_| format!("SSH prompt {id} was already cancelled"))
+    }
+
+    pub fn notify(&self, notice: SshHostKeyNotice) {
+        let _ = self.events.send(SshPromptEvent::Notice { notice });
+    }
+
+    pub fn sweep_cancelled(&self) {
+        let cancelled = {
+            let mut pending = self.pending.lock().unwrap();
+            let cancelled = pending
+                .iter()
+                .filter(|prompt| prompt.responder.is_closed())
+                .map(|prompt| prompt.request.id.clone())
+                .collect::<Vec<_>>();
+            pending.retain(|prompt| !prompt.responder.is_closed());
+            cancelled
+        };
+
+        for id in cancelled {
+            let _ = self.events.send(SshPromptEvent::Dismiss { id });
+        }
+    }
+}
+
+pub fn install_web_ssh_prompt_bridge(hub: Arc<SshPromptHub>) {
+    let (prompt_tx, mut prompt_rx) = mpsc::channel(PROMPT_CHANNEL_SIZE);
+    dbx_core::db::ssh_prompt::install_ssh_prompt_gateway(prompt_tx);
+    let prompt_hub = hub.clone();
+    tokio::spawn(async move {
+        while let Some(envelope) = prompt_rx.recv().await {
+            prompt_hub.register(envelope);
+        }
+    });
+
+    let (notice_tx, mut notice_rx) = mpsc::channel(PROMPT_CHANNEL_SIZE);
+    dbx_core::db::ssh_prompt::install_ssh_notice_gateway(notice_tx);
+    let notice_hub = hub.clone();
+    tokio::spawn(async move {
+        while let Some(notice) = notice_rx.recv().await {
+            notice_hub.notify(notice);
+        }
+    });
+
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
+        loop {
+            interval.tick().await;
+            hub.sweep_cancelled();
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dbx_core::db::ssh_prompt::SshPromptKind;
+
+    fn request(id: &str) -> SshPromptRequest {
+        SshPromptRequest {
+            id: id.to_string(),
+            kind: SshPromptKind::HostKeyVerify,
+            host: "example.test".to_string(),
+            port: 22,
+            key_type: Some("ssh-ed25519".to_string()),
+            fingerprint: Some("SHA256:test".to_string()),
+            prompt: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn late_subscriber_replays_pending_prompt_and_resolution() {
+        let hub = SshPromptHub::new();
+        let (responder, receiver) = tokio::sync::oneshot::channel();
+        hub.register(SshPromptEnvelope { request: request("prompt-1"), responder });
+
+        let (replay, mut events) = hub.subscribe();
+        assert!(
+            matches!(replay.as_slice(), [SshPromptEvent::Sync { pending_ids }, SshPromptEvent::Prompt { request }] if pending_ids == &["prompt-1"] && request.id == "prompt-1")
+        );
+
+        hub.resolve("prompt-1", SshPromptAnswer::Accept { remember: true }).unwrap();
+        assert!(matches!(receiver.await.unwrap(), SshPromptAnswer::Accept { remember: true }));
+        assert!(matches!(events.recv().await.unwrap(), SshPromptEvent::Dismiss { id } if id == "prompt-1"));
+    }
+
+    #[tokio::test]
+    async fn invalid_resolution_keeps_prompt_pending() {
+        let hub = SshPromptHub::new();
+        let (responder, receiver) = tokio::sync::oneshot::channel();
+        hub.register(SshPromptEnvelope { request: request("prompt-2"), responder });
+
+        assert!(hub.resolve("missing", SshPromptAnswer::Reject).is_err());
+        let (replay, _) = hub.subscribe();
+        assert!(
+            matches!(replay.as_slice(), [SshPromptEvent::Sync { pending_ids }, SshPromptEvent::Prompt { request }] if pending_ids == &["prompt-2"] && request.id == "prompt-2")
+        );
+
+        hub.resolve("prompt-2", SshPromptAnswer::Reject).unwrap();
+        assert!(matches!(receiver.await.unwrap(), SshPromptAnswer::Reject));
+    }
+}

crates/dbx-web/src/state.rs

@@ -35,6 +35,7 @@ pub struct WebState {
     pub login_rate_limit: Mutex<LoginRateLimit>,
     /// Table export temp files: export_id -> (file_path, format)
     pub export_files: RwLock<HashMap<String, (String, String)>>,
+    pub ssh_prompts: Arc<crate::ssh_prompt::SshPromptHub>,
 }
 
 impl WebState {