diff --git a/api/openapi.yaml b/api/openapi.yaml index afffcb31..98099606 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -45,8 +45,7 @@ components: PathReaderType: type: string enum: - - hlsMuxer - - rpiCameraSecondary + - hlsSession - rtmpConn - rtmpsConn - rtspConn @@ -1196,6 +1195,40 @@ components: items: $ref: '#/components/schemas/HLSMuxer' + HLSSession: + type: object + properties: + id: + type: string + format: uuid + created: + type: string + remoteAddr: + type: string + path: + type: string + query: + type: string + user: + type: string + outboundBytes: + type: integer + format: uint64 + + HLSSessionList: + type: object + properties: + pageCount: + type: integer + format: int64 + itemCount: + type: integer + format: int64 + items: + type: array + items: + $ref: '#/components/schemas/HLSSession' + Recording: type: object properties: @@ -2291,6 +2324,123 @@ paths: schema: $ref: '#/components/schemas/Error' + /v3/hlssessions/list: + get: + operationId: hlssessionsList + tags: [HLS] + summary: returns all HLS sessions. + description: '' + parameters: + - name: page + in: query + description: page number. + schema: + type: integer + default: 0 + - name: itemsPerPage + in: query + description: items per page. + schema: + type: integer + default: 100 + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/HLSSessionList' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /v3/hlssessions/get/{id}: + get: + operationId: hlssessionsGet + tags: [HLS] + summary: returns a HLS session. + description: '' + parameters: + - name: id + in: path + required: true + description: ID of the session. + schema: + type: string + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/HLSSession' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: session not found. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /v3/hlssessions/kick/{id}: + post: + operationId: hlssessionsKick + tags: [HLS] + summary: kicks out a HLS session from the server. + description: '' + parameters: + - name: id + in: path + required: true + description: ID of the session. + schema: + type: string + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/OK' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: session not found. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /v3/paths/list: get: operationId: pathsList diff --git a/docs/2-features/21-metrics.md b/docs/2-features/21-metrics.md index ccc5e003..d3c30a00 100644 --- a/docs/2-features/21-metrics.md +++ b/docs/2-features/21-metrics.md @@ -20,6 +20,10 @@ paths_inbound_bytes{name="[path_name]",state="[state]"} 1234 paths_outbound_bytes{name="[path_name]",state="[state]"} 1234 paths_inbound_frames_in_error{name="[path_name]",state="[state]"} 1234 +# HLS sessions +hls_sessions{id="[id]",path="[path]",remoteAddr="[remoteAddr]"} 1 +hls_sessions_outbound_bytes{id="[id]",path="[path]",remoteAddr="[remoteAddr]"} 187 + # HLS muxers hls_muxers{name="[name]"} 1 hls_muxers_outbound_bytes{name="[name]"} 187 @@ -151,9 +155,10 @@ Bitrates are not provided directly as metrics because they can be computed from Metrics can be filtered by using HTTP query parameters: -- `type=[TYPE]`: show metrics of a certain type only. TYPE can be `paths`, `hls_muxers`, `rtsp_conns`, `rtsp_sessions`, `rtsps_conns`, `rtsps_sessions`, `rtmp_conns`, `rtmps_conns`, `srt_conns`, `webrtc_sessions`. +- `type=[TYPE]`: show metrics of a certain type only. TYPE can be `paths`, `hls_sessions`, `hls_muxers`, `rtsp_conns`, `rtsp_sessions`, `rtsps_conns`, `rtsps_sessions`, `rtmp_conns`, `rtmps_conns`, `srt_conns`, `webrtc_sessions`. - `path=[PATH]`: show metrics belonging to a specific path only - `hls_muxer=[PATH]`: show metrics belonging to a specific HLS muxer only +- `hls_session=[ID]`: show metrics belonging to a specific HLS session only - `rtsp_conn=[ID]` show metrics belonging to a specific RTSP connection only - `rtsp_session=[SESSION]`: show metrics belonging to a specific RTSP session only - `rtsps_conn=[ID]` show metrics belonging to a specific RTSPS connection only diff --git a/docs/4-read/13-web-browsers.md b/docs/4-read/13-web-browsers.md index bfb6a7cd..c0feff45 100644 --- a/docs/4-read/13-web-browsers.md +++ b/docs/4-read/13-web-browsers.md @@ -164,6 +164,8 @@ After the video tag, add a script that initializes the stream when the page is f if (Hls.isSupported()) { const hls = new Hls({ xhrSetup: function (xhr, url) { + xhr.withCredentials = true; + let user = ""; // fill if needed let pass = ""; // fill if needed let token = ""; // fill if needed diff --git a/internal/api/api.go b/internal/api/api.go index 512b04dd..2da565e3 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -120,6 +120,9 @@ func (a *API) Initialize() error { if !interfaceIsEmpty(a.HLSServer) { group.GET("/hlsmuxers/list", a.onHLSMuxersList) group.GET("/hlsmuxers/get/*name", a.onHLSMuxersGet) + group.GET("/hlssessions/list", a.onHLSSessionsList) + group.GET("/hlssessions/get/:id", a.onHLSSessionsGet) + group.POST("/hlssessions/kick/:id", a.onHLSSessionsKick) } if !interfaceIsEmpty(a.RTSPServer) { diff --git a/internal/api/api_hls.go b/internal/api/api_hls.go index 5b9abfd3..961512f5 100644 --- a/internal/api/api_hls.go +++ b/internal/api/api_hls.go @@ -8,6 +8,7 @@ import ( "github.com/bluenviron/mediamtx/internal/servers/hls" "github.com/gin-gonic/gin" + "github.com/google/uuid" ) func (a *API) onHLSMuxersList(ctx *gin.Context) { @@ -47,3 +48,63 @@ func (a *API) onHLSMuxersGet(ctx *gin.Context) { ctx.JSON(http.StatusOK, data) } + +func (a *API) onHLSSessionsList(ctx *gin.Context) { + data, err := a.HLSServer.APISessionsList() + if err != nil { + a.writeError(ctx, http.StatusInternalServerError, err) + return + } + + data.ItemCount = len(data.Items) + pageCount, err := paginate(&data.Items, ctx.Query("itemsPerPage"), ctx.Query("page")) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + data.PageCount = pageCount + + ctx.JSON(http.StatusOK, data) +} + +func (a *API) onHLSSessionsGet(ctx *gin.Context) { + id := ctx.Param("id") + + uuid, err := uuid.Parse(id) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + + data, err := a.HLSServer.APISessionsGet(uuid) + if err != nil { + if errors.Is(err, hls.ErrSessionNotFound) { + a.writeError(ctx, http.StatusNotFound, err) + } else { + a.writeError(ctx, http.StatusInternalServerError, err) + } + return + } + + ctx.JSON(http.StatusOK, data) +} + +func (a *API) onHLSSessionsKick(ctx *gin.Context) { + uuid, err := uuid.Parse(ctx.Param("id")) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + + err = a.HLSServer.APISessionsKick(uuid) + if err != nil { + if errors.Is(err, hls.ErrSessionNotFound) { + a.writeError(ctx, http.StatusNotFound, err) + } else { + a.writeError(ctx, http.StatusInternalServerError, err) + } + return + } + + a.writeOK(ctx) +} diff --git a/internal/api/api_hls_test.go b/internal/api/api_hls_test.go index 71d57000..f370dd26 100644 --- a/internal/api/api_hls_test.go +++ b/internal/api/api_hls_test.go @@ -1,7 +1,9 @@ package api //nolint:revive import ( + "fmt" "net/http" + "sort" "testing" "time" @@ -9,13 +11,17 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/servers/hls" "github.com/bluenviron/mediamtx/internal/test" + "github.com/google/uuid" "github.com/stretchr/testify/require" ) type testHLSServer struct { - muxers map[string]*defs.APIHLSMuxer + muxers map[string]*defs.APIHLSMuxer + sessions map[string]*defs.APIHLSSession } +var testTime = time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC) + func (s *testHLSServer) APIMuxersList() (*defs.APIHLSMuxerList, error) { items := make([]defs.APIHLSMuxer, 0, len(s.muxers)) for _, muxer := range s.muxers { @@ -32,8 +38,36 @@ func (s *testHLSServer) APIMuxersGet(name string) (*defs.APIHLSMuxer, error) { return muxer, nil } +func (s *testHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + items := make([]defs.APIHLSSession, 0, len(s.sessions)) + for _, session := range s.sessions { + items = append(items, *session) + } + sort.Slice(items, func(i, j int) bool { + return items[i].Created.Before(items[j].Created) + }) + return &defs.APIHLSSessionList{Items: items}, nil +} + +func (s *testHLSServer) APISessionsGet(id uuid.UUID) (*defs.APIHLSSession, error) { + session, ok := s.sessions[id.String()] + if !ok { + return nil, hls.ErrSessionNotFound + } + return session, nil +} + +func (s *testHLSServer) APISessionsKick(id uuid.UUID) error { + _, ok := s.sessions[id.String()] + if !ok { + return hls.ErrSessionNotFound + } + delete(s.sessions, id.String()) + return nil +} + func TestHLSMuxersList(t *testing.T) { - now := time.Now() + now := testTime hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "test1": { @@ -80,7 +114,7 @@ func TestHLSMuxersList(t *testing.T) { } func TestHLSMuxersGet(t *testing.T) { - now := time.Now() + now := testTime hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "mypath": { @@ -118,3 +152,185 @@ func TestHLSMuxersGet(t *testing.T) { require.Equal(t, uint64(12), out.OutboundFramesDiscarded) require.Equal(t, uint64(9999), out.BytesSent) } + +func TestHLSSessionsList(t *testing.T) { + now := testTime + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + "session1": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.1:5000", + Path: "stream1", + Query: "key=val1", + User: "user1", + OutboundBytes: 111, + }, + "session2": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb42"), + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + Path: "stream2", + Query: "key=val2", + User: "user2", + OutboundBytes: 222, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + var out defs.APIHLSSessionList + httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/hlssessions/list", nil, &out) + + require.Equal(t, 2, out.ItemCount) + require.Equal(t, 1, out.PageCount) + require.Len(t, out.Items, 2) + require.Equal(t, []defs.APIHLSSession{ + { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.1:5000", + Path: "stream1", + Query: "key=val1", + User: "user1", + OutboundBytes: 111, + }, + { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb42"), + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + Path: "stream2", + Query: "key=val2", + User: "user2", + OutboundBytes: 222, + }, + }, out.Items) +} + +func TestHLSSessionsGet(t *testing.T) { + now := testTime + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + "18294761-f9d1-4ea9-9a35-fe265b62eb41": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.100:5000", + Path: "mystream", + Query: "key=val", + User: "myuser", + OutboundBytes: 345, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + sessionID := "18294761-f9d1-4ea9-9a35-fe265b62eb41" + + var out defs.APIHLSSession + httpRequest(t, hc, http.MethodGet, fmt.Sprintf("http://localhost:9997/v3/hlssessions/get/%s", sessionID), nil, &out) + + require.Equal(t, uuid.MustParse(sessionID), out.ID) + require.Equal(t, "192.168.1.100:5000", out.RemoteAddr) + require.Equal(t, "mystream", out.Path) + require.Equal(t, "key=val", out.Query) + require.Equal(t, "myuser", out.User) + require.Equal(t, uint64(345), out.OutboundBytes) +} + +func TestHLSSessionsKick(t *testing.T) { + now := testTime + sessionID := uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41") + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + sessionID.String(): { + ID: sessionID, + Created: now, + RemoteAddr: "192.168.1.100:5000", + Path: "mystream", + OutboundBytes: 345, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + httpRequest(t, hc, http.MethodPost, fmt.Sprintf("http://localhost:9997/v3/hlssessions/kick/%s", sessionID), nil, nil) + + _, ok := hlsServer.sessions[sessionID.String()] + require.False(t, ok) +} + +func TestHLSSessionsKickNotFound(t *testing.T) { + hlsServer := &testHLSServer{} + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + req, err := http.NewRequest(http.MethodPost, + fmt.Sprintf("http://localhost:9997/v3/hlssessions/kick/%s", uuid.New()), nil) + require.NoError(t, err) + + res, err := hc.Do(req) + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusNotFound, res.StatusCode) + checkError(t, res.Body, "session not found") +} diff --git a/internal/api/api_paths_test.go b/internal/api/api_paths_test.go index a5d4ee74..e27dc985 100644 --- a/internal/api/api_paths_test.go +++ b/internal/api/api_paths_test.go @@ -108,7 +108,7 @@ func TestPathsGet(t *testing.T) { BytesReceived: 123456, BytesSent: 789012, Readers: []defs.APIPathReader{ - {Type: defs.APIPathReaderTypeHLSMuxer, ID: "muxer1"}, + {Type: defs.APIPathReaderTypeHLSSession, ID: "session6123"}, {Type: defs.APIPathReaderTypeWebRTCSession, ID: "session456"}, }, }, diff --git a/internal/auth/error.go b/internal/auth/error.go index 37922bdb..ce7bafd7 100644 --- a/internal/auth/error.go +++ b/internal/auth/error.go @@ -7,6 +7,6 @@ type Error struct { } // Error implements the error interface. -func (e Error) Error() string { +func (e *Error) Error() string { return "authentication failed: " + e.Wrapped.Error() } diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 3db5ff08..998a12ed 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -14,10 +14,13 @@ import ( "testing" "time" + "github.com/bluenviron/gohlslib/v2" "github.com/bluenviron/gortmplib" rtmpcodecs "github.com/bluenviron/gortmplib/pkg/codecs" "github.com/bluenviron/gortsplib/v5" "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/bluenviron/gortsplib/v5/pkg/format/rtph264" "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts" tscodecs "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts/codecs" srt "github.com/datarhei/gosrt" @@ -369,7 +372,8 @@ func TestAPIProtocolListGet(t *testing.T) { "rtsps sessions", "rtmp", "rtmps", - "hls", + "hls sessions", + "hls muxers", "webrtc", "srt", } { @@ -470,7 +474,7 @@ func TestAPIProtocolListGet(t *testing.T) { time.Sleep(500 * time.Millisecond) - case "hls": + case "hls sessions", "hls muxers": source := gortsplib.Client{} err = source.StartRecording("rtsp://localhost:8554/mypath", &description.Session{Medias: []*description.Media{medi}}) @@ -612,7 +616,10 @@ func TestAPIProtocolListGet(t *testing.T) { case "rtmps": pa = "rtmpsconns" - case "hls": + case "hls sessions": + pa = "hlssessions" + + case "hls muxers": pa = "hlsmuxers" case "webrtc": @@ -792,7 +799,24 @@ func TestAPIProtocolListGet(t *testing.T) { }, }, out1) - case "hls": + case "hls sessions": + require.Equal(t, map[string]any{ + "itemCount": float64(1), + "pageCount": float64(1), + "items": []any{ + map[string]any{ + "id": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["id"], + "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], + "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], + "path": "mypath", + "query": "", + "user": "", + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + }, + }, + }, out1) + + case "hls muxers": require.Equal(t, map[string]any{ "itemCount": float64(1), "pageCount": float64(1), @@ -919,7 +943,7 @@ func TestAPIProtocolListGet(t *testing.T) { var out2 any - if ca == "hls" { + if ca == "hls muxers" { httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/"+pa+"/get/"+ out1.(map[string]any)["items"].([]any)[0].(map[string]any)["path"].(string), nil, &out2) @@ -979,7 +1003,7 @@ func TestAPIProtocolListGet(t *testing.T) { out2.(map[string]any)["rtcpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsReceived"] out2.(map[string]any)["rtcpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsSent"] - case "hls": + case "hls muxers": out2.(map[string]any)["lastRequest"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["lastRequest"] } @@ -1004,7 +1028,8 @@ func TestAPIProtocolGetNotFound(t *testing.T) { "rtsps sessions", "rtmp", "rtmps", - "hls", + "hls sessions", + "hls muxers", "webrtc", "srt", } { @@ -1055,7 +1080,10 @@ func TestAPIProtocolGetNotFound(t *testing.T) { case "rtmps": pa = "rtmpsconns" - case "hls": + case "hls sessions": + pa = "hlssessions" + + case "hls muxers": pa = "hlsmuxers" case "webrtc": @@ -1081,10 +1109,10 @@ func TestAPIProtocolGetNotFound(t *testing.T) { case "rtsp conns", "rtsps conns", "rtmp", "rtmps", "srt": checkError(t, "connection not found", res.Body) - case "rtsp sessions", "rtsps sessions", "webrtc": + case "rtsp sessions", "rtsps sessions", "hls sessions", "webrtc": checkError(t, "session not found", res.Body) - case "hls": + case "hls muxers": checkError(t, "muxer not found", res.Body) } }() @@ -1105,6 +1133,7 @@ func TestAPIProtocolKick(t *testing.T) { "rtsp", "rtsps", "rtmp", + "hls", "webrtc", "srt", } { @@ -1134,7 +1163,6 @@ func TestAPIProtocolKick(t *testing.T) { switch ca { case "rtsp": source := gortsplib.Client{} - err = source.StartRecording("rtsp://localhost:8554/mypath", &description.Session{Medias: []*description.Media{medi}}) require.NoError(t, err) @@ -1144,7 +1172,6 @@ func TestAPIProtocolKick(t *testing.T) { source := gortsplib.Client{ TLSConfig: &tls.Config{InsecureSkipVerify: true}, } - err = source.StartRecording("rtsps://localhost:8322/mypath", &description.Session{Medias: []*description.Media{medi}}) require.NoError(t, err) @@ -1180,6 +1207,45 @@ func TestAPIProtocolKick(t *testing.T) { err = w.WriteH264(track, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) + case "hls": + source := gortsplib.Client{} + err = source.StartRecording("rtsp://localhost:8554/mypath", + &description.Session{Medias: []*description.Media{medi}}) + require.NoError(t, err) + defer source.Close() + + var enc *rtph264.Encoder + enc, err = medi.Formats[0].(*format.H264).CreateEncoder() + require.NoError(t, err) + + tracksReceived := make(chan struct{}) + + client := &gohlslib.Client{ + URI: "http://localhost:8888/mypath/index.m3u8", + OnTracks: func(_ []*gohlslib.Track) error { + close(tracksReceived) + return nil + }, + } + err = client.Start() + require.NoError(t, err) + defer client.Close() + + time.Sleep(500 * time.Millisecond) + + for i := range 2 { + var pkts []*rtp.Packet + pkts, err = enc.Encode([][]byte{{5, 2, 3, 4}}) + require.NoError(t, err) + + pkts[0].Timestamp = uint32(i * 90000) + + err = source.WritePacketRTP(medi, pkts[0]) + require.NoError(t, err) + } + + <-tracksReceived + case "webrtc": var u *url.URL u, err = url.Parse("http://localhost:8889/mypath/whip") @@ -1243,6 +1309,9 @@ func TestAPIProtocolKick(t *testing.T) { case "rtmp": pa = "rtmpconns" + case "hls": + pa = "hlssessions" + case "webrtc": pa = "webrtcsessions" @@ -1256,6 +1325,7 @@ func TestAPIProtocolKick(t *testing.T) { } `json:"items"` } httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/"+pa+"/list", nil, &out1) + require.NotEmpty(t, out1.Items) httpRequest(t, hc, http.MethodPost, "http://localhost:9997/v3/"+pa+"/kick/"+out1.Items[0].ID, nil, nil) @@ -1283,6 +1353,7 @@ func TestAPIProtocolKickNotFound(t *testing.T) { "rtsp", "rtsps", "rtmp", + "hls", "webrtc", "srt", } { @@ -1318,6 +1389,9 @@ func TestAPIProtocolKickNotFound(t *testing.T) { case "rtmp": pa = "rtmpconns" + case "hls": + pa = "hlssessions" + case "webrtc": pa = "webrtcsessions" @@ -1341,11 +1415,8 @@ func TestAPIProtocolKickNotFound(t *testing.T) { case "rtsp conns", "rtsps conns", "rtmp", "rtmps", "srt": checkError(t, "connection not found", res.Body) - case "rtsp sessions", "rtsps sessions", "webrtc": + case "rtsp sessions", "rtsps sessions", "hls", "webrtc": checkError(t, "session not found", res.Body) - - case "hls": - checkError(t, "muxer not found", res.Body) } }() }) diff --git a/internal/core/core.go b/internal/core/core.go index d4810d6a..bfc8f837 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -609,6 +609,7 @@ func (p *Core) createResources(initial bool) error { ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, MuxerCloseAfter: p.conf.HLSMuxerCloseAfter, + ExternalCmdPool: p.externalCmdPool, Metrics: p.metrics, PathManager: p.pathManager, Parent: p, diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index ee15af82..1c49565e 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -86,6 +86,10 @@ paths_bytes_received 0 paths_bytes_sent 0 paths_readers 0 +# HLS sessions +hls_sessions 0 +hls_sessions_outbound_bytes 0 + # HLS muxers hls_muxers 0 hls_muxers_outbound_bytes 0 diff --git a/internal/core/path.go b/internal/core/path.go index 8d9ae7e9..3a17d424 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -506,7 +506,7 @@ func (pa *path) doDescribe(req defs.PathDescribeReq) { return } - req.Res <- defs.PathDescribeRes{Err: defs.PathNoStreamAvailableError{PathName: pa.name}} + req.Res <- defs.PathDescribeRes{Err: &defs.PathNoStreamAvailableError{PathName: pa.name}} } func (pa *path) doRemovePublisher(req defs.PathRemovePublisherReq) { @@ -598,7 +598,7 @@ func (pa *path) doAddReader(req defs.PathAddReaderReq) { return } - req.Res <- defs.PathAddReaderRes{Err: defs.PathNoStreamAvailableError{PathName: pa.name}} + req.Res <- defs.PathAddReaderRes{Err: &defs.PathNoStreamAvailableError{PathName: pa.name}} } func (pa *path) doRemoveReader(req defs.PathRemoveReaderReq) { @@ -699,11 +699,13 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { return pa.stream.OutboundBytes() }(), Readers: func() []defs.APIPathReader { - ret := make([]defs.APIPathReader, len(pa.readers)) - i := 0 + ret := make([]defs.APIPathReader, 0, len(pa.readers)) + for r := range pa.readers { - ret[i] = *r.APIReaderDescribe() - i++ + desc := *r.APIReaderDescribe() + if desc.Type != defs.APIPathReaderTypeHidden { + ret = append(ret, desc) + } } sort.Slice(ret, func(i, j int) bool { diff --git a/internal/defs/api_hls.go b/internal/defs/api_hls.go index be12489a..241be1a0 100644 --- a/internal/defs/api_hls.go +++ b/internal/defs/api_hls.go @@ -1,13 +1,45 @@ package defs -import "time" +import ( + "time" + + "github.com/google/uuid" +) // APIHLSServer contains methods used by the API and Metrics server. type APIHLSServer interface { + APISessionsList() (*APIHLSSessionList, error) + APISessionsGet(uuid.UUID) (*APIHLSSession, error) + APISessionsKick(uuid.UUID) error APIMuxersList() (*APIHLSMuxerList, error) APIMuxersGet(string) (*APIHLSMuxer, error) } +// APIHLSSessionList is a list of HLS sessions. +type APIHLSSessionList struct { + ItemCount int `json:"itemCount"` + PageCount int `json:"pageCount"` + Items []APIHLSSession `json:"items"` +} + +// APIHLSSession is an HLS session. +type APIHLSSession struct { + ID uuid.UUID `json:"id"` + Created time.Time `json:"created"` + RemoteAddr string `json:"remoteAddr"` + Path string `json:"path"` + Query string `json:"query"` + User string `json:"user"` + OutboundBytes uint64 `json:"outboundBytes"` +} + +// APIHLSMuxerList is a list of HLS muxers. +type APIHLSMuxerList struct { + ItemCount int `json:"itemCount"` + PageCount int `json:"pageCount"` + Items []APIHLSMuxer `json:"items"` +} + // APIHLSMuxer is an HLS muxer. type APIHLSMuxer struct { Path string `json:"path"` @@ -18,10 +50,3 @@ type APIHLSMuxer struct { // deprecated BytesSent uint64 `json:"bytesSent" deprecated:"true"` } - -// APIHLSMuxerList is a list of HLS muxers. -type APIHLSMuxerList struct { - ItemCount int `json:"itemCount"` - PageCount int `json:"pageCount"` - Items []APIHLSMuxer `json:"items"` -} diff --git a/internal/defs/api_path.go b/internal/defs/api_path.go index 830766c1..c3e0ccb0 100644 --- a/internal/defs/api_path.go +++ b/internal/defs/api_path.go @@ -43,16 +43,16 @@ type APIPathReaderType string // reader types. const ( - APIPathReaderTypeHLSMuxer APIPathReaderType = "hlsMuxer" - APIPathReaderTypeRTMPConn APIPathReaderType = "rtmpConn" - APIPathReaderTypeRTMPSConn APIPathReaderType = "rtmpsConn" - APIPathReaderTypeRTSPConn APIPathReaderType = "rtspConn" - APIPathReaderTypeRPICameraSecondary APIPathReaderType = "rpiCameraSecondary" - APIPathReaderTypeRTSPSession APIPathReaderType = "rtspSession" - APIPathReaderTypeRTSPSConn APIPathReaderType = "rtspsConn" - APIPathReaderTypeRTSPSSession APIPathReaderType = "rtspsSession" - APIPathReaderTypeSRTConn APIPathReaderType = "srtConn" - APIPathReaderTypeWebRTCSession APIPathReaderType = "webRTCSession" + APIPathReaderTypeHLSSession APIPathReaderType = "hlsSession" + APIPathReaderTypeRTMPConn APIPathReaderType = "rtmpConn" + APIPathReaderTypeRTMPSConn APIPathReaderType = "rtmpsConn" + APIPathReaderTypeRTSPConn APIPathReaderType = "rtspConn" + APIPathReaderTypeRTSPSession APIPathReaderType = "rtspSession" + APIPathReaderTypeRTSPSConn APIPathReaderType = "rtspsConn" + APIPathReaderTypeRTSPSSession APIPathReaderType = "rtspsSession" + APIPathReaderTypeSRTConn APIPathReaderType = "srtConn" + APIPathReaderTypeWebRTCSession APIPathReaderType = "webRTCSession" + APIPathReaderTypeHidden APIPathReaderType = "hidden" ) // APIPathReader is a reader. diff --git a/internal/defs/path.go b/internal/defs/path.go index 605a8ee0..343cdd46 100644 --- a/internal/defs/path.go +++ b/internal/defs/path.go @@ -16,7 +16,7 @@ type PathNoStreamAvailableError struct { } // Error implements the error interface. -func (e PathNoStreamAvailableError) Error() string { +func (e *PathNoStreamAvailableError) Error() string { return fmt.Sprintf("no stream is available on path '%s'", e.PathName) } diff --git a/internal/linters/go2api/go2api_test.go b/internal/linters/go2api/go2api_test.go index 7e8ee180..e6b161ec 100644 --- a/internal/linters/go2api/go2api_test.go +++ b/internal/linters/go2api/go2api_test.go @@ -175,8 +175,7 @@ func goEnumToApi(rt reflect.Type) (openAPISchema, bool) { case reflect.TypeOf(defs.APIPathReaderType("")): return openAPISchema{Type: "string", Enum: []string{ - "hlsMuxer", - "rpiCameraSecondary", + "hlsSession", "rtmpConn", "rtmpsConn", "rtspConn", @@ -352,6 +351,14 @@ func TestGo2API(t *testing.T) { "HLSMuxerList", defs.APIHLSMuxerList{}, }, + { + "HLSSession", + defs.APIHLSSession{}, + }, + { + "HLSSessionList", + defs.APIHLSSessionList{}, + }, { "Info", defs.APIInfo{}, diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 214f85da..ac63cde5 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -74,6 +74,7 @@ type metricsType string const ( metricsTypePaths metricsType = "paths" + metricsTypeHLSSessions metricsType = "hls_sessions" metricsTypeHLSMuxers metricsType = "hls_muxers" metricsTypeRTSPConns metricsType = "rtsp_conns" metricsTypeRTSPSessions metricsType = "rtsp_sessions" @@ -227,6 +228,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { typ := metricsType(ctx.Query("type")) pathFilter := ctx.Query("path") hlsMuxerFilter := ctx.Query("hls_muxer") + hlsSessionFilter := ctx.Query("hls_session") rtspConnFilter := ctx.Query("rtsp_conn") rtspSessionFilter := ctx.Query("rtsp_session") rtspsConnFilter := ctx.Query("rtsps_conn") @@ -238,6 +240,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { anyFilterActive := pathFilter != "" || hlsMuxerFilter != "" || + hlsSessionFilter != "" || rtspConnFilter != "" || rtspSessionFilter != "" || rtspsConnFilter != "" || @@ -336,6 +339,32 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(hlsServer) { + if (typ == "" || typ == metricsTypeHLSSessions) && (!anyFilterActive || hlsSessionFilter != "") { + var data *defs.APIHLSSessionList + data, err := hlsServer.APISessionsList() + if err == nil && len(data.Items) != 0 { + out.WriteString("# HLS sessions\n") + for _, i := range data.Items { + if hlsSessionFilter == "" || hlsSessionFilter == i.ID.String() { + ta := tags(map[string]string{ + "id": i.ID.String(), + "path": i.Path, + "remoteAddr": i.RemoteAddr, + }) + + metric(&out, "hls_sessions", ta, 1) + metric(&out, "hls_sessions_outbound_bytes", ta, int64(i.OutboundBytes)) + } + } + out.WriteString("\n") + } else if hlsSessionFilter == "" { + out.WriteString("# HLS sessions\n") + metric(&out, "hls_sessions", "", 0) + metric(&out, "hls_sessions_outbound_bytes", "", 0) + out.WriteString("\n") + } + } + if (typ == "" || typ == metricsTypeHLSMuxers) && (!anyFilterActive || hlsMuxerFilter != "") { var data *defs.APIHLSMuxerList data, err := hlsServer.APIMuxersList() diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 376f6640..92ce23eb 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -89,6 +89,28 @@ func (dummyHLSServer) APIMuxersGet(string) (*defs.APIHLSMuxer, error) { panic("unused") } +func (dummyHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + return &defs.APIHLSSessionList{ + ItemCount: 1, + PageCount: 1, + Items: []defs.APIHLSSession{{ + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), + RemoteAddr: "124.5.5.5:34542", + Path: "mypath", + OutboundBytes: 187, + }}, + }, nil +} + +func (dummyHLSServer) APISessionsGet(uuid.UUID) (*defs.APIHLSSession, error) { + panic("unused") +} + +func (dummyHLSServer) APISessionsKick(uuid.UUID) error { + panic("unused") +} + type dummyRTSPServer struct{} func (dummyRTSPServer) APIConnsList() (*defs.APIRTSPConnsList, error) { @@ -332,6 +354,18 @@ func (emptyHLSServer) APIMuxersGet(string) (*defs.APIHLSMuxer, error) { panic("unused") } +func (emptyHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + return &defs.APIHLSSessionList{}, nil +} + +func (emptyHLSServer) APISessionsGet(uuid.UUID) (*defs.APIHLSSession, error) { + panic("unused") +} + +func (emptyHLSServer) APISessionsKick(uuid.UUID) error { + panic("unused") +} + type emptyRTSPServer struct{} func (emptyRTSPServer) APIConnsList() (*defs.APIRTSPConnsList, error) { @@ -478,6 +512,12 @@ func TestMetrics(t *testing.T) { "paths_bytes_received{name=\"mypath\",state=\"ready\"} 123\n"+ "paths_bytes_sent{name=\"mypath\",state=\"ready\"} 456\n"+ "\n"+ + "# HLS sessions\n"+ + "hls_sessions{id=\"18294761-f9d1-4ea9-9a35-fe265b62eb41\",path=\"mypath\","+ + "remoteAddr=\"124.5.5.5:34542\"} 1\n"+ + "hls_sessions_outbound_bytes{id=\"18294761-f9d1-4ea9-9a35-fe265b62eb41\",path=\"mypath\","+ + "remoteAddr=\"124.5.5.5:34542\"} 187\n"+ + "\n"+ "# HLS muxers\n"+ "hls_muxers{name=\"mypath\"} 1\n"+ "hls_muxers_outbound_bytes{name=\"mypath\"} 789\n"+ @@ -835,6 +875,10 @@ func TestZeroMetricsFallback(t *testing.T) { "paths_bytes_sent 0\n"+ "paths_readers 0\n"+ "\n"+ + "# HLS sessions\n"+ + "hls_sessions 0\n"+ + `hls_sessions_outbound_bytes 0`+"\n"+ + "\n"+ "# HLS muxers\n"+ "hls_muxers 0\n"+ "hls_muxers_outbound_bytes 0\n"+ @@ -965,6 +1009,7 @@ func TestFilter(t *testing.T) { for _, ca := range []string{ "path", "hls_muxer", + "hls_session", "rtsp_conn", "rtsp_session", "rtsps_conn", @@ -1007,6 +1052,8 @@ func TestFilter(t *testing.T) { u += "?path=mypath" case "hls_muxer": u += "?hls_muxer=mypath" + case "hls_session": + u += "?hls_session=18294761-f9d1-4ea9-9a35-fe265b62eb41" case "rtsp_conn": u += "?rtsp_conn=18294761-f9d1-4ea9-9a35-fe265b62eb41" case "rtsp_session": @@ -1059,6 +1106,16 @@ func TestFilter(t *testing.T) { `hls_muxers_bytes_sent{name="mypath"} 789`+"\n\n", string(byts)) + case "hls_session": + require.Equal(t, + "# HLS sessions\n"+ + `hls_sessions{id="18294761-f9d1-4ea9-9a35-fe265b62eb41",path="mypath",`+ + `remoteAddr="124.5.5.5:34542"} 1`+"\n"+ + `hls_sessions_outbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41",path="mypath",`+ + `remoteAddr="124.5.5.5:34542"} 187`+"\n"+ + "\n", + string(byts)) + case "rtsp_conn": require.Equal(t, "# RTSP connections\n"+ diff --git a/internal/protocols/httpp/handler_origin.go b/internal/protocols/httpp/handler_origin.go index f7f6a8e1..1f7bbc1c 100644 --- a/internal/protocols/httpp/handler_origin.go +++ b/internal/protocols/httpp/handler_origin.go @@ -15,7 +15,10 @@ func isOriginAllowed(origin string, allowOrigins []string) (string, bool) { for _, o := range allowOrigins { if o == "*" { - return o, true + if origin != "" { + return origin, true + } + return "*", true } } diff --git a/internal/protocols/httpp/handler_origin_test.go b/internal/protocols/httpp/handler_origin_test.go index 1f07878d..4e6379f0 100644 --- a/internal/protocols/httpp/handler_origin_test.go +++ b/internal/protocols/httpp/handler_origin_test.go @@ -38,7 +38,7 @@ func TestHandlerOrigin(t *testing.T) { "everything allowed, with origin", "https://example.com", []string{"*"}, - "*", + "https://example.com", }, { "allowed", diff --git a/internal/recorder/recorder_instance.go b/internal/recorder/recorder_instance.go index 2c44ec70..d2ffea2b 100644 --- a/internal/recorder/recorder_instance.go +++ b/internal/recorder/recorder_instance.go @@ -47,8 +47,8 @@ func (ri *recorderInstance) initialize() { ri.format, ) ri.reader = &stream.Reader{ - SkipBytesSent: true, - Parent: ri, + SkipOutboundBytes: true, + Parent: ri, } ri.terminate = make(chan struct{}) diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 546ca224..a2ab17c1 100644 --- a/internal/servers/hls/http_server.go +++ b/internal/servers/hls/http_server.go @@ -35,6 +35,12 @@ func mergePathAndQuery(path string, rawQuery string) string { return res } +func isIOS(userAgent string) bool { + return strings.Contains(userAgent, "iPad") || + strings.Contains(userAgent, "iPhone") || + strings.Contains(userAgent, "iPod") +} + type httpServer struct { address string dumpPackets bool @@ -124,6 +130,17 @@ func (s *httpServer) onRequest(ctx *gin.Context) { var dir string var fname string + type contentType int + + const ( + index contentType = iota + multivariantPlaylist + mediaPlaylist + segment + ) + + var contentTyp contentType + switch { case strings.HasSuffix(pa, "/hls.min.js"): ctx.Header("Cache-Control", "max-age=3600") @@ -135,8 +152,16 @@ func (s *httpServer) onRequest(ctx *gin.Context) { case pa == "", pa == "favicon.ico", strings.HasSuffix(pa, "/hls.min.js.map"): return - case strings.HasSuffix(pa, ".m3u8") || - strings.HasSuffix(pa, ".ts") || + case strings.HasSuffix(pa, ".m3u8"): + dir, fname = gopath.Dir(pa), gopath.Base(pa) + + if fname == "index.m3u8" { + contentTyp = multivariantPlaylist + } else { + contentTyp = mediaPlaylist + } + + case strings.HasSuffix(pa, ".ts") || strings.HasSuffix(pa, ".mp4") || strings.HasSuffix(pa, ".mp"): dir, fname = gopath.Dir(pa), gopath.Base(pa) @@ -145,42 +170,169 @@ func (s *httpServer) onRequest(ctx *gin.Context) { fname += "4" } + contentTyp = segment + default: - dir, fname = pa, "" + dir = pa if !strings.HasSuffix(dir, "/") { ctx.Header("Location", mergePathAndQuery(ctx.Request.URL.Path+"/", ctx.Request.URL.RawQuery)) ctx.Writer.WriteHeader(http.StatusMovedPermanently) return } + + dir = dir[:len(dir)-1] + contentTyp = index } - dir = strings.TrimSuffix(dir, "/") - if dir == "" { - return - } + switch contentTyp { + case index: + _, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{ + AccessRequest: defs.PathAccessRequest{ + Name: dir, + Query: ctx.Request.URL.RawQuery, + Publish: false, + Proto: auth.ProtocolHLS, + Credentials: httpp.Credentials(ctx.Request), + IP: net.ParseIP(ctx.ClientIP()), + }, + }) + if err != nil { + var terr *auth.Error + if errors.As(err, &terr) { + if terr.AskCredentials { + ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) - res, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{ - AccessRequest: defs.PathAccessRequest{ - Name: dir, - Query: ctx.Request.URL.RawQuery, - Publish: false, - Proto: auth.ProtocolHLS, - Credentials: httpp.Credentials(ctx.Request), - IP: net.ParseIP(ctx.ClientIP()), - }, - }) - if err != nil { - var terr *auth.Error - if errors.As(err, &terr) { - if terr.AskCredentials { - ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) return } - s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + ctx.Header("Cache-Control", "max-age=3600") + ctx.Header("Content-Type", "text/html") + ctx.Writer.WriteHeader(http.StatusOK) + ctx.Writer.Write(hlsIndex) + + case multivariantPlaylist: + if ctx.Request.URL.Query().Get("cookieCheck") != "1" { + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: "cookieCheck", + Value: "1", + }) + + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: "cookieCheck", + Value: "1", + SameSite: http.SameSiteNoneMode, + Secure: true, + Partitioned: true, + HttpOnly: true, + }) + + q := ctx.Request.URL.Query() + q.Set("cookieCheck", "1") + ctx.Request.URL.RawQuery = q.Encode() + ctx.Writer.Header().Set("Location", mergePathAndQuery(ctx.Request.URL.Path, ctx.Request.URL.RawQuery)) + + ctx.Writer.WriteHeader(http.StatusFound) + return + } + + if _, err := ctx.Request.Cookie("cookieCheck"); err != nil && isIOS(ctx.Request.UserAgent()) { + s.writeErrorNoLog(ctx, http.StatusBadRequest, fmt.Errorf("HLS on iOS requires the server to set and read cookies")) + return + } + + q := ctx.Request.URL.Query() + q.Del("cookieCheck") + ctx.Request.URL.RawQuery = q.Encode() + + sx := &session{ + remoteAddr: httpp.RemoteAddr(ctx), + pathName: dir, + externalCmdPool: s.parent.ExternalCmdPool, + pathManager: s.pathManager, + server: s.parent, + } + err := sx.initialize(ctx) + if err != nil { + var terr *auth.Error + if errors.As(err, &terr) { + if terr.AskCredentials { + ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) + + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + var terr2 *defs.PathNoStreamAvailableError + if errors.As(err, &terr2) { + s.writeErrorNoLog(ctx, http.StatusNotFound, err) + return + } + + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + if cookie, err2 := ctx.Request.Cookie("cookieCheck"); err2 == nil && cookie.Value == "1" { + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: sessionCookieName, + Value: sx.secret.String(), + }) + + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: sessionCookieName, + Value: sx.secret.String(), + SameSite: http.SameSiteNoneMode, + Secure: true, + Partitioned: true, + HttpOnly: true, + }) + } else { + q = ctx.Request.URL.Query() + q.Set(sessionQueryParamName, sx.secret.String()) + ctx.Request.URL.RawQuery = q.Encode() + } + + ctx.Writer = &responseWriterCounter{ + ResponseWriter: ctx.Writer, + bytesSent: &sx.bytesSent, + } + + ctx.Request.URL.Path = fname + + err = sx.muxer.handleRequest(ctx) + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + default: + muxer, err := s.parent.getMuxer(serverGetMuxerReq{ + path: dir, + create: false, + }) + if err != nil { // wait some seconds to delay brute force attacks <-time.After(auth.PauseAfterError) @@ -188,31 +340,26 @@ func (s *httpServer) onRequest(ctx *gin.Context) { return } - s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) - return - } + sx := muxer.findSession(ctx) + if sx == nil { + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) - switch fname { - case "": - ctx.Header("Cache-Control", "max-age=3600") - ctx.Header("Content-Type", "text/html") - ctx.Writer.WriteHeader(http.StatusOK) - ctx.Writer.Write(hlsIndex) - - default: - var mux *muxer - mux, err = s.parent.getMuxer(serverGetMuxerReq{ - path: dir, - remoteAddr: httpp.RemoteAddr(ctx), - query: ctx.Request.URL.RawQuery, - sourceOnDemand: res.Conf.SourceOnDemand, - }) - if err != nil { - ctx.Writer.WriteHeader(http.StatusNotFound) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) return } + ctx.Writer = &responseWriterCounter{ + ResponseWriter: ctx.Writer, + bytesSent: &sx.bytesSent, + } + ctx.Request.URL.Path = fname - mux.handleRequest(ctx) + + err = muxer.handleRequest(ctx) + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } } } diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 5c5b2e90..cb85f2b0 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -4,22 +4,22 @@ import ( "context" "errors" "fmt" - "net/http" "sync" "sync/atomic" "time" + "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" "github.com/gin-gonic/gin" + "github.com/google/uuid" ) const ( - closeCheckPeriod = 1 * time.Second - recreatePause = 10 * time.Second + recreateInstancePause = 10 * time.Second ) func emptyTimer() *time.Timer { @@ -28,20 +28,9 @@ func emptyTimer() *time.Timer { return t } -type responseWriterWithCounter struct { - http.ResponseWriter - bytesSent *atomic.Uint64 -} - -func (w *responseWriterWithCounter) Write(p []byte) (int, error) { - n, err := w.ResponseWriter.Write(p) - w.bytesSent.Add(uint64(n)) - return n, err -} - -type muxerGetInstanceRes struct { - instance *muxerInstance - cumulatedOutboundFramesDiscarded uint64 +type muxerCloseInstanceReq struct { + instance *muxerInstance + err error } type muxer struct { @@ -67,9 +56,12 @@ type muxer struct { lastRequestTime atomic.Int64 bytesSent atomic.Uint64 - instanceMutex sync.RWMutex + mutex sync.RWMutex instance *muxerInstance cumulatedOutboundFramesDiscarded uint64 + sessionsBySecret map[uuid.UUID]*session + + chCloseInstance chan muxerCloseInstanceReq } func (m *muxer) initialize() { @@ -79,7 +71,8 @@ func (m *muxer) initialize() { m.ctxCancel = ctxCancel m.created = time.Now() m.lastRequestTime.Store(time.Now().UnixNano()) - m.bytesSent.Store(0) + m.sessionsBySecret = make(map[uuid.UUID]*session) + m.chCloseInstance = make(chan muxerCloseInstanceReq) m.Log(logger.Info, "created %s", func() string { if m.remoteAddr == "" { @@ -89,7 +82,7 @@ func (m *muxer) initialize() { }()) // block first request to getInstance() until the first instance is available - m.instanceMutex.Lock() + m.mutex.Lock() m.wg.Add(1) go m.run() @@ -116,9 +109,23 @@ func (m *muxer) run() { m.ctxCancel() - m.parent.closeMuxer(m) + if m.instance != nil { + m.instance.close() + } + + m.mutex.Lock() + + m.instance = nil + + for _, sx := range m.sessionsBySecret { + sx.close2(fmt.Errorf("muxer destroyed")) + } + + m.mutex.Unlock() m.Log(logger.Info, "destroyed: %v", err) + + m.parent.closeMuxer(m) } func (m *muxer) runInner() error { @@ -131,7 +138,7 @@ func (m *muxer) runInner() error { }, }) if err != nil { - m.instanceMutex.Unlock() + m.mutex.Unlock() return err } @@ -142,71 +149,101 @@ func (m *muxer) runInner() error { tmp, err := m.createInstance(res.Stream) if err != nil { if m.remoteAddr != "" || errors.Is(err, hls.ErrNoSupportedCodecs) { - m.instanceMutex.Unlock() + m.mutex.Unlock() return err } - m.Log(logger.Error, err.Error()) + m.Log(logger.Error, "muxer instance crashed: %v", err) } m.instance = tmp - m.instanceMutex.Unlock() + m.mutex.Unlock() - defer func() { - if m.instance != nil { - m.closeInstance() - } - }() - - var instanceError chan error - var recreateTimer *time.Timer + var recreateInstanceTimer *time.Timer if m.instance != nil { - instanceError = m.instance.errorChan() - recreateTimer = emptyTimer() + recreateInstanceTimer = emptyTimer() } else { - instanceError = make(chan error) - recreateTimer = time.NewTimer(recreatePause) + recreateInstanceTimer = time.NewTimer(recreateInstancePause) } + defer func() { + recreateInstanceTimer.Stop() + }() + + sessionCleanupTicker := time.NewTicker(sessionCleanupPeriod) + defer sessionCleanupTicker.Stop() + var activityCheckTimer *time.Timer if m.remoteAddr != "" { - activityCheckTimer = time.NewTimer(closeCheckPeriod) + activityCheckTimer = time.NewTimer(max(time.Duration(m.closeAfter)/3, 1*time.Second)) } else { activityCheckTimer = emptyTimer() } + defer func() { + activityCheckTimer.Stop() + }() + for { select { - case err = <-instanceError: - if m.remoteAddr != "" { - return err + case req := <-m.chCloseInstance: + if m.instance != req.instance { + continue } - m.Log(logger.Error, err.Error()) - m.closeInstance() - instanceError = make(chan error) - recreateTimer = time.NewTimer(recreatePause) + m.mutex.Lock() + m.cumulatedOutboundFramesDiscarded += m.instance.reader.OutboundFramesDiscarded() + m.instance = nil + m.mutex.Unlock() - case <-recreateTimer.C: + if m.remoteAddr != "" { + return req.err + } else { + m.mutex.Lock() + for _, sx := range m.sessionsBySecret { + sx.close2(fmt.Errorf("muxer instance crashed")) + } + m.sessionsBySecret = make(map[uuid.UUID]*session) + m.mutex.Unlock() + + m.Log(logger.Error, "muxer instance crashed: %v", req.err) + } + + recreateInstanceTimer = time.NewTimer(recreateInstancePause) + + case <-recreateInstanceTimer.C: tmp, err = m.createInstance(res.Stream) if err != nil { - m.Log(logger.Error, err.Error()) - recreateTimer = time.NewTimer(recreatePause) - } else { - m.instanceMutex.Lock() - m.instance = tmp - m.instanceMutex.Unlock() - - instanceError = m.instance.errorChan() + m.Log(logger.Error, "muxer instance crashed: %v", err) + recreateInstanceTimer = time.NewTimer(recreateInstancePause) + continue } + m.mutex.Lock() + m.instance = tmp + m.mutex.Unlock() + + case <-sessionCleanupTicker.C: + now := time.Now() + + m.mutex.Lock() + for secret, sx := range m.sessionsBySecret { + lastRequest := time.Unix(0, sx.lastRequestTime.Load()) + + if now.Sub(lastRequest) >= sessionCloseAfter { + delete(m.sessionsBySecret, secret) + sx.close2(fmt.Errorf("inactive")) + } + } + m.mutex.Unlock() + case <-activityCheckTimer.C: t := time.Unix(0, m.lastRequestTime.Load()) if time.Since(t) >= time.Duration(m.closeAfter) { return fmt.Errorf("not used anymore") } - activityCheckTimer = time.NewTimer(closeCheckPeriod) + activityCheckTimer = time.NewTimer(max(time.Duration(m.closeAfter)/3, 1*time.Second)) case <-m.ctx.Done(): return errors.New("terminated") @@ -214,16 +251,6 @@ func (m *muxer) runInner() error { } } -func (m *muxer) closeInstance() { - m.instanceMutex.Lock() - m.cumulatedOutboundFramesDiscarded += m.instance.reader.OutboundFramesDiscarded() - var tmp *muxerInstance - tmp, m.instance = m.instance, nil - m.instanceMutex.Unlock() - - tmp.close() -} - func (m *muxer) createInstance(strm *stream.Stream) (*muxerInstance, error) { mi := &muxerInstance{ variant: m.variant, @@ -233,54 +260,107 @@ func (m *muxer) createInstance(strm *stream.Stream) (*muxerInstance, error) { segmentMaxSize: m.segmentMaxSize, directory: m.directory, pathName: m.pathName, + bytesSent: &m.bytesSent, + wg: m.wg, stream: strm, + server: m.parent, parent: m, } err := mi.initialize() - return mi, err + if err != nil { + return nil, err + } + return mi, nil } -func (m *muxer) getInstance() muxerGetInstanceRes { - m.instanceMutex.RLock() - defer m.instanceMutex.RUnlock() - - return muxerGetInstanceRes{ - instance: m.instance, - cumulatedOutboundFramesDiscarded: m.cumulatedOutboundFramesDiscarded, +func (m *muxer) closeInstance(mi *muxerInstance, err error) { + select { + case m.chCloseInstance <- muxerCloseInstanceReq{instance: mi, err: err}: + case <-m.ctx.Done(): } } // APIReaderDescribe implements reader. func (m *muxer) APIReaderDescribe() *defs.APIPathReader { return &defs.APIPathReader{ - Type: defs.APIPathReaderTypeHLSMuxer, + Type: defs.APIPathReaderTypeHidden, ID: "", } } -func (m *muxer) handleRequest(ctx *gin.Context) { +func (m *muxer) addSession(sx *session) ([]format.Format, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + select { + case <-m.ctx.Done(): + return nil, fmt.Errorf("terminated") + default: + } + + if m.instance == nil { + return nil, fmt.Errorf("muxer instance not available") + } + + m.sessionsBySecret[sx.secret] = sx + return m.instance.reader.Formats(), nil +} + +func (m *muxer) findSession(ctx *gin.Context) *session { + var rawSecret string + if cookie, err := ctx.Request.Cookie(sessionCookieName); err == nil { + rawSecret = cookie.Value + } else { + q := ctx.Request.URL.Query() + rawSecret = q.Get(sessionQueryParamName) + } + + secret, err := uuid.Parse(rawSecret) + if err != nil { + return nil + } + + m.mutex.RLock() + defer m.mutex.RUnlock() + + sx, ok := m.sessionsBySecret[secret] + if !ok { + return nil + } + + if ctx.ClientIP() != sx.ip { + return nil + } + + sx.lastRequestTime.Store(time.Now().UnixNano()) + + return sx +} + +func (m *muxer) handleRequest(ctx *gin.Context) error { m.lastRequestTime.Store(time.Now().UnixNano()) - res := m.getInstance() - if res.instance == nil { - ctx.Writer.WriteHeader(http.StatusNotFound) - return + m.mutex.RLock() + instance := m.instance + m.mutex.RUnlock() + + if instance == nil { + return fmt.Errorf("muxer instance not available") } - w := &responseWriterWithCounter{ - ResponseWriter: ctx.Writer, - bytesSent: &m.bytesSent, - } - - res.instance.handleRequest(w, ctx.Request) + instance.handleRequest(ctx) + return nil } func (m *muxer) apiItem() *defs.APIHLSMuxer { - res := m.getInstance() + m.mutex.RLock() + instance := m.instance + cumulatedOutboundFramesDiscarded := m.cumulatedOutboundFramesDiscarded + m.mutex.RUnlock() - outboundFramesDiscarded := res.cumulatedOutboundFramesDiscarded - if res.instance != nil { - outboundFramesDiscarded += res.instance.reader.OutboundFramesDiscarded() + outboundFramesDiscarded := cumulatedOutboundFramesDiscarded + if instance != nil { + outboundFramesDiscarded += instance.reader.OutboundFramesDiscarded() } return &defs.APIHLSMuxer{ @@ -292,3 +372,52 @@ func (m *muxer) apiItem() *defs.APIHLSMuxer { BytesSent: m.bytesSent.Load(), } } + +func (m *muxer) apiSessionsList() []defs.APIHLSSession { + m.mutex.RLock() + defer m.mutex.RUnlock() + + sessions := make([]defs.APIHLSSession, 0, len(m.sessionsBySecret)) + + for _, sx := range m.sessionsBySecret { + sessions = append(sessions, *sx.apiItem()) + } + + return sessions +} + +func (m *muxer) findSessionByUUID(uuid uuid.UUID) *session { + for _, sx := range m.sessionsBySecret { + if sx.uuid == uuid { + return sx + } + } + return nil +} + +func (m *muxer) apiSessionsGet(uuid uuid.UUID) (*defs.APIHLSSession, bool) { + m.mutex.RLock() + sx := m.findSessionByUUID(uuid) + if sx == nil { + m.mutex.RUnlock() + return nil, false + } + m.mutex.RUnlock() + + return sx.apiItem(), true +} + +func (m *muxer) apiSessionsKick(uuid uuid.UUID) bool { + m.mutex.Lock() + defer m.mutex.Unlock() + + sx := m.findSessionByUUID(uuid) + if sx == nil { + return false + } + + sx.close2(fmt.Errorf("kicked")) + delete(m.sessionsBySecret, sx.secret) + + return true +} diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 9af7d5c1..bf291ded 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -1,9 +1,12 @@ package hls import ( - "net/http" + "context" + "fmt" "os" "path/filepath" + "sync" + "sync/atomic" "time" "github.com/bluenviron/gohlslib/v2" @@ -12,8 +15,21 @@ import ( "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" + "github.com/gin-gonic/gin" ) +const ( + sessionCookieName = "hlsSession" + sessionQueryParamName = "session" + sessionCloseAfter = 30 * time.Second + sessionCleanupPeriod = sessionCloseAfter / 3 +) + +type instanceParent interface { + logger.Writer + closeInstance(*muxerInstance, error) +} + type muxerInstance struct { variant conf.HLSVariant segmentCount int @@ -22,14 +38,21 @@ type muxerInstance struct { segmentMaxSize conf.StringSize directory string pathName string + bytesSent *atomic.Uint64 + wg *sync.WaitGroup stream *stream.Stream - parent logger.Writer + server logger.Writer + parent instanceParent - hmuxer *gohlslib.Muxer - reader *stream.Reader + ctx context.Context + ctxCancel func() + hmuxer *gohlslib.Muxer + reader *stream.Reader } func (mi *muxerInstance) initialize() error { + mi.Log(logger.Debug, "instance created") + var muxerDirectory string if mi.directory != "" { muxerDirectory = filepath.Join(mi.directory, mi.pathName) @@ -49,8 +72,8 @@ func (mi *muxerInstance) initialize() error { } mi.reader = &stream.Reader{ - SkipBytesSent: true, - Parent: mi, + SkipOutboundBytes: true, + Parent: mi, } err := hls.FromStream(mi.stream.Desc, mi.reader, mi.hmuxer) @@ -68,6 +91,11 @@ func (mi *muxerInstance) initialize() error { mi.stream.AddReader(mi.reader) + mi.ctx, mi.ctxCancel = context.WithCancel(context.Background()) + + mi.wg.Add(1) + go mi.run() + return nil } @@ -77,17 +105,50 @@ func (mi *muxerInstance) Log(level logger.Level, format string, args ...any) { } func (mi *muxerInstance) close() { + mi.ctxCancel() +} + +func (mi *muxerInstance) run() { + defer mi.wg.Done() + + err := mi.runInner() + + mi.ctxCancel() + mi.stream.RemoveReader(mi.reader) + mi.hmuxer.Close() + if mi.hmuxer.Directory != "" { os.Remove(mi.hmuxer.Directory) } + + mi.Log(logger.Debug, "instance destroyed: %v", err) + + mi.parent.closeInstance(mi, err) } -func (mi *muxerInstance) errorChan() chan error { - return mi.reader.Error() +func (mi *muxerInstance) runInner() error { + for { + select { + case <-mi.ctx.Done(): + return fmt.Errorf("terminated") + + case err := <-mi.reader.Error(): + return err + } + } } -func (mi *muxerInstance) handleRequest(w http.ResponseWriter, r *http.Request) { - mi.hmuxer.Handle(w, r) +func (mi *muxerInstance) handleRequest(ctx *gin.Context) { + w := ctx.Writer + + w = &responseWriterNoCache{ResponseWriter: w} + + w = &responseWriterCounter{ + ResponseWriter: w, + bytesSent: mi.bytesSent, + } + + mi.hmuxer.Handle(w, ctx.Request) } diff --git a/internal/servers/hls/response_writer_counter.go b/internal/servers/hls/response_writer_counter.go new file mode 100644 index 00000000..485d2bb1 --- /dev/null +++ b/internal/servers/hls/response_writer_counter.go @@ -0,0 +1,18 @@ +package hls + +import ( + "sync/atomic" + + "github.com/gin-gonic/gin" +) + +type responseWriterCounter struct { + gin.ResponseWriter + bytesSent *atomic.Uint64 +} + +func (w *responseWriterCounter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + w.bytesSent.Add(uint64(n)) + return n, err +} diff --git a/internal/servers/hls/response_writer_no_cache.go b/internal/servers/hls/response_writer_no_cache.go new file mode 100644 index 00000000..12208043 --- /dev/null +++ b/internal/servers/hls/response_writer_no_cache.go @@ -0,0 +1,19 @@ +package hls + +import ( + "net/http" + + "github.com/gin-gonic/gin" +) + +type responseWriterNoCache struct { + gin.ResponseWriter +} + +func (w *responseWriterNoCache) WriteHeader(statusCode int) { + if statusCode == http.StatusOK { + w.ResponseWriter.Header().Set("Cache-Control", "private, no-cache") + } + + w.ResponseWriter.WriteHeader(statusCode) +} diff --git a/internal/servers/hls/server.go b/internal/servers/hls/server.go index aca58c6f..c845b0e8 100644 --- a/internal/servers/hls/server.go +++ b/internal/servers/hls/server.go @@ -11,12 +11,17 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/google/uuid" ) // ErrMuxerNotFound is returned when a muxer is not found. var ErrMuxerNotFound = errors.New("muxer not found") +// ErrSessionNotFound is returned when a session is not found. +var ErrSessionNotFound = errors.New("session not found") + func interfaceIsEmpty(i any) bool { return reflect.ValueOf(i).Kind() != reflect.Pointer || reflect.ValueOf(i).IsNil() } @@ -28,9 +33,10 @@ type serverGetMuxerRes struct { type serverGetMuxerReq struct { path string - remoteAddr string - query string - sourceOnDemand bool + create bool + remoteAddr string // only if create == true + query string // only if create == true + sourceOnDemand bool // only if create == true res chan serverGetMuxerRes } @@ -53,6 +59,34 @@ type serverAPIMuxersGetReq struct { res chan serverAPIMuxersGetRes } +type serverAPISessionsListRes struct { + data *defs.APIHLSSessionList + err error +} + +type serverAPISessionsListReq struct { + res chan serverAPISessionsListRes +} + +type serverAPISessionsGetRes struct { + data *defs.APIHLSSession + err error +} + +type serverAPISessionsGetReq struct { + uuid uuid.UUID + res chan serverAPISessionsGetRes +} + +type serverAPISessionsKickRes struct { + err error +} + +type serverAPISessionsKickReq struct { + uuid uuid.UUID + res chan serverAPISessionsKickRes +} + type serverMetrics interface { SetHLSServer(defs.APIHLSServer) } @@ -86,6 +120,7 @@ type Server struct { ReadTimeout conf.Duration WriteTimeout conf.Duration MuxerCloseAfter conf.Duration + ExternalCmdPool *externalcmd.Pool Metrics serverMetrics PathManager serverPathManager Parent serverParent @@ -97,12 +132,15 @@ type Server struct { muxers map[string]*muxer // in - chPathReady chan defs.Path - chPathNotReady chan defs.Path - chGetMuxer chan serverGetMuxerReq - 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 + chAPISessionsList chan serverAPISessionsListReq + chAPISessionsGet chan serverAPISessionsGetReq + chAPISessionsKick chan serverAPISessionsKickReq } // Initialize initializes the server. @@ -118,6 +156,9 @@ func (s *Server) Initialize() error { s.chCloseMuxer = make(chan *muxer) s.chAPIMuxerList = make(chan serverAPIMuxersListReq) s.chAPIMuxerGet = make(chan serverAPIMuxersGetReq) + s.chAPISessionsList = make(chan serverAPISessionsListReq) + s.chAPISessionsGet = make(chan serverAPISessionsGetReq) + s.chAPISessionsKick = make(chan serverAPISessionsKickReq) s.httpServer = &httpServer{ address: s.Address, @@ -211,6 +252,8 @@ outer: switch { case ok: req.res <- serverGetMuxerRes{muxer: mux} + case !req.create: + req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer not found")} case s.AlwaysRemux && !req.sourceOnDemand: req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer is waiting to be created")} default: @@ -248,6 +291,43 @@ outer: req.res <- serverAPIMuxersGetRes{data: muxer.apiItem()} + case req := <-s.chAPISessionsList: + data := &defs.APIHLSSessionList{ + Items: []defs.APIHLSSession{}, + } + + for _, muxer := range s.muxers { + data.Items = append(data.Items, muxer.apiSessionsList()...) + } + + sort.Slice(data.Items, func(i, j int) bool { + return data.Items[i].Created.Before(data.Items[j].Created) + }) + + req.res <- serverAPISessionsListRes{data: data} + + case req := <-s.chAPISessionsGet: + for _, muxer := range s.muxers { + session, ok := muxer.apiSessionsGet(req.uuid) + if ok { + req.res <- serverAPISessionsGetRes{data: session} + continue outer + } + } + + req.res <- serverAPISessionsGetRes{err: ErrSessionNotFound} + + case req := <-s.chAPISessionsKick: + for _, muxer := range s.muxers { + ok := muxer.apiSessionsKick(req.uuid) + if ok { + req.res <- serverAPISessionsKickRes{} + continue outer + } + } + + req.res <- serverAPISessionsKickRes{err: ErrSessionNotFound} + case <-s.ctx.Done(): break outer } @@ -349,3 +429,53 @@ func (s *Server) APIMuxersGet(name string) (*defs.APIHLSMuxer, error) { return nil, fmt.Errorf("terminated") } } + +// APISessionsList implements defs.APIHLSServer. +func (s *Server) APISessionsList() (*defs.APIHLSSessionList, error) { + req := serverAPISessionsListReq{ + res: make(chan serverAPISessionsListRes), + } + + select { + case s.chAPISessionsList <- req: + res := <-req.res + return res.data, res.err + + case <-s.ctx.Done(): + return nil, fmt.Errorf("terminated") + } +} + +// APISessionsGet implements defs.APIHLSServer. +func (s *Server) APISessionsGet(uuid uuid.UUID) (*defs.APIHLSSession, error) { + req := serverAPISessionsGetReq{ + uuid: uuid, + res: make(chan serverAPISessionsGetRes), + } + + select { + case s.chAPISessionsGet <- req: + res := <-req.res + return res.data, res.err + + case <-s.ctx.Done(): + return nil, fmt.Errorf("terminated") + } +} + +// APISessionsKick implements defs.APIHLSServer. +func (s *Server) APISessionsKick(uuid uuid.UUID) error { + req := serverAPISessionsKickReq{ + uuid: uuid, + res: make(chan serverAPISessionsKickRes), + } + + select { + case s.chAPISessionsKick <- req: + res := <-req.res + return res.err + + case <-s.ctx.Done(): + return fmt.Errorf("terminated") + } +} diff --git a/internal/servers/hls/server_test.go b/internal/servers/hls/server_test.go index 841461aa..20f64dc9 100644 --- a/internal/servers/hls/server_test.go +++ b/internal/servers/hls/server_test.go @@ -13,6 +13,7 @@ import ( "github.com/bluenviron/gohlslib/v2" "github.com/bluenviron/gohlslib/v2/pkg/codecs" "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" "github.com/bluenviron/mediamtx/internal/auth" "github.com/bluenviron/mediamtx/internal/conf" @@ -158,7 +159,7 @@ func TestServerNotFound(t *testing.T) { }, addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { require.Equal(t, "nonexisting", req.AccessRequest.Name) - return nil, fmt.Errorf("not found") + return nil, &defs.PathNoStreamAvailableError{} }, } @@ -253,11 +254,22 @@ func TestServerRead(t *testing.T) { }, addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { require.Equal(t, "teststream", req.AccessRequest.Name) - if ca == "always remux off" { + + switch req.Author.(type) { + case (*muxer): + if ca == "always remux off" { + require.Equal(t, "param=value", req.AccessRequest.Query) + } else { + require.Equal(t, "", req.AccessRequest.Query) + } + + case *session: require.Equal(t, "param=value", req.AccessRequest.Query) - } else { - require.Equal(t, "", req.AccessRequest.Query) + + default: + t.Errorf("should not happen") } + return &defs.PathAddReaderRes{Path: &dummyPath{}, Stream: strm}, nil }, } @@ -579,7 +591,7 @@ func TestAuthError(t *testing.T) { ReadTimeout: conf.Duration(10 * time.Second), WriteTimeout: conf.Duration(10 * time.Second), PathManager: &dummyPathManager{ - findPathConfImpl: func(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { if req.AccessRequest.Credentials.User == "" && req.AccessRequest.Credentials.Pass == "" { return nil, &auth.Error{AskCredentials: true, Wrapped: fmt.Errorf("auth error")} } @@ -625,3 +637,133 @@ func TestAuthError(t *testing.T) { require.Equal(t, 2, n) } + +func TestAuthQueryPreservedAcrossRedirect(t *testing.T) { + s := &Server{ + Address: "127.0.0.1:8888", + Encryption: false, + ServerKey: "", + ServerCert: "", + AlwaysRemux: true, + Variant: conf.HLSVariant(gohlslib.MuxerVariantMPEGTS), + SegmentCount: 7, + SegmentDuration: conf.Duration(1 * time.Second), + PartDuration: conf.Duration(200 * time.Millisecond), + SegmentMaxSize: 50 * 1024 * 1024, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + PathManager: &dummyPathManager{ + findPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + return nil, &auth.Error{AskCredentials: true, Wrapped: fmt.Errorf("auth error")} + }, + }, + Parent: test.NilLogger, + } + err := s.Initialize() + require.NoError(t, err) + defer s.Close() + + client := &http.Client{ + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + } + + res, err := client.Get("http://127.0.0.1:8888/stream/index.m3u8?jwt=mytoken") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusFound, res.StatusCode) + require.Equal(t, "/stream/index.m3u8?cookieCheck=1&jwt=mytoken", res.Header.Get("Location")) +} + +func TestServerNoSupportedCodecs(t *testing.T) { + for _, ca := range []string{ + "always remux off", + "always remux on", + } { + t.Run(ca, func(t *testing.T) { + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}} + + strm := &stream.Stream{ + Desc: desc, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + + pm := &dummyPathManager{ + findPathConfImpl: func(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + require.Equal(t, "teststream", req.AccessRequest.Name) + return &defs.PathFindPathConfRes{Conf: &conf.Path{}, User: req.AccessRequest.Credentials.User}, nil + }, + addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { + require.Equal(t, "teststream", req.AccessRequest.Name) + return &defs.PathAddReaderRes{Path: &dummyPath{}, Stream: strm}, nil + }, + } + + s := &Server{ + Address: "127.0.0.1:8888", + AlwaysRemux: (ca == "always remux on"), + Variant: conf.HLSVariant(gohlslib.MuxerVariantMPEGTS), + SegmentCount: 7, + SegmentDuration: conf.Duration(1 * time.Second), + PartDuration: conf.Duration(200 * time.Millisecond), + SegmentMaxSize: 50 * 1024 * 1024, + TrustedProxies: conf.IPNetworks{}, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + PathManager: pm, + Parent: test.NilLogger, + } + err = s.Initialize() + require.NoError(t, err) + defer s.Close() + + client := &http.Client{ + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + } + + res, err := client.Get("http://127.0.0.1:8888/teststream/index.m3u8") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusFound, res.StatusCode) + require.Equal(t, "/teststream/index.m3u8?cookieCheck=1", res.Header.Get("Location")) + + res, err = client.Get("http://127.0.0.1:8888/teststream/index.m3u8?cookieCheck=1") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusInternalServerError, res.StatusCode) + require.Contains(t, res.Header.Get("Content-Type"), "application/json") + + byts, err := io.ReadAll(res.Body) + require.NoError(t, err) + + var payload defs.APIError + err = json.Unmarshal(byts, &payload) + require.NoError(t, err) + + if ca == "always remux off" { + require.Equal(t, defs.APIError{ + Status: defs.APIErrorStatusError, + Error: "terminated", + }, payload) + } else { + require.Equal(t, defs.APIError{ + Status: defs.APIErrorStatusError, + Error: "muxer is waiting to be created", + }, payload) + } + }) + } +} diff --git a/internal/servers/hls/session.go b/internal/servers/hls/session.go new file mode 100644 index 00000000..accf6544 --- /dev/null +++ b/internal/servers/hls/session.go @@ -0,0 +1,168 @@ +package hls + +import ( + "encoding/hex" + "net" + "slices" + "sync/atomic" + "time" + + "github.com/bluenviron/mediamtx/internal/auth" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/externalcmd" + "github.com/bluenviron/mediamtx/internal/hooks" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/httpp" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/gin-gonic/gin" + "github.com/google/uuid" +) + +type sessionServer interface { + logger.Writer + getMuxer(serverGetMuxerReq) (*muxer, error) +} + +type session struct { + remoteAddr string + pathName string + externalCmdPool *externalcmd.Pool + pathManager serverPathManager + server sessionServer + + uuid uuid.UUID + secret uuid.UUID + ip string + created time.Time + query string + user string + lastRequestTime atomic.Int64 + bytesSent atomic.Uint64 + path defs.Path + stream *stream.Stream + muxer *muxer + reader *stream.Reader + onUnreadHook func() +} + +func (s *session) initialize(ctx *gin.Context) error { + s.uuid = uuid.New() + s.secret = uuid.New() + s.ip, _, _ = net.SplitHostPort(s.remoteAddr) + s.created = time.Now() + s.query = ctx.Request.URL.RawQuery + s.lastRequestTime.Store(time.Now().UnixNano()) + + res, err := s.pathManager.AddReader(defs.PathAddReaderReq{ + Author: s, + AccessRequest: defs.PathAccessRequest{ + Name: s.pathName, + Query: s.query, + Publish: false, + Proto: auth.ProtocolHLS, + ID: &s.uuid, + Credentials: httpp.Credentials(ctx.Request), + IP: net.ParseIP(ctx.ClientIP()), + }, + }) + if err != nil { + return err + } + + s.path = res.Path + s.stream = res.Stream + s.user = res.User + + muxer, err := s.server.getMuxer(serverGetMuxerReq{ + path: s.pathName, + create: true, + remoteAddr: s.remoteAddr, + query: s.query, + sourceOnDemand: res.Path.SafeConf().SourceOnDemand, + }) + if err != nil { + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + return err + } + + s.muxer = muxer + + muxerFormats, err := s.muxer.addSession(s) + if err != nil { + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + return err + } + + s.reader = &stream.Reader{ + Parent: s, + } + + // all of this is needed to allow Stream to increase outbound bytes for every HLS session + for _, medi := range res.Stream.Desc.Medias { + for _, forma := range medi.Formats { + if slices.Contains(muxerFormats, forma) { + s.reader.OnData(medi, forma, func(_ *unit.Unit) error { + return nil + }) + } + } + } + + res.Stream.AddReader(s.reader) + + s.Log(logger.Info, "created by %s, reading from muxer '%s'", s.remoteAddr, s.pathName) + + s.onUnreadHook = hooks.OnRead(hooks.OnReadParams{ + Logger: s, + ExternalCmdPool: s.externalCmdPool, + Conf: res.Path.SafeConf(), + ExternalCmdEnv: res.Path.ExternalCmdEnv(), + Reader: *s.APIReaderDescribe(), + Query: s.query, + }) + + return nil +} + +// called by path or path manager. +// not implemented since closing the Muxer is enough to close every associated session. +func (s *session) Close() { +} + +func (s *session) close2(err error) { + s.stream.RemoveReader(s.reader) + + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + + s.onUnreadHook() + + s.Log(logger.Info, "closed: %v", err) +} + +// Log implements logger.Writer. +func (s *session) Log(level logger.Level, format string, args ...any) { + id := hex.EncodeToString(s.uuid[:4]) + s.server.Log(level, "[session %v] "+format, append([]any{id}, args...)...) +} + +func (s *session) apiItem() *defs.APIHLSSession { + outboundBytes := s.bytesSent.Load() + + return &defs.APIHLSSession{ + ID: s.uuid, + Created: s.created, + RemoteAddr: s.remoteAddr, + Path: s.pathName, + Query: s.query, + User: s.user, + OutboundBytes: outboundBytes, + } +} + +func (s *session) APIReaderDescribe() *defs.APIPathReader { + return &defs.APIPathReader{ + Type: defs.APIPathReaderTypeHLSSession, + ID: s.uuid.String(), + } +} diff --git a/internal/servers/rtsp/conn.go b/internal/servers/rtsp/conn.go index e4c48520..a189a821 100644 --- a/internal/servers/rtsp/conn.go +++ b/internal/servers/rtsp/conn.go @@ -170,7 +170,7 @@ func (c *conn) onDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, return res, nil, err2 } - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(res.Err, &terr2) { return &base.Response{ StatusCode: base.StatusNotFound, diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index e3bd488b..976a1387 100644 --- a/internal/servers/rtsp/session.go +++ b/internal/servers/rtsp/session.go @@ -287,7 +287,7 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx, return res, nil, err2 } - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(err, &terr2) { return &base.Response{ StatusCode: base.StatusNotFound, diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index d32717eb..0c280ec9 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -710,7 +710,7 @@ func TestServerReadNotFound(t *testing.T) { return &defs.PathFindPathConfRes{Conf: &conf.Path{}, User: req.AccessRequest.Credentials.User}, nil }, AddReaderImpl: func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { - return nil, defs.PathNoStreamAvailableError{} + return nil, &defs.PathNoStreamAvailableError{} }, } diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 157d5f02..925d0395 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -279,21 +279,19 @@ func (s *session) runPublish() (int, error) { func (s *session) runRead() (int, error) { ip, _, _ := net.SplitHostPort(s.req.remoteAddr) - req := defs.PathAccessRequest{ - Name: s.req.pathName, - Query: s.req.httpRequest.URL.RawQuery, - Proto: auth.ProtocolWebRTC, - ID: &s.uuid, - Credentials: httpp.Credentials(s.req.httpRequest), - IP: net.ParseIP(ip), - } - res, err := s.pathManager.AddReader(defs.PathAddReaderReq{ - Author: s, - AccessRequest: req, + Author: s, + AccessRequest: defs.PathAccessRequest{ + Name: s.req.pathName, + Query: s.req.httpRequest.URL.RawQuery, + Proto: auth.ProtocolWebRTC, + ID: &s.uuid, + Credentials: httpp.Credentials(s.req.httpRequest), + IP: net.ParseIP(ip), + }, }) if err != nil { - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(err, &terr2) { return http.StatusNotFound, err } diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index 20e0d80a..c5b81017 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -95,7 +95,7 @@ func (r *secondaryReader) Close() { // APIReaderDescribe implements reader. func (*secondaryReader) APIReaderDescribe() *defs.APIPathReader { return &defs.APIPathReader{ - Type: defs.APIPathReaderTypeRPICameraSecondary, + Type: defs.APIPathReaderTypeHidden, ID: "", } } @@ -341,7 +341,7 @@ func (s *Source) waitForPrimary( }, }) if err != nil { - var err2 defs.PathNoStreamAvailableError + var err2 *defs.PathNoStreamAvailableError if errors.As(err, &err2) { select { case <-time.After(pauseBetweenErrors): diff --git a/internal/stream/reader.go b/internal/stream/reader.go index fe9d647c..56b7386c 100644 --- a/internal/stream/reader.go +++ b/internal/stream/reader.go @@ -16,8 +16,8 @@ type OnDataFunc func(*unit.Unit) error // Reader is a stream reader. type Reader struct { - SkipBytesSent bool - Parent logger.Writer + SkipOutboundBytes bool + Parent logger.Writer onDatas map[*description.Media]map[format.Format]OnDataFunc queueSize int diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 57d2d353..e4c2527b 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -394,8 +394,8 @@ func (s *Stream) Initialize() error { alwaysAvailable: s.AlwaysAvailable, rtpMaxPayloadSize: s.RTPMaxPayloadSize, replaceNTP: s.ReplaceNTP, - addInboundBytes: s.addInboundBytes, - addOutboundBytes: s.addOutboundBytes, + inboundBytes: &s.inboundBytes, + outboundBytes: &s.outboundBytes, updateLastTime: s.updateLastTime, writeRTSP: s.writeRTSP, inboundFramesInError: s.inboundFramesInError, @@ -574,14 +574,6 @@ func (s *Stream) WaitForReaders() { <-s.hasReaders } -func (s *Stream) addInboundBytes(v uint64) { - s.inboundBytes.Add(v) -} - -func (s *Stream) addOutboundBytes(v uint64) { - s.outboundBytes.Add(v) -} - func (s *Stream) updateLastTime(pts time.Duration) { s.timeMutex.Lock() defer s.timeMutex.Unlock() diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index 522b7dc4..c057989b 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -2,6 +2,7 @@ package stream import ( "crypto/rand" + "sync/atomic" "time" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -44,8 +45,8 @@ type streamFormat struct { rtpMaxPayloadSize int replaceNTP bool inboundFramesInError *errordumper.Dumper - addInboundBytes func(uint64) - addOutboundBytes func(uint64) + inboundBytes *atomic.Uint64 + outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) parent logger.Writer diff --git a/internal/stream/stream_media.go b/internal/stream/stream_media.go index c7e4dff0..f3ccb170 100644 --- a/internal/stream/stream_media.go +++ b/internal/stream/stream_media.go @@ -1,6 +1,7 @@ package stream import ( + "sync/atomic" "time" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -15,8 +16,8 @@ type streamMedia struct { alwaysAvailable bool rtpMaxPayloadSize int replaceNTP bool - addInboundBytes func(uint64) - addOutboundBytes func(uint64) + inboundBytes *atomic.Uint64 + outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) inboundFramesInError *errordumper.Dumper @@ -36,8 +37,8 @@ func (sm *streamMedia) initialize() error { rtpMaxPayloadSize: sm.rtpMaxPayloadSize, replaceNTP: sm.replaceNTP, inboundFramesInError: sm.inboundFramesInError, - addInboundBytes: sm.addInboundBytes, - addOutboundBytes: sm.addOutboundBytes, + inboundBytes: sm.inboundBytes, + outboundBytes: sm.outboundBytes, updateLastTime: sm.updateLastTime, writeRTSP: sm.writeRTSP, parent: sm.parent, diff --git a/internal/stream/stream_standard_test.go b/internal/stream/stream_standard_test.go index 7594cd1d..0cfbb53e 100644 --- a/internal/stream/stream_standard_test.go +++ b/internal/stream/stream_standard_test.go @@ -69,7 +69,7 @@ func TestStream(t *testing.T) { require.Equal(t, uint64(14), strm.OutboundBytes()) } -func TestStreamSkipBytesSent(t *testing.T) { +func TestStreamSkipOutboundBytes(t *testing.T) { desc := &description.Session{Medias: []*description.Media{ { Type: description.MediaTypeVideo, @@ -98,7 +98,7 @@ func TestStreamSkipBytesSent(t *testing.T) { require.NoError(t, err) r := &Reader{ - SkipBytesSent: true, + SkipOutboundBytes: true, } recv := make(chan struct{}) diff --git a/internal/stream/sub_stream_format.go b/internal/stream/sub_stream_format.go index 4cd45611..2a9c3e7f 100644 --- a/internal/stream/sub_stream_format.go +++ b/internal/stream/sub_stream_format.go @@ -170,7 +170,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { } size := unitSize(u) - ssf.streamFormat.addInboundBytes(size) + ssf.streamFormat.inboundBytes.Add(size) ssf.streamFormat.writeRTSP(ssf.streamFormat.media, u.RTPPackets, u.NTP) @@ -178,8 +178,8 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { csr := sr cOnData := onData sr.push(func() error { - if !csr.SkipBytesSent { - ssf.streamFormat.addOutboundBytes(size) + if !csr.SkipOutboundBytes { + ssf.streamFormat.outboundBytes.Add(size) } return cOnData(u) })