Repository: lf-edge/ekuiper
Pull request #941 (2021-08-16 07:33:59)

fix(*): reflect.TypeOf must check nil to avoid panic
A lot of places are using reflect.TypeOf(i).Kind() to check interface types. If i is nil, the TypeOf may return nil so that this will cause a panic. Fix all potential problems.

Closes: #938

Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/internal/plugin/sourceMeta.go b/internal/plugin/sourceMeta.go
index 80e88b41..e071df64 100644
--- a/internal/plugin/sourceMeta.go
+++ b/internal/plugin/sourceMeta.go
@@ -304,7 +304,7 @@ func recursionDelMap(cf, fields map[string]interface{}, language string) error {
 			delete(auxCf, delKey)
 			continue
 		}
-		if reflect.Map == reflect.TypeOf(v).Kind() {
+		if reflect.TypeOf(v) != nil && reflect.Map == reflect.TypeOf(v).Kind() {
 			var auxCf, auxFields map[string]interface{}
 			if err := cast.MapToStruct(cf[k], &auxCf); nil != err {
 				return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "type_conversion_fail"), k, v)
@@ -364,7 +364,7 @@ func recursionNewFields(template []field, conf map[string]interface{}, ret *[]fi
 		if nil == v {
 			p.Default = v
 		} else {
-			if reflect.Map == reflect.TypeOf(v).Kind() {
+			if reflect.TypeOf(v) != nil && reflect.Map == reflect.TypeOf(v).Kind() {
 				var nextCf map[string]interface{}
 				if tmp, ok := v.(map[interface{}]interface{}); ok {
 					nextCf = cast.ConvertMap(tmp)
diff --git a/internal/topo/sink/edgex_sink.go b/internal/topo/sink/edgex_sink.go
index 6ba24e6e..af0e2457 100644
--- a/internal/topo/sink/edgex_sink.go
+++ b/internal/topo/sink/edgex_sink.go
@@ -172,7 +172,8 @@ func (ems *EdgexMsgBusSink) produceEvents(ctx api.StreamContext, result []byte)
 		}
 		for _, v := range m {
 			for k1, v1 := range v {
-				if k1 == ems.metadata {
+				// Ignore nil values
+				if k1 == ems.metadata || v1 == nil {
 					continue
 				} else {
 					var (
@@ -216,7 +217,11 @@ func (ems *EdgexMsgBusSink) produceEvents(ctx api.StreamContext, result []byte)
 }
 
 func getValueType(v interface{}) (string, interface{}, error) {
-	k := reflect.TypeOf(v).Kind()
+	vt := reflect.TypeOf(v)
+	if vt == nil {
+		return "", nil, fmt.Errorf("unsupported value nil")
+	}
+	k := vt.Kind()
 	switch k {
 	case reflect.Bool:
 		return v2.ValueTypeBool, v, nil
@@ -232,8 +237,11 @@ func getValueType(v interface{}) (string, interface{}, error) {
 		switch arrayValue := v.(type) {
 		case []interface{}:
 			if len(arrayValue) > 0 {
-				ka := reflect.TypeOf(arrayValue[0]).Kind()
-				switch ka {
+				kt := reflect.TypeOf(arrayValue[0])
+				if kt == nil {
+					return "", nil, fmt.Errorf("unsupported value %v(%s), the first element is nil", v, k)
+				}
+				switch kt.Kind() {
 				case reflect.Bool:
 					result := make([]bool, len(arrayValue))
 					for i, av := range arrayValue {
diff --git a/internal/topo/source/mqtt_type.go b/internal/topo/source/mqtt_type.go
index 0d48210c..b9daaaae 100644
--- a/internal/topo/source/mqtt_type.go
+++ b/internal/topo/source/mqtt_type.go
@@ -39,7 +39,7 @@ type modelVersion interface {
 	checkType(map[string]interface{}, string) []string
 }
 
-func modelFactory(version string) modelVersion {
+func modelFactory(_ string) modelVersion {
 	return new(deviceModel)
 }
 func (this *property) getName() string {
@@ -81,8 +81,11 @@ func intToBool(i int) bool {
 }
 func changeType(modelType string, data interface{}) (interface{}, string) {
 	var err error
-	dataType := reflect.TypeOf(data).Kind()
-	switch dataType {
+	dt := reflect.TypeOf(data)
+	if dt == nil {
+		return data, fmt.Sprintf("not support type : %v", nil)
+	}
+	switch dt.Kind() {
 	case reflect.Bool:
 		b, _ := data.(bool)
 		switch modelType {
@@ -138,7 +141,7 @@ func changeType(modelType string, data interface{}) (interface{}, string) {
 			return data, fmt.Sprintf("not support modelType : %s", modelType)
 		}
 	default:
-		return data, fmt.Sprintf("not support type : %v", dataType)
+		return data, fmt.Sprintf("not support type : %v", dt.Kind())
 	}
 	return data, ""
 }
diff --git a/internal/xsql/funcsMisc.go b/internal/xsql/funcsMisc.go
index 8634eb5d..76dd811f 100644
--- a/internal/xsql/funcsMisc.go
+++ b/internal/xsql/funcsMisc.go
@@ -209,7 +209,10 @@ func hashCall(name string, args []interface{}) (interface{}, bool) {
 	default:
 		return fmt.Errorf("unknown hash function name %s", name), false
 	}
-	io.WriteString(h, arg0)
+	_, err := io.WriteString(h, arg0)
+	if err != nil {
+		return err, false
+	}
 	return fmt.Sprintf("%x", h.Sum(nil)), true
 }
 
@@ -227,12 +230,11 @@ func otherCall(name string, args []interface{}) (interface{}, bool) {
 				return false, true
 			}
 		}
-		return false, true
 	case "newuuid":
-		if uuid, err := uuid.NewUUID(); err != nil {
+		if newUUID, err := uuid.NewUUID(); err != nil {
 			return err, false
 		} else {
-			return uuid.String(), true
+			return newUUID.String(), true
 		}
 	case "tstamp":
 		return conf.GetNowInMilli(), true
@@ -256,20 +258,26 @@ func otherCall(name string, args []interface{}) (interface{}, bool) {
 
 func jsonCall(name string, args []interface{}) (interface{}, bool) {
 	var input interface{}
-	switch reflect.TypeOf(args[0]).Kind() {
-	case reflect.Map:
-		input = convertToInterfaceArr(args[0].(map[string]interface{}))
-	case reflect.Slice:
-		input = convertSlice(args[0])
-	case reflect.String:
-		v, _ := args[0].(string)
-		err := json.Unmarshal([]byte(v), &input)
-		if err != nil {
-			return fmt.Errorf("%s function error: the first argument '%v' is not a valid json string", name, args[0]), false
+	at := reflect.TypeOf(args[0])
+	if at != nil {
+		switch at.Kind() {
+		case reflect.Map:
+			input = convertToInterfaceArr(args[0].(map[string]interface{}))
+		case reflect.Slice:
+			input = convertSlice(args[0])
+		case reflect.String:
+			v, _ := args[0].(string)
+			err := json.Unmarshal([]byte(v), &input)
+			if err != nil {
+				return fmt.Errorf("%s function error: the first argument '%v' is not a valid json string", name, args[0]), false
+			}
+		default:
+			return fmt.Errorf("%s function error: the first argument must be a map but got %v", name, args[0]), false
 		}
-	default:
-		return fmt.Errorf("%s function error: the first argument must be a map but got %v", name, args[0]), false
+	} else {
+		return fmt.Errorf("%s function error: the first argument must be a map but got nil", name), false
 	}
+
 	builder := gval.Full(jsonpath.PlaceholderExtension())
 	path, err := builder.NewEvaluable(args[1].(string))
 	if err != nil {
@@ -292,6 +300,9 @@ func jsonCall(name string, args []interface{}) (interface{}, bool) {
 			return fmt.Errorf("%s function error: query result (%v) is not an array", name, result), false
 		}
 	case "json_path_exists":
+		if result == nil {
+			return false, true
+		}
 		e := true
 		switch reflect.TypeOf(result).Kind() {
 		case reflect.Slice, reflect.Array:
@@ -307,7 +318,12 @@ func jsonCall(name string, args []interface{}) (interface{}, bool) {
 func convertToInterfaceArr(orig map[string]interface{}) map[string]interface{} {
 	result := make(map[string]interface{})
 	for k, v := range orig {
-		switch reflect.TypeOf(v).Kind() {
+		vt := reflect.TypeOf(v)
+		if vt == nil {
+			result[k] = nil
+			continue
+		}
+		switch vt.Kind() {
 		case reflect.Slice:
 			result[k] = convertSlice(v)
 		case reflect.Map:
diff --git a/pkg/cast/cast.go b/pkg/cast/cast.go
index f9f6f083..714e885c 100644
--- a/pkg/cast/cast.go
+++ b/pkg/cast/cast.go
@@ -837,10 +837,11 @@ func ToTypedSlice(input interface{}, conv func(interface{}, Strictness) (interfa
 		return nil, fmt.Errorf("cannot convert %[1]T(%[1]v) to %s slice)", input, eleType)
 	}
 	ele, err := conv(s.Index(0).Interface(), sn)
-	if err != nil {
+	et := reflect.TypeOf(ele)
+	if err != nil || et == nil {
 		return nil, fmt.Errorf("cannot convert %[1]T(%[1]v) to %s slice for the %d element: %v", input, eleType, 0, err)
 	}
-	result := reflect.MakeSlice(reflect.SliceOf(reflect.TypeOf(ele)), s.Len(), s.Len())
+	result := reflect.MakeSlice(reflect.SliceOf(et), s.Len(), s.Len())
 	result.Index(0).Set(reflect.ValueOf(ele))
 	for i := 1; i < s.Len(); i++ {
 		ele, err := conv(s.Index(i).Interface(), sn)


Test patch:
diff --git a/internal/topo/operator/misc_func_test.go b/internal/topo/operator/misc_func_test.go
index 1b15e602..c3b8a5be 100644
--- a/internal/topo/operator/misc_func_test.go
+++ b/internal/topo/operator/misc_func_test.go
@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"reflect"
@@ -433,6 +434,7 @@ func TestJsonPathFunc_Apply1(t *testing.T) {
 		sql    string
 		data   interface{}
 		result interface{}
+		err    string
 	}{
 		{
 			sql: `SELECT json_path_query(equipment, "$.arm_right") AS a FROM test`,
@@ -785,6 +787,28 @@ func TestJsonPathFunc_Apply1(t *testing.T) {
 			result: []map[string]interface{}{{
 				"powerOnTs": float64(1000),
 			}},
+		}, {
+			sql: `SELECT json_path_query(equipment, "$.arm_right") AS a FROM test`,
+			data: &xsql.Tuple{
+				Emitter: "test",
+				Message: xsql.Message{
+					"class": "warrior",
+					"equipment2": map[string]interface{}{
+						"rings": []map[string]interface{}{
+							{
+								"name":   "ring of despair",
+								"weight": 0.1,
+							}, {
+								"name":   "ring of strength",
+								"weight": 2.4,
+							},
+						},
+						"arm_right": "Sword of flame",
+						"arm_left":  "Shield of faith",
+					},
+				},
+			},
+			err: "run Select error: call func json_path_query error: json_path_query function error: the first argument must be a map but got nil",
 		},
 	}
 
@@ -799,20 +823,30 @@ func TestJsonPathFunc_Apply1(t *testing.T) {
 		pp := &ProjectOp{Fields: stmt.Fields}
 		fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters)
 		result := pp.Apply(ctx, tt.data, fv, afv)
-		var mapRes []map[string]interface{}
-		if v, ok := result.([]byte); ok {
-			err := json.Unmarshal(v, &mapRes)
-			if err != nil {
-				t.Errorf("Failed to parse the input into map.\n")
-				continue
+		switch rt := result.(type) {
+		case []byte:
+			if tt.err == "" {
+				var mapRes []map[string]interface{}
+				err := json.Unmarshal(rt, &mapRes)
+				if err != nil {
+					t.Errorf("Failed to parse the input into map.\n")
+					continue
+				}
+				if !reflect.DeepEqual(tt.result, mapRes) {
+					t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
+				}
+			} else {
+				t.Errorf("%d: invalid result:\n  exp error %s\n  got=%s\n\n", i, tt.err, result)
 			}
-			//fmt.Printf("%t\n", mapRes["kuiper_field_0"])
-
-			if !reflect.DeepEqual(tt.result, mapRes) {
-				t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
+		case error:
+			if tt.err == "" {
+				t.Errorf("%d: got error:\n  exp=%s\n  got=%s\n\n", i, tt.result, err)
+			} else if !reflect.DeepEqual(tt.err, testx.Errstring(rt)) {
+				t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.err, err)
 			}
-		} else {
-			t.Errorf("The returned result is not type of []byte but found %v", result)
+		default:
+			t.Errorf("%d: Invalid returned result found %v", i, result)
 		}
+
 	}
 }
diff --git a/internal/topo/sink/edgex_sink_test.go b/internal/topo/sink/edgex_sink_test.go
index 05b8a2cd..11ead5b4 100644
--- a/internal/topo/sink/edgex_sink_test.go
+++ b/internal/topo/sink/edgex_sink_test.go
@@ -108,7 +108,8 @@ func TestProduceEvents(t1 *testing.T) {
 							"temperature":{"deviceName":"test device name2","id":"22","origin":24}
 							}
 						},
-						{"h1":100}
+						{"h1":100},
+						{"h2":null}
 					]`,
 			expected: &dtos.Event{
 				Id:          "abc",
diff --git a/internal/topo/source/mqtt_type_test.go b/internal/topo/source/mqtt_type_test.go
index 5e014d6b..c648c731 100644
--- a/internal/topo/source/mqtt_type_test.go
+++ b/internal/topo/source/mqtt_type_test.go
@@ -30,6 +30,7 @@ func TestCheckType(t *testing.T) {
 		{"temperature": "1", "humidity": "1"},
 		{"temperature": 1, "humidity": "1"},
 		{"temperature": "1", "humidity": 1},
+		{"temperature": nil, "humidity": "1"},
 	}
 
 	topics := []string{`$ke/events/device/device1/data/update`, `$ke/events/device/device2/data/update`}
diff --git a/pkg/cast/cast_test.go b/pkg/cast/cast_test.go
index e6b62b5d..dceb2520 100644
--- a/pkg/cast/cast_test.go
+++ b/pkg/cast/cast_test.go
@@ -15,6 +15,7 @@
 package cast
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 )
@@ -45,3 +46,47 @@ func TestMapConvert_Funcs(t *testing.T) {
 		t.Errorf("result mismatch:\n\nexp=%s\n\ngot=%s\n\n", exp, got)
 	}
 }
+
+func TestToTypedSlice(t *testing.T) {
+	var tests = []struct {
+		s interface{}
+		r interface{}
+		e string
+	}{
+		{
+			s: []interface{}{"abc", 123},
+			r: []string{"abc", "123"},
+		},
+		{
+			s: []interface{}{"addd", "bbb"},
+			r: []string{"addd", "bbb"},
+		},
+		{
+			s: []interface{}{nil, "bbb", "ddd"},
+			e: "cannot convert []interface {}([<nil> bbb ddd]) to string slice for the 0 element: <nil>",
+		},
+	}
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, tt := range tests {
+		result, err := ToTypedSlice(tt.s, func(input interface{}, ssn Strictness) (interface{}, error) {
+			if input == nil {
+				return nil, nil
+			} else {
+				return fmt.Sprintf("%v", input), nil
+			}
+		}, "string", CONVERT_SAMEKIND)
+
+		if !reflect.DeepEqual(tt.e, errstring(err)) {
+			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
+		} else if tt.e == "" && !reflect.DeepEqual(tt.r, result) {
+			t.Errorf("%d\n\nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.r, result)
+		}
+	}
+}
+
+func errstring(err error) string {
+	if err != nil {
+		return err.Error()
+	}
+	return ""
+}