diff --git a/crates/analyzer/src/lib.rs b/crates/analyzer/src/lib.rs index 9b45f92..3362428 100644 --- a/crates/analyzer/src/lib.rs +++ b/crates/analyzer/src/lib.rs @@ -418,7 +418,7 @@ impl LanguageServer for Backend { .request::(lsp_types::GotoDefinitionParams { text_document_position_params: lsp_types::TextDocumentPositionParams { position: lsp_types::Position { - line: line as u32 - 1, + line: line as u32, character: character as u32, }, text_document: lsp_types::TextDocumentIdentifier { uri }, @@ -504,6 +504,19 @@ impl LanguageServer for Backend { } })) } + + async fn completion_resolve(&self, params: CompletionItem) -> Result { + match self + .rust_analyzer + .lock() + .await + .request::(params.clone()) + .await + { + Ok(o) => Ok(o), + _ => Ok(params), + } + } } #[tokio::main] @@ -601,18 +614,13 @@ fn find_row_col(rope: &Rope, needle: &str) -> Option<(usize, usize)> { let col_idx = char_idx - line_start; - Some((line_idx + 1, col_idx + 1)) + Some((line_idx, col_idx + 1)) } fn simplify_item(mut item: CompletionItem) -> CompletionItem { item.text_edit = None; item.additional_text_edits = None; item.command = None; - item.data = None; - - item.insert_text = Some(item.label.clone()); - - item.insert_text_format = Some(InsertTextFormat::PLAIN_TEXT); item } diff --git a/crates/analyzer/src/rust_analyzer.rs b/crates/analyzer/src/rust_analyzer.rs index b7e584d..ec92b8e 100644 --- a/crates/analyzer/src/rust_analyzer.rs +++ b/crates/analyzer/src/rust_analyzer.rs @@ -1,10 +1,11 @@ -use std::{collections::HashMap, path::PathBuf, process::Stdio, sync::Arc}; +use std::{collections::HashMap, path::PathBuf, process::Stdio, sync::Arc, time::Duration}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use serde_json::{Value, json}; use tokio::{ io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}, sync::{Mutex, oneshot}, + time::timeout, }; use tower_lsp::lsp_types::{ self, ClientCapabilities, InitializeParams, InitializedParams, Url, WorkspaceFolder, @@ -12,10 +13,13 @@ use tower_lsp::lsp_types::{ request::{self, Request}, }; +const MAX_CONTENT_LENGTH: usize = 50 * 1024 * 1024; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + #[derive(Debug, Deserialize)] pub struct JsonRpcResponse { pub jsonrpc: String, - pub id: Option, + pub id: Option, pub result: Option, pub error: Option, } @@ -27,13 +31,16 @@ pub struct JsonRpcError { pub data: Option, } -type PendingMap = Arc>>>; +type PendingMap = Arc>>>>; #[derive(Debug)] pub struct RustAnalyzer { - stdin: tokio::process::ChildStdin, + // Wrapped in a Mutex to support safe concurrent sharing if the design expands + stdin: Arc>, pending: PendingMap, id: usize, + // Keep child handler to explicitly manage child process lifecycle and prevent zombie processes + _child: tokio::process::Child, } async fn send_lsp_message( @@ -49,27 +56,37 @@ async fn send_lsp_message( async fn read_lsp_message( reader: &mut R, -) -> Result> { +) -> Result> { let mut line = String::new(); let mut content_length = 0; - // Read headers until we hit the empty separator line (\r\n) - loop { + // Guard against infinite header reading attacks/bugs (Max 100 headers) + for _ in 0..100 { line.clear(); - reader.read_line(&mut line).await?; - if line == "\r\n" || line.is_empty() { + let bytes_read = reader.read_line(&mut line).await?; + if bytes_read == 0 || line == "\r\n" || line.is_empty() { break; } - if line.to_lowercase().starts_with("content-length:") { - content_length = line["content-length:".len()..].trim().parse::()?; + if line.to_ascii_lowercase().starts_with("content-length:") { + if let Some(val_str) = line.split(':').nth(1) { + content_length = val_str.trim().parse::()?; + } } } + // Explode early if payload size violates strict guard rails to prevent memory-exhaustion (OOM) if content_length == 0 { - return Err("Missing or invalid Content-Length header".into()); + return Err("Missing, invalid, or zero Content-Length header".into()); + } + if content_length > MAX_CONTENT_LENGTH { + return Err(format!( + "Content-Length {} exceeds maximum threshold", + content_length + ) + .into()); } - // Read the exact byte buffer payload + // Explicitly secure pre-allocation limit let mut buffer = vec![0u8; content_length]; reader.read_exact(&mut buffer).await?; @@ -77,19 +94,26 @@ async fn read_lsp_message( } impl RustAnalyzer { - pub fn new() -> Result> { + pub fn new() -> Result> { let mut child = tokio::process::Command::new("rust-analyzer") .stdin(Stdio::piped()) .stdout(Stdio::piped()) - .stderr(Stdio::null()) // Ignore logs for simplicity + .stderr(Stdio::null()) .spawn()?; - let stdin = child.stdin.take().unwrap(); - let stdout = child.stdout.take().unwrap(); + let stdin = child + .stdin + .take() + .ok_or("Failed to open child stdin pipe")?; + let stdout = child + .stdout + .take() + .ok_or("Failed to open child stdout pipe")?; let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); let pending_clone = pending.clone(); + // Background supervisor task loop tokio::spawn(async move { let mut stdout = BufReader::new(stdout); @@ -97,7 +121,12 @@ impl RustAnalyzer { let raw = match read_lsp_message(&mut stdout).await { Ok(v) => v, Err(err) => { - eprintln!("LSP read error: {err}"); + eprintln!("LSP fatal stream read failure: {err}"); + // CRITICAL: Notify all pending channels that the bridge broke down + let mut lock = pending_clone.lock().await; + for (_, tx) in lock.drain() { + let _ = tx.send(Err(format!("LSP reader task dropped: {}", err))); + } break; } }; @@ -105,51 +134,43 @@ impl RustAnalyzer { let value: Value = match serde_json::from_str(&raw) { Ok(v) => v, Err(err) => { - eprintln!("Invalid JSON from rust-analyzer: {err}"); - continue; + eprintln!("Corrupted JSON received: {err}"); + continue; // Keep the connection running despite malformed frame } }; - if value.get("method").is_some() { + // Filter server notification frames + if value.get("method").is_some() && value.get("id").is_none() { continue; } let id = value.get("id").and_then(|v| v.as_u64()).map(|v| v as usize); - match id { - Some(id) => { - let tx = pending_clone.lock().await.remove(&id); - - match tx { - Some(tx) => { - let _ = tx.send(value); - } - None => { - eprintln!("({:?}) {}", pending_clone, value); - eprintln!("Received response for unknown request id {id}"); - } - } - } - - None => { - // notification or server request - eprintln!("Received server notification/request: {raw}"); + if let Some(id) = id { + let tx = pending_clone.lock().await.remove(&id); + if let Some(tx) = tx { + let _ = tx.send(Ok(value)); + } else { + eprintln!("Received orphaned or delayed frame for ID: {id}"); } + } else { + eprintln!("Received unhandled protocol notification framework: {raw}"); } } }); Ok(Self { - stdin, + stdin: Arc::new(Mutex::new(stdin)), pending, id: 0, + _child: child, }) } pub async fn request( &mut self, params: R::Params, - ) -> Result> + ) -> Result> where R::Result: DeserializeOwned, { @@ -160,47 +181,86 @@ impl RustAnalyzer { let (tx, rx) = oneshot::channel(); - self.pending.lock().await.insert(id, tx); - - send_lsp_message( - &mut self.stdin, - &json!({ - "jsonrpc": "2.0", - "id": id, - "method": R::METHOD, - "params": params, - }), - ) - .await?; - - let value = rx.await?; - - let envelope: JsonRpcResponse = serde_json::from_value(value.clone())?; - - if let Some(err) = envelope.error { - return Err(format!("LSP Error ({}): {}", err.code, err.message).into()); + // Scope the lock allocation tightly + { + self.pending.lock().await.insert(id, tx); } - envelope - .result - .ok_or_else(|| format!("missing result: {:?}", value).into()) + let payload = json!({ + "jsonrpc": "2.0", + "id": id, + "method": R::METHOD, + "params": params, + }); + + // Acquire lock on writing stream to ensure thread safety + let mut stdin_lock = self.stdin.lock().await; + + if let Err(err) = send_lsp_message(&mut *stdin_lock, &payload).await { + // Rollback the pending map operation to avoid internal memory memory-leaks if serialization/IO errors trigger + self.pending.lock().await.remove(&id); + return Err(Box::new(err)); + } + + // Explicit drop of write lock early so other operations can pipe messages synchronously + drop(stdin_lock); + + // Enforce an absolute time constraint limit to break free from hanging processes + let response_payload = match timeout(REQUEST_TIMEOUT, rx).await { + Ok(Ok(Ok(value))) => value, + Ok(Ok(Err(task_err))) => return Err(task_err.into()), + Ok(Err(_oneshot_canceled)) => { + return Err( + "Bridge connection closed down; reader channel dropped unexpectedly".into(), + ); + } + Err(_timeout_elapsed) => { + // Clear state tracking entries dynamically upon expiration failure + self.pending.lock().await.remove(&id); + return Err( + format!("Request ID {} timed out after {:?}", id, REQUEST_TIMEOUT).into(), + ); + } + }; + + let envelope: JsonRpcResponse = + serde_json::from_value(response_payload.clone())?; + + if let Some(err) = envelope.error { + return Err(format!("LSP Engine Error ({}): {}", err.code, err.message).into()); + } + + envelope.result.ok_or_else(|| { + format!( + "Missing inner structural payload result: {:?}", + response_payload + ) + .into() + }) } - pub async fn notify(&mut self, method: &str, req: T) -> std::io::Result<()> { - send_lsp_message( - &mut self.stdin, - &json!({ - "jsonrpc": "2.0", - "method": method, - "params": req, - }), - ) - .await + pub async fn notify( + &mut self, + method: &str, + req: T, + ) -> Result<(), Box> { + let payload = json!({ + "jsonrpc": "2.0", + "method": method, + "params": req, + }); + + let mut stdin_lock = self.stdin.lock().await; + send_lsp_message(&mut *stdin_lock, &payload).await?; + Ok(()) } } impl RustAnalyzer { - pub async fn initialize(&mut self, root: &PathBuf) -> Result<(), Box> { + pub async fn initialize( + &mut self, + root: &PathBuf, + ) -> Result<(), Box> { let project_uri = Url::from_directory_path(root) .map_err(|_| "Failed to convert path to valid file:// URL")?; @@ -226,14 +286,12 @@ impl RustAnalyzer { }; self.request::(init_params).await?; - Ok(()) } - pub async fn initialized(&mut self) -> Result<(), Box> { + pub async fn initialized(&mut self) -> Result<(), Box> { self.notify(Initialized::METHOD, InitializedParams {}) .await?; - Ok(()) } }