astagent: prevent panic in handleChannelDestroyed

agents/astagent.go

@@ -312,8 +312,23 @@ func (sma *AsteriskAgent) handleChannelDestroyed(ev *SMAsteriskEvent) {
 	cgrEvDisp, hasIt := sma.eventsCache[chID]
 	sma.evCacheMux.RUnlock()
 	if !hasIt {
-		if cgrReqType, _ := ev.ariEv["channel"].(map[string]any)["channelvars"].(map[string]any)[utils.CGRReqType].(string); cgrReqType == utils.EmptyString {
-			return // Not handled by us
+		channelVar, ok := ev.ariEv["channel"].(map[string]any)
+		if !ok {
+			utils.Logger.Warning(fmt.Sprintf(
+				"<%s> missing or invalid 'channel' field in event: %s",
+				utils.AsteriskAgent, utils.ToJSON(ev.ariEv)))
+			return
+		}
+
+		channelVars, ok := channelVar["channelvars"].(map[string]any)
+		if !ok {
+			utils.Logger.Warning(fmt.Sprintf(
+				"<%s> missing or invalid 'channelvars' field in 'channel': %s",
+				utils.AsteriskAgent, utils.ToJSON(channelVar)))
+			return
+		}
+		if cgrReqType := utils.IfaceAsString(channelVars[utils.CGRReqType]); cgrReqType == "" {
+			return
 		}
 		// convert received event to CGREvent
 		var err error

agents/astagent_test.go

@@ -18,6 +18,10 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package agents
 
 import (
+	"bytes"
+	"log"
+	"os"
+	"strings"
 	"testing"
 
 	"github.com/cgrates/birpc"
@@ -27,6 +31,7 @@ import (
 	"github.com/cgrates/cgrates/sessions"
 	"github.com/cgrates/cgrates/utils"
 	"github.com/cgrates/rpcclient"
+	"github.com/google/go-cmp/cmp"
 )
 
 func TestAAsSessionSClientIface(t *testing.T) {
@@ -83,10 +88,11 @@ func TestHandleChannelDestroyedFail(t *testing.T) {
 		"type":      "ChannelDestroyed",
 	}
 	ev := NewSMAsteriskEvent(ariEv, "127.0.0.1", utils.EmptyString)
-	evCopy := ev
+	evCopy := ev.Clone()
+	evCopy.cachedFields = map[string]string{"channelID": "1714719185.3"}
 	sma.handleChannelDestroyed(ev)
-	if ev != evCopy {
-		t.Errorf("Expected ev to not change, received <%v>", utils.ToJSON(ev))
+	if diff := cmp.Diff(evCopy, ev, cmp.AllowUnexported(SMAsteriskEvent{})); diff != "" {
+		t.Errorf("handleChannelDestroyed modified SMAsteriskEvent unexpectedly (-want +got): \n%s", diff)
 	}
 }
 
@@ -119,3 +125,68 @@ func TestAsteriskAgentV1AlterSession(t *testing.T) {
 		t.Errorf("Expected error: %v, got: %v", utils.ErrNotImplemented, err)
 	}
 }
+
+func TestHandleChannelDestroyedCases(t *testing.T) {
+	cfg := config.NewDefaultCGRConfig()
+	internalSessionSChan := make(chan birpc.ClientConnector, 1)
+	cM := engine.NewConnManager(cfg, map[string]chan context.ClientConnector{
+		utils.ConcatenatedKey(rpcclient.BiRPCInternal, utils.MetaSessionS): internalSessionSChan,
+	})
+	sma, err := NewAsteriskAgent(cfg, 1, cM)
+	if err != nil {
+		t.Error(err)
+	}
+
+	utils.Logger.SetLogLevel(4)
+	utils.Logger.SetSyslog(nil)
+
+	t.Cleanup(func() {
+		utils.Logger.SetLogLevel(0)
+		log.SetOutput(os.Stderr)
+	})
+
+	testCases := []struct {
+		name   string
+		ariEv  map[string]any
+		expLog string
+	}{
+		{
+			name:   "Missing Channel",
+			ariEv:  map[string]any{},
+			expLog: "<AsteriskAgent> missing or invalid 'channel' field in event: {}",
+		},
+		{
+			name:   "Invalid Channel",
+			ariEv:  map[string]any{"channel": "invalid"},
+			expLog: `<AsteriskAgent> missing or invalid 'channel' field in event: {"channel":"invalid"}`,
+		},
+		{
+			name:   "Missing ChannelVars",
+			ariEv:  map[string]any{"channel": map[string]any{"channel": "1"}},
+			expLog: `<AsteriskAgent> missing or invalid 'channelvars' field in 'channel': {"channel":"1"}`,
+		},
+		{
+			name:   "Invalid ChannelVars",
+			ariEv:  map[string]any{"channel": map[string]any{"channelvars": "invalid"}},
+			expLog: `<AsteriskAgent> missing or invalid 'channelvars' field in 'channel': {"channelvars":"invalid"}`,
+		},
+		{
+			name:   "Valid ChannelVars",
+			ariEv:  map[string]any{"channel": map[string]any{"channelvars": map[string]any{utils.CGRReqType: "test type"}}},
+			expLog: "",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := &bytes.Buffer{}
+			log.SetOutput(buf)
+			ev := NewSMAsteriskEvent(tc.ariEv, "127.0.0.1", utils.EmptyString)
+			sma.handleChannelDestroyed(ev)
+			if !strings.Contains(buf.String(), tc.expLog) {
+				t.Errorf("expected log warning %s", buf)
+			}
+
+		})
+	}
+
+}

agents/asterisk_event.go

@@ -48,6 +48,29 @@ type SMAsteriskEvent struct { // Standalone struct so we can cache the fields wh
 	opts          map[string]any
 }
 
+// Clone returns a deep copy of SMAsteriskEvent.
+func (e *SMAsteriskEvent) Clone() *SMAsteriskEvent {
+	ariEvClone := make(map[string]any, len(e.ariEv))
+	for k, v := range e.ariEv {
+		ariEvClone[k] = v
+	}
+	cachedFieldsClone := make(map[string]string, len(e.cachedFields))
+	for k, v := range e.cachedFields {
+		cachedFieldsClone[k] = v
+	}
+	optsClone := make(map[string]any, len(e.opts))
+	for k, v := range e.opts {
+		optsClone[k] = v
+	}
+	return &SMAsteriskEvent{
+		ariEv:         ariEvClone,
+		asteriskIP:    e.asteriskIP,
+		asteriskAlias: e.asteriskAlias,
+		cachedFields:  cachedFieldsClone,
+		opts:          optsClone,
+	}
+}
+
 // parseStasisArgs will convert the args passed to Stasis into CGRateS attribute/value pairs understood by CGRateS and store them in cachedFields
 // args need to be in the form of []string{"key=value", "key2=value2"}
 func (smaEv *SMAsteriskEvent) parseStasisArgs() {

agents/asterisk_event_test.go

@@ -663,3 +663,36 @@ func TestRestoreAndUpdateFieldsFail(t *testing.T) {
 	}
 
 }
+
+func TestSMAsteriskEventClone(t *testing.T) {
+	e := &SMAsteriskEvent{
+		ariEv: map[string]any{
+			"EV": "ID1",
+			"Id": 1001,
+		},
+		asteriskIP:    "192.168.1.1",
+		asteriskAlias: "pbx-1",
+		cachedFields: map[string]string{
+			"eventType": "ARIChannelStateChange",
+		},
+		opts: map[string]any{
+			"opt1": true,
+		},
+	}
+	clone := e.Clone()
+	if !reflect.DeepEqual(clone.ariEv, e.ariEv) {
+		t.Errorf("ariEv maps are not deeply equal")
+	}
+	clone.ariEv["EV"] = "modified"
+	clone.cachedFields["eventType"] = "modified"
+	clone.opts["opt1"] = false
+	if e.ariEv["EV"] == "modified" {
+		t.Errorf("Modifying clone affected original ariEv")
+	}
+	if e.cachedFields["eventType"] == "modified" {
+		t.Errorf("Modifying clone affected original cachedFields")
+	}
+	if e.opts["opt1"] == false {
+		t.Errorf("Modifying clone affected original opts")
+	}
+}