From c2883f2ce9764ea35fb8387bd16b6b8f38da2592 Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Tue, 13 Feb 2024 17:37:03 +0100 Subject: [PATCH] hls: when alwaysRemux is true, prevent clients from creating muxers (#3015) --- internal/servers/hls/http_server.go | 22 +++++++++--- internal/servers/hls/muxer.go | 36 ++++++++++---------- internal/servers/hls/server.go | 52 +++++++++++++++++------------ 3 files changed, 66 insertions(+), 44 deletions(-) diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 5de849d1..8d6398b0 100644 --- a/internal/servers/hls/http_server.go +++ b/internal/servers/hls/http_server.go @@ -191,10 +191,24 @@ func (s *httpServer) onRequest(ctx *gin.Context) { ctx.Writer.Write(hlsIndex) default: - s.parent.handleRequest(muxerHandleRequestReq{ - path: dir, - file: fname, - ctx: ctx, + mux, err := s.parent.getMuxer(serverGetMuxerReq{ + path: dir, + remoteAddr: httpp.RemoteAddr(ctx), }) + if err != nil { + s.Log(logger.Error, err.Error()) + ctx.Writer.WriteHeader(http.StatusInternalServerError) + return + } + + err = mux.processRequest(muxerProcessRequestReq{}) + if err != nil { + s.Log(logger.Error, err.Error()) + ctx.Writer.WriteHeader(http.StatusInternalServerError) + return + } + + ctx.Request.URL.Path = fname + mux.handleRequest(ctx) } } diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 2f7ec443..9fa1733d 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -51,11 +51,8 @@ func (w *responseWriterWithCounter) Write(p []byte) (int, error) { return n, err } -type muxerHandleRequestReq struct { - path string - file string - ctx *gin.Context - res chan *muxer +type muxerProcessRequestReq struct { + res chan error } type muxer struct { @@ -81,11 +78,11 @@ type muxer struct { writer *asyncwriter.Writer lastRequestTime *int64 muxer *gohlslib.Muxer - requests []*muxerHandleRequestReq + requests []*muxerProcessRequestReq bytesSent *uint64 // in - chRequest chan *muxerHandleRequestReq + chProcessRequest chan muxerProcessRequestReq } func (m *muxer) initialize() { @@ -96,7 +93,7 @@ func (m *muxer) initialize() { m.created = time.Now() m.lastRequestTime = int64Ptr(time.Now().UnixNano()) m.bytesSent = new(uint64) - m.chRequest = make(chan *muxerHandleRequestReq) + m.chProcessRequest = make(chan muxerProcessRequestReq) m.Log(logger.Info, "created %s", func() string { if m.remoteAddr == "" { @@ -156,22 +153,22 @@ func (m *muxer) run() { } return errors.New("terminated") - case req := <-m.chRequest: + case req := <-m.chProcessRequest: switch { case isRecreating: - req.res <- nil + req.res <- fmt.Errorf("recreating") case isReady: - req.res <- m + req.res <- nil default: - m.requests = append(m.requests, req) + m.requests = append(m.requests, &req) } case <-innerReady: isReady = true for _, req := range m.requests { - req.res <- m + req.res <- nil } m.requests = nil @@ -206,7 +203,7 @@ func (m *muxer) run() { func (m *muxer) clearQueuedRequests() { for _, req := range m.requests { - req.res <- nil + req.res <- fmt.Errorf("terminated") } m.requests = nil } @@ -479,12 +476,15 @@ func (m *muxer) handleRequest(ctx *gin.Context) { m.muxer.Handle(w, ctx.Request) } -// processRequest is called by hlsserver.Server (forwarded from ServeHTTP). -func (m *muxer) processRequest(req *muxerHandleRequestReq) { +func (m *muxer) processRequest(req muxerProcessRequestReq) error { + req.res = make(chan error) + select { - case m.chRequest <- req: + case m.chProcessRequest <- req: + return <-req.res + case <-m.ctx.Done(): - req.res <- nil + return fmt.Errorf("terminated") } } diff --git a/internal/servers/hls/server.go b/internal/servers/hls/server.go index a673e6cf..98856226 100644 --- a/internal/servers/hls/server.go +++ b/internal/servers/hls/server.go @@ -11,12 +11,22 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" - "github.com/bluenviron/mediamtx/internal/protocols/httpp" ) // ErrMuxerNotFound is returned when a muxer is not found. var ErrMuxerNotFound = errors.New("muxer not found") +type serverGetMuxerRes struct { + muxer *muxer + err error +} + +type serverGetMuxerReq struct { + path string + remoteAddr string + res chan serverGetMuxerRes +} + type serverAPIMuxersListRes struct { data *defs.APIHLSMuxerList err error @@ -68,12 +78,12 @@ type Server struct { muxers map[string]*muxer // in - chPathReady chan defs.Path - chPathNotReady chan defs.Path - chHandleRequest chan muxerHandleRequestReq - chCloseMuxer chan *muxer - chAPIMuxerList chan serverAPIMuxersListReq - chAPIMuxerGet chan serverAPIMuxersGetReq + chPathReady chan defs.Path + chPathNotReady chan defs.Path + chGetMuxer chan serverGetMuxerReq + chCloseMuxer chan *muxer + chAPIMuxerList chan serverAPIMuxersListReq + chAPIMuxerGet chan serverAPIMuxersGetReq } // Initialize initializes the server. @@ -85,7 +95,7 @@ func (s *Server) Initialize() error { s.muxers = make(map[string]*muxer) s.chPathReady = make(chan defs.Path) s.chPathNotReady = make(chan defs.Path) - s.chHandleRequest = make(chan muxerHandleRequestReq) + s.chGetMuxer = make(chan serverGetMuxerReq) s.chCloseMuxer = make(chan *muxer) s.chAPIMuxerList = make(chan serverAPIMuxersListReq) s.chAPIMuxerGet = make(chan serverAPIMuxersGetReq) @@ -147,15 +157,15 @@ outer: delete(s.muxers, pa.Name()) } - case req := <-s.chHandleRequest: - r, ok := s.muxers[req.path] + case req := <-s.chGetMuxer: + mux, ok := s.muxers[req.path] switch { case ok: - r.processRequest(&req) - + req.res <- serverGetMuxerRes{muxer: mux} + case !s.AlwaysRemux: + req.res <- serverGetMuxerRes{muxer: s.createMuxer(req.path, req.remoteAddr)} default: - r := s.createMuxer(req.path, httpp.RemoteAddr(req.ctx)) - r.processRequest(&req) + req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer is waiting to be created")} } case c := <-s.chCloseMuxer: @@ -230,18 +240,16 @@ func (s *Server) closeMuxer(c *muxer) { } } -func (s *Server) handleRequest(req muxerHandleRequestReq) { - req.res = make(chan *muxer) +func (s *Server) getMuxer(req serverGetMuxerReq) (*muxer, error) { + req.res = make(chan serverGetMuxerRes) select { - case s.chHandleRequest <- req: - muxer := <-req.res - if muxer != nil { - req.ctx.Request.URL.Path = req.file - muxer.handleRequest(req.ctx) - } + case s.chGetMuxer <- req: + res := <-req.res + return res.muxer, res.err case <-s.ctx.Done(): + return nil, fmt.Errorf("terminated") } }