Add UniFi Protect Opus talkback source

README.md

@@ -199,6 +199,7 @@ A summary table of all modules and features can be found [here](internal/README.
 - [`tapo`](internal/tapo/README.md) - [TP-Link Tapo](https://www.tapo.com/) cameras with two-way audio support.
 - [`vigi`](internal/tapo/README.md#tp-link-vigi) - TP-Link Vigi cameras.
 - [`tuya`](internal/tuya/README.md) - [Tuya](https://www.tuya.com/) ecosystem cameras with two-way audio support.
+- [`unifi`](internal/unifi/README.md) - UniFi Protect cameras with Opus talkback support.
 - [`webtorrent`](internal/webtorrent/README.md) - Stream from another go2rtc via [WebTorrent](https://en.wikipedia.org/wiki/WebTorrent) protocol.
 - [`wyze`](internal/wyze/README.md) - [Wyze](https://wyze.com/) cameras using native P2P protocol
 - [`xiaomi`](internal/xiaomi/README.md) - [Xiaomi Mi Home](https://home.mi.com/) ecosystem cameras with two-way audio support.
@@ -282,6 +283,7 @@ Supported for:
 [`rtsp`](internal/rtsp/README.md#two-way-audio), 
 [`tapo`](internal/tapo/README.md), 
 [`tuya`](internal/tuya/README.md), 
+[`unifi`](internal/unifi/README.md), 
 [`webrtc`](internal/webrtc/README.md), 
 [`wyze`](internal/wyze/README.md), 
 [`xiaomi`](internal/xiaomi/README.md).

internal/README.md

@@ -59,6 +59,7 @@ Some formats and protocols go2rtc supports exclusively. They have no equivalent
 | [`rtsp`]       | `rtsp`          | `rtsp`           | yes   | yes    | yes    | yes     |
 | [`tapo`]       | `mpegts`        | `http`           | yes   |        |        | yes     |
 | [`tuya`]       | `srtp`          | `webrtc`         | yes   |        |        | yes     |
+| [`unifi`]      | `rtp`           | `http`           |       |        |        | yes     |
 | [`v4l2`]       | `rawvideo`      | `ioctl`          | yes   |        |        |         |
 | [`webrtc`]     | `srtp`          | `webrtc`         | yes   | yes    | yes    | yes     |
 | [`webtorrent`] | `srtp`          | `webrtc`         | yes   | yes    |        |         |
@@ -104,6 +105,7 @@ Some formats and protocols go2rtc supports exclusively. They have no equivalent
 [`streams`]: streams/README.md
 [`tapo`]: tapo/README.md
 [`tuya`]: tuya/README.md
+[`unifi`]: unifi/README.md
 [`v4l2`]: v4l2/README.md
 [`webrtc`]: webrtc/README.md
 [`webtorrent`]: webtorrent/README.md

internal/app/README.md

@@ -94,4 +94,4 @@ log:
   api: trace   # module name: log level
 ```
 
-Modules: `api`, `streams`, `rtsp`, `webrtc`, `mp4`, `hls`, `mjpeg`, `hass`, `homekit`, `onvif`, `rtmp`, `webtorrent`, `wyoming`, `echo`, `exec`, `expr`, `ffmpeg`, `wyze`, `xiaomi`.
+Modules: `api`, `streams`, `rtsp`, `webrtc`, `mp4`, `hls`, `mjpeg`, `hass`, `homekit`, `onvif`, `rtmp`, `webtorrent`, `wyoming`, `echo`, `exec`, `expr`, `ffmpeg`, `unifi`, `wyze`, `xiaomi`.

internal/streams/producer.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/creds"
 )
 
 type state byte
@@ -135,7 +136,7 @@ func (p *Producer) MarshalJSON() ([]byte, error) {
 	if conn := p.conn; conn != nil {
 		return json.Marshal(conn)
 	}
-	info := map[string]string{"url": p.url}
+	info := map[string]string{"url": creds.SecretString(p.url)}
 	return json.Marshal(info)
 }
 
@@ -149,7 +150,7 @@ func (p *Producer) start() {
 		return
 	}
 
-	log.Debug().Msgf("[streams] start producer url=%s", p.url)
+	log.Debug().Msgf("[streams] start producer url=%s", creds.SecretString(p.url))
 
 	p.state = stateStart
 	p.workerID++
@@ -167,7 +168,7 @@ func (p *Producer) worker(conn core.Producer, workerID int) {
 			return
 		}
 
-		log.Warn().Err(err).Str("url", p.url).Caller().Send()
+		log.Warn().Err(err).Str("url", creds.SecretString(p.url)).Caller().Send()
 	}
 
 	p.reconnect(workerID, 0)
@@ -178,11 +179,11 @@ func (p *Producer) reconnect(workerID, retry int) {
 	defer p.mu.Unlock()
 
 	if p.workerID != workerID {
-		log.Trace().Msgf("[streams] stop reconnect url=%s", p.url)
+		log.Trace().Msgf("[streams] stop reconnect url=%s", creds.SecretString(p.url))
 		return
 	}
 
-	log.Debug().Msgf("[streams] retry=%d to url=%s", retry, p.url)
+	log.Debug().Msgf("[streams] retry=%d to url=%s", retry, creds.SecretString(p.url))
 
 	conn, err := GetProducer(p.url)
 	if err != nil {
@@ -257,7 +258,7 @@ func (p *Producer) stop() {
 		p.workerID++
 	}
 
-	log.Debug().Msgf("[streams] stop producer url=%s", p.url)
+	log.Debug().Msgf("[streams] stop producer url=%s", creds.SecretString(p.url))
 
 	if p.conn != nil {
 		_ = p.conn.Stop()

internal/unifi/README.md

@@ -0,0 +1,24 @@
+# UniFi Protect
+
+UniFi Protect talkback is supported as a native go2rtc audio backchannel.
+
+## Configuration
+
+```yaml
+streams:
+  gate:
+    - rtspx://192.168.1.1:7441/<alias>#backchannel=0
+    - ffmpeg:gate#audio=opus
+    - unifi-talkback:https://192.168.1.1?camera_id=<id>&api_key=<secret>
+```
+
+The `unifi-talkback:` source uses the UniFi Protect public API:
+
+- `GET /proxy/protect/integration/v1/cameras/{camera_id}` checks whether the camera has a speaker
+- `POST /proxy/protect/integration/v1/cameras/{camera_id}/talkback-session` starts a talkback session
+
+Normal `video+audio` viewing does not activate talkback. go2rtc opens the UniFi talkback session only after a client sends microphone RTP into the audio backchannel.
+
+Push-to-talk clients should start sending microphone audio when talk begins and stop or disconnect that microphone send session when talk ends. go2rtc closes the FFmpeg RTP output when the backchannel producer stops.
+
+Only UniFi talkback sessions that report `codec: "opus"` are supported.

internal/unifi/unifi.go

@@ -0,0 +1,22 @@
+package unifi
+
+import (
+	"github.com/AlexxIT/go2rtc/internal/app"
+	"github.com/AlexxIT/go2rtc/internal/streams"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/unifi"
+)
+
+func Init() {
+	unifi.SetLogger(app.GetLogger("unifi"))
+
+	streams.HandleFunc(unifi.SchemeTalkback, func(source string) (core.Producer, error) {
+		return unifi.DialTalkback(source)
+	})
+
+	for _, sources := range streams.GetAllSources() {
+		for _, source := range sources {
+			unifi.RegisterTalkbackSecrets(source)
+		}
+	}
+}

main.go

@@ -41,6 +41,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/streams"
 	"github.com/AlexxIT/go2rtc/internal/tapo"
 	"github.com/AlexxIT/go2rtc/internal/tuya"
+	"github.com/AlexxIT/go2rtc/internal/unifi"
 	"github.com/AlexxIT/go2rtc/internal/v4l2"
 	"github.com/AlexxIT/go2rtc/internal/webrtc"
 	"github.com/AlexxIT/go2rtc/internal/webtorrent"
@@ -105,6 +106,7 @@ func main() {
 		{"roborock", roborock.Init},
 		{"tapo", tapo.Init},
 		{"tuya", tuya.Init},
+		{"unifi", unifi.Init},
 		{"wyze", wyze.Init},
 		{"xiaomi", xiaomi.Init},
 		{"yandex", yandex.Init},

pkg/README.md

@@ -42,6 +42,7 @@ Some formats and protocols go2rtc supports exclusively. They have no equivalent
 | Net (priv) | roborock     | webrtc          |         | h264, opus                      | opus                | `roborock:`   |
 | Net (priv) | tapo         | http            |         | h264, pcma                      | pcm_alaw            | `tapo:`       |
 | Net (priv) | tuya         | webrtc          |         |                                 |                     | `tuya:`       |
+| Net (priv) | unifi        | http, rtp       |         |                                 | opus                | `unifi-talkback:` |
 | Net (priv) | vigi         | http            |         |                                 |                     | `vigi:`       |
 | Net (priv) | webtorrent   | webrtc          | TODO    | TODO                            | TODO                | `webtorrent:` |
 | Net (priv) | xiaomi*      | cs2, tutk       |         |                                 |                     | `xiaomi:`     |

pkg/creds/secrets.go

@@ -35,8 +35,13 @@ func getReplacer() *strings.Replacer {
 	defer secretsMu.Unlock()
 
 	if secretsReplacer == nil {
-		oldnew := make([]string, 0, 2*len(secrets))
-		for _, s := range secrets {
+		values := slices.Clone(secrets)
+		slices.SortFunc(values, func(a, b string) int {
+			return len(b) - len(a)
+		})
+
+		oldnew := make([]string, 0, 2*len(values))
+		for _, s := range values {
 			oldnew = append(oldnew, s, "***")
 		}
 		secretsReplacer = strings.NewReplacer(oldnew...)
@@ -59,16 +64,26 @@ const (
 
 func SecretString(s string) string {
 	re := getReplacer()
-	s = userinfoRegexp.ReplaceAllString(s, `://***@`)
+	s = secretUserinfo(s)
 	return re.Replace(s)
 }
 
 func SecretWrite(w io.Writer, s string) (n int, err error) {
 	re := getReplacer()
-	s = userinfoRegexp.ReplaceAllString(s, `://***@`)
+	s = secretUserinfo(s)
 	return re.WriteString(w, s)
 }
 
+func secretUserinfo(s string) string {
+	return userinfoRegexp.ReplaceAllStringFunc(s, func(match string) string {
+		userinfo := match[3 : len(match)-1]
+		if strings.Contains(userinfo, ":") {
+			return `://***:***@`
+		}
+		return `://***@`
+	})
+}
+
 func SecretWriter(w io.Writer) io.Writer {
 	return &secretWriter{w}
 }

pkg/creds/secrets_test.go

@@ -13,3 +13,11 @@ func TestString(t *testing.T) {
 	s := SecretString("rtsp://admin:pa$$word@192.168.1.123/stream1")
 	require.Equal(t, "rtsp://***:***@192.168.1.123/stream1", s)
 }
+
+func TestStringOverlappingSecrets(t *testing.T) {
+	AddSecret("rtp://127.0.0.1:6500")
+	AddSecret("rtp://127.0.0.1:6500/talkback?token=secret")
+
+	s := SecretString(`{"url":"rtp://127.0.0.1:6500/talkback?token=secret"}`)
+	require.Equal(t, `{"url":"***"}`, s)
+}