From 35e1f486c9b618cc67cf8edfd18eef2b8d54ce3a Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sun, 22 Feb 2026 13:34:05 +0100 Subject: [PATCH] add integrated packet dumper (#5488) this allows to dump any incoming and outgoing packet, to disk, in pcapng format. --- .github/ISSUE_TEMPLATE/bug.yml | 20 +- api/openapi.yaml | 2 + go.mod | 5 +- go.sum | 18 +- internal/api/api.go | 21 +- internal/conf/conf.go | 1 + internal/core/core.go | 24 +- internal/core/path.go | 2 + internal/core/path_manager.go | 4 +- internal/metrics/metrics.go | 21 +- internal/packetdumper/conn.go | 258 ++++++++++++++++++ internal/packetdumper/conn_test.go | 193 +++++++++++++ internal/packetdumper/dial_context.go | 32 +++ internal/packetdumper/listen.go | 24 ++ internal/packetdumper/listen_packet.go | 30 ++ internal/packetdumper/listener.go | 42 +++ internal/packetdumper/packet_conn.go | 198 ++++++++++++++ internal/packetdumper/packet_conn_test.go | 202 ++++++++++++++ internal/playback/server.go | 21 +- internal/pprof/pprof.go | 21 +- internal/protocols/httpp/server.go | 28 +- internal/protocols/udp/listen.go | 161 ----------- internal/protocols/udp/listener.go | 174 ++++++++++++ .../udp/{listen_test.go => listener_test.go} | 23 +- internal/protocols/udp/params.go | 19 ++ internal/protocols/unix/listen.go | 138 ---------- internal/protocols/unix/listener.go | 141 ++++++++++ .../unix/{listen_test.go => listener_test.go} | 15 +- internal/protocols/unix/params.go | 22 ++ internal/servers/hls/http_server.go | 21 +- internal/servers/hls/server.go | 2 + internal/servers/rtmp/server.go | 31 ++- internal/servers/rtsp/server.go | 15 + internal/servers/webrtc/http_server.go | 21 +- internal/servers/webrtc/server.go | 2 + internal/staticsources/handler.go | 7 + internal/staticsources/hls/source.go | 13 + internal/staticsources/mpegts/source.go | 42 ++- internal/staticsources/rtmp/source.go | 19 +- internal/staticsources/rtp/source.go | 42 ++- internal/staticsources/rtsp/source.go | 44 +-- internal/staticsources/webrtc/source.go | 13 + mediamtx.yml | 2 + 43 files changed, 1699 insertions(+), 435 deletions(-) create mode 100644 internal/packetdumper/conn.go create mode 100644 internal/packetdumper/conn_test.go create mode 100644 internal/packetdumper/dial_context.go create mode 100644 internal/packetdumper/listen.go create mode 100644 internal/packetdumper/listen_packet.go create mode 100644 internal/packetdumper/listener.go create mode 100644 internal/packetdumper/packet_conn.go create mode 100644 internal/packetdumper/packet_conn_test.go delete mode 100644 internal/protocols/udp/listen.go create mode 100644 internal/protocols/udp/listener.go rename internal/protocols/udp/{listen_test.go => listener_test.go} (63%) create mode 100644 internal/protocols/udp/params.go delete mode 100644 internal/protocols/unix/listen.go create mode 100644 internal/protocols/unix/listener.go rename internal/protocols/unix/{listen_test.go => listener_test.go} (78%) create mode 100644 internal/protocols/unix/params.go diff --git a/.github/ISSUE_TEMPLATE/bug.yml b/.github/ISSUE_TEMPLATE/bug.yml index cf7078bc..2b0b5d28 100644 --- a/.github/ISSUE_TEMPLATE/bug.yml +++ b/.github/ISSUE_TEMPLATE/bug.yml @@ -71,21 +71,19 @@ body: attributes: label: MediaMTX logs description: | - MediaMTX logs are often useful to identify the issue. If you think this is the case, set the parameter 'logLevel' to 'debug' and attach logs. + MediaMTX logs are often useful to identify the issue. If you think this is the case, set 'logLevel' to 'debug' and attach logs. placeholder: Paste or drag the log file here - type: textarea id: network attributes: - label: Network dump + label: Packet dump description: | - If the bug arises when using MediaMTX with external hardware or software, the most helpful content you can provide is a dump of the data exchanged between the server and the target (network dump). + If the bug arises when using MediaMTX with external hardware or software, the most helpful information you can provide is a packet dump, that can be generated in this way: - That can be generated in this way: - 1. Download wireshark (https://www.wireshark.org/) - 2. Start capturing on the interface used for exchanging packets - * If the server and the external hardware or software are both installed on your pc, the interface is probably "loopback", otherwise it's the one of your network card. - 3. Start the server and replicate the issue - 4. Stop capturing, save the result in .pcap format - 5. Attach - placeholder: Attach the pcap file by dragging it here + 1. In mediamtx.yml, set 'dumpPackets' to 'true' + 2. Start the server and replicate the issue + 3. Stop the server, find the generated .pcapng files in the current directory + 4. Attach the pcapng files by dragging them here + + placeholder: Attach the pcapng files by dragging them here diff --git a/api/openapi.yaml b/api/openapi.yaml index cea1f795..47a6bcba 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -79,6 +79,8 @@ components: type: string sysLogPrefix: type: string + dumpPackets: + type: boolean readTimeout: type: string writeTimeout: diff --git a/go.mod b/go.mod index 867bcfaa..b71bc4d0 100644 --- a/go.mod +++ b/go.mod @@ -11,8 +11,8 @@ require ( github.com/alecthomas/kong v1.14.0 github.com/asticode/go-astits v1.14.0 github.com/bluenviron/gohlslib/v2 v2.2.5 - github.com/bluenviron/gortmplib v0.2.1 - github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260211202618-9aea8d6fc9d6 + github.com/bluenviron/gortmplib v0.2.2-0.20260218152940-e1378f261371 + github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260222121945-18b06189ef23 github.com/bluenviron/mediacommon/v2 v2.7.2 github.com/datarhei/gosrt v0.10.0 github.com/fsnotify/fsnotify v1.9.0 @@ -22,6 +22,7 @@ require ( github.com/go-git/go-git/v5 v5.16.5 github.com/goccy/go-yaml v1.19.2 github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/google/gopacket v1.1.19 github.com/google/uuid v1.6.0 github.com/gookit/color v1.6.0 github.com/gorilla/websocket v1.5.3 diff --git a/go.sum b/go.sum index d8c29463..e70366c9 100644 --- a/go.sum +++ b/go.sum @@ -35,10 +35,10 @@ github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c h1:8XZeJrs4+ZYh github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c/go.mod h1:x1vxHcL/9AVzuk5HOloOEPrtJY0MaalYr78afXZ+pWI= github.com/bluenviron/gohlslib/v2 v2.2.5 h1:gQyVmymInXSwm7QgMBwaq6eZZAM0FaT+hX41iueMSyw= github.com/bluenviron/gohlslib/v2 v2.2.5/go.mod h1:4ZQMhQ1m9SMFWfh65lptaz0YwHbRxDIza7mM9OZsydo= -github.com/bluenviron/gortmplib v0.2.1 h1:ep+Aq4WTqbXUsJqMMenaRth2ugG1u3cm/fG0jHoriYI= -github.com/bluenviron/gortmplib v0.2.1/go.mod h1:AMqbShthdHCzF6OFjcJ4LHTBbeZ5jSfDZvZBW1zaToA= -github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260211202618-9aea8d6fc9d6 h1:Vnp5Lw784UU1jl7Xf0fcYkp2HMQDpOJpHQQRncD4ILg= -github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260211202618-9aea8d6fc9d6/go.mod h1:pMia36dVdPoRMBxV/T1Z2xYxwI8bFkkd7xc2WM5KsOM= +github.com/bluenviron/gortmplib v0.2.2-0.20260218152940-e1378f261371 h1:xoizvABlBF+G2N3CO7IuV0/IrLzL964Av9IkkR6lE+s= +github.com/bluenviron/gortmplib v0.2.2-0.20260218152940-e1378f261371/go.mod h1:HWpYpjVjl9pZlT3VRY9zciLrG1+JvYVY16VC6R7kB6M= +github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260222121945-18b06189ef23 h1:ZtXURmQHjNIfG/raJyfoNgTAGniD1LKrrpk3Ze5O2VI= +github.com/bluenviron/gortsplib/v5 v5.3.2-0.20260222121945-18b06189ef23/go.mod h1:k6NJcnpvsXDXGyVJpedGJ6K2ysA0xx9fiX9hA7pEua4= github.com/bluenviron/mediacommon/v2 v2.7.2 h1:BA7X7YOKq4VC6Ygm5FIg2D2IYTt/Ud6T43anXkHbDE8= github.com/bluenviron/mediacommon/v2 v2.7.2/go.mod h1:5V15TiOfeaNVmZPVuOqAwqQSWyvMV86/dijDKu5q9Zs= github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ= @@ -104,6 +104,8 @@ github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUv github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8= +github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo= github.com/google/pprof v0.0.0-20260202012954-cb029daf43ef h1:xpF9fUHpoIrrjX24DURVKiwHcFpw19ndIs+FwTSMbno= github.com/google/pprof v0.0.0-20260202012954-cb029daf43ef/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= @@ -238,6 +240,7 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/arch v0.20.0 h1:dx1zTU0MAE98U+TQ8BLl7XsJbgze2WnNKF/8tGp/Q6c= golang.org/x/arch v0.20.0/go.mod h1:bdwinDaKcfZUGpH09BB7ZmOfhalA8lQdzl62l8gGWsk= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= golang.org/x/crypto v0.0.0-20211209193657-4570a0811e8b/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= @@ -245,15 +248,20 @@ golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0LeHDbnYEryqj5Q1ug8= golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY= +golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= +golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -277,8 +285,10 @@ golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw= google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/api/api.go b/internal/api/api.go index eefc81df..e44c16a7 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -80,6 +80,7 @@ type API struct { Version string Started time.Time Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -183,15 +184,17 @@ func (a *API) Initialize() error { group.DELETE("/recordings/deletesegment", a.onRecordingDeleteSegment) a.httpServer = &httpp.Server{ - Address: a.Address, - AllowOrigins: a.AllowOrigins, - ReadTimeout: time.Duration(a.ReadTimeout), - WriteTimeout: time.Duration(a.WriteTimeout), - Encryption: a.Encryption, - ServerCert: a.ServerCert, - ServerKey: a.ServerKey, - Handler: router, - Parent: a, + Address: a.Address, + AllowOrigins: a.AllowOrigins, + DumpPackets: a.DumpPackets, + DumpPacketsPrefix: "api_server_conn", + ReadTimeout: time.Duration(a.ReadTimeout), + WriteTimeout: time.Duration(a.WriteTimeout), + Encryption: a.Encryption, + ServerCert: a.ServerCert, + ServerKey: a.ServerKey, + Handler: router, + Parent: a, } err := a.httpServer.Initialize() if err != nil { diff --git a/internal/conf/conf.go b/internal/conf/conf.go index 5eb74153..2a2871c7 100644 --- a/internal/conf/conf.go +++ b/internal/conf/conf.go @@ -245,6 +245,7 @@ type Conf struct { LogStructured bool `json:"logStructured"` LogFile string `json:"logFile"` SysLogPrefix string `json:"sysLogPrefix"` + DumpPackets bool `json:"dumpPackets"` ReadTimeout Duration `json:"readTimeout"` WriteTimeout Duration `json:"writeTimeout"` ReadBufferCount *int `json:"readBufferCount,omitempty"` // deprecated diff --git a/internal/core/core.go b/internal/core/core.go index 9b82331b..59d3c5fd 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -352,6 +352,7 @@ func (p *Core) createResources(initial bool) error { p.metrics == nil { i := &metrics.Metrics{ Address: p.conf.MetricsAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.MetricsEncryption, ServerKey: p.conf.MetricsServerKey, ServerCert: p.conf.MetricsServerCert, @@ -373,6 +374,7 @@ func (p *Core) createResources(initial bool) error { p.pprof == nil { i := &pprof.PPROF{ Address: p.conf.PPROFAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.PPROFEncryption, ServerKey: p.conf.PPROFServerKey, ServerCert: p.conf.PPROFServerCert, @@ -403,6 +405,7 @@ func (p *Core) createResources(initial bool) error { p.playbackServer == nil { i := &playback.Server{ Address: p.conf.PlaybackAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.PlaybackEncryption, ServerKey: p.conf.PlaybackServerKey, ServerCert: p.conf.PlaybackServerCert, @@ -426,7 +429,7 @@ func (p *Core) createResources(initial bool) error { p.pathManager = &pathManager{ logLevel: p.conf.LogLevel, - authManager: p.authManager, + dumpPackets: p.conf.DumpPackets, rtspAddress: p.conf.RTSPAddress, readTimeout: p.conf.ReadTimeout, writeTimeout: p.conf.WriteTimeout, @@ -434,6 +437,7 @@ func (p *Core) createResources(initial bool) error { udpReadBufferSize: p.conf.UDPReadBufferSize, rtpMaxPayloadSize: rtpMaxPayloadSize, pathConfs: p.conf.Paths, + authManager: p.authManager, externalCmdPool: p.externalCmdPool, metrics: p.metrics, parent: p, @@ -453,6 +457,7 @@ func (p *Core) createResources(initial bool) error { i := &rtsp.Server{ Address: p.conf.RTSPAddress, AuthMethods: p.conf.RTSPAuthMethods.ToAuthMethods(), + DumpPackets: p.conf.DumpPackets, UDPReadBufferSize: udpReadBufferSize, ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, @@ -495,6 +500,7 @@ func (p *Core) createResources(initial bool) error { i := &rtsp.Server{ Address: p.conf.RTSPSAddress, AuthMethods: p.conf.RTSPAuthMethods.ToAuthMethods(), + DumpPackets: p.conf.DumpPackets, UDPReadBufferSize: udpReadBufferSize, ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, @@ -531,6 +537,7 @@ func (p *Core) createResources(initial bool) error { p.rtmpServer == nil { i := &rtmp.Server{ Address: p.conf.RTMPAddress, + DumpPackets: p.conf.DumpPackets, ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, IsTLS: false, @@ -563,6 +570,7 @@ func (p *Core) createResources(initial bool) error { IsTLS: true, ServerCert: p.conf.RTMPServerCert, ServerKey: p.conf.RTMPServerKey, + DumpPackets: p.conf.DumpPackets, RTSPAddress: p.conf.RTSPAddress, RunOnConnect: p.conf.RunOnConnect, RunOnConnectRestart: p.conf.RunOnConnectRestart, @@ -583,6 +591,7 @@ func (p *Core) createResources(initial bool) error { p.hlsServer == nil { i := &hls.Server{ Address: p.conf.HLSAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.HLSEncryption, ServerKey: p.conf.HLSServerKey, ServerCert: p.conf.HLSServerCert, @@ -613,6 +622,7 @@ func (p *Core) createResources(initial bool) error { p.webRTCServer == nil { i := &webrtc.Server{ Address: p.conf.WebRTCAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.WebRTCEncryption, ServerKey: p.conf.WebRTCServerKey, ServerCert: p.conf.WebRTCServerCert, @@ -671,6 +681,7 @@ func (p *Core) createResources(initial bool) error { Version: string(version), Started: started, Address: p.conf.APIAddress, + DumpPackets: p.conf.DumpPackets, Encryption: p.conf.APIEncryption, ServerKey: p.conf.APIServerKey, ServerCert: p.conf.APIServerCert, @@ -742,6 +753,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { !reflect.DeepEqual(newConf.MetricsTrustedProxies, p.conf.MetricsTrustedProxies) || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || + newConf.DumpPackets != p.conf.DumpPackets || closeAuthManager || closeLogger @@ -755,6 +767,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { !reflect.DeepEqual(newConf.PPROFTrustedProxies, p.conf.PPROFTrustedProxies) || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || + newConf.DumpPackets != p.conf.DumpPackets || closeAuthManager || closeLogger @@ -775,6 +788,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { !reflect.DeepEqual(newConf.PlaybackTrustedProxies, p.conf.PlaybackTrustedProxies) || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || + newConf.DumpPackets != p.conf.DumpPackets || closeAuthManager || closeLogger if !closePlaybackServer && p.playbackServer != nil && !reflect.DeepEqual(newConf.Paths, p.conf.Paths) { @@ -783,6 +797,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { closePathManager := newConf == nil || newConf.LogLevel != p.conf.LogLevel || + newConf.DumpPackets != p.conf.DumpPackets || newConf.RTSPAddress != p.conf.RTSPAddress || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || @@ -803,6 +818,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTSPAddress != p.conf.RTSPAddress || !reflect.DeepEqual(newConf.RTSPAuthMethods, p.conf.RTSPAuthMethods) || newConf.RTSPUDPReadBufferSize != p.conf.RTSPUDPReadBufferSize || + newConf.DumpPackets != p.conf.DumpPackets || newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || @@ -826,6 +842,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTSPSAddress != p.conf.RTSPSAddress || !reflect.DeepEqual(newConf.RTSPAuthMethods, p.conf.RTSPAuthMethods) || newConf.RTSPUDPReadBufferSize != p.conf.RTSPUDPReadBufferSize || + newConf.DumpPackets != p.conf.DumpPackets || newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || @@ -845,6 +862,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTMP != p.conf.RTMP || newConf.RTMPEncryption != p.conf.RTMPEncryption || newConf.RTMPAddress != p.conf.RTMPAddress || + newConf.DumpPackets != p.conf.DumpPackets || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.RTSPAddress != p.conf.RTSPAddress || @@ -859,6 +877,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTMP != p.conf.RTMP || newConf.RTMPEncryption != p.conf.RTMPEncryption || newConf.RTMPSAddress != p.conf.RTMPSAddress || + newConf.DumpPackets != p.conf.DumpPackets || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.RTMPServerCert != p.conf.RTMPServerCert || @@ -889,6 +908,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.HLSMuxerCloseAfter != p.conf.HLSMuxerCloseAfter || + newConf.DumpPackets != p.conf.DumpPackets || closePathManager || closeMetrics || closeLogger @@ -913,6 +933,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.WebRTCSTUNGatherTimeout != p.conf.WebRTCSTUNGatherTimeout || newConf.WebRTCHandshakeTimeout != p.conf.WebRTCHandshakeTimeout || newConf.WebRTCTrackGatherTimeout != p.conf.WebRTCTrackGatherTimeout || + newConf.DumpPackets != p.conf.DumpPackets || closeMetrics || closePathManager || closeLogger @@ -940,6 +961,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { !reflect.DeepEqual(newConf.APITrustedProxies, p.conf.APITrustedProxies) || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || + newConf.DumpPackets != p.conf.DumpPackets || closeAuthManager || closePathManager || closeRTSPServer || diff --git a/internal/core/path.go b/internal/core/path.go index f4c8a1f1..55869b5c 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -68,6 +68,7 @@ type pathAPIPathsGetReq struct { type path struct { parentCtx context.Context logLevel conf.LogLevel + dumpPackets bool rtspAddress string readTimeout conf.Duration writeTimeout conf.Duration @@ -193,6 +194,7 @@ func (pa *path) run() { pa.source = &staticsources.Handler{ Conf: pa.conf, LogLevel: pa.logLevel, + DumpPackets: pa.dumpPackets, ReadTimeout: pa.readTimeout, WriteTimeout: pa.writeTimeout, WriteQueueSize: pa.writeQueueSize, diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index d9b94125..d9800957 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -71,14 +71,15 @@ type pathManagerParent interface { type pathManager struct { logLevel conf.LogLevel - authManager pathManagerAuthManager rtspAddress string + dumpPackets bool readTimeout conf.Duration writeTimeout conf.Duration writeQueueSize int udpReadBufferSize uint rtpMaxPayloadSize int pathConfs map[string]*conf.Path + authManager pathManagerAuthManager externalCmdPool *externalcmd.Pool metrics *metrics.Metrics parent pathManagerParent @@ -443,6 +444,7 @@ func (pm *pathManager) createPath( pa := &path{ parentCtx: pm.ctx, logLevel: pm.logLevel, + dumpPackets: pm.dumpPackets, rtspAddress: pm.rtspAddress, readTimeout: pm.readTimeout, writeTimeout: pm.writeTimeout, diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 266b7629..e6ed0875 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -71,6 +71,7 @@ type metricsParent interface { // Metrics is a metrics provider. type Metrics struct { Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -104,15 +105,17 @@ func (m *Metrics) Initialize() error { router.GET("/metrics", m.onMetrics) m.httpServer = &httpp.Server{ - Address: m.Address, - AllowOrigins: m.AllowOrigins, - ReadTimeout: time.Duration(m.ReadTimeout), - WriteTimeout: time.Duration(m.WriteTimeout), - Encryption: m.Encryption, - ServerCert: m.ServerCert, - ServerKey: m.ServerKey, - Handler: router, - Parent: m, + Address: m.Address, + AllowOrigins: m.AllowOrigins, + DumpPackets: m.DumpPackets, + DumpPacketsPrefix: "metrics_server_conn", + ReadTimeout: time.Duration(m.ReadTimeout), + WriteTimeout: time.Duration(m.WriteTimeout), + Encryption: m.Encryption, + ServerCert: m.ServerCert, + ServerKey: m.ServerKey, + Handler: router, + Parent: m, } err := m.httpServer.Initialize() if err != nil { diff --git a/internal/packetdumper/conn.go b/internal/packetdumper/conn.go new file mode 100644 index 00000000..68a77cfd --- /dev/null +++ b/internal/packetdumper/conn.go @@ -0,0 +1,258 @@ +// Package packetdumper provides utilities to dump packets to disk. +package packetdumper + +import ( + "fmt" + "net" + "os" + "sync" + "time" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcapgo" + "github.com/google/uuid" +) + +var _ net.Conn = (*Conn)(nil) + +type direction int + +const ( + dirRead direction = iota + dirWrite + dirHandshake +) + +type dumpEntry struct { + ntp time.Time + data []byte + direction direction +} + +// Conn is a wrapper around net.Conn that dumps packets to disk. +type Conn struct { + Prefix string + Conn net.Conn + ServerSide bool + + f *os.File + pw *pcapgo.NgWriter + once sync.Once + + queue chan dumpEntry + terminated chan struct{} + done chan struct{} +} + +// Initialize initializes Conn. +func (c *Conn) Initialize() error { + var err error + c.f, err = os.Create(fmt.Sprintf("%s_%d_%s.pcapng", c.Prefix, time.Now().UnixNano(), uuid.New().String())) + if err != nil { + return err + } + + c.pw, err = pcapgo.NewNgWriter(c.f, layers.LinkTypeEthernet) + if err != nil { + c.f.Close() + return err + } + + c.queue = make(chan dumpEntry, 64) + c.terminated = make(chan struct{}) + c.done = make(chan struct{}) + + go c.run() + + c.enqueue(dumpEntry{ntp: time.Now(), direction: dirHandshake}) + + return nil +} + +// Close implements net.Conn. +func (c *Conn) Close() error { + c.once.Do(func() { + close(c.terminated) + }) + <-c.done + return c.Conn.Close() +} + +func (c *Conn) run() { + defer close(c.done) + defer c.f.Close() + + local := c.Conn.LocalAddr().(*net.TCPAddr) + remote := c.Conn.RemoteAddr().(*net.TCPAddr) + + nextLocalSequence := uint32(1000) + nextRemoteSequence := uint32(2000) + + for { + select { + case e := <-c.queue: + c.processEntry(e, local, remote, &nextLocalSequence, &nextRemoteSequence) + + case <-c.terminated: + // Drain anything already in the queue before exiting. + for { + select { + case e := <-c.queue: + c.processEntry(e, local, remote, &nextLocalSequence, &nextRemoteSequence) + default: + c.pw.Flush() //nolint:errcheck + return + } + } + } + } +} + +func (c *Conn) processEntry( + e dumpEntry, + local, remote *net.TCPAddr, + nextLocalSequence, nextRemoteSequence *uint32, +) { + switch e.direction { + case dirHandshake: + clientAddr, serverAddr := local, remote // client side: local initiates + clientSeq, serverSeq := nextLocalSequence, nextRemoteSequence + if c.ServerSide { + clientAddr, serverAddr = remote, local // server side: remote initiated + clientSeq, serverSeq = nextRemoteSequence, nextLocalSequence + } + + // SYN (client -> server) + c.writePacket(e.ntp, clientAddr, serverAddr, + layers.TCP{SYN: true, Window: 65535, Seq: *clientSeq, Ack: 0}, nil) + *clientSeq++ + + // SYN-ACK (server -> client) + c.writePacket(e.ntp, serverAddr, clientAddr, + layers.TCP{SYN: true, ACK: true, Window: 65535, Seq: *serverSeq, Ack: *clientSeq}, nil) + *serverSeq++ + + // ACK (client -> server) + c.writePacket(e.ntp, clientAddr, serverAddr, + layers.TCP{ACK: true, Window: 65535, Seq: *clientSeq, Ack: *serverSeq}, nil) + + case dirRead: + tcpFlags := layers.TCP{ + PSH: true, + ACK: true, + Window: 14600, + Seq: *nextRemoteSequence, + Ack: *nextLocalSequence, + } + c.writePacket(e.ntp, remote, local, tcpFlags, e.data) + *nextRemoteSequence += uint32(len(e.data)) + + case dirWrite: + tcpFlags := layers.TCP{ + PSH: true, + ACK: true, + Window: 14600, + Seq: *nextLocalSequence, + Ack: *nextRemoteSequence, + } + c.writePacket(e.ntp, local, remote, tcpFlags, e.data) + *nextLocalSequence += uint32(len(e.data)) + } +} + +func (c *Conn) writePacket( + ntp time.Time, + src, dst *net.TCPAddr, + tcpFlags layers.TCP, + payload []byte, +) { + eth := &layers.Ethernet{ + SrcMAC: net.HardwareAddr{0, 0, 0, 0, 0, 0}, + DstMAC: net.HardwareAddr{0, 0, 0, 0, 0, 0}, + EthernetType: layers.EthernetTypeIPv6, + } + + ipv6 := &layers.IPv6{ + Version: 6, + SrcIP: src.IP.To16(), + DstIP: dst.IP.To16(), + NextHeader: layers.IPProtocolTCP, + HopLimit: 64, + } + + tcp := &layers.TCP{ + SrcPort: layers.TCPPort(src.Port), + DstPort: layers.TCPPort(dst.Port), + Seq: tcpFlags.Seq, + Ack: tcpFlags.Ack, + Window: tcpFlags.Window, + SYN: tcpFlags.SYN, + ACK: tcpFlags.ACK, + PSH: tcpFlags.PSH, + FIN: tcpFlags.FIN, + } + tcp.SetNetworkLayerForChecksum(ipv6) //nolint:errcheck + + buf := gopacket.NewSerializeBuffer() + opts := gopacket.SerializeOptions{FixLengths: true, ComputeChecksums: true} + gopacket.SerializeLayers(buf, opts, eth, ipv6, tcp, gopacket.Payload(payload)) //nolint:errcheck + + raw := buf.Bytes() + + c.pw.WritePacket(gopacket.CaptureInfo{ //nolint:errcheck + Timestamp: ntp, + CaptureLength: len(raw), + Length: len(raw), + }, raw) +} + +func (c *Conn) enqueue(e dumpEntry) { + select { + case c.queue <- e: + case <-c.terminated: + } +} + +func (c *Conn) Read(p []byte) (n int, err error) { + n, err = c.Conn.Read(p) + + if n != 0 { + c.enqueue(dumpEntry{ + ntp: time.Now(), + data: append([]byte(nil), p[:n]...), + direction: dirRead, + }) + } + + return n, err +} + +func (c *Conn) Write(p []byte) (n int, err error) { + n, err = c.Conn.Write(p) + + if err == nil { + c.enqueue(dumpEntry{ + ntp: time.Now(), + data: append([]byte(nil), p...), + direction: dirWrite, + }) + } + + return n, err +} + +// LocalAddr implements net.Conn. +func (c *Conn) LocalAddr() net.Addr { return c.Conn.LocalAddr() } + +// RemoteAddr implements net.Conn. +func (c *Conn) RemoteAddr() net.Addr { return c.Conn.RemoteAddr() } + +// SetDeadline implements net.Conn. +func (c *Conn) SetDeadline(t time.Time) error { return c.Conn.SetDeadline(t) } + +// SetReadDeadline implements net.Conn. +func (c *Conn) SetReadDeadline(t time.Time) error { return c.Conn.SetReadDeadline(t) } + +// SetWriteDeadline implements net.Conn. +func (c *Conn) SetWriteDeadline(t time.Time) error { return c.Conn.SetWriteDeadline(t) } diff --git a/internal/packetdumper/conn_test.go b/internal/packetdumper/conn_test.go new file mode 100644 index 00000000..89227ecb --- /dev/null +++ b/internal/packetdumper/conn_test.go @@ -0,0 +1,193 @@ +package packetdumper + +import ( + "io" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// startTCPPair dials a local TCP listener and returns both ends of the connection. +func startTCPPair(t *testing.T) (client, server net.Conn) { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + serverCh := make(chan net.Conn, 1) + go func() { + conn, err2 := ln.Accept() + if err2 == nil { + serverCh <- conn + } + }() + + client, err = net.Dial("tcp", ln.Addr().String()) + require.NoError(t, err) + + t.Cleanup(func() { ln.Close() }) + + select { + case server = <-serverCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for server connection") + } + + return client, server +} + +func cleanupPcapng(t *testing.T, prefix string) { + t.Helper() + + matches, err := filepath.Glob(prefix + "_*.pcapng") + require.NoError(t, err, "glob for pcapng files") + require.NotEmpty(t, matches, "expected at least one pcapng file to have been created") + + for _, f := range matches { + require.NoError(t, os.Remove(f), "removing pcapng file %s", f) + } +} + +func TestConnInitialize_CreatesFile(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck +} + +func TestConnWrite(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck + + n, err := c.Write([]byte("hello world")) + require.NoError(t, err) + require.Equal(t, 11, n) + + buf := make([]byte, 11) + _, err = io.ReadFull(server, buf) + require.NoError(t, err) + require.Equal(t, []byte("hello world"), buf) +} + +func TestConnRead(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck + + _, err := server.Write([]byte("incoming data")) + require.NoError(t, err) + + buf := make([]byte, 32) + n, err := c.Read(buf) + require.NoError(t, err) + require.Equal(t, []byte("incoming data"), buf[:n]) +} + +func TestConnServerSide(t *testing.T) { + client, server := startTCPPair(t) + defer client.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: server, ServerSide: true} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck + + n, err := c.Write([]byte("server response")) + require.NoError(t, err) + require.Equal(t, 15, n) + + buf := make([]byte, 15) + _, err = io.ReadFull(client, buf) + require.NoError(t, err) + require.Equal(t, []byte("server response"), buf) +} + +func TestConnMultipleWriteRead(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck + + for _, msg := range []string{"foo", "bar", "baz"} { + n, err := c.Write([]byte(msg)) + require.NoError(t, err) + require.Equal(t, len(msg), n) + } + + buf := make([]byte, len("foobarbaz")) + _, err := io.ReadFull(server, buf) + require.NoError(t, err) + require.Equal(t, []byte("foobarbaz"), buf) + + _, err = server.Write([]byte("abcde")) + require.NoError(t, err) + _, err = server.Write([]byte("fghij")) + require.NoError(t, err) + + readBuf := make([]byte, 10) + _, err = io.ReadFull(c, readBuf) + require.NoError(t, err) + require.Equal(t, []byte("abcdefghij"), readBuf) +} + +func TestConnCloseIdempotent(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + + defer c.Close() //nolint:errcheck + defer c.Close() //nolint:errcheck +} + +func TestConnDelegatesAddrMethods(t *testing.T) { + client, server := startTCPPair(t) + defer server.Close() + + prefix := filepath.Join(t.TempDir(), "capture") + c := &Conn{Prefix: prefix, Conn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapng(t, prefix) + defer c.Close() //nolint:errcheck + + require.Equal(t, client.LocalAddr(), c.LocalAddr()) + require.Equal(t, client.RemoteAddr(), c.RemoteAddr()) + + require.NoError(t, c.SetDeadline(time.Now().Add(time.Second))) + require.NoError(t, c.SetReadDeadline(time.Now().Add(time.Second))) + require.NoError(t, c.SetWriteDeadline(time.Now().Add(time.Second))) +} diff --git a/internal/packetdumper/dial_context.go b/internal/packetdumper/dial_context.go new file mode 100644 index 00000000..ec39d84c --- /dev/null +++ b/internal/packetdumper/dial_context.go @@ -0,0 +1,32 @@ +package packetdumper + +import ( + "context" + "net" +) + +// DialContext is a wrapper around net.Dialer.DialContext that dumps packets to disk. +type DialContext struct { + Prefix string + DialContext func(ctx context.Context, network, address string) (net.Conn, error) +} + +// Do mimics net.Dialer.DialContext. +func (d *DialContext) Do(ctx context.Context, network, address string) (net.Conn, error) { + conn, err := d.DialContext(ctx, network, address) + if err != nil { + return nil, err + } + + c := &Conn{ + Prefix: d.Prefix, + Conn: conn, + } + err = c.Initialize() + if err != nil { + conn.Close() + return nil, err + } + + return c, nil +} diff --git a/internal/packetdumper/listen.go b/internal/packetdumper/listen.go new file mode 100644 index 00000000..2f88d651 --- /dev/null +++ b/internal/packetdumper/listen.go @@ -0,0 +1,24 @@ +package packetdumper + +import ( + "net" +) + +// Listen is a wrapper around net.Listen that dumps packets to disk. +type Listen struct { + Prefix string + Listen func(network, address string) (net.Listener, error) +} + +// Do mimics net.Listen. +func (l *Listen) Do(network, address string) (net.Listener, error) { + ln, err := l.Listen(network, address) + if err != nil { + return nil, err + } + + return &Listener{ + Prefix: l.Prefix, + Listener: ln, + }, nil +} diff --git a/internal/packetdumper/listen_packet.go b/internal/packetdumper/listen_packet.go new file mode 100644 index 00000000..2e9824a0 --- /dev/null +++ b/internal/packetdumper/listen_packet.go @@ -0,0 +1,30 @@ +package packetdumper + +import ( + "net" +) + +// ListenPacket is a wrapper around net.ListenPacket that dumps packets to disk. +type ListenPacket struct { + Prefix string + ListenPacket func(network, address string) (net.PacketConn, error) +} + +// Do mimics net.ListenPacket +func (l *ListenPacket) Do(network, address string) (net.PacketConn, error) { + pc, err := l.ListenPacket(network, address) + if err != nil { + return nil, err + } + + d := &PacketConn{ + Prefix: l.Prefix, + PacketConn: pc, + } + err = d.Initialize() + if err != nil { + return nil, err + } + + return d, nil +} diff --git a/internal/packetdumper/listener.go b/internal/packetdumper/listener.go new file mode 100644 index 00000000..87bcd9ef --- /dev/null +++ b/internal/packetdumper/listener.go @@ -0,0 +1,42 @@ +package packetdumper + +import "net" + +var _ net.Listener = (*Listener)(nil) + +// Listener is a wrapper around net.Listener that dumps packets to disk. +type Listener struct { + Prefix string + Listener net.Listener +} + +// Accept implements net.Listener. +func (l *Listener) Accept() (net.Conn, error) { + conn, err := l.Listener.Accept() + if err != nil { + return nil, err + } + + cd := &Conn{ + Prefix: l.Prefix, + Conn: conn, + ServerSide: true, + } + err = cd.Initialize() + if err != nil { + conn.Close() //nolint:errcheck + return nil, err + } + + return cd, nil +} + +// Close implements net.Listener. +func (l *Listener) Close() error { + return l.Listener.Close() +} + +// Addr implements net.Listener. +func (l *Listener) Addr() net.Addr { + return l.Listener.Addr() +} diff --git a/internal/packetdumper/packet_conn.go b/internal/packetdumper/packet_conn.go new file mode 100644 index 00000000..b6805fcd --- /dev/null +++ b/internal/packetdumper/packet_conn.go @@ -0,0 +1,198 @@ +package packetdumper + +import ( + "fmt" + "net" + "os" + "sync" + "syscall" + "time" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcapgo" + "github.com/google/uuid" +) + +var _ net.PacketConn = (*PacketConn)(nil) + +type extendedPacketConn interface { + net.PacketConn + SetReadBuffer(bytes int) error + SyscallConn() (syscall.RawConn, error) +} + +type packetDumpEntry struct { + ntp time.Time + data []byte + src, dst *net.UDPAddr +} + +// PacketConn is a wrapper around net.PacketConn that dumps packets to disk. +type PacketConn struct { + Prefix string + PacketConn net.PacketConn + + f *os.File + pw *pcapgo.NgWriter + once sync.Once + + queue chan packetDumpEntry + terminated chan struct{} + done chan struct{} +} + +// Initialize initializes PacketConn. +func (c *PacketConn) Initialize() error { + var err error + c.f, err = os.Create(fmt.Sprintf("%s_%d_%s.pcapng", c.Prefix, time.Now().UnixNano(), uuid.New().String())) + if err != nil { + return err + } + + c.pw, err = pcapgo.NewNgWriter(c.f, layers.LinkTypeEthernet) + if err != nil { + c.f.Close() + return err + } + + c.queue = make(chan packetDumpEntry, 64) + c.terminated = make(chan struct{}) + c.done = make(chan struct{}) + + go c.run() + + return nil +} + +// Close implements net.PacketConn. +func (c *PacketConn) Close() error { + c.once.Do(func() { + close(c.terminated) + }) + <-c.done + return c.PacketConn.Close() +} + +func (c *PacketConn) run() { + defer close(c.done) + defer c.f.Close() + + for { + select { + case e := <-c.queue: + c.writePacket(e.ntp, e.src, e.dst, e.data) + + case <-c.terminated: + for { + select { + case e := <-c.queue: + c.writePacket(e.ntp, e.src, e.dst, e.data) + default: + c.pw.Flush() //nolint:errcheck + return + } + } + } + } +} + +func (c *PacketConn) writePacket(ntp time.Time, src, dst *net.UDPAddr, payload []byte) { + eth := &layers.Ethernet{ + SrcMAC: net.HardwareAddr{0, 0, 0, 0, 0, 0}, + DstMAC: net.HardwareAddr{0, 0, 0, 0, 0, 0}, + EthernetType: layers.EthernetTypeIPv6, + } + + ipv6 := &layers.IPv6{ + Version: 6, + SrcIP: src.IP.To16(), + DstIP: dst.IP.To16(), + NextHeader: layers.IPProtocolUDP, + HopLimit: 64, + } + + udp := &layers.UDP{ + SrcPort: layers.UDPPort(src.Port), + DstPort: layers.UDPPort(dst.Port), + } + udp.SetNetworkLayerForChecksum(ipv6) //nolint:errcheck + + buf := gopacket.NewSerializeBuffer() + opts := gopacket.SerializeOptions{FixLengths: true, ComputeChecksums: true} + gopacket.SerializeLayers(buf, opts, eth, ipv6, udp, gopacket.Payload(payload)) //nolint:errcheck + + raw := buf.Bytes() + c.pw.WritePacket(gopacket.CaptureInfo{ //nolint:errcheck + Timestamp: ntp, + CaptureLength: len(raw), + Length: len(raw), + }, raw) +} + +func (c *PacketConn) enqueue(e packetDumpEntry) { + select { + case c.queue <- e: + case <-c.terminated: + } +} + +// ReadFrom implements net.PacketConn. +func (c *PacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { + n, addr, err = c.PacketConn.ReadFrom(p) + + if n != 0 { + local := c.PacketConn.LocalAddr().(*net.UDPAddr) + remote := addr.(*net.UDPAddr) + + c.enqueue(packetDumpEntry{ + ntp: time.Now(), + data: append([]byte(nil), p[:n]...), + src: remote, + dst: local, + }) + } + + return n, addr, err +} + +// WriteTo implements net.PacketConn. +func (c *PacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) { + n, err = c.PacketConn.WriteTo(p, addr) + + if err == nil { + local := c.PacketConn.LocalAddr().(*net.UDPAddr) + remote := addr.(*net.UDPAddr) + + c.enqueue(packetDumpEntry{ + ntp: time.Now(), + data: append([]byte(nil), p...), + src: local, + dst: remote, + }) + } + + return n, err +} + +// LocalAddr implements net.PacketConn. +func (c *PacketConn) LocalAddr() net.Addr { return c.PacketConn.LocalAddr() } + +// SetDeadline implements net.PacketConn. +func (c *PacketConn) SetDeadline(t time.Time) error { return c.PacketConn.SetDeadline(t) } + +// SetReadDeadline implements net.PacketConn. +func (c *PacketConn) SetReadDeadline(t time.Time) error { return c.PacketConn.SetReadDeadline(t) } + +// SetWriteDeadline implements net.PacketConn. +func (c *PacketConn) SetWriteDeadline(t time.Time) error { return c.PacketConn.SetWriteDeadline(t) } + +// SetReadBuffer implements extendedPacketConn. +func (c *PacketConn) SetReadBuffer(bytes int) error { + return c.PacketConn.(extendedPacketConn).SetReadBuffer(bytes) +} + +// SyscallConn implements extendedPacketConn. +func (c *PacketConn) SyscallConn() (syscall.RawConn, error) { + return c.PacketConn.(extendedPacketConn).SyscallConn() +} diff --git a/internal/packetdumper/packet_conn_test.go b/internal/packetdumper/packet_conn_test.go new file mode 100644 index 00000000..df43baf5 --- /dev/null +++ b/internal/packetdumper/packet_conn_test.go @@ -0,0 +1,202 @@ +package packetdumper + +import ( + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// startUDPPair creates a pair of UDP connections and returns both ends. +func startUDPPair(t *testing.T) (client, server *net.UDPConn) { + t.Helper() + + serverAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0") + require.NoError(t, err) + + server, err = net.ListenUDP("udp", serverAddr) + require.NoError(t, err) + + clientAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0") + require.NoError(t, err) + + client, err = net.ListenUDP("udp", clientAddr) + require.NoError(t, err) + + t.Cleanup(func() { + server.Close() //nolint:errcheck + client.Close() //nolint:errcheck + }) + + return client, server +} + +func cleanupPcapngPacket(t *testing.T, prefix string) { + t.Helper() + + matches, err := filepath.Glob(prefix + "_*.pcapng") + require.NoError(t, err, "glob for pcapng files") + require.NotEmpty(t, matches, "expected at least one pcapng file to have been created") + + for _, f := range matches { + require.NoError(t, os.Remove(f), "removing pcapng file %s", f) + } +} + +func TestPacketConnInitialize_CreatesFile(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck +} + +func TestPacketConnWriteTo(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck + + n, err := c.WriteTo([]byte("hello world"), server.LocalAddr()) + require.NoError(t, err) + require.Equal(t, 11, n) + + buf := make([]byte, 32) + server.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + rn, _, err := server.ReadFromUDP(buf) + require.NoError(t, err) + require.Equal(t, []byte("hello world"), buf[:rn]) +} + +func TestPacketConnReadFrom(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck + + _, err := server.WriteTo([]byte("incoming data"), client.LocalAddr()) + require.NoError(t, err) + + buf := make([]byte, 32) + c.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + n, addr, err := c.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, []byte("incoming data"), buf[:n]) + require.NotNil(t, addr) +} + +func TestPacketConnMultipleWriteRead(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck + + serverAddr := server.LocalAddr() + for _, msg := range []string{"foo", "bar", "baz"} { + n, err := c.WriteTo([]byte(msg), serverAddr) + require.NoError(t, err) + require.Equal(t, len(msg), n) + } + + server.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + buf := make([]byte, 32) + received := make([]byte, 0, 9) + for range 3 { + n, _, err := server.ReadFromUDP(buf) + require.NoError(t, err) + received = append(received, buf[:n]...) + } + require.Equal(t, []byte("foobarbaz"), received) + + for _, msg := range []string{"abcde", "fghij"} { + _, err := server.WriteTo([]byte(msg), client.LocalAddr()) + require.NoError(t, err) + } + + c.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + readReceived := make([]byte, 0, 10) + for range 2 { + n, _, err := c.ReadFrom(buf) + require.NoError(t, err) + readReceived = append(readReceived, buf[:n]...) + } + require.Equal(t, []byte("abcdefghij"), readReceived) +} + +func TestPacketConnCloseIdempotent(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + + defer c.Close() //nolint:errcheck + defer c.Close() //nolint:errcheck +} + +func TestPacketConnDelegatesAddrMethods(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck + + require.Equal(t, client.LocalAddr(), c.LocalAddr()) + + require.NoError(t, c.SetDeadline(time.Now().Add(time.Second))) + require.NoError(t, c.SetReadDeadline(time.Now().Add(time.Second))) + require.NoError(t, c.SetWriteDeadline(time.Now().Add(time.Second))) +} + +func TestPacketConnReadFromRecordsSource(t *testing.T) { + client, server := startUDPPair(t) + defer server.Close() //nolint:errcheck + + prefix := filepath.Join(t.TempDir(), "capture") + c := &PacketConn{Prefix: prefix, PacketConn: client} + require.NoError(t, c.Initialize()) + + defer cleanupPcapngPacket(t, prefix) + defer c.Close() //nolint:errcheck + + _, err := server.WriteTo([]byte("ping"), client.LocalAddr()) + require.NoError(t, err) + + buf := make([]byte, 32) + c.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + n, addr, err := c.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, []byte("ping"), buf[:n]) + + // The reported source address should match the server's address. + require.Equal(t, server.LocalAddr().String(), addr.String()) +} diff --git a/internal/playback/server.go b/internal/playback/server.go index 15602b47..bdc7180c 100644 --- a/internal/playback/server.go +++ b/internal/playback/server.go @@ -22,6 +22,7 @@ type serverAuthManager interface { // Server is the playback server. type Server struct { Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -48,15 +49,17 @@ func (s *Server) Initialize() error { router.GET("/get", s.onGet) s.httpServer = &httpp.Server{ - Address: s.Address, - AllowOrigins: s.AllowOrigins, - ReadTimeout: time.Duration(s.ReadTimeout), - WriteTimeout: time.Duration(s.WriteTimeout), - Encryption: s.Encryption, - ServerCert: s.ServerCert, - ServerKey: s.ServerKey, - Handler: router, - Parent: s, + Address: s.Address, + AllowOrigins: s.AllowOrigins, + DumpPackets: s.DumpPackets, + DumpPacketsPrefix: "playback_server_conn", + ReadTimeout: time.Duration(s.ReadTimeout), + WriteTimeout: time.Duration(s.WriteTimeout), + Encryption: s.Encryption, + ServerCert: s.ServerCert, + ServerKey: s.ServerKey, + Handler: router, + Parent: s, } err := s.httpServer.Initialize() if err != nil { diff --git a/internal/pprof/pprof.go b/internal/pprof/pprof.go index f15142dd..12b03650 100644 --- a/internal/pprof/pprof.go +++ b/internal/pprof/pprof.go @@ -27,6 +27,7 @@ type pprofParent interface { // PPROF is a pprof exporter. type PPROF struct { Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -51,15 +52,17 @@ func (pp *PPROF) Initialize() error { pprof.Register(router) pp.httpServer = &httpp.Server{ - Address: pp.Address, - AllowOrigins: pp.AllowOrigins, - ReadTimeout: time.Duration(pp.ReadTimeout), - WriteTimeout: time.Duration(pp.WriteTimeout), - Encryption: pp.Encryption, - ServerCert: pp.ServerCert, - ServerKey: pp.ServerKey, - Handler: router, - Parent: pp, + Address: pp.Address, + DumpPackets: pp.DumpPackets, + AllowOrigins: pp.AllowOrigins, + DumpPacketsPrefix: "pprof_server_conn", + ReadTimeout: time.Duration(pp.ReadTimeout), + WriteTimeout: time.Duration(pp.WriteTimeout), + Encryption: pp.Encryption, + ServerCert: pp.ServerCert, + ServerKey: pp.ServerKey, + Handler: router, + Parent: pp, } err := pp.httpServer.Initialize() if err != nil { diff --git a/internal/protocols/httpp/server.go b/internal/protocols/httpp/server.go index c2305b98..85a2d0f4 100644 --- a/internal/protocols/httpp/server.go +++ b/internal/protocols/httpp/server.go @@ -13,6 +13,7 @@ import ( "github.com/bluenviron/mediamtx/internal/certloader" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/restrictnetwork" ) @@ -30,15 +31,17 @@ func (nilWriter) Write(p []byte) (int, error) { // - server header // - filtering of invalid requests type Server struct { - Address string - AllowOrigins []string - ReadTimeout time.Duration - WriteTimeout time.Duration - Encryption bool - ServerCert string - ServerKey string - Handler http.Handler - Parent logger.Writer + Address string + AllowOrigins []string + DumpPackets bool + DumpPacketsPrefix string + ReadTimeout time.Duration + WriteTimeout time.Duration + Encryption bool + ServerCert string + ServerKey string + Handler http.Handler + Parent logger.Writer ln net.Listener inner *http.Server @@ -96,6 +99,13 @@ func (s *Server) Initialize() error { return err } + if s.DumpPackets { + s.ln = &packetdumper.Listener{ + Prefix: s.DumpPacketsPrefix, + Listener: s.ln, + } + } + if network == "unix" { os.Chmod(address, 0o755) //nolint:errcheck } diff --git a/internal/protocols/udp/listen.go b/internal/protocols/udp/listen.go deleted file mode 100644 index 7d598bea..00000000 --- a/internal/protocols/udp/listen.go +++ /dev/null @@ -1,161 +0,0 @@ -// Package udp contains utilities to work with the UDP protocol. -package udp - -import ( - "fmt" - "net" - "net/url" - "syscall" - "time" - - "github.com/bluenviron/gortsplib/v5/pkg/multicast" - "github.com/bluenviron/gortsplib/v5/pkg/readbuffer" - "github.com/bluenviron/mediamtx/internal/restrictnetwork" -) - -type packetConn interface { - net.PacketConn - SetReadBuffer(bytes int) error - SyscallConn() (syscall.RawConn, error) -} - -type udpConn struct { - pc net.PacketConn - sourceIP net.IP -} - -func (r *udpConn) Close() error { - return r.pc.Close() -} - -func (r *udpConn) Read(p []byte) (int, error) { - for { - n, addr, err := r.pc.ReadFrom(p) - - if r.sourceIP != nil && addr != nil && !addr.(*net.UDPAddr).IP.Equal(r.sourceIP) { - continue - } - - return n, err - } -} - -func (r *udpConn) Write(_ []byte) (int, error) { - panic("unimplemented") -} - -func (r *udpConn) LocalAddr() net.Addr { - panic("unimplemented") -} - -func (r *udpConn) RemoteAddr() net.Addr { - panic("unimplemented") -} - -func (r *udpConn) SetDeadline(_ time.Time) error { - panic("unimplemented") -} - -func (r *udpConn) SetReadDeadline(t time.Time) error { - return r.pc.SetReadDeadline(t) -} - -func (r *udpConn) SetWriteDeadline(_ time.Time) error { - panic("unimplemented") -} - -func defaultInterfaceForMulticast(multicastAddr *net.UDPAddr) (*net.Interface, error) { - conn, err := net.Dial("udp4", multicastAddr.String()) - if err != nil { - return nil, err - } - localAddr := conn.LocalAddr().(*net.UDPAddr) - conn.Close() - - interfaces, err := net.Interfaces() - if err != nil { - return nil, err - } - - for _, iface := range interfaces { - var addrs []net.Addr - addrs, err = iface.Addrs() - if err != nil { - continue - } - - for _, addr := range addrs { - var ip net.IP - switch v := addr.(type) { - case *net.IPNet: - ip = v.IP - case *net.IPAddr: - ip = v.IP - } - - if ip != nil && ip.Equal(localAddr.IP) { - return &iface, nil - } - } - } - - return nil, fmt.Errorf("could not find any interface for using multicast address %s", multicastAddr) -} - -// Listen creates a UDP listener on the given URL. -func Listen(u *url.URL, udpReadBufferSize int) (net.Conn, error) { - q := u.Query() - var sourceIP net.IP - - if src := q.Get("source"); src != "" { - sourceIP = net.ParseIP(src) - if sourceIP == nil { - return nil, fmt.Errorf("invalid source IP") - } - } - - addr, err := net.ResolveUDPAddr("udp", u.Host) - if err != nil { - return nil, err - } - - var pc packetConn - - if ip4 := addr.IP.To4(); ip4 != nil && addr.IP.IsMulticast() { - var intf *net.Interface - - if intfName := q.Get("interface"); intfName != "" { - intf, err = net.InterfaceByName(intfName) - if err != nil { - return nil, err - } - } else { - intf, err = defaultInterfaceForMulticast(addr) - if err != nil { - return nil, err - } - } - - pc, err = multicast.NewSingleConn(intf, addr.String(), net.ListenPacket) - if err != nil { - return nil, err - } - } else { - var tmp net.PacketConn - tmp, err = net.ListenPacket(restrictnetwork.Restrict("udp", addr.String())) - if err != nil { - return nil, err - } - pc = tmp.(*net.UDPConn) - } - - if udpReadBufferSize != 0 { - err = readbuffer.SetReadBuffer(pc, udpReadBufferSize) - if err != nil { - pc.Close() - return nil, err - } - } - - return &udpConn{pc: pc, sourceIP: sourceIP}, nil -} diff --git a/internal/protocols/udp/listener.go b/internal/protocols/udp/listener.go new file mode 100644 index 00000000..3699e457 --- /dev/null +++ b/internal/protocols/udp/listener.go @@ -0,0 +1,174 @@ +// Package udp contains utilities to work with the UDP protocol. +package udp + +import ( + "fmt" + "net" + "syscall" + "time" + + "github.com/bluenviron/gortsplib/v5/pkg/multicast" + "github.com/bluenviron/gortsplib/v5/pkg/readbuffer" + "github.com/bluenviron/mediamtx/internal/restrictnetwork" +) + +type packetConn interface { + net.PacketConn + SetReadBuffer(bytes int) error + SyscallConn() (syscall.RawConn, error) +} + +func defaultInterfaceForMulticast(multicastAddr *net.UDPAddr) (*net.Interface, error) { + conn, err := net.Dial("udp4", multicastAddr.String()) + if err != nil { + return nil, err + } + localAddr := conn.LocalAddr().(*net.UDPAddr) + conn.Close() + + interfaces, err := net.Interfaces() + if err != nil { + return nil, err + } + + for _, iface := range interfaces { + var addrs []net.Addr + addrs, err = iface.Addrs() + if err != nil { + continue + } + + for _, addr := range addrs { + var ip net.IP + switch v := addr.(type) { + case *net.IPNet: + ip = v.IP + case *net.IPAddr: + ip = v.IP + } + + if ip != nil && ip.Equal(localAddr.IP) { + return &iface, nil + } + } + } + + return nil, fmt.Errorf("could not find any interface for using multicast address %s", multicastAddr) +} + +// Listener is a listener on a UDP socket. +type Listener struct { + Address string + Source string + IntfName string + UDPReadBufferSize int + ListenPacket func(network, address string) (net.PacketConn, error) + + pc packetConn + sourceIP net.IP +} + +// Initialize initializes the listener. +func (l *Listener) Initialize() error { + if l.ListenPacket == nil { + l.ListenPacket = net.ListenPacket + } + + if l.Source != "" { + l.sourceIP = net.ParseIP(l.Source) + if l.sourceIP == nil { + return fmt.Errorf("invalid source IP") + } + } + + addr, err := net.ResolveUDPAddr("udp", l.Address) + if err != nil { + return err + } + + if ip4 := addr.IP.To4(); ip4 != nil && addr.IP.IsMulticast() { + var intf *net.Interface + + if l.IntfName != "" { + intf, err = net.InterfaceByName(l.IntfName) + if err != nil { + return err + } + } else { + intf, err = defaultInterfaceForMulticast(addr) + if err != nil { + return err + } + } + + l.pc, err = multicast.NewSingleConn(intf, addr.String(), l.ListenPacket) + if err != nil { + return err + } + } else { + var tmp net.PacketConn + tmp, err = l.ListenPacket(restrictnetwork.Restrict("udp", addr.String())) + if err != nil { + return err + } + l.pc = tmp.(packetConn) + } + + if l.UDPReadBufferSize != 0 { + err = readbuffer.SetReadBuffer(l.pc, l.UDPReadBufferSize) + if err != nil { + l.pc.Close() + return err + } + } + + return nil +} + +// Close closes the listener. +func (l *Listener) Close() error { + return l.pc.Close() +} + +// Read implements net.Conn. +func (l *Listener) Read(p []byte) (int, error) { + for { + n, addr, err := l.pc.ReadFrom(p) + + if l.sourceIP != nil && addr != nil && !addr.(*net.UDPAddr).IP.Equal(l.sourceIP) { + continue + } + + return n, err + } +} + +// Write implements net.Conn. +func (l *Listener) Write(_ []byte) (int, error) { + panic("unimplemented") +} + +// LocalAddr implements net.Conn. +func (l *Listener) LocalAddr() net.Addr { + panic("unimplemented") +} + +// RemoteAddr implements net.Conn. +func (l *Listener) RemoteAddr() net.Addr { + panic("unimplemented") +} + +// SetDeadline implements net.Conn. +func (l *Listener) SetDeadline(_ time.Time) error { + panic("unimplemented") +} + +// SetReadDeadline implements net.Conn. +func (l *Listener) SetReadDeadline(t time.Time) error { + return l.pc.SetReadDeadline(t) +} + +// SetWriteDeadline implements net.Conn. +func (l *Listener) SetWriteDeadline(_ time.Time) error { + panic("unimplemented") +} diff --git a/internal/protocols/udp/listen_test.go b/internal/protocols/udp/listener_test.go similarity index 63% rename from internal/protocols/udp/listen_test.go rename to internal/protocols/udp/listener_test.go index 49802227..5bb1eca7 100644 --- a/internal/protocols/udp/listen_test.go +++ b/internal/protocols/udp/listener_test.go @@ -2,7 +2,6 @@ package udp import ( "net" - "net/url" "testing" "time" @@ -10,26 +9,30 @@ import ( ) func TestListen(t *testing.T) { - u, err := url.Parse("udp://127.0.0.1:0") + l := &Listener{ + Address: "127.0.0.1:0", + UDPReadBufferSize: 4096, + } + err := l.Initialize() require.NoError(t, err) - - conn, err := Listen(u, 4096) - require.NoError(t, err) - defer conn.Close() + defer l.Close() //nolint:errcheck done := make(chan struct{}) go func() { defer close(done) - buf := make([]byte, 1024) - conn.SetReadDeadline(time.Now().Add(2 * time.Second)) - n, err2 := conn.Read(buf) + err2 := l.SetReadDeadline(time.Now().Add(2 * time.Second)) require.NoError(t, err2) + + buf := make([]byte, 1024) + n, err2 := l.Read(buf) + require.NoError(t, err2) + require.Equal(t, []byte("testing"), buf[:n]) }() - localAddr := conn.(*udpConn).pc.LocalAddr().(*net.UDPAddr) + localAddr := l.pc.LocalAddr().(*net.UDPAddr) clientConn, err := net.DialUDP("udp", nil, localAddr) require.NoError(t, err) diff --git a/internal/protocols/udp/params.go b/internal/protocols/udp/params.go new file mode 100644 index 00000000..efe17782 --- /dev/null +++ b/internal/protocols/udp/params.go @@ -0,0 +1,19 @@ +package udp + +import "net/url" + +// Params are the parameters of a UDP listener. +type Params struct { + Address string + Source string + IntfName string +} + +// URLToParams converts a URL to Params. +func URLToParams(u *url.URL) *Params { + return &Params{ + Address: u.Host, + Source: u.Query().Get("source"), + IntfName: u.Query().Get("interface"), + } +} diff --git a/internal/protocols/unix/listen.go b/internal/protocols/unix/listen.go deleted file mode 100644 index d320ec52..00000000 --- a/internal/protocols/unix/listen.go +++ /dev/null @@ -1,138 +0,0 @@ -// Package unix contains utilities to work with Unix sockets. -package unix - -import ( - "fmt" - "net" - "net/url" - "os" - "sync" - "time" -) - -type unixConn struct { - l net.Listener - c net.Conn - mutex sync.Mutex - closed bool - deadline time.Time -} - -func (r *unixConn) Close() error { - r.mutex.Lock() - defer r.mutex.Unlock() - - r.closed = true - - r.l.Close() - - if r.c != nil { - r.c.Close() - } - - return nil -} - -func (r *unixConn) acceptWithDeadline() (net.Conn, error) { - done := make(chan struct{}) - defer func() { <-done }() - - terminate := make(chan struct{}) - defer close(terminate) - - go func() { - defer close(done) - select { - case <-time.After(time.Until(r.deadline)): - r.l.Close() - case <-terminate: - return - } - }() - - c, err := r.l.Accept() - if err != nil { - if time.Now().After(r.deadline) { - return nil, fmt.Errorf("deadline exceeded") - } - return nil, err - } - return c, nil -} - -func (r *unixConn) setConn(c net.Conn) error { - r.mutex.Lock() - defer r.mutex.Unlock() - - if r.closed { - return fmt.Errorf("closed") - } - - r.c = c - return nil -} - -func (r *unixConn) Read(p []byte) (int, error) { - if r.c == nil { - c, err := r.acceptWithDeadline() - if err != nil { - return 0, err - } - - err = r.setConn(c) - if err != nil { - return 0, err - } - } - - r.c.SetReadDeadline(r.deadline) - return r.c.Read(p) -} - -func (r *unixConn) Write(_ []byte) (int, error) { - panic("unimplemented") -} - -func (r *unixConn) LocalAddr() net.Addr { - panic("unimplemented") -} - -func (r *unixConn) RemoteAddr() net.Addr { - panic("unimplemented") -} - -func (r *unixConn) SetDeadline(_ time.Time) error { - panic("unimplemented") -} - -func (r *unixConn) SetReadDeadline(t time.Time) error { - r.deadline = t - return nil -} - -func (r *unixConn) SetWriteDeadline(_ time.Time) error { - panic("unimplemented") -} - -// Listen creates a Unix listener on the given URL. -func Listen(u *url.URL) (net.Conn, error) { - var pa string - if u.Path != "" { - pa = u.Path - } else { - pa = u.Host - } - - if pa == "" { - return nil, fmt.Errorf("invalid unix path") - } - - os.Remove(pa) - - socket, err := net.Listen("unix", pa) - if err != nil { - return nil, err - } - - return &unixConn{l: socket}, nil -} diff --git a/internal/protocols/unix/listener.go b/internal/protocols/unix/listener.go new file mode 100644 index 00000000..6963b901 --- /dev/null +++ b/internal/protocols/unix/listener.go @@ -0,0 +1,141 @@ +// Package unix contains utilities to work with Unix sockets. +package unix + +import ( + "fmt" + "net" + "os" + "sync" + "time" +) + +// Listener is a listener on a Unix socket. +type Listener struct { + Path string + + l net.Listener + c net.Conn + mutex sync.Mutex + closed bool + deadline time.Time +} + +// Initialize initializes the listener. +func (l *Listener) Initialize() error { + if l.Path == "" { + return fmt.Errorf("invalid unix path") + } + + os.Remove(l.Path) + + var err error + l.l, err = net.Listen("unix", l.Path) + if err != nil { + return err + } + + return nil +} + +// Close closes the listener. +func (l *Listener) Close() error { + l.mutex.Lock() + defer l.mutex.Unlock() + + l.closed = true + + l.l.Close() + + if l.c != nil { + l.c.Close() + } + + return nil +} + +func (l *Listener) acceptWithDeadline() (net.Conn, error) { + done := make(chan struct{}) + defer func() { <-done }() + + terminate := make(chan struct{}) + defer close(terminate) + + go func() { + defer close(done) + select { + case <-time.After(time.Until(l.deadline)): + l.l.Close() + case <-terminate: + return + } + }() + + c, err := l.l.Accept() + if err != nil { + if time.Now().After(l.deadline) { + return nil, fmt.Errorf("deadline exceeded") + } + return nil, err + } + return c, nil +} + +func (l *Listener) setConn(c net.Conn) error { + l.mutex.Lock() + defer l.mutex.Unlock() + + if l.closed { + return fmt.Errorf("closed") + } + + l.c = c + return nil +} + +func (l *Listener) Read(p []byte) (int, error) { + if l.c == nil { + c, err := l.acceptWithDeadline() + if err != nil { + return 0, err + } + + err = l.setConn(c) + if err != nil { + return 0, err + } + } + + l.c.SetReadDeadline(l.deadline) + return l.c.Read(p) +} + +// Write implements net.Conn. +func (l *Listener) Write(_ []byte) (int, error) { + panic("unimplemented") +} + +// LocalAddr implements net.Conn. +func (l *Listener) LocalAddr() net.Addr { + panic("unimplemented") +} + +// RemoteAddr implements net.Conn. +func (l *Listener) RemoteAddr() net.Addr { + panic("unimplemented") +} + +// SetDeadline implements net.Conn. +func (l *Listener) SetDeadline(_ time.Time) error { + panic("unimplemented") +} + +// SetReadDeadline implements net.Conn. +func (l *Listener) SetReadDeadline(t time.Time) error { + l.deadline = t + return nil +} + +// SetWriteDeadline implements net.Conn. +func (l *Listener) SetWriteDeadline(_ time.Time) error { + panic("unimplemented") +} diff --git a/internal/protocols/unix/listen_test.go b/internal/protocols/unix/listener_test.go similarity index 78% rename from internal/protocols/unix/listen_test.go rename to internal/protocols/unix/listener_test.go index b0c194ae..729742f1 100644 --- a/internal/protocols/unix/listen_test.go +++ b/internal/protocols/unix/listener_test.go @@ -2,7 +2,6 @@ package unix import ( "net" - "net/url" "os" "testing" "time" @@ -16,12 +15,12 @@ func TestListen(t *testing.T) { socket.Close() defer os.Remove(socket.Name()) - u, err := url.Parse("unix://" + socket.Name()) + l := &Listener{ + Path: socket.Name(), + } + err = l.Initialize() require.NoError(t, err) - - conn, err := Listen(u) - require.NoError(t, err) - defer conn.Close() + defer l.Close() //nolint:errcheck done := make(chan struct{}) @@ -29,8 +28,8 @@ func TestListen(t *testing.T) { defer close(done) buf := make([]byte, 1024) - conn.SetReadDeadline(time.Now().Add(2 * time.Second)) - n, err2 := conn.Read(buf) + l.SetReadDeadline(time.Now().Add(2 * time.Second)) //nolint:errcheck + n, err2 := l.Read(buf) require.NoError(t, err2) require.Equal(t, []byte("testing"), buf[:n]) }() diff --git a/internal/protocols/unix/params.go b/internal/protocols/unix/params.go new file mode 100644 index 00000000..0a67b83e --- /dev/null +++ b/internal/protocols/unix/params.go @@ -0,0 +1,22 @@ +package unix + +import "net/url" + +// Params are the parameters of a unix listener. +type Params struct { + Path string +} + +// URLToParams converts a URL to Params. +func URLToParams(u *url.URL) *Params { + var pa string + if u.Path != "" { + pa = u.Path + } else { + pa = u.Host + } + + return &Params{ + Path: pa, + } +} diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 07b7f1f7..fb48b80e 100644 --- a/internal/servers/hls/http_server.go +++ b/internal/servers/hls/http_server.go @@ -36,6 +36,7 @@ func mergePathAndQuery(path string, rawQuery string) string { type httpServer struct { address string + dumpPackets bool encryption bool serverKey string serverCert string @@ -58,15 +59,17 @@ func (s *httpServer) initialize() error { router.Use(s.onRequest) s.inner = &httpp.Server{ - Address: s.address, - AllowOrigins: s.allowOrigins, - ReadTimeout: time.Duration(s.readTimeout), - WriteTimeout: time.Duration(s.writeTimeout), - Encryption: s.encryption, - ServerCert: s.serverCert, - ServerKey: s.serverKey, - Handler: router, - Parent: s, + Address: s.address, + AllowOrigins: s.allowOrigins, + DumpPackets: s.dumpPackets, + DumpPacketsPrefix: "hls_server_conn", + ReadTimeout: time.Duration(s.readTimeout), + WriteTimeout: time.Duration(s.writeTimeout), + Encryption: s.encryption, + ServerCert: s.serverCert, + ServerKey: s.serverKey, + Handler: router, + Parent: s, } err := s.inner.Initialize() if err != nil { diff --git a/internal/servers/hls/server.go b/internal/servers/hls/server.go index 180ea1cb..fcbce238 100644 --- a/internal/servers/hls/server.go +++ b/internal/servers/hls/server.go @@ -71,6 +71,7 @@ type serverParent interface { // Server is a HLS server. type Server struct { Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -121,6 +122,7 @@ func (s *Server) Initialize() error { s.httpServer = &httpServer{ address: s.Address, + dumpPackets: s.DumpPackets, encryption: s.Encryption, serverKey: s.ServerKey, serverCert: s.ServerCert, diff --git a/internal/servers/rtmp/server.go b/internal/servers/rtmp/server.go index c024497a..8d45f65c 100644 --- a/internal/servers/rtmp/server.go +++ b/internal/servers/rtmp/server.go @@ -18,6 +18,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/restrictnetwork" "github.com/bluenviron/mediamtx/internal/stream" ) @@ -74,6 +75,7 @@ type serverParent interface { // Server is a RTMP server. type Server struct { Address string + DumpPackets bool ReadTimeout conf.Duration WriteTimeout conf.Duration IsTLS bool @@ -106,31 +108,36 @@ type Server struct { // Initialize initializes the server. func (s *Server) Initialize() error { - ln, err := func() (net.Listener, error) { - if !s.IsTLS { - return net.Listen(restrictnetwork.Restrict("tcp", s.Address)) - } + var err error + s.ln, err = net.Listen(restrictnetwork.Restrict("tcp", s.Address)) + if err != nil { + return err + } + if s.DumpPackets { + s.ln = &packetdumper.Listener{ + Prefix: "rtmp_server_conn", + Listener: s.ln, + } + } + + if s.IsTLS { s.loader = &certloader.CertLoader{ CertPath: s.ServerCert, KeyPath: s.ServerKey, Parent: s.Parent, } - err := s.loader.Initialize() + err = s.loader.Initialize() if err != nil { - return nil, err + s.ln.Close() + return err } - network, address := restrictnetwork.Restrict("tcp", s.Address) - return tls.Listen(network, address, &tls.Config{GetCertificate: s.loader.GetCertificate()}) - }() - if err != nil { - return err + s.ln = tls.NewListener(s.ln, &tls.Config{GetCertificate: s.loader.GetCertificate()}) } s.ctx, s.ctxCancel = context.WithCancel(context.Background()) - s.ln = ln s.conns = make(map[*conn]struct{}) s.chNewConn = make(chan net.Conn) s.chAcceptErr = make(chan error) diff --git a/internal/servers/rtsp/server.go b/internal/servers/rtsp/server.go index 159b34ab..f2124d46 100644 --- a/internal/servers/rtsp/server.go +++ b/internal/servers/rtsp/server.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "errors" "fmt" + "net" "reflect" "sort" "strings" @@ -23,6 +24,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/stream" ) @@ -90,6 +92,7 @@ type serverParent interface { type Server struct { Address string AuthMethods []auth.VerifyMethod + DumpPackets bool UDPReadBufferSize uint ReadTimeout conf.Duration WriteTimeout conf.Duration @@ -165,6 +168,18 @@ func (s *Server) Initialize() error { s.srv.TLSConfig = &tls.Config{GetCertificate: s.loader.GetCertificate()} } + if s.DumpPackets { + s.srv.Listen = (&packetdumper.Listen{ + Prefix: "rtsp_server_conn", + Listen: net.Listen, + }).Do + + s.srv.ListenPacket = (&packetdumper.ListenPacket{ + Prefix: "rtsp_server_packetconn", + ListenPacket: net.ListenPacket, + }).Do + } + err := s.srv.Start() if err != nil { return err diff --git a/internal/servers/webrtc/http_server.go b/internal/servers/webrtc/http_server.go index 7dbdaa20..2a7b198e 100644 --- a/internal/servers/webrtc/http_server.go +++ b/internal/servers/webrtc/http_server.go @@ -74,6 +74,7 @@ func sessionLocation(publish bool, path string, rawQuery string, secret uuid.UUI type httpServer struct { address string + dumpPackets bool encryption bool serverKey string serverCert string @@ -96,15 +97,17 @@ func (s *httpServer) initialize() error { router.Use(s.onRequest) s.inner = &httpp.Server{ - Address: s.address, - AllowOrigins: s.allowOrigins, - ReadTimeout: time.Duration(s.readTimeout), - WriteTimeout: time.Duration(s.writeTimeout), - Encryption: s.encryption, - ServerCert: s.serverCert, - ServerKey: s.serverKey, - Handler: router, - Parent: s, + Address: s.address, + AllowOrigins: s.allowOrigins, + DumpPackets: s.dumpPackets, + DumpPacketsPrefix: "webrtc_server_conn", + ReadTimeout: time.Duration(s.readTimeout), + WriteTimeout: time.Duration(s.writeTimeout), + Encryption: s.encryption, + ServerCert: s.serverCert, + ServerKey: s.serverKey, + Handler: router, + Parent: s, } err := s.inner.Initialize() if err != nil { diff --git a/internal/servers/webrtc/server.go b/internal/servers/webrtc/server.go index ae07a26d..979d9260 100644 --- a/internal/servers/webrtc/server.go +++ b/internal/servers/webrtc/server.go @@ -187,6 +187,7 @@ type serverParent interface { // Server is a WebRTC server. type Server struct { Address string + DumpPackets bool Encryption bool ServerKey string ServerCert string @@ -251,6 +252,7 @@ func (s *Server) Initialize() error { s.httpServer = &httpServer{ address: s.Address, + dumpPackets: s.DumpPackets, encryption: s.Encryption, serverKey: s.ServerKey, serverCert: s.ServerCert, diff --git a/internal/staticsources/handler.go b/internal/staticsources/handler.go index 1d436575..64600f2b 100644 --- a/internal/staticsources/handler.go +++ b/internal/staticsources/handler.go @@ -64,6 +64,7 @@ type handlerParent interface { type Handler struct { Conf *conf.Path LogLevel conf.LogLevel + DumpPackets bool ReadTimeout conf.Duration WriteTimeout conf.Duration WriteQueueSize int @@ -102,6 +103,7 @@ func (s *Handler) Initialize() { strings.HasPrefix(s.Conf.Source, "rtsp+ws://") || strings.HasPrefix(s.Conf.Source, "rtsps+ws://"): s.instance = &ssrtsp.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, WriteTimeout: s.WriteTimeout, WriteQueueSize: s.WriteQueueSize, @@ -112,6 +114,7 @@ func (s *Handler) Initialize() { case strings.HasPrefix(s.Conf.Source, "rtmp://") || strings.HasPrefix(s.Conf.Source, "rtmps://"): s.instance = &ssrtmp.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, WriteTimeout: s.WriteTimeout, Parent: s, @@ -120,6 +123,7 @@ func (s *Handler) Initialize() { case strings.HasPrefix(s.Conf.Source, "http://") || strings.HasPrefix(s.Conf.Source, "https://"): s.instance = &sshls.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, Parent: s, } @@ -128,6 +132,7 @@ func (s *Handler) Initialize() { strings.HasPrefix(s.Conf.Source, "udp+mpegts://") || strings.HasPrefix(s.Conf.Source, "unix+mpegts://"): s.instance = &ssmpegts.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, UDPReadBufferSize: s.UDPReadBufferSize, Parent: s, @@ -142,6 +147,7 @@ func (s *Handler) Initialize() { case strings.HasPrefix(s.Conf.Source, "whep://") || strings.HasPrefix(s.Conf.Source, "wheps://"): s.instance = &sswebrtc.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, UDPReadBufferSize: s.UDPReadBufferSize, Parent: s, @@ -150,6 +156,7 @@ func (s *Handler) Initialize() { case strings.HasPrefix(s.Conf.Source, "udp+rtp://") || strings.HasPrefix(s.Conf.Source, "unix+rtp://"): s.instance = &ssrtp.Source{ + DumpPackets: s.DumpPackets, ReadTimeout: s.ReadTimeout, UDPReadBufferSize: s.UDPReadBufferSize, Parent: s, diff --git a/internal/staticsources/hls/source.go b/internal/staticsources/hls/source.go index cde35ab9..b7617fbc 100644 --- a/internal/staticsources/hls/source.go +++ b/internal/staticsources/hls/source.go @@ -2,6 +2,7 @@ package hls import ( + "net" "net/http" "net/http/cookiejar" "net/url" @@ -14,6 +15,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/errordumper" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/stream" @@ -27,6 +29,7 @@ type parent interface { // Source is a HLS static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration Parent parent } @@ -64,7 +67,17 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { return err } + dialContext := (&net.Dialer{}).DialContext + + if s.DumpPackets { + dialContext = (&packetdumper.DialContext{ + Prefix: "hls_source_conn", + DialContext: dialContext, + }).Do + } + tr := &http.Transport{ + DialContext: dialContext, TLSClientConfig: tls.MakeConfig(u.Hostname(), params.Conf.SourceFingerprint), } defer tr.CloseIdleConnections() diff --git a/internal/staticsources/mpegts/source.go b/internal/staticsources/mpegts/source.go index 5181204e..a17b4312 100644 --- a/internal/staticsources/mpegts/source.go +++ b/internal/staticsources/mpegts/source.go @@ -13,6 +13,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/errordumper" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/mpegts" "github.com/bluenviron/mediamtx/internal/protocols/udp" "github.com/bluenviron/mediamtx/internal/protocols/unix" @@ -27,6 +28,7 @@ type parent interface { // Source is a MPEG-TS static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration UDPReadBufferSize uint Parent parent @@ -50,10 +52,15 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { switch u.Scheme { case "unix+mpegts": - nc, err = unix.Listen(u) + params := unix.URLToParams(u) + l := &unix.Listener{ + Path: params.Path, + } + err = l.Initialize() if err != nil { return err } + nc = l default: udpReadBufferSize := s.UDPReadBufferSize @@ -61,10 +68,41 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { udpReadBufferSize = *params.Conf.MPEGTSUDPReadBufferSize } - nc, err = udp.Listen(u, int(udpReadBufferSize)) + listenPacket := net.ListenPacket + + if s.DumpPackets { + listenPacket = func(network, address string) (net.PacketConn, error) { + pc, err2 := net.ListenPacket(network, address) + if err2 != nil { + return nil, err2 + } + + d := &packetdumper.PacketConn{ + Prefix: "mpegts_source_packetconn", + PacketConn: pc, + } + err2 = d.Initialize() + if err2 != nil { + return nil, err2 + } + + return d, nil + } + } + + params := udp.URLToParams(u) + l := &udp.Listener{ + Address: params.Address, + Source: params.Source, + IntfName: params.IntfName, + UDPReadBufferSize: int(udpReadBufferSize), + ListenPacket: listenPacket, + } + err = l.Initialize() if err != nil { return err } + nc = l } readerErr := make(chan error) diff --git a/internal/staticsources/rtmp/source.go b/internal/staticsources/rtmp/source.go index ad5c7ae7..082fc5e2 100644 --- a/internal/staticsources/rtmp/source.go +++ b/internal/staticsources/rtmp/source.go @@ -14,6 +14,7 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/rtmp" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/stream" @@ -27,6 +28,7 @@ type parent interface { // Source is a RTMP static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration WriteTimeout conf.Duration Parent parent @@ -56,11 +58,22 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { } } + dialContext := (&net.Dialer{}).DialContext + + if s.DumpPackets { + dialContext = (&packetdumper.DialContext{ + Prefix: "rtmp_source_conn", + DialContext: dialContext, + }).Do + } + connectCtx, connectCtxCancel := context.WithTimeout(params.Context, time.Duration(s.ReadTimeout)) + conn := &gortmplib.Client{ - URL: u, - TLSConfig: tls.MakeConfig(u.Hostname(), params.Conf.SourceFingerprint), - Publish: false, + URL: u, + TLSConfig: tls.MakeConfig(u.Hostname(), params.Conf.SourceFingerprint), + Publish: false, + DialContext: dialContext, } err = conn.Initialize(connectCtx) connectCtxCancel() diff --git a/internal/staticsources/rtp/source.go b/internal/staticsources/rtp/source.go index f55f3731..95ac6d3c 100644 --- a/internal/staticsources/rtp/source.go +++ b/internal/staticsources/rtp/source.go @@ -15,6 +15,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/errordumper" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/udp" "github.com/bluenviron/mediamtx/internal/protocols/unix" "github.com/bluenviron/mediamtx/internal/stream" @@ -30,6 +31,7 @@ type parent interface { // Source is a RTP static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration UDPReadBufferSize uint Parent parent @@ -65,10 +67,15 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { switch u.Scheme { case "unix+rtp": - nc, err = unix.Listen(u) + params := unix.URLToParams(u) + l := &unix.Listener{ + Path: params.Path, + } + err = l.Initialize() if err != nil { return err } + nc = l default: udpReadBufferSize := s.UDPReadBufferSize @@ -76,10 +83,41 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { udpReadBufferSize = *params.Conf.RTPUDPReadBufferSize } - nc, err = udp.Listen(u, int(udpReadBufferSize)) + listenPacket := net.ListenPacket + + if s.DumpPackets { + listenPacket = func(network, address string) (net.PacketConn, error) { + pc, err2 := net.ListenPacket(network, address) + if err2 != nil { + return nil, err2 + } + + d := &packetdumper.PacketConn{ + Prefix: "rtp_source_packetconn", + PacketConn: pc, + } + err2 = d.Initialize() + if err2 != nil { + return nil, err2 + } + + return d, nil + } + } + + params := udp.URLToParams(u) + l := &udp.Listener{ + Address: params.Address, + Source: params.Source, + IntfName: params.IntfName, + UDPReadBufferSize: int(udpReadBufferSize), + ListenPacket: listenPacket, + } + err = l.Initialize() if err != nil { return err } + nc = l } readerErr := make(chan error) diff --git a/internal/staticsources/rtsp/source.go b/internal/staticsources/rtsp/source.go index cd2fa1e4..d8b09a34 100644 --- a/internal/staticsources/rtsp/source.go +++ b/internal/staticsources/rtsp/source.go @@ -3,6 +3,7 @@ package rtsp import ( "fmt" + "net" "net/url" "regexp" "time" @@ -17,6 +18,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/errordumper" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/rtsp" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/stream" @@ -75,6 +77,7 @@ type parent interface { // Source is a RTSP static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration WriteTimeout conf.Duration WriteQueueSize int @@ -132,36 +135,20 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { scheme = "rtsps" } - var tunnel gortsplib.Tunnel - switch u0.Scheme { - case "rtsp+http", "rtsps+http": - tunnel = gortsplib.TunnelHTTP - case "rtsp+ws", "rtsps+ws": - tunnel = gortsplib.TunnelWebSocket - default: - tunnel = gortsplib.TunnelNone - } - u, err := base.ParseURL(regexp.MustCompile("^.*?://").ReplaceAllString(params.ResolvedSource, "rtsp://")) if err != nil { return err } - udpReadBufferSize := s.UDPReadBufferSize - if params.Conf.RTSPUDPReadBufferSize != nil { - udpReadBufferSize = *params.Conf.RTSPUDPReadBufferSize - } - c := &gortsplib.Client{ Scheme: scheme, Host: u.Host, - Tunnel: tunnel, Protocol: params.Conf.RTSPTransport.Protocol, TLSConfig: tls.MakeConfig(u.Hostname(), params.Conf.SourceFingerprint), ReadTimeout: time.Duration(s.ReadTimeout), WriteTimeout: time.Duration(s.WriteTimeout), + UDPReadBufferSize: int(s.UDPReadBufferSize), WriteQueueSize: s.WriteQueueSize, - UDPReadBufferSize: int(udpReadBufferSize), AnyPortEnable: params.Conf.RTSPAnyPort, UDPSourcePortRange: [2]uint16{ uint16(params.Conf.RTSPUDPSourcePortRange[0]), @@ -184,6 +171,29 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { }, } + switch u0.Scheme { + case "rtsp+http", "rtsps+http": + c.Tunnel = gortsplib.TunnelHTTP + case "rtsp+ws", "rtsps+ws": + c.Tunnel = gortsplib.TunnelWebSocket + } + + if params.Conf.RTSPUDPReadBufferSize != nil { + s.UDPReadBufferSize = *params.Conf.RTSPUDPReadBufferSize + } + + if s.DumpPackets { + c.DialContext = (&packetdumper.DialContext{ + Prefix: "rtsp_source_conn", + DialContext: (&net.Dialer{}).DialContext, + }).Do + + c.ListenPacket = (&packetdumper.ListenPacket{ + Prefix: "rtsp_source_packetconn", + ListenPacket: net.ListenPacket, + }).Do + } + err = c.Start() if err != nil { return err diff --git a/internal/staticsources/webrtc/source.go b/internal/staticsources/webrtc/source.go index c53e3ca3..2344f17c 100644 --- a/internal/staticsources/webrtc/source.go +++ b/internal/staticsources/webrtc/source.go @@ -3,6 +3,7 @@ package webrtc import ( "fmt" + "net" "net/http" "net/url" "strings" @@ -13,6 +14,7 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/packetdumper" "github.com/bluenviron/mediamtx/internal/protocols/tls" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" "github.com/bluenviron/mediamtx/internal/protocols/whip" @@ -27,6 +29,7 @@ type parent interface { // Source is a WebRTC static source. type Source struct { + DumpPackets bool ReadTimeout conf.Duration UDPReadBufferSize uint Parent parent @@ -48,7 +51,17 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { u.Scheme = strings.ReplaceAll(u.Scheme, "whep", "http") + dialContext := (&net.Dialer{}).DialContext + + if s.DumpPackets { + dialContext = (&packetdumper.DialContext{ + Prefix: "webrtc_source_conn", + DialContext: dialContext, + }).Do + } + tr := &http.Transport{ + DialContext: dialContext, TLSClientConfig: tls.MakeConfig(u.Hostname(), params.Conf.SourceFingerprint), } defer tr.CloseIdleConnections() diff --git a/mediamtx.yml b/mediamtx.yml index 81ce37be..36c3e3f1 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -16,6 +16,8 @@ logStructured: false logFile: mediamtx.log # When "syslog" is in logDestinations, use prefix for logs. sysLogPrefix: mediamtx +# Dump packets to disk. This is useful for debugging. +dumpPackets: false # Timeout of read operations. readTimeout: 10s