Fix: check sql component required fields (#19418)

Author: Lynn-Inf

internal/agent/component/dynamic_params.go

@@ -92,6 +92,10 @@ func validateDynamicParams(component string, params map[string]any) error {
 		return validateVariableAggregatorGroups(component, params)
 	case "userfillup":
 		return validateInputOptions(component, params)
+	case "exesql":
+		if err := validateExeSQLParams(params); err != nil {
+			return fmt.Errorf("[%s] %w", component, err)
+		}
 	}
 	return nil
 }

internal/agent/component/exesql_params.go

@@ -0,0 +1,131 @@
+// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package component
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+)
+
+// exeSQLDBTypes is the whitelist enforced at save time for an ExeSQL
+// component's db_type. It mirrors Python ExeSQLParam.check
+// (agent/tools/exesql.py), where ExeSQLParam defaults db_type to "mysql"
+// when the DSL omits it.
+var exeSQLDBTypes = []string{"mysql", "postgres", "mariadb", "mssql", "IBM DB2", "trino", "oceanbase"}
+
+// validateExeSQLParams enforces the required connection settings of an
+// ExeSQL component at save time. The Go agent PUT path validates DSL
+// component parameters through validateDynamicParams; Python performs the
+// same checks via Canvas.validate_component_parameters ->
+// ExeSQLParam.check. Without them a canvas can be persisted with a broken
+// DB tool that only fails at runtime.
+func validateExeSQLParams(params map[string]any) error {
+	dbType, explicit := params["db_type"]
+	if !explicit {
+		// Python default when the DSL omits db_type.
+		dbType = "mysql"
+	}
+	dbTypeString, ok := dbType.(string)
+	if !ok || !exeSQLDBTypeAllowed(dbTypeString) {
+		return fmt.Errorf("Choose DB type %v is not supported, it should be in %v", dbType, exeSQLDBTypes)
+	}
+
+	// Connection identity fields have no Python defaults and must be present.
+	for _, field := range []struct {
+		key  string
+		desc string
+	}{
+		{"database", "Database name"},
+		{"username", "database username"},
+		{"host", "IP Address"},
+	} {
+		if !isNonBlankString(params[field.key]) {
+			return fmt.Errorf("%s does not support empty value", field.desc)
+		}
+	}
+
+	if port, present := params["port"]; present {
+		if !isPositiveInteger(port) {
+			return fmt.Errorf("IP Port %v not supported, should be positive integer", port)
+		}
+	}
+
+	// Trino connects without credentials; every other supported DB type
+	// requires a password.
+	if dbTypeString != "trino" && !isNonBlankString(params["password"]) {
+		return fmt.Errorf("Database password does not support empty value")
+	}
+
+	if maxRecords, present := params["max_records"]; present {
+		if !isPositiveInteger(maxRecords) {
+			return fmt.Errorf("Maximum number of records %v not supported, should be positive integer", maxRecords)
+		}
+	}
+
+	// Refuse configs aimed at RAGFlow's own metadata database, mirroring
+	// Python ExeSQLParam.check's security guard.
+	if trimmedDBString(params["database"]) == "rag_flow" &&
+		(trimmedDBString(params["host"]) == "ragflow-mysql" || trimmedDBString(params["password"]) == "infini_rag_flow") {
+		return fmt.Errorf("For the security reason, it does not support database named rag_flow.")
+	}
+	return nil
+}
+
+// exeSQLDBTypeAllowed reports whether dbType is one of exeSQLDBTypes.
+func exeSQLDBTypeAllowed(dbType string) bool {
+	for _, candidate := range exeSQLDBTypes {
+		if candidate == dbType {
+			return true
+		}
+	}
+	return false
+}
+
+// trimmedDBString returns the whitespace-trimmed value of a string-ish
+// parameter, or "" when it is absent or not a string.
+func trimmedDBString(value any) string {
+	s, _ := value.(string)
+	return strings.TrimSpace(s)
+}
+
+// isPositiveInteger reports whether value is a whole number greater than
+// zero. json.Number lexemes survive the JSON decode and whole-valued
+// float64/float32 are accepted for consistency with isIntegerNumber.
+func isPositiveInteger(value any) bool {
+	if !isIntegerNumber(value) {
+		return false
+	}
+	switch n := value.(type) {
+	case json.Number:
+		i, err := n.Int64()
+		return err == nil && i > 0
+	case float64:
+		return n > 0
+	case float32:
+		return n > 0
+	default:
+		v := reflect.ValueOf(value)
+		switch v.Kind() {
+		case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+			return v.Int() > 0
+		case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+			return v.Uint() > 0
+		default:
+			return false
+		}
+	}
+}

