fix: harden SMTP, SQL, and tokenizer boundaries (#21558)

internal/agent/tool/email.go

@@ -43,8 +43,8 @@ const (
 	emailSessionTimeout = 30 * time.Second
 )
 
-// emailParams contains both the Python model-call inputs and the Canvas node
-// configuration used to deliver the message. Info exposes only the former.
+// emailParams contains the resolved message fields and node-level SMTP
+// configuration used to deliver the message.
 type emailParams struct {
 	SMTPServer   string `json:"smtp_server"`
 	SMTPPort     int    `json:"smtp_port"`
@@ -58,6 +58,15 @@ type emailParams struct {
 	Subject      string `json:"subject"`
 }
 
+// emailCallParams is the subset of emailParams that may be supplied by the
+// model. SMTP configuration must come only from the Canvas node.
+type emailCallParams struct {
+	ToEmail string `json:"to_email"`
+	CCEmail string `json:"cc_email"`
+	Content string `json:"content"`
+	Subject string `json:"subject"`
+}
+
 // emailEnvelope is what the model sees.
 type emailEnvelope struct {
 	OK    bool   `json:"ok"`
@@ -171,10 +180,20 @@ var sendEmail = sendEmailSMTP
 // InvokableRun sends the email.
 func (e *EmailTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) {
 	p := e.defaults
-	if err := json.Unmarshal([]byte(argsJSON), &p); err != nil {
+	callParams := emailCallParams{
+		ToEmail: p.ToEmail,
+		CCEmail: p.CCEmail,
+		Content: p.Content,
+		Subject: p.Subject,
+	}
+	if err := json.Unmarshal([]byte(argsJSON), &callParams); err != nil {
 		return emailErrJSON(fmt.Errorf("email: parse arguments: %w", err)),
 			fmt.Errorf("email: parse arguments: %w", err)
 	}
+	p.ToEmail = callParams.ToEmail
+	p.CCEmail = callParams.CCEmail
+	p.Content = callParams.Content
+	p.Subject = callParams.Subject
 	state, _ := runtime.GetStateFromContext(ctx)
 	p.ToEmail = runtime.ResolveTemplateForDisplay(p.ToEmail, state)
 	p.CCEmail = runtime.ResolveTemplateForDisplay(p.CCEmail, state)

internal/agent/tool/email_test.go

@@ -119,6 +119,34 @@ func TestEmail_SendBuildsDistinctHeadersAndEnvelopeRecipients(t *testing.T) {
 	}
 }
 
+func TestEmail_ModelArgsCannotOverrideNodeSMTPConfig(t *testing.T) {
+	ctx := t.Context()
+	originalSendEmail := sendEmail
+	t.Cleanup(func() { sendEmail = originalSendEmail })
+	var sentParams emailParams
+	sendEmail = func(_ context.Context, p emailParams, _ []byte) error {
+		sentParams = p
+		return nil
+	}
+
+	tool := newEmailTool(emailParams{
+		SMTPServer:   "smtp.trusted.example",
+		SMTPPort:     587,
+		Email:        "sender@example.com",
+		SMTPUsername: "trusted-user",
+		Password:     "trusted-password",
+		SenderName:   "Trusted Sender",
+	})
+	args := `{"to_email":"recipient@example.com","subject":"hello","content":"body","smtp_server":"attacker.example","smtp_port":2525,"email":"attacker@example.com","smtp_username":"attacker","password":"stolen-password","sender_name":"Attacker"}`
+	if _, err := tool.InvokableRun(ctx, args); err != nil {
+		t.Fatalf("InvokableRun: %v", err)
+	}
+
+	if sentParams.SMTPServer != "smtp.trusted.example" || sentParams.SMTPPort != 587 || sentParams.Email != "sender@example.com" || sentParams.SMTPUsername != "trusted-user" || sentParams.Password != "trusted-password" || sentParams.SenderName != "Trusted Sender" {
+		t.Fatalf("model arguments overrode node SMTP config: %#v", sentParams)
+	}
+}
+
 func TestEmail_STARTTLSRequiredBeforeSubmission(t *testing.T) {
 	ctx := t.Context()
 	t.Parallel()

internal/binding/cpp/Makefile

@@ -1,6 +1,6 @@
 # Makefile for RAG Tokenizer with CGO bindings
 
-.PHONY: all clean build c_api c_api_debug c_api_asan test_go test_memory valgrind asan
+.PHONY: all clean build c_api c_api_debug c_api_asan test_string_utils test_go test_memory valgrind asan
 
 BUILD_DIR := build
 ASAN_BUILD_DIR := build-asan
@@ -8,6 +8,13 @@ C_API_LIB := $(BUILD_DIR)/librag_tokenizer_c_api.a
 C_API_ASAN_LIB := $(ASAN_BUILD_DIR)/librag_tokenizer_c_api.a
 C_API_DEBUG_LIB := $(BUILD_DIR)/librag_tokenizer_c_api_debug.a
 
+ifeq ($(shell uname -s),Darwin)
+HOMEBREW_PREFIX ?= $(shell brew --prefix 2>/dev/null)
+ifneq ($(strip $(HOMEBREW_PREFIX)),)
+SIMDE_CPPFLAGS := -isystem $(HOMEBREW_PREFIX)/include
+endif
+endif
+
 all: build c_api
 
 # Create build directory
@@ -30,6 +37,11 @@ c_api_asan: $(ASAN_BUILD_DIR)
 	cd $(ASAN_BUILD_DIR) && cmake .. -DENABLE_ASAN=ON && make rag_tokenizer_c_api -j$$(nproc)
 	@echo "ASan library built: $(C_API_ASAN_LIB)"
 
+test_string_utils:
+	mkdir -p $(BUILD_DIR)
+	$(CXX) $(CPPFLAGS) $(SIMDE_CPPFLAGS) $(CXXFLAGS) -std=c++20 -fsanitize=address -fno-omit-frame-pointer -g -I. string_utils_test.cpp -o $(BUILD_DIR)/string_utils_test
+	ASAN_OPTIONS=detect_leaks=0 ./$(BUILD_DIR)/string_utils_test
+
 # Build debug version of C API library with memory tracking
 c_api_debug: $(BUILD_DIR)
 	cd $(BUILD_DIR) && \

internal/binding/cpp/rag_analyzer.cpp

@@ -1500,8 +1500,9 @@ void RAGAnalyzer::MergeWithPosition(const std::vector<std::string> &tokens,
 void RAGAnalyzer::EnglishNormalize(const std::vector<std::string> &tokens, std::vector<std::string> &res) {
     for (auto &t : tokens) {
         if (re2::RE2::PartialMatch(t, pattern1_)) { //"[a-zA-Z_-]+$"
+            lowercase_string_buffer_.resize(t.size() + 1);
             char *lowercase_term = lowercase_string_buffer_.data();
-            ToLower(t.c_str(), t.size(), lowercase_term, term_string_buffer_limit_);
+            ToLower(t.c_str(), t.size(), lowercase_term, lowercase_string_buffer_.size());
             std::string term_to_stem;
             if (use_lemmatizer_) {
                 term_to_stem = wordnet_lemma_->Lemmatize(lowercase_term);
@@ -1886,8 +1887,9 @@ std::string RAGAnalyzer::Tokenize(const std::string &line) {
                 NLTKWordTokenizer::GetInstance().Tokenize(sentence, term_list);
             }
             for (unsigned i = 0; i < term_list.size(); ++i) {
+                lowercase_string_buffer_.resize(term_list[i].size() + 1);
                 char *lowercase_term = lowercase_string_buffer_.data();
-                ToLower(term_list[i].c_str(), term_list[i].size(), lowercase_term, term_string_buffer_limit_);
+                ToLower(term_list[i].c_str(), term_list[i].size(), lowercase_term, lowercase_string_buffer_.size());
                 std::string term_to_stem;
                 if (use_lemmatizer_) {
                     term_to_stem = wordnet_lemma_->Lemmatize(lowercase_term);
@@ -2034,8 +2036,9 @@ std::pair<std::vector<std::string>, std::vector<std::pair<unsigned, unsigned>>>
                         unsigned start_pos = sentence_start_pos + static_cast<unsigned>(pos_in_sentence);
                         unsigned end_pos = start_pos + static_cast<unsigned>(term.size());
                         // Apply lowercase before lemmatization to match Python NLTK behavior
+                        lowercase_string_buffer_.resize(term.size() + 1);
                         char *lowercase_term = lowercase_string_buffer_.data();
-                        ToLower(term.c_str(), term.size(), lowercase_term, term_string_buffer_limit_);
+                        ToLower(term.c_str(), term.size(), lowercase_term, lowercase_string_buffer_.size());
                         std::string term_to_stem;
                         if (use_lemmatizer_) {
                             term_to_stem = wordnet_lemma_->Lemmatize(lowercase_term);
@@ -2367,8 +2370,9 @@ void RAGAnalyzer::EnglishNormalizeWithPosition(const std::vector<std::string> &t
         const auto &[start_pos, end_pos] = positions[i];
 
         if (re2::RE2::PartialMatch(token, pattern1_)) { //"[a-zA-Z_-]+$"
+            lowercase_string_buffer_.resize(token.size() + 1);
             char *lowercase_term = lowercase_string_buffer_.data();
-            ToLower(token.c_str(), token.size(), lowercase_term, term_string_buffer_limit_);
+            ToLower(token.c_str(), token.size(), lowercase_term, lowercase_string_buffer_.size());
             std::string term_to_stem;
             if (use_lemmatizer_) {
                 term_to_stem = wordnet_lemma_->Lemmatize(lowercase_term);

internal/binding/cpp/string_utils.h

@@ -13,6 +13,8 @@
 // limitations under the License.
 #pragma once
 
+#include <algorithm>
+
 #if defined(__GNUC__) && (defined(__x86_64__) || defined(__i386__))
 #include <immintrin.h>
 #elif defined(__GNUC__) && defined(__aarch64__)
@@ -52,17 +54,21 @@ inline void ToLower(char* data, size_t len)
 
 inline void ToLower(const char* data, size_t len, char* out, size_t out_limit)
 {
-    memcpy(out, data, len);
+    if (out == nullptr || out_limit == 0)
+        return;
+    const size_t output_len = std::min(len, out_limit - 1);
+    if (output_len > 0)
+        memcpy(out, data, output_len);
     char* begin = out;
-    char* end = out + len;
+    char* end = out + output_len;
     char* p = begin;
 #if defined(__SSE2__)
     static constexpr int SSE2_BYTES = sizeof(__m128i);
-    const char* sse2_end = begin + (len & ~(SSE2_BYTES - 1));
+    const char* sse2_end = begin + (output_len & ~(SSE2_BYTES - 1));
     const auto a_minus1 = _mm_set1_epi8('A' - 1);
     const auto z_plus1 = _mm_set1_epi8('Z' + 1);
     const auto delta = _mm_set1_epi8('a' - 'A');
-    for (; p > sse2_end; p += SSE2_BYTES)
+    for (; p < sse2_end; p += SSE2_BYTES)
     {
         auto bytes = _mm_loadu_si128((const __m128i*)p);
         _mm_maskmoveu_si128(_mm_xor_si128(bytes, delta),

internal/binding/cpp/string_utils_test.cpp

@@ -0,0 +1,28 @@
+#include <cassert>
+#include <string>
+#include <vector>
+
+#include "string_utils.h"
+
+int main() {
+    constexpr size_t output_limit = 4096 * 3;
+    std::string input(output_limit * 2, 'A');
+    std::vector<char> guarded_output(output_limit + 2, 'x');
+    char* output = guarded_output.data() + 1;
+
+    ToLower(input.data(), input.size(), output, output_limit);
+
+    for (size_t i = 0; i < output_limit - 1; ++i) {
+        assert(output[i] == 'a');
+    }
+    assert(output[output_limit - 1] == '\0');
+    assert(guarded_output.front() == 'x');
+    assert(guarded_output.back() == 'x');
+
+    ToLower(input.data(), input.size(), nullptr, 0);
+
+    std::string long_token(output_limit * 2, 'B');
+    std::vector<char> full_output(long_token.size() + 1);
+    ToLower(long_token.data(), long_token.size(), full_output.data(), full_output.size());
+    assert(std::string(full_output.data()) == std::string(long_token.size(), 'b'));
+}

internal/engine/serenedb/client.go

@@ -167,8 +167,16 @@ func (e *serenedbEngine) exec(ctx context.Context, query string, args ...interfa
 
 // queryMaps runs a query and shapes each row into a map keyed by column name,
 // decoding JSON and array columns to structured values.
+type sqlQueryer interface {
+	QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
+}
+
 func (e *serenedbEngine) queryMaps(ctx context.Context, query string, args ...interface{}) ([]map[string]interface{}, error) {
-	rows, err := e.db.QueryContext(ctx, query, args...)
+	return queryMapsWith(ctx, e.db, query, args...)
+}
+
+func queryMapsWith(ctx context.Context, db sqlQueryer, query string, args ...interface{}) ([]map[string]interface{}, error) {
+	rows, err := db.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, err
 	}

internal/engine/serenedb/serenedb_test.go

@@ -20,10 +20,14 @@
 package serenedb
 
 import (
+	"context"
 	"reflect"
+	"regexp"
 	"strings"
 	"testing"
 
+	"github.com/DATA-DOG/go-sqlmock"
+
 	"ragflow/internal/engine/types"
 	"ragflow/internal/server/config"
 )
@@ -386,14 +390,86 @@ func TestParseMatchExprs(t *testing.T) {
 }
 
 func TestBoundRunSQL(t *testing.T) {
-	if got := boundRunSQL("SELECT * FROM t;", 1024); got != "SELECT * FROM t LIMIT 1024" {
-		t.Errorf("bound = %q", got)
+	if got, err := boundRunSQL("SELECT * FROM t;", 1024); err != nil || got != "SELECT * FROM t\nLIMIT 1024" {
+		t.Errorf("bound = %q, error = %v", got, err)
 	}
-	if got := boundRunSQL("SELECT * FROM t LIMIT 5", 1024); got != "SELECT * FROM t LIMIT 5" {
-		t.Errorf("existing limit must be kept: %q", got)
+	if got, err := boundRunSQL("SELECT * FROM t LIMIT 5", 1024); err != nil || got != "SELECT * FROM t LIMIT 5" {
+		t.Errorf("existing limit must be kept: %q, error = %v", got, err)
 	}
-	if got := boundRunSQL("UPDATE t SET x=1", 1024); got != "UPDATE t SET x=1" {
-		t.Errorf("non-select must be untouched: %q", got)
+	if _, err := boundRunSQL("UPDATE t SET x=1", 1024); err == nil {
+		t.Error("UPDATE should be rejected")
+	}
+	if _, err := boundRunSQL("WITH changed AS (DELETE FROM t RETURNING id) SELECT * FROM changed", 1024); err == nil {
+		t.Error("data-modifying CTE should be rejected")
+	}
+	if _, err := boundRunSQL("SELECT 1; DROP TABLE t", 1024); err == nil {
+		t.Error("multiple statements should be rejected")
+	}
+	if _, err := boundRunSQL("SELECT 1 --\r; DROP TABLE t", 1024); err == nil {
+		t.Error("CR-terminated comment must not hide a second statement")
+	}
+	if _, err := boundRunSQL(`SELECT E'\'' ; DROP TABLE t; --'`, 1024); err == nil {
+		t.Error("backslash escape in E string must not hide a second statement")
+	}
+	if _, err := boundRunSQL("SELECT 'x\\'; DROP TABLE t; --'", 1024); err == nil {
+		t.Error("ambiguous backslash escape in a plain string must be rejected")
+	}
+	if got, err := boundRunSQL("SELECT 'UPDATE; DROP'; -- ignored trailing comment", 1024); err != nil || got != "SELECT 'UPDATE; DROP'\nLIMIT 1024" {
+		t.Errorf("quoted commands should not be interpreted as SQL: %q, error = %v", got, err)
+	}
+	if got, err := boundRunSQL("SELECT * FROM t -- trailing comment", 1024); err != nil || got != "SELECT * FROM t -- trailing comment\nLIMIT 1024" {
+		t.Errorf("limit should follow a trailing line comment: %q, error = %v", got, err)
+	}
+}
+
+func TestRunSQLRejectsMutationStatementsBeforeExecution(t *testing.T) {
+	for _, query := range []string{
+		"DROP TABLE chunks",
+		"UPDATE chunks SET content = 'changed'",
+		"SELECT 1 INTO chunks",
+		"WITH changed AS (DELETE FROM chunks RETURNING id) SELECT * FROM changed",
+		"SELECT 1; DROP TABLE chunks",
+	} {
+		t.Run(query, func(t *testing.T) {
+			db, _, err := sqlmock.New()
+			if err != nil {
+				t.Fatalf("sqlmock.New: %v", err)
+			}
+			defer db.Close()
+
+			engine := &serenedbEngine{db: db}
+			_, err = engine.RunSQL(context.Background(), "chunks", query, nil, "json")
+			if err == nil {
+				t.Fatalf("RunSQL(%q) error = %v, want read-only SELECT rejection before query execution", query, err)
+			}
+			if !strings.Contains(err.Error(), "only read-only SELECT statements") && !strings.Contains(err.Error(), "multiple SQL statements are not allowed") {
+				t.Fatalf("RunSQL(%q) error = %v, want read-only SELECT rejection before query execution", query, err)
+			}
+		})
+	}
+}
+
+func TestRunSQLUsesReadOnlyTransaction(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer db.Close()
+
+	mock.ExpectBegin()
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1\nLIMIT 1024")).WillReturnRows(sqlmock.NewRows([]string{"value"}).AddRow(1))
+	mock.ExpectCommit()
+
+	engine := &serenedbEngine{db: db}
+	rows, err := engine.RunSQL(context.Background(), "chunks", "SELECT 1", nil, "json")
+	if err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["value"] != int64(1) {
+		t.Fatalf("RunSQL rows = %#v", rows)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("SQL expectations: %v", err)
 	}
 }
 

internal/engine/serenedb/sql.go

@@ -18,39 +18,220 @@ package serenedb
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
-	"regexp"
 	"strings"
 )
 
 const runSQLDefaultLimit = 1024
 
-var (
-	selectPrefixRe = regexp.MustCompile(`(?i)^(select|with)\b`)
-	hasLimitRe     = regexp.MustCompile(`(?i)\blimit\b`)
-)
+type runSQLToken struct {
+	text  string
+	word  bool
+	start int
+}
 
-// boundRunSQL trims a statement and appends a default LIMIT to unbounded
-// SELECT/WITH queries so the text-to-SQL path cannot stream an entire table.
-// Pure for tests.
-func boundRunSQL(sqlText string, limit int) string {
-	txt := strings.TrimSpace(sqlText)
-	txt = strings.TrimRight(txt, ";")
-	if limit > 0 && selectPrefixRe.MatchString(txt) && !hasLimitRe.MatchString(txt) {
-		txt = fmt.Sprintf("%s LIMIT %d", txt, limit)
+var runSQLMutationWords = map[string]struct{}{
+	"ALTER": {}, "ANALYZE": {}, "BEGIN": {}, "CALL": {}, "CHECKPOINT": {}, "CLUSTER": {}, "COMMENT": {},
+	"COMMIT": {}, "COPY": {}, "CREATE": {}, "DEALLOCATE": {}, "DELETE": {}, "DISCARD": {}, "DO": {},
+	"DROP": {}, "EXECUTE": {}, "EXPORT": {}, "GRANT": {}, "IMPORT": {}, "INSERT": {}, "INTO": {}, "LISTEN": {},
+	"LOCK": {}, "MERGE": {}, "NOTIFY": {}, "PREPARE": {}, "REASSIGN": {}, "REINDEX": {}, "RELEASE": {},
+	"REFRESH": {}, "RESET": {}, "REVOKE": {}, "ROLLBACK": {}, "SAVEPOINT": {}, "SET": {}, "START": {},
+	"TRUNCATE": {}, "UNLISTEN": {}, "UPDATE": {}, "VACUUM": {},
+}
+
+// scanRunSQLTokens separates SQL words from quoted strings, quoted identifiers,
+// and comments so command checks cannot be hidden in comments or string values.
+func scanRunSQLTokens(sqlText string) ([]runSQLToken, error) {
+	tokens := make([]runSQLToken, 0, len(sqlText)/4)
+	for i := 0; i < len(sqlText); {
+		c := sqlText[i]
+		if c == ' ' || c == '\t' || c == '\n' || c == '\r' || c == '\f' {
+			i++
+			continue
+		}
+		if c == '-' && i+1 < len(sqlText) && sqlText[i+1] == '-' {
+			i += 2
+			for i < len(sqlText) && sqlText[i] != '\n' && sqlText[i] != '\r' {
+				i++
+			}
+			continue
+		}
+		if c == '/' && i+1 < len(sqlText) && sqlText[i+1] == '*' {
+			commentStart, depth := i, 1
+			i += 2
+			for i < len(sqlText) && depth > 0 {
+				switch {
+				case i+1 < len(sqlText) && sqlText[i] == '/' && sqlText[i+1] == '*':
+					depth++
+					i += 2
+				case i+1 < len(sqlText) && sqlText[i] == '*' && sqlText[i+1] == '/':
+					depth--
+					i += 2
+				default:
+					i++
+				}
+			}
+			if depth != 0 {
+				return nil, fmt.Errorf("unterminated SQL comment at byte %d", commentStart)
+			}
+			continue
+		}
+		if c == '\'' || c == '"' {
+			quote, start := c, i
+			backslashEscapes := quote == '\'' && i > 0 && (sqlText[i-1] == 'e' || sqlText[i-1] == 'E') &&
+				(i == 1 || !isRunSQLWordByte(sqlText[i-2]))
+			i++
+			closed := false
+			for i < len(sqlText) {
+				if sqlText[i] == '\\' && i+1 < len(sqlText) {
+					if backslashEscapes {
+						i += 2
+						continue
+					}
+					// Plain-string escaping depends on standard_conforming_strings; reject
+					// ambiguous quote escapes rather than guessing the server setting.
+					if quote == '\'' && sqlText[i+1] == '\'' {
+						return nil, fmt.Errorf("ambiguous backslash escape in SQL string at byte %d", i)
+					}
+				}
+				if sqlText[i] != quote {
+					i++
+					continue
+				}
+				if i+1 < len(sqlText) && sqlText[i+1] == quote {
+					i += 2
+					continue
+				}
+				i++
+				closed = true
+				break
+			}
+			if !closed {
+				return nil, fmt.Errorf("unterminated SQL quote at byte %d", start)
+			}
+			tokens = append(tokens, runSQLToken{start: start})
+			continue
+		}
+		if c == '$' {
+			if delimiter, ok := runSQLDollarQuoteDelimiter(sqlText, i); ok {
+				bodyStart := i + len(delimiter)
+				closeOffset := strings.Index(sqlText[bodyStart:], delimiter)
+				if closeOffset < 0 {
+					return nil, fmt.Errorf("unterminated SQL dollar quote at byte %d", i)
+				}
+				i = bodyStart + closeOffset + len(delimiter)
+				tokens = append(tokens, runSQLToken{start: bodyStart - len(delimiter)})
+				continue
+			}
+		}
+		if isRunSQLWordStart(c) {
+			start := i
+			i++
+			for i < len(sqlText) && isRunSQLWordByte(sqlText[i]) {
+				i++
+			}
+			tokens = append(tokens, runSQLToken{text: strings.ToUpper(sqlText[start:i]), word: true, start: start})
+			continue
+		}
+		tokens = append(tokens, runSQLToken{text: string(c), start: i})
+		i++
 	}
-	return txt
+	return tokens, nil
+}
+
+func runSQLDollarQuoteDelimiter(sqlText string, start int) (string, bool) {
+	i := start + 1
+	if i < len(sqlText) && sqlText[i] == '$' {
+		return "$$", true
+	}
+	if i >= len(sqlText) || !isRunSQLWordStart(sqlText[i]) {
+		return "", false
+	}
+	for i < len(sqlText) && (isRunSQLWordStart(sqlText[i]) || sqlText[i] >= '0' && sqlText[i] <= '9') {
+		i++
+	}
+	if i >= len(sqlText) || sqlText[i] != '$' {
+		return "", false
+	}
+	return sqlText[start : i+1], true
+}
+
+func isRunSQLWordStart(c byte) bool {
+	return c == '_' || c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z'
+}
+
+func isRunSQLWordByte(c byte) bool {
+	return isRunSQLWordStart(c) || c >= '0' && c <= '9' || c == '$'
+}
+
+// boundRunSQL accepts one read-only SELECT statement, including SELECT CTEs,
+// and appends a default row limit when the query has none. The lexical checks
+// deliberately reject command keywords anywhere outside quoted data, which
+// also blocks data-modifying CTEs.
+func boundRunSQL(sqlText string, limit int) (string, error) {
+	txt := strings.TrimSpace(sqlText)
+	tokens, err := scanRunSQLTokens(txt)
+	if err != nil {
+		return "", err
+	}
+	if len(tokens) == 0 {
+		return "", fmt.Errorf("only read-only SELECT statements are allowed")
+	}
+	if last := tokens[len(tokens)-1]; last.text == ";" {
+		txt = strings.TrimSpace(txt[:last.start])
+		tokens = tokens[:len(tokens)-1]
+	}
+	for _, token := range tokens {
+		if token.text == ";" {
+			return "", fmt.Errorf("multiple SQL statements are not allowed")
+		}
+	}
+	if len(tokens) == 0 || !tokens[0].word || (tokens[0].text != "SELECT" && tokens[0].text != "WITH") {
+		return "", fmt.Errorf("only read-only SELECT statements are allowed")
+	}
+	for _, token := range tokens {
+		if token.word {
+			if _, mutating := runSQLMutationWords[token.text]; mutating {
+				return "", fmt.Errorf("only read-only SELECT statements are allowed")
+			}
+		}
+	}
+	if limit > 0 && !runSQLHasLimit(tokens) {
+		txt = fmt.Sprintf("%s\nLIMIT %d", txt, limit)
+	}
+	return txt, nil
+}
+
+func runSQLHasLimit(tokens []runSQLToken) bool {
+	for _, token := range tokens {
+		if token.word && token.text == "LIMIT" {
+			return true
+		}
+	}
+	return false
 }
 
 // RunSQL executes a text-to-SQL query directly against SereneDB and returns the
 // rows as maps. SereneDB speaks SQL natively, so unlike the Infinity engine no
 // psql subprocess or field-alias rewrite is needed.
 func (e *serenedbEngine) RunSQL(ctx context.Context, tableName string, sqlText string, kbIDs []string, format string) ([]map[string]interface{}, error) {
-	query := boundRunSQL(sqlText, runSQLDefaultLimit)
-	rows, err := e.queryMaps(ctx, query)
+	query, err := boundRunSQL(sqlText, runSQLDefaultLimit)
 	if err != nil {
 		return nil, fmt.Errorf("serenedb: run sql: %w", err)
 	}
+	tx, err := e.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
+	if err != nil {
+		return nil, fmt.Errorf("serenedb: run sql: begin read-only transaction: %w", err)
+	}
+	defer tx.Rollback()
+	rows, err := queryMapsWith(ctx, tx, query)
+	if err != nil {
+		return nil, fmt.Errorf("serenedb: run sql: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, fmt.Errorf("serenedb: run sql: commit read-only transaction: %w", err)
+	}
 	if rows == nil {
 		rows = []map[string]interface{}{}
 	}