Merge pull request #277 from kerberos-io/feature/support-configmaps

feature/support-configmaps

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
 

README.md

@@ -65,6 +65,7 @@ There are a myriad of cameras out there (USB, IP and other cameras), and it migh
 
 ### Contributing
 
+1. [Security vulnerability reporting](#security-vulnerability-reporting)
 1. [Contribute with Codespaces](#contribute-with-codespaces)
 2. [Develop and build](#develop-and-build)
 3. [Building from source](#building-from-source)
@@ -301,6 +302,10 @@ If we talk about video encoders and decoders (codecs) there are 2 major video co
 
 Conclusion: depending on the use case you might choose one over the other, and you can use both at the same time. For example you can use H264 (main stream) for livestreaming, and H265 (sub stream) for recording. If you wish to play recordings in a cross-platform and cross-browser environment, you might opt for H264 for better support.
 
+## Security vulnerability reporting
+
+If you found a potential security vulnerability, please use the private channels described in [SECURITY.md](SECURITY.md). Avoid opening public GitHub issues for sensitive findings.
+
 ## Contribute with Codespaces
 
 One of the major blockers for letting you contribute to an Open Source project is to set up your local development machine. Why? Because you might already have some tools and libraries installed that are used for other projects, and the libraries you would need for Kerberos Agent, for example FFmpeg, might require a different version. Welcome to dependency hell...

SECURITY.md

@@ -0,0 +1,40 @@
+# Security Policy
+
+## Supported Versions
+
+We only provide security fixes for the latest release series on the `master` branch.
+
+## Reporting a Vulnerability
+
+Please do **not** open a public GitHub issue for potential security vulnerabilities.
+
+Use one of the private channels below:
+
+1. Preferred: GitHub private vulnerability reporting
+   - https://github.com/kerberos-io/agent/security/advisories/new
+2. Fallback: Email
+   - support@kerberos.io
+   - Optional CC: support@uug.ai
+
+Please include:
+
+- A short summary and impact.
+- Reproduction steps or proof of concept.
+- Affected version(s), commit hash, or deployment details.
+- Any proposed mitigation/workaround.
+- Your preferred attribution name.
+
+For faster triage, use this subject format in email:
+
+`[Security][Kerberos Agent] <short title>`
+
+## Response Expectations
+
+- Acknowledgement target: within 3 business days.
+- Triage/update target: within 7 business days after acknowledgement.
+
+If you do not receive a response in time, please resend your report and include your original timestamp.
+
+## Disclosure and Credits
+
+We follow coordinated disclosure. After a fix is available, we will credit reporters unless they prefer to stay anonymous.

machinery/README.md

@@ -22,4 +22,8 @@ https://brianmacdonald.github.io/Ethonate/address#0xf4a759C9436E2280Ea9cdd23d314
 
 [**Docker Hub**](https://hub.docker.com/r/kerberos/agent) | [**Documentation**](https://doc.kerberos.io) | [**Website**](https://kerberos.io)
 
-Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+
+## Security reporting
+
+For sensitive vulnerabilities, use private disclosure channels documented in [../SECURITY.md](../SECURITY.md).

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/config/main.go

@@ -193,8 +193,78 @@ 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.
+		customWrap := &models.Configuration{Config: configuration.CustomConfig}
+		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)
+	}
+}
+
+// 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 +273,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 +616,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/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/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/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