Merge commit from fork

fix(cloud): strip Hub credential headers on cross-host redirect

Dockerfile

@@ -60,7 +60,7 @@ RUN cp -r /agent ./
 
 RUN /dist/agent/main version
 
-FROM node:18.14.0-alpine3.16 AS build-ui
+FROM node:22-alpine AS build-ui
 
 RUN apk update && apk upgrade --available && sync
 

Dockerfile.arm64

@@ -60,7 +60,7 @@ RUN cp -r /agent ./
 
 RUN /dist/agent/main version
 
-FROM node:18.14.0-alpine3.16 AS build-ui
+FROM node:22-alpine AS build-ui
 
 RUN apk update && apk upgrade --available && sync
 

machinery/src/cloud/kerberos_hub.go

@@ -68,9 +68,9 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 		tr := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
-		client = &http.Client{Transport: tr}
+		client = &http.Client{Transport: tr, CheckRedirect: stripHubCredentialsOnCrossHostRedirect}
 	} else {
-		client = &http.Client{}
+		client = &http.Client{CheckRedirect: stripHubCredentialsOnCrossHostRedirect}
 	}
 
 	resp, err := client.Do(req)
@@ -129,3 +129,20 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	log.Log.Info(errorMessage)
 	return false, true, errors.New(errorMessage)
 }
+
+// stripHubCredentialsOnCrossHostRedirect removes the custom Kerberos Hub
+// credential headers on a redirect that crosses to a different host. net/http
+// already strips the standard sensitive headers (Authorization, Cookie,
+// WWW-Authenticate) on a cross-host redirect, but it does NOT strip
+// custom-named headers, so without this the Hub private/public keys would be
+// forwarded to any host the configured HubURI redirects to.
+func stripHubCredentialsOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) == 0 {
+		return nil
+	}
+	if req.URL.Host != via[0].URL.Host {
+		req.Header.Del("X-Kerberos-Hub-PrivateKey")
+		req.Header.Del("X-Kerberos-Hub-PublicKey")
+	}
+	return nil
+}

machinery/src/routers/mqtt/main.go

@@ -11,6 +11,7 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -170,11 +171,45 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 	return nil
 }
 
