package gortsplib import ( "bytes" "crypto/tls" "net" "strconv" "testing" "time" "github.com/pion/rtcp" "github.com/pion/rtp" psdp "github.com/pion/sdp/v3" "github.com/stretchr/testify/require" "github.com/bluenviron/gortsplib/v3/pkg/base" "github.com/bluenviron/gortsplib/v3/pkg/conn" "github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/headers" "github.com/bluenviron/gortsplib/v3/pkg/media" "github.com/bluenviron/gortsplib/v3/pkg/sdp" ) func invalidURLAnnounceReq(t *testing.T, control string) base.Request { return base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: func() []byte { medi := testH264Media medi.Control = control sout := &sdp.SessionDescription{ SessionName: psdp.SessionName("Stream"), Origin: psdp.Origin{ Username: "-", NetworkType: "IN", AddressType: "IP4", UnicastAddress: "127.0.0.1", }, TimeDescriptions: []psdp.TimeDescription{ {Timing: psdp.Timing{0, 0}}, //nolint:govet }, MediaDescriptions: []*psdp.MediaDescription{medi.Marshal()}, } byts, _ := sout.Marshal() return byts }(), } } func TestServerRecordErrorAnnounce(t *testing.T) { for _, ca := range []struct { name string req base.Request err string }{ { "missing content-type", base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, }, }, "Content-Type header is missing", }, { "invalid content-type", base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"aa"}, }, }, "unsupported Content-Type header '[aa]'", }, { "invalid medias", base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: []byte{0x01, 0x02, 0x03, 0x04}, }, "invalid SDP: invalid line: (\x01\x02\x03\x04)", }, { "invalid URL 1", invalidURLAnnounceReq(t, "rtsp:// aaaaa"), "unable to generate media URL", }, { "invalid URL 3", invalidURLAnnounceReq(t, "rtsp://host/otherpath"), "invalid media path: must begin with '/teststream', but is '/otherpath'", }, } { t.Run(ca.name, func(t *testing.T) { nconnClosed := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { require.EqualError(t, ctx.Error, ca.err) close(nconnClosed) }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) _, err = writeReqReadRes(conn, ca.req) require.NoError(t, err) <-nconnClosed }) } } func TestServerRecordPath(t *testing.T) { for _, ca := range []struct { name string control string announceURL string setupURL string path string query string }{ { "normal", "bbb=ccc", "rtsp://localhost:8554/teststream", "rtsp://localhost:8554/teststream/bbb=ccc", "/teststream", "", }, { "subpath", "ddd=eee", "rtsp://localhost:8554/test/stream", "rtsp://localhost:8554/test/stream/ddd=eee", "/test/stream", "", }, { "subpath and query", "fff=ggg", "rtsp://localhost:8554/test/stream?testing=0", "rtsp://localhost:8554/test/stream?testing=0/fff=ggg", "/test/stream", "testing=0", }, { "no path", "streamid=1", "rtsp://localhost:8554", "rtsp://localhost:8554/streamid=1", "", "", }, { "single slash", "streamid=1", "rtsp://localhost:8554/", "rtsp://localhost:8554//streamid=1", "/", "", }, } { t.Run(ca.name, func(t *testing.T) { s := &Server{ Handler: &testServerHandler{ onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { // make sure that media URLs are not overridden by NewServerStream() stream := NewServerStream(ctx.Medias) defer stream.Close() return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { require.Equal(t, ca.path, ctx.Path) require.Equal(t, ca.query, ctx.Query) return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { require.Equal(t, ca.path, ctx.Path) require.Equal(t, ca.query, ctx.Query) return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) media := testH264Media media.Control = ca.control sout := &sdp.SessionDescription{ SessionName: psdp.SessionName("Stream"), Origin: psdp.Origin{ Username: "-", NetworkType: "IN", AddressType: "IP4", UnicastAddress: "127.0.0.1", }, TimeDescriptions: []psdp.TimeDescription{ {Timing: psdp.Timing{0, 0}}, //nolint:govet }, MediaDescriptions: []*psdp.MediaDescription{media.Marshal()}, } byts, _ := sout.Marshal() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL(ca.announceURL), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: byts, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) th := &headers.Transport{ Protocol: headers.TransportProtocolTCP, Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), InterleavedIDs: &[2]int{0, 1}, } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL(ca.setupURL), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": th.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL(ca.announceURL), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) }) } } func TestServerRecordErrorSetupMediaTwice(t *testing.T) { serverErr := make(chan error) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { serverErr <- ctx.Error }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, }, RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": headers.Transport{ Protocol: headers.TransportProtocolTCP, Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), InterleavedIDs: &[2]int{0, 1}, }.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Transport": headers.Transport{ Protocol: headers.TransportProtocolTCP, Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), InterleavedIDs: &[2]int{2, 3}, }.Marshal(), "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusBadRequest, res.StatusCode) err = <-serverErr require.EqualError(t, err, "media has already been setup") } func TestServerRecordErrorRecordPartialMedias(t *testing.T) { serverErr := make(chan error) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { serverErr <- ctx.Error }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) forma := &formats.Generic{ PayloadTyp: 96, RTPMap: "private/90000", } err = forma.Init() require.NoError(t, err) medias := media.Medias{ &media.Media{ Type: "application", Formats: []formats.Format{forma}, }, &media.Media{ Type: "application", Formats: []formats.Format{forma}, }, } medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) th := &headers.Transport{ Protocol: headers.TransportProtocolTCP, Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), InterleavedIDs: &[2]int{0, 1}, } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": th.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusBadRequest, res.StatusCode) err = <-serverErr require.EqualError(t, err, "not all announced medias have been setup") } func TestServerRecord(t *testing.T) { for _, transport := range []string{ "udp", "tcp", "tls", } { t.Run(transport, func(t *testing.T) { nconnOpened := make(chan struct{}) nconnClosed := make(chan struct{}) sessionOpened := make(chan struct{}) sessionClosed := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onConnOpen: func(ctx *ServerHandlerOnConnOpenCtx) { close(nconnOpened) }, onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { close(nconnClosed) }, onSessionOpen: func(ctx *ServerHandlerOnSessionOpenCtx) { close(sessionOpened) }, onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) { close(sessionClosed) }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { // send RTCP packets directly to the session. // these are sent after the response, only if onRecord returns StatusOK. ctx.Session.WritePacketRTCP(ctx.Session.AnnouncedMedias()[0], &testRTCPPacket) ctx.Session.WritePacketRTCP(ctx.Session.AnnouncedMedias()[1], &testRTCPPacket) for i := 0; i < 2; i++ { ctx.Session.OnPacketRTP( ctx.Session.AnnouncedMedias()[i], ctx.Session.AnnouncedMedias()[i].Formats[0], func(pkt *rtp.Packet) { require.Equal(t, &testRTPPacket, pkt) }) ci := i ctx.Session.OnPacketRTCP( ctx.Session.AnnouncedMedias()[i], func(pkt rtcp.Packet) { require.Equal(t, &testRTCPPacket, pkt) ctx.Session.WritePacketRTCP(ctx.Session.AnnouncedMedias()[ci], &testRTCPPacket) }) } return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, RTSPAddress: "localhost:8554", } switch transport { case "udp": s.UDPRTPAddress = "127.0.0.1:8000" s.UDPRTCPAddress = "127.0.0.1:8001" case "tls": cert, err := tls.X509KeyPair(serverCert, serverKey) require.NoError(t, err) s.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}} } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() nconn = func() net.Conn { if transport == "tls" { return tls.Client(nconn, &tls.Config{InsecureSkipVerify: true}) } return nconn }() conn := conn.NewConn(nconn) <-nconnOpened medias := media.Medias{ &media.Media{ Type: media.TypeVideo, Formats: []formats.Format{&formats.H264{ PayloadTyp: 96, SPS: []byte{0x01, 0x02, 0x03, 0x04}, PPS: []byte{0x01, 0x02, 0x03, 0x04}, PacketizationMode: 1, }}, }, &media.Media{ Type: media.TypeVideo, Formats: []formats.Format{&formats.H264{ PayloadTyp: 96, SPS: []byte{0x01, 0x02, 0x03, 0x04}, PPS: []byte{0x01, 0x02, 0x03, 0x04}, PacketizationMode: 1, }}, }, } medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) <-sessionOpened var l1s [2]net.PacketConn var l2s [2]net.PacketConn var sx headers.Session var serverPorts [2]*[2]int for i := 0; i < 2; i++ { inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), } if transport == "udp" { inTH.Protocol = headers.TransportProtocolUDP inTH.ClientPorts = &[2]int{35466 + i*2, 35467 + i*2} l1s[i], err = net.ListenPacket("udp", "localhost:"+strconv.FormatInt(int64(inTH.ClientPorts[0]), 10)) require.NoError(t, err) defer l1s[i].Close() l2s[i], err = net.ListenPacket("udp", "localhost:"+strconv.FormatInt(int64(inTH.ClientPorts[1]), 10)) require.NoError(t, err) defer l2s[i].Close() } else { inTH.Protocol = headers.TransportProtocolTCP inTH.InterleavedIDs = &[2]int{2 + i*2, 3 + i*2} } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[i].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var th headers.Transport err = th.Unmarshal(res.Header["Transport"]) require.NoError(t, err) if transport == "udp" { serverPorts[i] = th.ServerPorts } } res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) for i := 0; i < 2; i++ { // server -> client (direct) if transport == "udp" { buf := make([]byte, 2048) n, _, err := l2s[i].ReadFrom(buf) require.NoError(t, err) require.Equal(t, testRTCPPacketMarshaled, buf[:n]) } else { f, err := conn.ReadInterleavedFrame() require.NoError(t, err) require.Equal(t, 3+i*2, f.Channel) require.Equal(t, testRTCPPacketMarshaled, f.Payload) } // skip firewall opening if transport == "udp" { buf := make([]byte, 2048) _, _, err := l2s[i].ReadFrom(buf) require.NoError(t, err) } // client -> server if transport == "udp" { l1s[i].WriteTo(testRTPPacketMarshaled, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: serverPorts[i][0], }) l2s[i].WriteTo(testRTCPPacketMarshaled, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: serverPorts[i][1], }) } else { err := conn.WriteInterleavedFrame(&base.InterleavedFrame{ Channel: 2 + i*2, Payload: testRTPPacketMarshaled, }, make([]byte, 1024)) require.NoError(t, err) err = conn.WriteInterleavedFrame(&base.InterleavedFrame{ Channel: 3 + i*2, Payload: testRTCPPacketMarshaled, }, make([]byte, 1024)) require.NoError(t, err) } } for i := 0; i < 2; i++ { // server -> client (RTCP) if transport == "udp" { buf := make([]byte, 2048) n, _, err := l2s[i].ReadFrom(buf) require.NoError(t, err) require.Equal(t, testRTCPPacketMarshaled, buf[:n]) } else { f, err := conn.ReadInterleavedFrame() require.NoError(t, err) require.Equal(t, 3+i*2, f.Channel) require.Equal(t, testRTCPPacketMarshaled, f.Payload) } } res, err = writeReqReadRes(conn, base.Request{ Method: base.Teardown, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) <-sessionClosed nconn.Close() <-nconnClosed }) } } func TestServerRecordErrorInvalidProtocol(t *testing.T) { errorRecv := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { require.EqualError(t, ctx.Error, "received unexpected interleaved frame") close(errorRecv) }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, UDPRTPAddress: "127.0.0.1:8000", UDPRTCPAddress: "127.0.0.1:8001", RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), Protocol: headers.TransportProtocolUDP, ClientPorts: &[2]int{35466, 35467}, } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var th headers.Transport err = th.Unmarshal(res.Header["Transport"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) err = conn.WriteInterleavedFrame(&base.InterleavedFrame{ Channel: 0, Payload: []byte{0x01, 0x02, 0x03, 0x04}, }, make([]byte, 1024)) require.NoError(t, err) <-errorRecv } func TestServerRecordRTCPReport(t *testing.T) { s := &Server{ Handler: &testServerHandler{ onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, udpReceiverReportPeriod: 1 * time.Second, UDPRTPAddress: "127.0.0.1:8000", UDPRTCPAddress: "127.0.0.1:8001", RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) l1, err := net.ListenPacket("udp", "localhost:34556") require.NoError(t, err) defer l1.Close() l2, err := net.ListenPacket("udp", "localhost:34557") require.NoError(t, err) defer l2.Close() res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), Protocol: headers.TransportProtocolUDP, ClientPorts: &[2]int{34556, 34557}, }.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var th headers.Transport err = th.Unmarshal(res.Header["Transport"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) byts, _ := (&rtp.Packet{ Header: rtp.Header{ Version: 2, Marker: true, PayloadType: 96, SequenceNumber: 534, Timestamp: 54352, SSRC: 753621, }, Payload: []byte{0x01, 0x02, 0x03, 0x04}, }).Marshal() _, err = l1.WriteTo(byts, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: th.ServerPorts[0], }) require.NoError(t, err) // wait for the packet's SSRC to be saved time.Sleep(500 * time.Millisecond) byts, _ = (&rtcp.SenderReport{ SSRC: 753621, NTPTime: 0xcbddcc34999997ff, RTPTime: 54352, PacketCount: 1, OctetCount: 4, }).Marshal() _, err = l2.WriteTo(byts, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: th.ServerPorts[1], }) require.NoError(t, err) // skip firewall opening buf := make([]byte, 2048) _, _, err = l2.ReadFrom(buf) require.NoError(t, err) buf = make([]byte, 2048) n, _, err := l2.ReadFrom(buf) require.NoError(t, err) pkts, err := rtcp.Unmarshal(buf[:n]) require.NoError(t, err) rr, ok := pkts[0].(*rtcp.ReceiverReport) require.True(t, ok) require.Equal(t, &rtcp.ReceiverReport{ SSRC: rr.SSRC, Reports: []rtcp.ReceptionReport{ { SSRC: rr.Reports[0].SSRC, LastSequenceNumber: 534, LastSenderReport: rr.Reports[0].LastSenderReport, Delay: rr.Reports[0].Delay, }, }, ProfileExtensions: []uint8{}, }, rr) } func TestServerRecordTimeout(t *testing.T) { for _, transport := range []string{ "udp", "tcp", } { t.Run(transport, func(t *testing.T) { nconnClosed := make(chan struct{}) sessionClosed := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { close(nconnClosed) }, onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) { close(sessionClosed) }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, ReadTimeout: 1 * time.Second, RTSPAddress: "localhost:8554", checkStreamPeriod: 500 * time.Millisecond, } if transport == "udp" { s.UDPRTPAddress = "127.0.0.1:8000" s.UDPRTCPAddress = "127.0.0.1:8001" } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), } if transport == "udp" { inTH.Protocol = headers.TransportProtocolUDP inTH.ClientPorts = &[2]int{35466, 35467} } else { inTH.Protocol = headers.TransportProtocolTCP inTH.InterleavedIDs = &[2]int{0, 1} } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var th headers.Transport err = th.Unmarshal(res.Header["Transport"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) <-sessionClosed if transport == "tcp" { <-nconnClosed } }) } } func TestServerRecordWithoutTeardown(t *testing.T) { for _, transport := range []string{ "udp", "tcp", } { t.Run(transport, func(t *testing.T) { nconnClosed := make(chan struct{}) sessionClosed := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) { close(nconnClosed) }, onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) { close(sessionClosed) }, onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, ReadTimeout: 1 * time.Second, RTSPAddress: "localhost:8554", } if transport == "udp" { s.UDPRTPAddress = "127.0.0.1:8000" s.UDPRTCPAddress = "127.0.0.1:8001" } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), } if transport == "udp" { inTH.Protocol = headers.TransportProtocolUDP inTH.ClientPorts = &[2]int{35466, 35467} } else { inTH.Protocol = headers.TransportProtocolTCP inTH.InterleavedIDs = &[2]int{0, 1} } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var th headers.Transport err = th.Unmarshal(res.Header["Transport"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) nconn.Close() <-sessionClosed <-nconnClosed }) } } func TestServerRecordUDPChangeConn(t *testing.T) { s := &Server{ Handler: &testServerHandler{ onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onGetParameter: func(ctx *ServerHandlerOnGetParameterCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, }, UDPRTPAddress: "127.0.0.1:8000", UDPRTCPAddress: "127.0.0.1:8001", RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() sxID := "" func() { nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{testH264Media} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), Protocol: headers.TransportProtocolUDP, ClientPorts: &[2]int{35466, 35467}, } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) sxID = sx.Session }() func() { nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) res, err := writeReqReadRes(conn, base.Request{ Method: base.GetParameter, URL: mustParseURL("rtsp://localhost:8554/teststream/"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Session": base.HeaderValue{sxID}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) }() } func TestServerRecordDecodeErrors(t *testing.T) { for _, ca := range []struct { proto string name string }{ {"udp", "rtp invalid"}, {"udp", "rtcp invalid"}, {"udp", "rtp packets lost"}, {"udp", "rtp too big"}, {"udp", "rtcp too big"}, {"tcp", "rtcp invalid"}, {"tcp", "rtcp too big"}, } { t.Run(ca.proto+" "+ca.name, func(t *testing.T) { errorRecv := make(chan struct{}) s := &Server{ Handler: &testServerHandler{ onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil, nil }, onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) { return &base.Response{ StatusCode: base.StatusOK, }, nil }, onPacketLost: func(ctx *ServerHandlerOnPacketLostCtx) { if ca.proto == "udp" && ca.name == "rtp packets lost" { require.EqualError(t, ctx.Error, "69 RTP packet(s) lost") } close(errorRecv) }, onDecodeError: func(ctx *ServerHandlerOnDecodeErrorCtx) { switch { case ca.proto == "udp" && ca.name == "rtp invalid": require.EqualError(t, ctx.Error, "RTP header size insufficient: 2 < 4") case ca.proto == "udp" && ca.name == "rtcp invalid": require.EqualError(t, ctx.Error, "rtcp: packet too short") case ca.proto == "udp" && ca.name == "rtp too big": require.EqualError(t, ctx.Error, "RTP packet is too big to be read with UDP") case ca.proto == "udp" && ca.name == "rtcp too big": require.EqualError(t, ctx.Error, "RTCP packet is too big to be read with UDP") case ca.proto == "tcp" && ca.name == "rtcp invalid": require.EqualError(t, ctx.Error, "rtcp: packet too short") case ca.proto == "tcp" && ca.name == "rtcp too big": require.EqualError(t, ctx.Error, "RTCP packet size (2000) is greater than maximum allowed (1472)") } close(errorRecv) }, }, UDPRTPAddress: "127.0.0.1:8000", UDPRTCPAddress: "127.0.0.1:8001", RTSPAddress: "localhost:8554", } err := s.Start() require.NoError(t, err) defer s.Close() nconn, err := net.Dial("tcp", "localhost:8554") require.NoError(t, err) defer nconn.Close() conn := conn.NewConn(nconn) medias := media.Medias{&media.Media{ Type: media.TypeApplication, Formats: []formats.Format{&formats.Generic{ PayloadTyp: 97, RTPMap: "private/90000", }}, }} medias.SetControls() res, err := writeReqReadRes(conn, base.Request{ Method: base.Announce, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"1"}, "Content-Type": base.HeaderValue{"application/sdp"}, }, Body: mustMarshalMedias(medias), }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) inTH := &headers.Transport{ Delivery: func() *headers.TransportDelivery { v := headers.TransportDeliveryUnicast return &v }(), Mode: func() *headers.TransportMode { v := headers.TransportModeRecord return &v }(), } if ca.proto == "udp" { inTH.Protocol = headers.TransportProtocolUDP inTH.ClientPorts = &[2]int{35466, 35467} } else { inTH.Protocol = headers.TransportProtocolTCP inTH.InterleavedIDs = &[2]int{0, 1} } var l1 net.PacketConn var l2 net.PacketConn if ca.proto == "udp" { l1, err = net.ListenPacket("udp", "127.0.0.1:35466") require.NoError(t, err) defer l1.Close() l2, err = net.ListenPacket("udp", "127.0.0.1:35467") require.NoError(t, err) defer l2.Close() } res, err = writeReqReadRes(conn, base.Request{ Method: base.Setup, URL: mustParseURL("rtsp://localhost:8554/teststream/" + medias[0].Control), Header: base.Header{ "CSeq": base.HeaderValue{"2"}, "Transport": inTH.Marshal(), }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) var sx headers.Session err = sx.Unmarshal(res.Header["Session"]) require.NoError(t, err) var resTH headers.Transport err = resTH.Unmarshal(res.Header["Transport"]) require.NoError(t, err) res, err = writeReqReadRes(conn, base.Request{ Method: base.Record, URL: mustParseURL("rtsp://localhost:8554/teststream"), Header: base.Header{ "CSeq": base.HeaderValue{"3"}, "Session": base.HeaderValue{sx.Session}, }, }) require.NoError(t, err) require.Equal(t, base.StatusOK, res.StatusCode) switch { //nolint:dupl case ca.proto == "udp" && ca.name == "rtp invalid": l1.WriteTo([]byte{0x01, 0x02}, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[0], }) case ca.proto == "udp" && ca.name == "rtcp invalid": l2.WriteTo([]byte{0x01, 0x02}, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[1], }) case ca.proto == "udp" && ca.name == "rtp packets lost": byts, _ := rtp.Packet{ Header: rtp.Header{ PayloadType: 97, SequenceNumber: 30, }, }.Marshal() l1.WriteTo(byts, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[0], }) byts, _ = rtp.Packet{ Header: rtp.Header{ PayloadType: 97, SequenceNumber: 100, }, }.Marshal() l1.WriteTo(byts, &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[0], }) case ca.proto == "udp" && ca.name == "rtp too big": l1.WriteTo(bytes.Repeat([]byte{0x01, 0x02}, 2000/2), &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[0], }) case ca.proto == "udp" && ca.name == "rtcp too big": l2.WriteTo(bytes.Repeat([]byte{0x01, 0x02}, 2000/2), &net.UDPAddr{ IP: net.ParseIP("127.0.0.1"), Port: resTH.ServerPorts[1], }) case ca.proto == "tcp" && ca.name == "rtcp invalid": err = conn.WriteInterleavedFrame(&base.InterleavedFrame{ Channel: 1, Payload: []byte{0x01, 0x02}, }, make([]byte, 2048)) require.NoError(t, err) case ca.proto == "tcp" && ca.name == "rtcp too big": err = conn.WriteInterleavedFrame(&base.InterleavedFrame{ Channel: 1, Payload: bytes.Repeat([]byte{0x01, 0x02}, 2000/2), }, make([]byte, 2048)) require.NoError(t, err) } <-errorRecv }) } }