Merge pull request #280 from kerberos-io/fix/dts-pts-correction

fix/dts-pts-correction

machinery/src/capture/gortsplib.go

@@ -517,6 +517,35 @@ func (g *Golibrtsp) ConnectBackChannel(ctx context.Context, ctxRunAgent context.
 	return
 }
 
+// dtsExtractor abstracts the codec-specific DTS extractors from mediacommon
+// (h264.DTSExtractor2 and h265.DTSExtractor2), which expose the same method.
+type dtsExtractor interface {
+	Extract(au [][]byte, pts int64) (int64, error)
+}
+
+// compositionOffsetMs returns the composition time offset (PTS - DTS) in
+// milliseconds for a coded access unit. Streams that contain B-frames deliver
+// access units in decode order with non-monotonic PTS; the fragmented MP4
+// writer needs a monotonic DTS timeline plus a per-sample composition offset
+// so browsers (Media Source Extensions) can decode the chained segments.
+//
+// It returns 0 when the codec has no frame reordering (the common case, e.g.
+// baseline "IPPP" streams) or when extraction fails, making it a safe no-op.
+func compositionOffsetMs(ext dtsExtractor, au [][]byte, pts int64, clockRate int) int64 {
+	if ext == nil || clockRate <= 0 {
+		return 0
+	}
+	dts, err := ext.Extract(au, pts)
+	if err != nil {
+		return 0
+	}
+	offset := pts - dts
+	if offset <= 0 {
+		return 0
+	}
+	return offset * 1000 / int64(clockRate)
+}
+
 // Start the RTSP client, and start reading packets.
 func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets.Queue, configuration *models.Configuration, communication *models.Communication) (err error) {
 	log.Log.Debug("capture.golibrtsp.Start(): started")
@@ -602,7 +631,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 	var filteredAU [][]byte
 	if g.VideoH264Media != nil && g.VideoH264Forma != nil {
 
-		//dtsExtractor := h264.NewDTSExtractor2()
+		// Extracts DTS from the bitstream to support B-frame H264 streams.
+		// Created once per stream (tracks reorder state across access units).
+		h264DTSExtractor := h264.NewDTSExtractor2()
 
 		g.Client.OnPacketRTP(g.VideoH264Media, g.VideoH264Forma, func(rtppkt *rtp.Packet) {
 
@@ -742,6 +773,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				// Composition time offset (PTS - DTS) in milliseconds. Non-zero
+				// only for streams with B-frames; the MP4 writer uses it to keep a
+				// monotonic decode timeline and present frames in PTS order.
+				compositionOffset := compositionOffsetMs(h264DTSExtractor, au, pts2, g.VideoH264Forma.ClockRate())
+
 				pkt := packets.Packet{
 					IsKeyFrame:      idrPresent,
 					Packet:          rtppkt,
@@ -749,7 +785,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					Time:            pts2,
 					TimeLegacy:      pts,
 					CurrentTime:     time.Now().UnixMilli(),
-					CompositionTime: pts2,
+					CompositionTime: compositionOffset,
 					Idx:             g.VideoH264Index,
 					IsVideo:         true,
 					IsAudio:         false,
@@ -817,6 +853,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 
 	// called when a video RTP packet arrives for H265
 	if g.VideoH265Media != nil && g.VideoH265Forma != nil {
+
+		// Extracts DTS from the bitstream to support B-frame H265 streams.
+		// Created once per stream (tracks reorder state across access units).
+		h265DTSExtractor := h265.NewDTSExtractor2()
+
 		g.Client.OnPacketRTP(g.VideoH265Media, g.VideoH265Forma, func(rtppkt *rtp.Packet) {
 
 			// This will check if we need to stop the thread,
@@ -860,6 +901,10 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					}
 				}
 
+				// Preserve the decoded access unit (in decode order) for DTS
+				// extraction before we rewrite it into the filtered/annexb form.
+				decodedAU := au
+
 				filteredAU = [][]byte{
 					{byte(h265.NALUType_AUD_NUT) << 1, 1, 0x50},
 				}
@@ -902,6 +947,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				// Composition time offset (PTS - DTS) in milliseconds; see H264 handler.
+				compositionOffset := compositionOffsetMs(h265DTSExtractor, decodedAU, pts2, g.VideoH265Forma.ClockRate())
+
 				pkt := packets.Packet{
 					IsKeyFrame:      isRandomAccess,
 					Packet:          rtppkt,
@@ -909,7 +957,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					Time:            pts2,
 					TimeLegacy:      pts,
 					CurrentTime:     time.Now().UnixMilli(),
-					CompositionTime: pts2,
+					CompositionTime: compositionOffset,
 					Idx:             g.VideoH265Index,
 					IsVideo:         true,
 					IsAudio:         false,

machinery/src/capture/main.go

@@ -140,23 +140,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				if start && // If already recording and current frame is a keyframe and we should stop recording
 					nextPkt.IsKeyFrame && (startRecording+postRecording-now <= 0 || now-startRecording > maxRecordingPeriod-500) {
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						// Write the last packet
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						// Write the last packet
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					// Write the last packet before closing the recording.
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 
 					// Close mp4
 					if len(mp4Video.SPSNALUs) == 0 && len(configuration.Config.Capture.IPCamera.SPSNALUs) > 0 {
@@ -311,43 +296,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
 					}
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							// We might need to use ffmpeg to transcode the audio to AAC.
-							// For now we will skip the audio track.
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					recordingStatus = "started"
 
 				} else if start {
 
-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						// New method using new mp4 library
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						if pkt.Codec == "AAC" {
-							if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-							}
-						} else if pkt.Codec == "PCM_MULAW" {
-							// TODO: transcode to AAC, some work to do..
-							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-						}
-					}
+					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 				}
 				pkt = nextPkt
 			}
@@ -571,29 +525,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						start = true
 					}
 					if start {
-						pts := convertPTS(pkt.TimeLegacy)
-						if pkt.IsVideo {
-							log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add video sample")
-							if mp4Video != nil {
-								if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-									log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
-								}
-							}
-						} else if pkt.IsAudio {
-							log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add audio sample")
-							if pkt.Codec == "AAC" {
-								if mp4Video != nil {
-									if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-										log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
-									}
-								}
-							} else if pkt.Codec == "PCM_MULAW" {
-								// TODO: transcode to AAC, some work to do..
-								// We might need to use ffmpeg to transcode the audio to AAC.
-								// For now we will skip the audio track.
-								log.Log.Debug("capture.main.HandleRecordStream(motiondetection): no AAC audio codec detected, skipping audio track.")
-							}
-						}
+						writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					}
 
 					pkt = nextPkt
@@ -867,6 +799,41 @@ func convertPTS(v time.Duration) uint64 {
 	return uint64(v.Milliseconds())
 }
 
+// writeSampleToMP4 writes a single capture packet to the fragmented MP4.
+//
+// For video it derives the decode timestamp (DTS) from the packet PTS using the
+// per-packet composition offset (PTS - DTS), which is non-zero only for streams
+// that contain B-frames. Passing the monotonic DTS as the sample timestamp keeps
+// the fragment timeline (tfdt/sidx) monotonic, while the composition offset is
+// forwarded so frames are still presented in PTS order.
+func writeSampleToMP4(mp4Video *video.MP4, videoTrack, audioTrack uint32, pkt packets.Packet) {
+	if mp4Video == nil {
+		return
+	}
+
+	pts := convertPTS(pkt.TimeLegacy)
+
+	if pkt.IsVideo {
+		compositionOffset := pkt.CompositionTime
+		dts := pts
+		if compositionOffset > 0 && uint64(compositionOffset) <= pts {
+			dts = pts - uint64(compositionOffset)
+		}
+		if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, dts, compositionOffset); err != nil {
+			log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
+		}
+	} else if pkt.IsAudio {
+		if pkt.Codec == "AAC" {
+			if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts, 0); err != nil {
+				log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
+			}
+		} else if pkt.Codec == "PCM_MULAW" {
+			// TODO: transcode to AAC, some work to do..
+			log.Log.Debug("capture.main.writeSampleToMP4(): no AAC audio codec detected, skipping audio track.")
+		}
+	}
+}
+
 /*func convertPTS2(v int64) uint64 {
 	return uint64(v) / 100
 }*/

