Validate received SDP earlier.

api_signaling.go

@@ -32,6 +32,7 @@ import (
 	"time"
 
 	"github.com/golang-jwt/jwt/v4"
+	"github.com/pion/sdp/v3"
 )
 
 const (
@@ -42,6 +43,11 @@ const (
 	HelloVersionV2 = "2.0"
 )
 
+var (
+	ErrNoSdp      = NewError("no_sdp", "Payload does not contain a SDP.")
+	ErrInvalidSdp = NewError("invalid_sdp", "Payload does not contain a valid SDP.")
+)
+
 // ClientMessage is a message that is sent from a client to the server.
 type ClientMessage struct {
 	json.Marshaler
@@ -563,12 +569,39 @@ type MessageClientMessageData struct {
 	RoomType string                 `json:"roomType"`
 	Bitrate  int                    `json:"bitrate,omitempty"`
 	Payload  map[string]interface{} `json:"payload"`
+
+	offerSdp  *sdp.SessionDescription // Only set if Type == "offer"
+	answerSdp *sdp.SessionDescription // Only set if Type == "answer"
 }
 
 func (m *MessageClientMessageData) CheckValid() error {
 	if !IsValidStreamType(m.RoomType) {
 		return fmt.Errorf("invalid room type: %s", m.RoomType)
 	}
+	if m.Type == "offer" || m.Type == "answer" {
+		sdpValue, found := m.Payload["sdp"]
+		if !found {
+			return ErrNoSdp
+		}
+		sdpText, ok := sdpValue.(string)
+		if !ok {
+			return ErrInvalidSdp
+		}
+
+		var sdp sdp.SessionDescription
+		if err := sdp.Unmarshal([]byte(sdpText)); err != nil {
+			return NewErrorDetail("invalid_sdp", "Error parsing SDP from payload.", map[string]interface{}{
+				"error": err.Error(),
+			})
+		}
+
+		switch m.Type {
+		case "offer":
+			m.offerSdp = &sdp
+		case "answer":
+			m.answerSdp = &sdp
+		}
+	}
 	return nil
 }
 

clientsession.go

@@ -730,23 +730,6 @@ func (s *ClientSession) SubscriberClosed(subscriber McuSubscriber) {
 	}
 }
 
-type SdpError struct {
-	message string
-}
-
-func (e *SdpError) Error() string {
-	return e.message
-}
-
-type WrappedSdpError struct {
-	SdpError
-	err error
-}
-
-func (e *WrappedSdpError) Unwrap() error {
-	return e.err
-}
-
 type PermissionError struct {
 	permission Permission
 }
@@ -759,23 +742,10 @@ func (e *PermissionError) Error() string {
 	return fmt.Sprintf("permission \"%s\" not found", e.permission)
 }
 
-func (s *ClientSession) isSdpAllowedToSendLocked(payload map[string]interface{}) (MediaType, error) {
-	sdpValue, found := payload["sdp"]
-	if !found {
-		return 0, &SdpError{"payload does not contain a sdp"}
-	}
-	sdpText, ok := sdpValue.(string)
-	if !ok {
-		return 0, &SdpError{"payload does not contain a valid sdp"}
-	}
-	var sdp sdp.SessionDescription
-	if err := sdp.Unmarshal([]byte(sdpText)); err != nil {
-		return 0, &WrappedSdpError{
-			SdpError: SdpError{
-				message: fmt.Sprintf("could not parse sdp: %s", err),
-			},
-			err: err,
-		}
+func (s *ClientSession) isSdpAllowedToSendLocked(sdp *sdp.SessionDescription) (MediaType, error) {
+	if sdp == nil {
+		// Should have already been checked when data was validated.
+		return 0, ErrNoSdp
 	}
 
 	var mediaTypes MediaType
@@ -813,8 +783,8 @@ func (s *ClientSession) IsAllowedToSend(data *MessageClientMessageData) error {
 		// Client is allowed to publish any media (audio / video).
 		return nil
 	} else if data != nil && data.Type == "offer" {
-		// Parse SDP to check what user is trying to publish and check permissions accordingly.
-		if _, err := s.isSdpAllowedToSendLocked(data.Payload); err != nil {
+		// Check what user is trying to publish and check permissions accordingly.
+		if _, err := s.isSdpAllowedToSendLocked(data.offerSdp); err != nil {
 			return err
 		}
 
@@ -844,7 +814,7 @@ func (s *ClientSession) checkOfferTypeLocked(streamType StreamType, data *Messag
 
 		return MediaTypeScreen, nil
 	} else if data != nil && data.Type == "offer" {
-		mediaTypes, err := s.isSdpAllowedToSendLocked(data.Payload)
+		mediaTypes, err := s.isSdpAllowedToSendLocked(data.offerSdp)
 		if err != nil {
 			return 0, err
 		}

hub.go

@@ -2334,11 +2334,6 @@ func (h *Hub) processMcuMessage(session *ClientSession, client_message *ClientMe
 			sendNotAllowed(session, client_message, "Not allowed to publish.")
 			return
 		}
-		if err, ok := err.(*SdpError); ok {
-			log.Printf("Session %s sent unsupported offer %s, ignoring (%s)", session.PublicId(), data.RoomType, err)
-			sendNotAllowed(session, client_message, "Not allowed to publish.")
-			return
-		}
 	case "selectStream":
 		if session.PublicId() == message.Recipient.SessionId {
 			log.Printf("Not selecting substream for own %s stream in session %s", data.RoomType, session.PublicId())

hub_test.go

@@ -4697,6 +4697,30 @@ func TestClientRequestOfferNotInRoom(t *testing.T) {
 			if err := client2.RunUntilOffer(ctx, MockSdpOfferAudioAndVideo); err != nil {
 				t.Fatal(err)
 			}
+
+			if err := client2.SendMessage(MessageClientMessageRecipient{
+				Type:      "session",
+				SessionId: hello1.Hello.SessionId,
+			}, MessageClientMessageData{
+				Type:     "answer",
+				Sid:      "12345",
+				RoomType: "screen",
+				Payload: map[string]interface{}{
+					"sdp": MockSdpAnswerAudioAndVideo,
+				},
+			}); err != nil {
+				t.Fatal(err)
+			}
+
+			// The sender won't get a reply...
+			ctx2, cancel2 := context.WithTimeout(context.Background(), 100*time.Millisecond)
+			defer cancel2()
+
+			if message, err := client2.RunUntilMessage(ctx2); err != nil && err != ErrNoMessageReceived && err != context.DeadlineExceeded {
+				t.Error(err)
+			} else if message != nil {
+				t.Errorf("Expected no message, got %+v", message)
+			}
 		})
 	}
 }

mcu_test.go

@@ -253,6 +253,8 @@ func (s *TestMCUSubscriber) SendMessage(ctx context.Context, message *MessageCli
 				"type": "offer",
 				"sdp":  sdp,
 			})
+		case "answer":
+			callback(nil, nil)
 		default:
 			callback(fmt.Errorf("Message type %s is not implemented", data.Type), nil)
 		}

proxy/proxy_server.go

@@ -777,9 +777,10 @@ func (s *ProxyServer) processPayload(ctx context.Context, client *ProxyClient, s
 		fallthrough
 	case "candidate":
 		mcuData = &signaling.MessageClientMessageData{
-			Type:    payload.Type,
-			Sid:     payload.Sid,
-			Payload: payload.Payload,
+			RoomType: string(mcuClient.StreamType()),
+			Type:     payload.Type,
+			Sid:      payload.Sid,
+			Payload:  payload.Payload,
 		}
 	case "endOfCandidates":
 		// Ignore but confirm, not passed along to Janus anyway.
@@ -796,14 +797,21 @@ func (s *ProxyServer) processPayload(ctx context.Context, client *ProxyClient, s
 		fallthrough
 	case "sendoffer":
 		mcuData = &signaling.MessageClientMessageData{
-			Type: payload.Type,
-			Sid:  payload.Sid,
+			RoomType: string(mcuClient.StreamType()),
+			Type:     payload.Type,
+			Sid:      payload.Sid,
 		}
 	default:
 		session.sendMessage(message.NewErrorServerMessage(UnsupportedPayload))
 		return
 	}
 
+	if err := mcuData.CheckValid(); err != nil {
+		log.Printf("Received invalid payload %+v for %s client %s: %s", mcuData, mcuClient.StreamType(), payload.ClientId, err)
+		session.sendMessage(message.NewErrorServerMessage(UnsupportedPayload))
+		return
+	}
+
 	mcuClient.SendMessage(ctx, nil, mcuData, func(err error, response map[string]interface{}) {
 		var responseMsg *signaling.ProxyServerMessage
 		if err != nil {