Security advisory GHSA-r5ph-4jxm-6j9p: LF Edge eKuiper has a SQL Injection in sqlKvStore
Package: ekuiper (pypi), vulnerable version: 1.14.1, fixed in: 1.14.2
CWE: CWE-89
Files changed by the fix:
internal/pkg/store/sql/sqlKv.go
internal/pkg/store/sql/sqlTs.go

Fix diff:
diff --git a/internal/pkg/store/sql/sqlKv.go b/internal/pkg/store/sql/sqlKv.go
index ba58862710..ad93f50b2f 100644
--- a/internal/pkg/store/sql/sqlKv.go
+++ b/internal/pkg/store/sql/sqlKv.go
@@ -89,10 +89,15 @@ func (kv *sqlKvStore) Set(key string, value interface{}) error {
 func (kv *sqlKvStore) Get(key string, value interface{}) (bool, error) {
 	result := false
 	err := kv.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("SELECT val FROM '%s' WHERE key='%s';", kv.table, key)
-		row := db.QueryRow(query)
+		query := fmt.Sprintf("SELECT val FROM '%s' WHERE key=?;", kv.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			result = false
+			return nil
+		}
+		row := stmt.QueryRow(key)
 		var tmp []byte
-		err := row.Scan(&tmp)
+		err = row.Scan(&tmp)
 		if err != nil {
 			result = false
 			return nil
@@ -110,9 +115,13 @@ func (kv *sqlKvStore) Get(key string, value interface{}) (bool, error) {
 func (kv *sqlKvStore) GetKeyedState(key string) (interface{}, error) {
 	var value interface{}
 	err := kv.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("SELECT val FROM '%s' WHERE key='%s';", kv.table, key)
-		row := db.QueryRow(query)
-		err := row.Scan(&value)
+		query := fmt.Sprintf("SELECT val FROM '%s' WHERE key=?;", kv.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		row := stmt.QueryRow(key)
+		err = row.Scan(&value)
 		if err != nil {
 			return err
 		}
@@ -137,10 +146,14 @@ func (kv *sqlKvStore) SetKeyedState(key string, value interface{}) error {
 
 func (kv *sqlKvStore) Delete(key string) error {
 	return kv.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("SELECT key FROM '%s' WHERE key='%s';", kv.table, key)
-		row := db.QueryRow(query)
+		query := fmt.Sprintf("SELECT key FROM '%s' WHERE key=?;", kv.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		row := stmt.QueryRow(key)
 		var tmp []byte
-		err := row.Scan(&tmp)
+		err = row.Scan(&tmp)
 		if nil != err || 0 == len(tmp) {
 			return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("%s is not found", key))
 		}
diff --git a/internal/pkg/store/sql/sqlTs.go b/internal/pkg/store/sql/sqlTs.go
index 2267a22c8a..284e5a0e06 100644
--- a/internal/pkg/store/sql/sqlTs.go
+++ b/internal/pkg/store/sql/sqlTs.go
@@ -77,8 +77,12 @@ func (t *ts) Set(key int64, value interface{}) (bool, error) {
 func (t ts) Get(key int64, value interface{}) (bool, error) {
 	result := false
 	err := t.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("SELECT val FROM %s WHERE key=%d;", t.table, key)
-		row := db.QueryRow(query)
+		query := fmt.Sprintf("SELECT val FROM %s WHERE key=?;", t.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		row := stmt.QueryRow(key)
 		var tmp []byte
 		switch err := row.Scan(&tmp); err {
 		case sql.ErrNoRows:
@@ -111,16 +115,24 @@ func (t ts) Last(value interface{}) (int64, error) {
 
 func (t ts) Delete(key int64) error {
 	return t.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("DELETE FROM %s WHERE key=%d;", t.table, key)
-		_, err := db.Exec(query)
+		query := fmt.Sprintf("DELETE FROM %s WHERE key=?;", t.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		_, err = stmt.Exec(key)
 		return err
 	})
 }
 
 func (t ts) DeleteBefore(key int64) error {
 	return t.database.Apply(func(db *sql.DB) error {
-		query := fmt.Sprintf("DELETE FROM %s WHERE key<%d;", t.table, key)
-		_, err := db.Exec(query)
+		query := fmt.Sprintf("DELETE FROM %s WHERE key<?;", t.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		_, err = stmt.Exec(key)
 		return err
 	})
 }