machinery/src/components/Kerberos.go

@@ -752,10 +752,24 @@ func GetSnapshotRaw(c *gin.Context, captureDevice *capture.Capture, configuratio
 // @Description Get the current configuration.
 // @Success 200
 func GetConfig(c *gin.Context, captureDevice *capture.Capture, configuration *models.Configuration, communication *models.Communication) {
-	// We'll try to get a snapshot from the camera.
-	base64Image := capture.Base64Image(captureDevice, communication, configuration)
-	if base64Image != "" {
-		communication.Image = base64Image
+	// We'll try to get a fresh snapshot from the camera. Capturing a snapshot
+	// reads a keyframe from the live stream, which blocks until one arrives.
+	// When the camera is offline or the stream is stalled (no packets being
+	// received) this would block the /config endpoint indefinitely, making the
+	// agent appear unreachable even though its HTTP server is healthy. We
+	// therefore bound the snapshot fetch with a short timeout and fall back to
+	// the last cached snapshot, so /config always responds promptly.
+	snapshot := make(chan string, 1)
+	go func() {
+		snapshot <- capture.Base64Image(captureDevice, communication, configuration)
+	}()
+	select {
+	case base64Image := <-snapshot:
+		if base64Image != "" {
+			communication.Image = base64Image
+		}
+	case <-time.After(2 * time.Second):
+		log.Log.Info("components.Kerberos.GetConfig(): snapshot timed out (stream stalled or camera offline), returning configuration with the last cached snapshot.")
 	}
 
 	c.JSON(200, gin.H{

machinery/src/config/main.go

@@ -193,8 +193,104 @@ func OpenConfig(configDirectory string, configuration *models.Configuration) {
 	return
 }
 
-// This function will override the configuration with environment variables.
+// OverrideWithEnvironmentVariables builds the effective configuration from the
+// environment variables.
+//
+// In ConfigMap/standalone mode (DEPLOYMENT empty or "agent") the global
+// configuration is delivered as GLOBAL_AGENT_* environment variables and the
+// per-agent configuration as AGENT_* environment variables. We parse them into
+// the separate global and custom configurations and build the effective
+// configuration as "global overridden by custom", mirroring the MongoDB-backed
+// factory behaviour. This keeps the global and per-agent (custom) configuration
+// separated so the factory edit page can distinguish inherited global settings
+// from per-agent overrides.
 func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
+	if os.Getenv("DEPLOYMENT") == "" || os.Getenv("DEPLOYMENT") == "agent" {
+		initConfigPointers(&configuration.Config)
+
+		// Parse the global configuration from the GLOBAL_AGENT_* variables.
+		globalWrap := &models.Configuration{Config: configuration.GlobalConfig}
+		initConfigPointers(&globalWrap.Config)
+		applyAgentEnvVars(globalWrap, "GLOBAL_", false)
+		configuration.GlobalConfig = globalWrap.Config
+
+		// Parse the per-agent (custom) configuration from the AGENT_* variables.
+		// In ConfigMap mode the per-agent overrides are delivered exclusively
+		// through AGENT_* environment variables, so we must start from an empty
+		// configuration rather than the bundled config.json that OpenConfig loaded
+		// into CustomConfig. Otherwise defaults from that file (e.g. cloud="s3")
+		// would leak into the custom config and be mistaken for explicit per-agent
+		// overrides, hiding inherited global settings (the factory edit page would
+		// show the local default instead of the inherited global persistence).
+		customBase := configuration.CustomConfig
+		if isConfigMapMode() {
+			customBase = models.Config{}
+		}
+		customWrap := &models.Configuration{Config: customBase}
+		initConfigPointers(&customWrap.Config)
+		applyAgentEnvVars(customWrap, "", false)
+		configuration.CustomConfig = customWrap.Config
+
+		// Build the effective configuration: global base, then per-agent
+		// overrides on top. Defaults (e.g. signing) are applied on the last
+		// pass only.
+		applyAgentEnvVars(configuration, "GLOBAL_", false)
+		applyAgentEnvVars(configuration, "", true)
+	} else {
+		// Factory/MongoDB mode: the global and custom configurations are already
+		// loaded and merged from MongoDB; we only override the effective
+		// configuration with any AGENT_* environment variables.
+		applyAgentEnvVars(configuration, "", true)
+	}
+}
+
+// isConfigMapMode reports whether the agent is running in ConfigMap mode, i.e.
+// whether a global configuration layer is delivered separately through
+// GLOBAL_AGENT_* environment variables. In that mode the per-agent (custom)
+// configuration must be built solely from the AGENT_* overrides and must not be
+// seeded with the bundled config.json defaults, so that inherited global
+// settings remain distinguishable from explicit per-agent overrides.
+func isConfigMapMode() bool {
+	for _, env := range os.Environ() {
+		if strings.HasPrefix(env, "GLOBAL_AGENT_") {
+			return true
+		}
+	}
+	return false
+}
+
+// initConfigPointers ensures all pointer sub-structs are non-nil so that the
+// environment-variable parsing can assign into them without dereferencing a nil
+// pointer.
+func initConfigPointers(config *models.Config) {
+	if config.KStorage == nil {
+		config.KStorage = &models.KStorage{}
+	}
+	if config.KStorageSecondary == nil {
+		config.KStorageSecondary = &models.KStorage{}
+	}
+	if config.S3 == nil {
+		config.S3 = &models.S3{}
+	}
+	if config.Encryption == nil {
+		config.Encryption = &models.Encryption{}
+	}
+	if config.Signing == nil {
+		config.Signing = &models.Signing{}
+	}
+	if config.Dropbox == nil {
+		config.Dropbox = &models.Dropbox{}
+	}
+	if config.Region == nil {
+		config.Region = &models.Region{}
+	}
+}
+
+// applyAgentEnvVars applies the AGENT_* environment variables (optionally
+// carrying the given prefix, e.g. "GLOBAL_") onto configuration.Config. When
+// applyDefaults is true, defaults (such as the signing key) are applied after
+// parsing; this should only be done for the effective configuration.
+func applyAgentEnvVars(configuration *models.Configuration, prefix string, applyDefaults bool) {
 	environmentVariables := os.Environ()
 
 	// Initialize the configuration for some new fields.
@@ -203,9 +299,10 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 	}
 
 	for _, env := range environmentVariables {
-		if strings.Contains(env, "AGENT_") {
-			key := strings.Split(env, "=")[0]
-			value := os.Getenv(key)
+		fullKey := strings.SplitN(env, "=", 2)[0]
+		if strings.HasPrefix(fullKey, prefix+"AGENT_") && !(prefix == "" && strings.HasPrefix(fullKey, "GLOBAL_AGENT_")) {
+			key := strings.TrimPrefix(fullKey, prefix)
+			value := os.Getenv(fullKey)
 			switch key {
 
 			/* General configuration */
@@ -545,13 +642,20 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 		}
 	}
 
-	// Signing is a new feature, so if empty we set default values.
-	if configuration.Config.Signing == nil || configuration.Config.Signing.PrivateKey == "" {
+	// Signing is a new feature, so if empty we set default values. Only applied
+	// for the effective configuration (applyDefaults), not for the separate
+	// global/custom views.
+	if applyDefaults && (configuration.Config.Signing == nil || configuration.Config.Signing.PrivateKey == "") {
 		configuration.Config.Signing = &models.Signing{
 			Enabled:    "true",
 			PrivateKey: "-----BEGIN PRIVATE KEY-----\nMIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQDoSxjyw08lRxF4Yoqmcaewjq3XjB55dMy4tlN5MGLdr8aAPuNR9Mwh3jlh1bDpwQXNgZkHDV/q9bpdPGGi7SQo2xw+rDuo5Y1f3wdzz+iuCTPbzoGFalE+1PZlU5TEtUtlbt7MRc4pxTaLP3u0P3EtW3KnzcUarcJWZJYxzv7gqVNCA/47BN+1ptqjwz3LAlah5yaftEvVjkaANOsafUswbS4VT44XfSlbKgebORCKDuNgQiyhuV5gU+J0TOaqRWwwMAWV0UoScyJLfhHRBCrUwrCUTwqH9jfkB7pgRFsYoZJd4MKMeHJjFSum+QXCBqInSnwu8c2kJChiLMWqJ+mhpTdfUAmSkeUSStfbbcavIPbDABvMgzOcmYMIVXXe57twU0xdu3AqWLtc9kw1BkUgZblM9pSSpYrIDheEyMs2/hiLgXsIaM0nVQtqwrA7rbeEGuPblzA6hvHgwN9K6HaBqdlGSlpYZ0v3SWIMwmxRB+kIojlyuggm8Qa4mqL97GFDGl6gOBGlNUFTBUVEa3EaJ7NJpGobRGsh/9dXzcW4aYmT9WxlzTlIKksI1ro6KdRfuVWfEs4AnG8bVEJmofK8EUrueB9IdXlcJZB49xolnOZPFohtMe/0U7evQOQP3sZnX+KotCsE7OXJvL09oF58JKoqmK9lPp0+pFBU4g6NjQIDAQABAoICAA+RSWph1t+q5R3nxUxFTYMrhv5IjQe2mDxJpF3B409zolC9OHxgGUisobTY3pBqs0DtKbxUeH2A0ehUH/axEosWHcz3cmIbgxHE9kdlJ9B3Lmss6j/uw+PWutu1sgm5phaIFIvuNNRWhPB6yXUwU4sLRat1+Z9vTmIQiKdtLIrtJz/n2VDvrJxn1N+yAsE20fnrksFKyZuxVsJaZPiX/t5Yv1/z0LjFjVoL7GUA5/Si7csN4ftqEhUrkNr2BvcZlTyffrF4lZCXrtl76RNUaxhqIu3H0gFbV2UfBpuckkfAhNRpXJ4iFSxm4nQbk4ojV8+l21RFOBeDN2Z7Ocu6auP5MnzpopR66vmDCmPoid498VGgDzFQEVkOar8WAa4v9h85QgLKrth6FunmaWJUT6OggQD3yY58GSwp5+ARMETMBP2x6Eld+PGgqoJvPT1+l/e9gOw7/SJ+Wz6hRXZAm/eiXMppHtB7sfea5rscNanPjJkK9NvPM0MX9cq/iA6QjXuETkMbubjo+Cxk3ydZiIQmWQDAx/OgxTyHbeRCVhLPcAphX0clykCuHZpI9Mvvj643/LoE0mjTByWJXf/WuGJA8ElHkjSdokVJ7jumz8OZZHfq0+V7+la2opsObeQANHW5MLWrnHlRVzTGV0IRZDXh7h1ptUJ4ubdvw/GJ2NeTAoIBAQD0lXXdjYKWC4uZ4YlgydP8b1CGda9cBV5RcPt7q9Ya1R2E4ieYyohmzltopvdaOXdsTZzhtdzOzKF+2qNcbBKhBTleYZ8GN5RKbo7HwXWpzfCTjseKHOD/QPwvBKXzLVWNtXn1NrLR79Rv0wbkYF6DtoqpEPf5kMs4bx79yW+mz8FUgdEeMjKphx6Jd5RYlTUxS64K6bnK7gjHNCF2cwdxsh4B6EB649GKeNz4JXi+oQBmOcX5ncXnkJrbju+IjtCkQ40HINVNdX7XeEaaw6KGaImVjw61toPUuDaioYUojufayoyXaUJnDbHQ2tNekEpq5iwnenZCbUKWmSeRe7dLAoIBAQDzIscYujsrmPxiTj2prhG0v36NRNP99mShnnJGowiIs+UBS0EMdOmBFa2sC9uFs/VnreQNYPDJdfr7O5VK9kfbH/PSiiKJ+wVebfdAlWkJYH27JN2Kl2l/OsvRVelNvF3BWIYF46qzGxIM0axaz3T2ZAJ9SrUgeAYhak6uyM4fbexEWXxDgPGu6C0jB6IAzmHJnnh+j5+4ZXqjVyUxBYtUsWXF/TXomVcT9jxj7aUmS2/Us0XTVOVNpALqqYcekrzsX/wX0OEi5HkivYXHcNaDHx3NuUf6KdYof5DwPUM76qe+5/kWlSIHP3M6rIFK3pYFUnkHn2E8jNWcO97Aio+HAoIBAA+bcff/TbPxbKkXIUMR3fsfx02tONFwbkJYKVQM9Q6lRsrx+4Dee7HDvUWCUgpp3FsG4NnuVvbDTBLiNMZzBwVLZgvFwvYMmePeBjJs/+sj/xQLamQ/z4O6S91cOJK589mlGPEy2lpXKYExQCFWnPFetp5vPMOqH62sOZgMQJmubDHOTt/UaDM1Mhenj8nPS6OnpqV/oKF4awr7Ip+CW5k/unZ4sZSl8PsbF06mZXwUngfn6+Av1y8dpSQZjONz6ZBx1w/7YmEc/EkXnbnGfhqBlTX7+P5TdTofvyzFjc+2vsjRYANRbjFRSGWBcTd5kaYcpfim8eDvQ+6EO2gnMt0CggEAH2ln1Y8B5AEQ4lZ/avOdP//ZhsDUrqPtnl/NHckkahzrwj4JumVEYbP+SxMBGoYEd4+kvgG/OhfvBBRPlm65G9tF8fZ8vdzbdba5UfO7rUV1GP+LS8OCErjy6imySaPDbR5Vul8Oh7NAor1YCidxUf/bvnovanF3QUvtvHEfCDp4YuA4yLPZBaLjaforePUw9w5tPNSravRZYs74dBvmQ1vj7S9ojpN5B5AxfyuNwaPPX+iFZec69MvywISEe3Ozysof1Kfc3lgsOkvIA9tVK32SqSh93xkWnQbWH+OaUxxe7bAko0FDMzKEXZk53wVg1nEwR8bUljEPy+6EOdXs8wKCAQEAsEOWYMY5m7HkeG2XTTvX7ECmmdGl/c4ZDVwzB4IPxqUG7XfLmtsON8YoKOEUpJoc4ANafLXzmU+esUGbH4Ph22IWgP9jzws7jxaN/Zoku64qrSjgEZFTRIpKyhFk/ImWbS9laBW4l+m0tqTTRqoE0QEJf/2uv/04q65zrA70X9z2+KTrAtqOiRQPWl/IxRe9U4OEeGL+oD+YlXKCDsnJ3rwUIOZgJx0HWZg7K35DKwqs1nVi56FBdljiTRKAjVLRedjgDCSfGS1yUZ3krHzpaPt1qgnT3rdtYcIdbYDr66V2/gEEaz6XMGHuTk/ewjzUJxq9UTVeXOCbkRPXgVJg1w==\n-----END PRIVATE KEY-----",
 		}
 	}
+
+	// When the agent is configured through environment variables the global and
+	// custom configurations were already parsed separately (see
+	// OverrideWithEnvironmentVariables), so there is no need to mirror the
+	// effective configuration into CustomConfig anymore.
 }
 
 func SaveConfig(configDirectory string, config models.Config, configuration *models.Configuration, communication *models.Communication) error {

machinery/src/packets/packet.go

@@ -14,7 +14,7 @@ type Packet struct {
 	IsKeyFrame      bool   // video packet is key frame
 	Idx             int8   // stream index in container format
 	Codec           string // codec name
-	CompositionTime int64  // packet presentation time minus decode time for H264 B-Frame
+	CompositionTime int64  // composition offset (PTS - DTS) in milliseconds, non-zero for H264/H265 B-frames
 	Time            int64  // packet decode time
 	TimeLegacy      time.Duration
 	CurrentTime     int64  // current time in milliseconds (UNIX timestamp)

machinery/src/routers/http/jwt_middleware.go

@@ -14,7 +14,14 @@ import (
 func JWTMiddleWare() jwt.GinJWTMiddleware {
 
 	identityKey := "id"
-	myKey := "TOBECHANGED"
+	// Allow the JWT signing secret to be configured through an environment
+	// variable so that tokens issued by another service (e.g. the Kerberos
+	// Factory) can be validated by the agent. Falls back to the historic
+	// default to preserve backwards compatibility.
+	myKey := os.Getenv("AGENT_JWT_SECRET")
+	if myKey == "" {
+		myKey = "TOBECHANGED"
+	}
 
 	m := jwt.GinJWTMiddleware{
 		Realm:       "kerberosio",
@@ -106,7 +113,11 @@ func JWTMiddleWare() jwt.GinJWTMiddleware {
 		// - "query:<name>"
 		// - "cookie:<name>"
 		// - "param:<name>"
-		TokenLookup: "header: Authorization, query: token, cookie: jwt",
+		// X-Authorization is included because requests proxied through the
+		// Kubernetes apiserver service-proxy have their Authorization header
+		// consumed by the apiserver; the original bearer token is forwarded in
+		// the X-Authorization header instead.
+		TokenLookup: "header: Authorization, header: X-Authorization, query: token, cookie: jwt",
 		// TokenLookup: "query:token",
 		// TokenLookup: "cookie:token",
 

machinery/src/video/mp4.go

@@ -266,7 +266,16 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	return true
 }
 
-func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64) error {
+// AddSampleToTrack appends a sample to the given track.
+//
+// For video, pts is the decode timestamp (DTS, in milliseconds) and
+// compositionOffset is the composition time offset (PTS - DTS, in milliseconds).
+// The offset is non-zero only for streams that contain B-frames; it is written
+// as the sample's signed composition time offset so the decoder presents frames
+// in PTS order while the fragment timeline stays monotonic in DTS.
+//
+// For audio, pts is the sample timestamp and compositionOffset should be 0.
+func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64, compositionOffset int64) error {
 
 	if isKeyframe && trackID == uint32(mp4.VideoTrack) {
 		mp4.TotalKeyframesReceived++
@@ -375,7 +384,7 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 				fullSample.Sample = mp4ff.Sample{
 					Size:                  uint32(len(fullSample.Data)),
 					Flags:                 flags,
-					CompositionTimeOffset: 0, // No composition time offset for video
+					CompositionTimeOffset: int32(compositionOffset), // PTS-DTS, non-zero for B-frames
 				}
 				mp4.VideoFullSample = &fullSample
 				mp4.PendingSampleIsKeyframe = isKeyframe

machinery/src/video/mp4_duration_test.go

@@ -49,7 +49,7 @@ func TestMP4Duration(t *testing.T) {
 	for i := 0; i < numFrames; i++ {
 		pts := uint64(i) * frameDuration
 		isKeyframe := i%gopSize == 0
-		err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts)
+		err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts, 0)
 		if err != nil {
 			t.Fatalf("AddSampleToTrack failed at frame %d: %v", i, err)
 		}