internal/agent/component/exesql_params_test.go

@@ -0,0 +1,124 @@
+package component
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidateDynamicEntriesExeSQL(t *testing.T) {
+	valid := dslWithComponents(
+		componentDSL("ExeSQL", map[string]any{
+			"db_type":     "mysql",
+			"database":    "orders",
+			"username":    "report",
+			"host":        "10.0.0.5",
+			"port":        3306,
+			"password":    "secret",
+			"max_records": 100,
+		}),
+	)
+	if err := ValidateDynamicEntries(valid); err != nil {
+		t.Fatalf("valid ExeSQL DSL rejected: %v", err)
+	}
+
+	// db_type is optional: ExeSQLParam defaults it to "mysql".
+	noDBType := dslWithComponents(componentDSL("ExeSQL", map[string]any{
+		"database": "orders",
+		"username": "report",
+		"host":     "10.0.0.5",
+		"password": "secret",
+	}))
+	if err := ValidateDynamicEntries(noDBType); err != nil {
+		t.Fatalf("ExeSQL DSL without db_type rejected: %v", err)
+	}
+
+	// Trino connects without credentials.
+	trino := dslWithComponents(componentDSL("ExeSQL", map[string]any{
+		"db_type":  "trino",
+		"database": "catalog",
+		"username": "report",
+		"host":     "10.0.0.5",
+		"password": "",
+	}))
+	if err := ValidateDynamicEntries(trino); err != nil {
+		t.Fatalf("valid trino ExeSQL DSL rejected: %v", err)
+	}
+
+	tests := []struct {
+		name      string
+		component map[string]any
+		want      string
+	}{
+		{
+			"unsupported db_type",
+			componentDSL("ExeSQL", map[string]any{"db_type": "oracle", "database": "d", "username": "u", "host": "h", "password": "p"}),
+			"Choose DB type oracle is not supported, it should be in",
+		},
+		{
+			"non-string db_type",
+			componentDSL("ExeSQL", map[string]any{"db_type": 1, "database": "d", "username": "u", "host": "h", "password": "p"}),
+			"Choose DB type 1 is not supported, it should be in",
+		},
+		{
+			"empty database",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "", "username": "u", "host": "h", "password": "p"}),
+			"[ExeSQL] Database name does not support empty value",
+		},
+		{
+			"missing database",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "username": "u", "host": "h", "password": "p"}),
+			"[ExeSQL] Database name does not support empty value",
+		},
+		{
+			"empty username",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "", "host": "h", "password": "p"}),
+			"[ExeSQL] database username does not support empty value",
+		},
+		{
+			"empty host",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "u", "host": " ", "password": "p"}),
+			"[ExeSQL] IP Address does not support empty value",
+		},
+		{
+			"missing password",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "u", "host": "h"}),
+			"[ExeSQL] Database password does not support empty value",
+		},
+		{
+			"zero port",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "u", "host": "h", "password": "p", "port": 0}),
+			"[ExeSQL] IP Port 0 not supported, should be positive integer",
+		},
+		{
+			"fractional port",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "u", "host": "h", "password": "p", "port": 3306.5}),
+			"[ExeSQL] IP Port 3306.5 not supported, should be positive integer",
+		},
+		{
+			"negative max_records",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "d", "username": "u", "host": "h", "password": "p", "max_records": -1}),
+			"[ExeSQL] Maximum number of records -1 not supported, should be positive integer",
+		},
+		{
+			"rag_flow database with default mysql host",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "rag_flow", "username": "u", "host": "ragflow-mysql", "password": "secret"}),
+			"[ExeSQL] For the security reason, it does not support database named rag_flow.",
+		},
+		{
+			"rag_flow database with default mysql password",
+			componentDSL("ExeSQL", map[string]any{"db_type": "mysql", "database": "rag_flow", "username": "u", "host": "10.0.0.5", "password": "infini_rag_flow"}),
+			"[ExeSQL] For the security reason, it does not support database named rag_flow.",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateDynamicEntries(dslWithComponents(tt.component))
+			if err == nil {
+				t.Fatalf("expected error containing %q, got nil", tt.want)
+			}
+			if !strings.Contains(err.Error(), tt.want) {
+				t.Fatalf("error %q does not contain %q", err.Error(), tt.want)
+			}
+		})
+	}
+}