fix: MCP guard, socket host, RPC client, sandbox runner, file launch hardening (#1565)

lib/core/extensions/rpc/json_rpc_stdio_client.dart

@@ -92,8 +92,9 @@ class JsonRpcStdioClient {
       if (params != null) 'params': params,
     };
 
-    _stdin.writeln(jsonEncode(payload));
     try {
+      // A closed sink throws here, before the flush below.
+      _stdin.writeln(jsonEncode(payload));
       await _stdin.flush().timeout(const Duration(seconds: 3));
     } on TimeoutException {
       _pending.remove(id);

lib/core/extensions/sandbox/sandbox_process_runner.dart

@@ -125,24 +125,31 @@ class SandboxProcessRunner {
       baseDirectory: scratchBaseDirectory,
     );
 
-    final detected = bwrapAvailable ?? await detectBwrapAvailability();
-    final usesBwrap = detected;
-    if (bwrapAvailable == null && !usesBwrap) {
-      debugPrint(
-        'SandboxProcessRunner: bubblewrap unavailable or cannot set up user '
-        'namespaces on this system; $pluginId requires consent to launch '
-        'without OS sandbox.',
+    final SandboxLaunchCommand command;
+    try {
+      final detected = bwrapAvailable ?? await detectBwrapAvailability();
+      final usesBwrap = detected;
+      if (bwrapAvailable == null && !usesBwrap) {
+        debugPrint(
+          'SandboxProcessRunner: bubblewrap unavailable or cannot set up user '
+          'namespaces on this system; $pluginId requires consent to launch '
+          'without OS sandbox.',
+        );
+      }
+      command = SandboxLaunchCommand.build(
+        pluginExecutable: pluginExecutable,
+        pluginArguments: pluginArguments,
+        scratchPath: scratch.path,
+        extensionRoot: extensionRoot,
+        capabilities: capabilities,
+        platformOverride: platformOverride,
+        bwrapAvailable: usesBwrap,
       );
+    } catch (_) {
+      // The scratch directory exists before the command is built.
+      await scratch.delete();
+      rethrow;
     }
-    final command = SandboxLaunchCommand.build(
-      pluginExecutable: pluginExecutable,
-      pluginArguments: pluginArguments,
-      scratchPath: scratch.path,
-      extensionRoot: extensionRoot,
-      capabilities: capabilities,
-      platformOverride: platformOverride,
-      bwrapAvailable: usesBwrap,
-    );
 
     final isolationIssue =
         SandboxOsIsolation.exceptionForLaunchCommand(command);

lib/core/mcp/mcp_socket_host.dart

@@ -16,16 +16,25 @@ class McpSocketHost {
     required this.version,
     required this.onSession,
     this.handshakeTimeout = const Duration(seconds: 5),
+    this.maxUnauthenticated = 16,
+    this.maxHandshakeBytes = 4096,
   });
 
   final File endpointFile;
   final String version;
   final Future<void> Function(StreamChannel<String> channel) onSession;
   final Duration handshakeTimeout;
 
+  /// Sockets that may wait for their token at once; more are turned away.
+  final int maxUnauthenticated;
+
+  /// Bytes accepted before the token line is complete (a token is ~44).
+  final int maxHandshakeBytes;
+
   ServerSocket? _server;
   String? _token;
   final _sockets = <Socket>{};
+  var _unauthenticated = 0;
 
   bool get isRunning => _server != null;
   int? get port => _server?.port;
@@ -77,19 +86,46 @@ class McpSocketHost {
   }
 
   void _accept(Socket socket) {
+    if (_unauthenticated >= maxUnauthenticated) {
+      socket.destroy();
+      return;
+    }
     _sockets.add(socket);
+    _unauthenticated++;
     // A client that disappears mid-write must not surface as an uncaught error.
     unawaited(socket.done.then((_) {}, onError: (Object _) {}));
     final incoming = StreamController<String>();
     final outgoing = StreamController<String>();
     var authed = false;
+    var waiting = true;
+    var handshakeBytes = 0;
+    void stopWaiting() {
+      if (!waiting) return;
+      waiting = false;
+      _unauthenticated--;
+    }
+
     final timer = Timer(handshakeTimeout, () {
-      if (!authed) _reject(socket, 'Handshake timed out.');
+      if (!authed) {
+        stopWaiting();
+        _reject(socket, 'Handshake timed out.');
+      }
     });
 
     late final StreamSubscription<String> sub;
     sub = socket
         .cast<List<int>>()
+        .map((chunk) {
+          // Without a newline the line splitter would buffer without bound.
+          if (!authed && (handshakeBytes += chunk.length) > maxHandshakeBytes) {
+            timer.cancel();
+            sub.cancel();
+            stopWaiting();
+            _reject(socket, 'Handshake too long.');
+            return const <int>[];
+          }
+          return chunk;
+        })
         .transform(utf8.decoder)
         .transform(const LineSplitter())
         .listen(
@@ -102,10 +138,12 @@ class McpSocketHost {
         if (token == null || !McpEndpoint.tokensEqual(line.trim(), token)) {
           timer.cancel();
           sub.cancel();
+          stopWaiting();
           _reject(socket, 'Invalid Querya MCP token. Restart the MCP client.');
           return;
         }
         authed = true;
+        stopWaiting();
         timer.cancel();
         _sessions++;
         _changes.add(null);
@@ -116,8 +154,14 @@ class McpSocketHost {
         unawaited(onSession(StreamChannel.withCloseGuarantee(
             incoming.stream, outgoing.sink)));
       },
-      onDone: () => _closed(socket, incoming, outgoing, authed),
-      onError: (_) => _closed(socket, incoming, outgoing, authed),
+      onDone: () {
+        stopWaiting();
+        _closed(socket, incoming, outgoing, authed);
+      },
+      onError: (_) {
+        stopWaiting();
+        _closed(socket, incoming, outgoing, authed);
+      },
       cancelOnError: true,
     );
   }

lib/core/mcp/mcp_sql_guard.dart

@@ -38,14 +38,16 @@ abstract final class McpSqlGuard {
 
   /// `EXPLAIN ANALYZE` executes the statement; the rest would explain a write.
   static final _explainForbidden = RegExp(
-    r'\b(ANALYZE|ANALYSE|INSERT|UPDATE|DELETE|MERGE|REPLACE|CREATE|DROP|ALTER|TRUNCATE|GRANT|REVOKE|CALL|DO|COPY)\b',
+    r'\b(ANALYZE|ANALYSE|INSERT|UPDATE|DELETE|MERGE|REPLACE\s+INTO|CREATE|DROP|ALTER|TRUNCATE|GRANT|REVOKE|CALL|DO|COPY)\b',
   );
 
   /// Functions that act on the server even inside a read-only transaction.
   static final _forbiddenFunctions = RegExp(
     r'\b(PG_TERMINATE_BACKEND|PG_CANCEL_BACKEND|PG_RELOAD_CONF|PG_ROTATE_LOGFILE|'
-    r'PG_READ_FILE|PG_READ_BINARY_FILE|PG_LS_DIR|PG_STAT_FILE|LO_IMPORT|LO_EXPORT|'
-    r'DBLINK\w*|SET_CONFIG|LOAD_FILE|LOAD_EXTENSION|WRITEFILE|READFILE|FSDIR)\s*\(',
+    r'PG_READ_FILE|PG_READ_BINARY_FILE|PG_LS_DIR|PG_STAT_FILE|LO_\w*|'
+    r'QUERY_TO_XML\w*|CURSOR_TO_XML|TABLE_TO_XML\w*|PG_(?:TRY_)?ADVISORY_\w*|PG_SLEEP\w*|'
+    r'NEXTVAL|SETVAL|DBLINK\w*|SET_CONFIG|LOAD_FILE|LOAD_EXTENSION|WRITEFILE|'
+    r'READFILE|FSDIR)\s*\(',
   );
 
   /// SQLite pragmas that only read the schema.

lib/core/platform/file_launch_service.dart

@@ -54,9 +54,6 @@ class FileLaunchService {
   FileLaunchTarget? get pendingTarget =>
       _pendingTargets.isNotEmpty ? _pendingTargets.first : null;
 
-  /// Whether there is at least one pending launch target.
-  bool get hasPendingTarget => _pendingTargets.isNotEmpty;
-
   /// Sets or clears the pending launch target (useful for testing or single-target flows).
   void setPendingTarget(FileLaunchTarget? target) {
     _pendingTargets.clear();
@@ -153,7 +150,7 @@ class FileLaunchService {
     for (final conn in connections) {
       if (conn.type == 'sqlite' && conn.host != null) {
         final existingNormalized = p.normalize(p.absolute(conn.host!));
-        if (existingNormalized == target.path) {
+        if (_samePath(existingNormalized, target.path)) {
           return conn;
         }
       }
@@ -174,6 +171,13 @@ class FileLaunchService {
     return newRow.copyWith(id: id);
   }
 
+  /// Windows and macOS volumes are case-insensitive by default, so
+  /// `C:\Data\A.db` and `c:\data\a.db` are one database there.
+  static bool _samePath(String a, String b) =>
+      Platform.isWindows || Platform.isMacOS
+          ? a.toLowerCase() == b.toLowerCase()
+          : a == b;
+
   /// Reads the full text of an SQL script for [target].
   Future<String> readSqlContent(FileLaunchTarget target) async {
     final file = File(target.path);

lib/features/main_screen/main_screen.dart

@@ -80,7 +80,7 @@ class _MainScreenState extends State<MainScreen> {
         final connections = await LocalDb.instance.getConnections();
         if (connections.isEmpty &&
             mounted &&
-            !FileLaunchService.instance.hasPendingTarget) {
+            !FileLaunchService.instance.hasPendingTargets) {
           _onOpenWelcomeTour();
         }
       }

test/core/mcp/mcp_query_service_test.dart

@@ -212,7 +212,7 @@ void main() {
     test('a timeout becomes a readable error', () async {
       db.onExecute = (_) => throw TimeoutException('slow');
       await expectLater(
-        service().runQuery(1, 'SELECT pg_sleep(60)'),
+        service().runQuery(1, 'SELECT count(*) FROM big_table'),
         throwsA(isA<McpToolException>().having(
             (e) => e.message, 'message', contains('time limit'))),
       );

test/core/mcp/mcp_socket_host_test.dart

@@ -0,0 +1,57 @@
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:path/path.dart' as p;
+import 'package:querya_desktop/core/mcp/mcp_socket_host.dart';
+
+void main() {
+  late Directory tempDir;
+  late McpSocketHost host;
+
+  Future<void> startHost({int maxUnauthenticated = 16}) async {
+    host = McpSocketHost(
+      endpointFile: File(p.join(tempDir.path, 'mcp.json')),
+      version: 'test',
+      onSession: (_) async {},
+      maxUnauthenticated: maxUnauthenticated,
+      maxHandshakeBytes: 64,
+    );
+    await host.start();
+  }
+
+  setUp(() async {
+    tempDir = await Directory.systemTemp.createTemp('querya_mcp_host_');
+  });
+
+  tearDown(() async {
+    if (host.isRunning) await host.stop();
+    await tempDir.delete(recursive: true);
+  });
+
+  test('a handshake line without a newline is cut off at the limit (#1565)',
+      () async {
+    await startHost();
+    final socket = await Socket.connect(InternetAddress.loopbackIPv4, host.port!);
+    final reply = utf8.decodeStream(socket);
+    socket.add(List.filled(200, 0x41));
+    await socket.flush();
+
+    expect(await reply.timeout(const Duration(seconds: 3)),
+        contains('Handshake too long'));
+    socket.destroy();
+  });
+
+  test('sockets beyond the unauthenticated cap are turned away (#1565)',
+      () async {
+    await startHost(maxUnauthenticated: 1);
+    final first = await Socket.connect(InternetAddress.loopbackIPv4, host.port!);
+    // Let the host register the first one before the second arrives.
+    await Future<void>.delayed(const Duration(milliseconds: 100));
+    final second = await Socket.connect(InternetAddress.loopbackIPv4, host.port!);
+
+    await second.drain<void>().timeout(const Duration(seconds: 3));
+    first.destroy();
+    second.destroy();
+  });
+}

test/core/mcp/mcp_sql_guard_test.dart

@@ -86,6 +86,24 @@ void main() {
     refused("SELECT load_extension('evil')", SqlDialect.sqlite);
   });
 
+  test('functions with side effects beyond the transaction are refused (#1565)',
+      () {
+    refused("SELECT query_to_xml('DELETE FROM y', true, true, '')");
+    refused("SELECT cursor_to_xml('c', 1, true, true, '')");
+    refused("SELECT table_to_xml('t', true, true, '')");
+    refused('SELECT pg_advisory_lock(1)');
+    refused('SELECT pg_try_advisory_xact_lock(1)');
+    refused('SELECT pg_sleep(60)');
+    refused("SELECT lo_get(1234)");
+    refused("SELECT nextval('s')");
+    refused("SELECT setval('s', 10)");
+  });
+
+  test('EXPLAIN of a SELECT that calls REPLACE() is allowed (#1565)', () {
+    allowed("EXPLAIN SELECT REPLACE(name, 'a', 'b') FROM users");
+    refused("EXPLAIN REPLACE INTO users VALUES (1)", SqlDialect.mysql);
+  });
+
   test('SQLite schema pragmas are allowed, others are refused', () {
     allowed('PRAGMA table_info(users)', SqlDialect.sqlite);
     allowed('PRAGMA index_list(users)', SqlDialect.sqlite);

test/core/platform/file_launch_service_test.dart

@@ -90,13 +90,13 @@ void main() {
         '/home/user/queries/reporting.sql',
       ]);
 
-      expect(FileLaunchService.instance.hasPendingTarget, isTrue);
+      expect(FileLaunchService.instance.hasPendingTargets, isTrue);
       expect(FileLaunchService.instance.pendingTarget?.fileName, 'reporting.sql');
       expect(FileLaunchService.instance.pendingTarget?.kind, FileLaunchKind.sql);
 
       final consumed = FileLaunchService.instance.consumePendingTarget();
       expect(consumed?.fileName, 'reporting.sql');
-      expect(FileLaunchService.instance.hasPendingTarget, isFalse);
+      expect(FileLaunchService.instance.hasPendingTargets, isFalse);
     });
   });
 

test/features/main_screen/main_screen_file_launch_test.dart

@@ -56,7 +56,7 @@ void main() {
       final target = FileLaunchTarget(path: dbFile.path, kind: FileLaunchKind.sqlite);
       FileLaunchService.instance.setPendingTarget(target);
 
-      expect(FileLaunchService.instance.hasPendingTarget, isTrue);
+      expect(FileLaunchService.instance.hasPendingTargets, isTrue);
 
       final consumed = FileLaunchService.instance.consumePendingTarget();
       expect(consumed, isNotNull);