-// maxSignalingAge is the maximum age of a WebRTC signaling message (request-hd-stream,
-// receive-hd-candidates) before it is considered stale and discarded. With CleanSession=false
-// the MQTT broker may replay queued messages from previous sessions; this prevents the agent
-// from setting up peer connections for viewers that are no longer waiting.
-const maxSignalingAge = 30 * time.Second
+// recentHDSessions tracks recently-seen WebRTC viewer session IDs so we can
+// dedupe duplicate request-hd-stream messages without relying on the broker's
+// (and the viewer's) wall clock. The viewer's offer-republish loop can fire
+// the same request several times for the same session_id while waiting for an
+// answer; the broker can also redeliver a message after a reconnect with
+// CleanSession=false. In both cases we want to handle the session exactly
+// once.
+//
+// Entries expire after recentHDSessionTTL. The map is small (one entry per
+// active viewer over the TTL window) so a periodic sweep is sufficient.
+const recentHDSessionTTL = 60 * time.Second
+
+var (
+	recentHDSessionsMu sync.Mutex
+	recentHDSessions   = make(map[string]time.Time)
+)
+
+// markHDSessionSeen returns true if this session_id was already processed
+// within the TTL window (i.e. this message should be treated as a duplicate).
+// It also opportunistically prunes expired entries.
+func markHDSessionSeen(sessionID string) bool {
+	if sessionID == "" {
+		return false
+	}
+	recentHDSessionsMu.Lock()
+	defer recentHDSessionsMu.Unlock()
+	now := time.Now()
+	// Lazy GC — cheap given the expected map size.
+	for k, t := range recentHDSessions {
+		if now.Sub(t) > recentHDSessionTTL {
+			delete(recentHDSessions, k)
+		}
+	}
+	if _, exists := recentHDSessions[sessionID]; exists {
+		return true
+	}
+	recentHDSessions[sessionID] = now
+	return false
+}
 
 func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory string, configuration *models.Configuration, communication *models.Communication) {
 	if hubKey == "" {
@@ -282,16 +317,13 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 				// We'll find out which message we received, and act accordingly.
 				log.Log.Info("routers.mqtt.main.MQTTListenerHandler(): received message with action: " + payload.Action)
 
-				// For time-sensitive WebRTC signaling messages, discard stale ones that may
-				// have been queued by the broker while CleanSession=false.
-				if payload.Action == "request-hd-stream" || payload.Action == "receive-hd-candidates" {
-					messageAge := time.Since(time.Unix(message.Timestamp, 0))
-					if messageAge > maxSignalingAge {
-						log.Log.Info("routers.mqtt.main.MQTTListenerHandler(): discarding stale " + payload.Action +
-							" message (age: " + messageAge.Round(time.Second).String() + ")")
-						return
-					}
-				}
+				// NOTE: We intentionally do NOT discard request-hd-stream /
+				// receive-hd-candidates messages based on a wall-clock age. The
+				// viewer and agent clocks can drift (especially on embedded
+				// devices), which previously caused valid requests to be
+				// silently dropped and forced the user to refresh the page.
+				// Duplicate handling for request-hd-stream is done by session_id
+				// inside HandleRequestHDStream (see markHDSessionSeen).
 
 				switch payload.Action {
 				case "record":
@@ -536,6 +568,15 @@ func HandleRequestHDStream(mqttClient mqtt.Client, hubKey string, payload models
 
 	if requestHDStreamPayload.Timestamp != 0 {
 		if communication.CameraConnected {
+			// Dedupe by session_id: the viewer republishes its offer while
+			// waiting for an answer (and the broker may redeliver), and we
+			// don't want to spawn multiple peer connections for the same
+			// browser session.
+			if markHDSessionSeen(requestHDStreamPayload.SessionID) {
+				log.Log.Info("routers.mqtt.main.HandleRequestHDStream(): duplicate request for session " +
+					requestHDStreamPayload.SessionID + ", ignoring")
+				return
+			}
 			// Set the Hub key, so we can send back the answer.
 			requestHDStreamPayload.HubKey = hubKey
 			if communication.HandleLiveHDHandshake == nil {

machinery/src/video/mp4.go

@@ -32,24 +32,6 @@ const MacEpochOffset uint64 = 2082844800
 // resulting in ~3 second fragments (assuming a typical GOP interval).
 const FragmentDurationMs = 3000
 
-// MinNormalGOPMs is the maximum spacing between two consecutive IDRs that
-// we still consider an anomalous "loop/restart seam". When two keyframes
-// arrive closer than this, we treat the second one as an upstream
-// restart/loop-seam and force a fresh fragment so the seam IDR cannot end
-// up as a mid-fragment sync sample. The check only runs when the current
-// fragment has not yet reached FragmentDurationMs.
-//
-// This must be set well below the smallest plausible *legitimate* GOP
-// length. Typical IP cameras use GOP intervals of 1000-2000 ms, and the
-// arrival timing of consecutive IDRs can jitter by a few hundred ms due to
-// network/RTSP buffering. A threshold close to 1 s (e.g. 950) caused
-// false positives on cameras with ~1 s GOPs (warnings like
-// "gap=800 ms / 300 ms / 200 ms" while the stream itself was healthy).
-// 400 ms is comfortably below any realistic GOP yet still catches the
-// virtual-rtsp / ffmpeg loop-seam pattern (seam IDRs typically arrive
-// 100-200 ms after the prior IDR).
-const MinNormalGOPMs = 400
-
 type MP4 struct {
 	// FileName is the name of the file
 	FileName                string
@@ -74,8 +56,6 @@ type MP4 struct {
 	FreeBoxSize             int64
 	FragmentStartRawPTS     uint64            // Raw PTS for timing when to flush fragments
 	FragmentStartDTS        uint64            // Accumulated VideoTotalDuration at fragment start (matches tfdt)
-	LastKeyframeRawPTS      uint64            // Raw PTS of the most recently seen keyframe (in any fragment)
-	LastKeyframeGapMs       uint64            // Gap (ms) between the previous two consecutive keyframes
 	MoofBoxes               int64             // Number of moof boxes in the file
 	MoofBoxSizes            []int64           // Sizes of each moof box
 	SegmentDurations        []uint64          // Duration of each segment in timescale units
@@ -253,24 +233,6 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	var duration uint64
 	if nextPTS > 0 && nextPTS > mp4.VideoFullSample.DecodeTime {
 		duration = nextPTS - mp4.VideoFullSample.DecodeTime
-		// Guard against forward PTS jumps (e.g. when looping a source MP4
-		// through virtual-rtsp the upstream ffmpeg may insert a large offset
-		// at the loop boundary, or the RTSP stream may stall briefly).
-		// Without this clamp the sample gets a huge duration which appears
-		// as a discontinuity in the trun/sidx/mvhd and causes browsers
-		// (Video.js / MSE) to abort playback with a "media corruption"
-		// error around the loop boundary.
-		var maxPlausible uint64 = 1000 // 1 second hard ceiling
-		if mp4.LastVideoSampleDTS > 0 && mp4.LastVideoSampleDTS*10 < maxPlausible {
-			maxPlausible = mp4.LastVideoSampleDTS * 10
-		}
-		if duration > maxPlausible {
-			log.Log.Warning(fmt.Sprintf("mp4.flushPendingVideoSample(): video PTS jumped forward (nextPTS=%d, prevDTS=%d, gap=%d ms) - clamping to %d ms (likely source loop/stall discontinuity)", nextPTS, mp4.VideoFullSample.DecodeTime, duration, maxPlausible))
-			duration = mp4.LastVideoSampleDTS
-			if duration == 0 {
-				duration = 33
-			}
-		}
 	} else {
 		// No valid nextPTS (Close case) or PTS went backwards (jitter/discontinuity)
 		if nextPTS > 0 {
@@ -327,37 +289,6 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 		}
 		shouldFlush := !mp4.Start || elapsed >= FragmentDurationMs
 
-		// Detect upstream source-loop / restart discontinuity. When an MP4 is
-		// looped through virtual-rtsp (ffmpeg `-stream_loop -1 -re`) the loop
-		// seam emits a fresh IDR much sooner than a normal GOP would. PTS keeps
-		// growing monotonically, so the timing-only `elapsed` check above does
-		// not catch it and the seam IDR ends up as a mid-fragment sync sample.
-		// MSE-based players (Video.js / Chromium / Firefox) reject the resulting
-		// fragment with a "media corruption" error because the inner IDR resets
-		// frame_num/POC inside what they expect to be a single GOP. Force a
-		// fragment boundary whenever two consecutive keyframes arrive much
-		// closer than a normal GOP.
-		//
-		// We only flag this as a seam when it is a *sudden* anomaly: the
-		// previous keyframe gap must have been healthy (>= MinNormalGOPMs).
-		// This avoids false positives on cameras that legitimately emit
-		// short-interval IDRs (short GOP, motion-triggered recovery IDRs,
-		// all-intra streams) where every keyframe would otherwise be flagged
-		// in a cascade, producing many tiny fragments and log spam.
-		if trackID == uint32(mp4.VideoTrack) && mp4.Start &&
-			mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS {
-			gap := pts - mp4.LastKeyframeRawPTS
-			if !shouldFlush && gap < MinNormalGOPMs &&
-				(mp4.LastKeyframeGapMs == 0 || mp4.LastKeyframeGapMs >= MinNormalGOPMs) {
-				log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): forcing fragment flush at unexpectedly close keyframe (gap=%d ms, fragment elapsed=%d ms) - likely upstream loop/restart discontinuity", gap, elapsed))
-				shouldFlush = true
-			}
-			mp4.LastKeyframeGapMs = gap
-		}
-		if trackID == uint32(mp4.VideoTrack) {
-			mp4.LastKeyframeRawPTS = pts
-		}
-
 		if shouldFlush {
 			// Write the previous segment to the file
 			if mp4.Start {
@@ -462,14 +393,6 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 					if started {
 						dts = 1
 					}
-					// Guard against forward PTS jumps (e.g. virtual-rtsp loop
-					// boundary or upstream stalls). Without this clamp the
-					// audio trun would carry an enormous sample duration that
-					// renders the recording unplayable in browsers.
-					if mp4.LastAudioSampleDTS > 0 && dts > mp4.LastAudioSampleDTS*10 {
-						log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): audio PTS jumped forward (pts=%d, prevDTS=%d, gap=%d) - clamping to last known duration", pts, mp4.AudioFullSample.DecodeTime, dts))
-						dts = mp4.LastAudioSampleDTS
-					}
 					mp4.LastAudioSampleDTS = dts
 					//fmt.Printf("Adding sample to track %d, PTS: %d, Duration: %d, size: %d\n", trackID, pts, dts, len(aac[7:]))
 					mp4.AudioTotalDuration += dts

machinery/src/video/mp4_loopseam_test.go

@@ -1,102 +0,0 @@
-package video
-
-import (
-	"fmt"
-	"os"
-	"testing"
-
-	mp4ff "github.com/Eyevinn/mp4ff/mp4"
-	"github.com/kerberos-io/agent/machinery/src/models"
-)
-
-// TestMP4LoopSeamIsolation reproduces the loop-seam pattern from the
-// failing virtual-rtsp recordings: ~1s GOPs, but at the source-MP4
-// loop boundary an IDR arrives prematurely (~200-870ms after the
-// previous IDR). Without the fix this seam IDR ends up bunched into
-// the same fragment as the prior GOP's IDR which trips macOS
-// VideoToolbox (kVTVideoDecoderBadDataErr / -12909). The fix forces
-// a fragment flush whenever two IDRs arrive closer than MinNormalGOPMs.
-func TestMP4LoopSeamIsolation(t *testing.T) {
-	tmpFile := "/tmp/test_loop_seam.mp4"
-	defer os.Remove(tmpFile)
-
-	sps := []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8}
-	pps := []byte{0x68, 0xce, 0x38, 0x80}
-	mp4Video := NewMP4(tmpFile, [][]byte{sps}, [][]byte{pps}, nil, 30)
-	mp4Video.SetWidth(1920)
-	mp4Video.SetHeight(1080)
-	v := mp4Video.AddVideoTrack("H264")
-
-	mk := func(k bool) []byte {
-		nt := byte(0x01)
-		if k {
-			nt = 0x65
-		}
-		f := []byte{0, 0, 0, 1, nt}
-		for i := 0; i < 200; i++ {
-			f = append(f, byte(i))
-		}
-		return f
-	}
-
-	frameDur := uint64(33)
-	pts := uint64(0)
-	emit := func(n int, gopLen int) {
-		for f := 0; f < n; f++ {
-			isKey := (f % gopLen) == 0
-			mp4Video.AddSampleToTrack(v, isKey, mk(isKey), pts)
-			pts += frameDur
-		}
-	}
-
-	// 17 seconds of normal content (last "good" IDR at sec 17).
-	emit(17*30, 30)
-	// Seam: IDR arrives ~150ms after previous (vs normal ~1000ms).
-	// This matches the realistic virtual-rtsp / ffmpeg `-stream_loop`
-	// loop boundary, where the new clip's first IDR is emitted shortly
-	// after the previous clip's final IDR.
-	pts -= 820
-	emit(13*30, 30)
-
-	mp4Video.Close(&models.Config{Signing: &models.Signing{PrivateKey: ""}})
-
-	f, _ := os.Open(tmpFile)
-	defer f.Close()
-	parsed, err := mp4ff.DecodeFile(f)
-	if err != nil {
-		t.Fatalf("decode: %v", err)
-	}
-
-	fragIdx := 0
-	for _, seg := range parsed.Segments {
-		for _, fr := range seg.Fragments {
-			for _, traf := range fr.Moof.Trafs {
-				if traf.Tfhd.TrackID != 1 {
-					continue
-				}
-				tfdt := traf.Tfdt.BaseMediaDecodeTime()
-				offset := uint64(0)
-				var keys []uint64
-				for _, trun := range traf.Truns {
-					for _, s := range trun.Samples {
-						if (s.Flags>>24)&0x03 == 0x02 {
-							keys = append(keys, offset)
-						}
-						offset += uint64(s.Dur)
-					}
-				}
-				fmt.Printf("frag %d tfdt=%d samples_dur=%d keys@%v\n",
-					fragIdx, tfdt, offset, keys)
-				for i := 1; i < len(keys); i++ {
-					gap := keys[i] - keys[i-1]
-					if gap < MinNormalGOPMs {
-						t.Errorf("frag %d (tfdt=%d): two IDRs only %d ms apart "+
-							"in same fragment (< %d) - seam was not isolated",
-							fragIdx, tfdt, gap, MinNormalGOPMs)
-					}
-				}
-				fragIdx++
-			}
-		}
-	}
-}

machinery/src/webrtc/main.go

@@ -26,7 +26,12 @@ import (
 
 const (
 	// Channel buffer sizes
-	candidateChannelBuffer = 100
+	// candidateChannelBuffer: large enough to absorb the burst of trickled ICE
+	// candidates that can arrive over MQTT before the SetRemoteDescription
+	// goroutine starts draining them. A small buffer caused candidates to be
+	// dropped silently on restrictive networks, leaving ICE stuck in
+	// "checking" until the viewer refreshed.
+	candidateChannelBuffer = 512
 	rtcpBufferSize         = 1500
 
 	// Timeouts and intervals
@@ -116,6 +121,22 @@ func (cm *ConnectionManager) RemovePeerConnection(sessionKey string) {
 	}
 }
 
+// CloseExistingPeerConnection closes and removes any peer connection currently
+// registered under sessionKey. Returns true if one was found. This is used to
+// reset state cleanly when a new request-hd-stream arrives for a session id
+// that the agent thinks is still active (for example after a viewer reload
+// where the previous PC hasn't yet been timed out by ICE).
+func (cm *ConnectionManager) CloseExistingPeerConnection(sessionKey string) bool {
+	cm.mu.RLock()
+	wrapper, exists := cm.peerConnections[sessionKey]
+	cm.mu.RUnlock()
+	if !exists || wrapper == nil {
+		return false
+	}
+	cleanupPeerConnection(sessionKey, wrapper)
+	return true
+}
+
 // QueueCandidate safely queues a candidate for a session without racing with channel closure.
 func (cm *ConnectionManager) QueueCandidate(sessionKey string, candidate string) bool {
 	cm.mu.Lock()
@@ -341,6 +362,17 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 	// We create a channel which will hold the candidates for this session.
 	sessionKey := config.Key + "/" + handshakePayload.SessionID
+
+	// If a previous peer connection for this exact session is still hanging
+	// around (e.g. a viewer reloaded before pion's ICE timeout fired) close it
+	// first so we start from a clean slate. Without this, the new request would
+	// race against a stale PC that still owns the per-peer broadcaster tracks.
+	if globalConnectionManager.CloseExistingPeerConnection(sessionKey) {
+		log.Log.Info("webrtc.main.InitializeWebRTCConnection(): closed stale peer connection for session " + handshakePayload.SessionID)
+	}
+	// Drain/reset the candidate channel too \u2014 leftover candidates from the
+	// prior session are not valid for the new ICE agent.
+	globalConnectionManager.CloseCandidateChannel(sessionKey)
 	candidateChannel := globalConnectionManager.GetOrCreateCandidateChannel(sessionKey)
 
 	// Set variables