Merge pull request #317 from kerberos-io/feature/backoff-moq-relay-session

feature/backoff-moq-relay-session

.vscode/launch.json

@@ -17,7 +17,7 @@
                 "8080"
             ],
             "envFile": "${workspaceFolder}/machinery/.env.local",
-            "buildFlags": "--tags dynamic",
+            "buildFlags": "--tags dynamic,moq",
             "env": {
                 "GOWORK": "off"
             },

README-RTSPS-TLS.md

@@ -377,7 +377,7 @@ For a process running directly in the same environment:
 AGENT_CAPTURE_IPCAMERA_RTSP="rtsps://<user>:<password>@10.0.30.11:9554/?inst=1"
 AGENT_CAPTURE_IPCAMERA_SUB_RTSP="rtsps://<user>:<password>@10.0.30.11:9554/?inst=2"
 AGENT_CAPTURE_IPCAMERA_RTSPS_INSECURE=false
-SSL_CERT_FILE=/home/agent/data/config/uug-camera-trust-bundle.pem
+AGENT_CAPTURE_IPCAMERA_RTSPS_CA_FILE=/home/agent/data/config/uug-camera-trust-bundle.pem
 ```
 
 For a container, mount the public bundle read-only at the exact path visible
@@ -387,20 +387,22 @@ includes Debian's `ca-certificates` package:
 ```bash
 docker run \
   -v /secure/config/uug-camera-trust-bundle.pem:/home/agent/data/config/uug-camera-trust-bundle.pem:ro \
-  -e SSL_CERT_FILE=/home/agent/data/config/uug-camera-trust-bundle.pem \
+  -e AGENT_CAPTURE_IPCAMERA_RTSPS_CA_FILE=/home/agent/data/config/uug-camera-trust-bundle.pem \
   -e AGENT_CAPTURE_IPCAMERA_RTSPS_INSECURE=false \
   -e 'AGENT_CAPTURE_IPCAMERA_RTSP=rtsps://<user>:<password>@10.0.30.11:9554/?inst=1' \
   -e 'AGENT_CAPTURE_IPCAMERA_SUB_RTSP=rtsps://<user>:<password>@10.0.30.11:9554/?inst=2' \
   kerberos/agent:latest
 ```
 
-Restart the Agent after changing trust files. Go can cache the process system
-certificate pool after its first use, so editing a file does not guarantee that
-an already-running process reloads it.
+`AGENT_CAPTURE_IPCAMERA_RTSPS_CA_FILE` starts with the operating system's roots
+and appends the camera bundle only to the gortsplib TLS configuration. Other
+clients, including MoQ, Hub, and Vault, retain the normal public CA chain.
+Restart the Agent after changing trust files.
 
-The default production mode retains the image's normal public roots in addition
-to the private camera CA. For a deliberately private-CA-only deployment, mount
-an empty directory and set `SSL_CERT_DIR` to its path:
+Do not set `SSL_CERT_FILE` or `SSL_CERT_DIR` in production solely for camera
+trust. They are process-wide and can prevent other clients from validating
+public services. For a deliberate process-wide isolation test, mount an empty
+directory and set `SSL_CERT_DIR` to its path:
 
 ```bash
 -v /secure/config/empty-ca-dir:/home/agent/data/config/empty-ca-dir:ro \
@@ -523,9 +525,10 @@ go run -tags moq . -action run -port 8080
 That fresh process must fail with `x509: certificate signed by unknown
 authority`.
 
-Use exactly one trust-distribution approach when possible:
+Use exactly one camera trust-distribution approach when possible:
 
-1. Mount a private trust bundle and set `SSL_CERT_FILE`; or
+1. Mount a private trust bundle and set
+  `AGENT_CAPTURE_IPCAMERA_RTSPS_CA_FILE`; or
 2. Install the CA certificates into the operating-system trust store.
 
 Using both is valid, but makes isolation tests less obvious.

machinery/data/tus/1786715908_3-737_encrypted_0-0-0-0_-1_25498.mp4.primary.json

@@ -0,0 +1 @@
+{"upload_url":"https://vault.kerberos.io/api/storage/tus/19e42fbc666a38064904caf8c46d182a","vault_uri":"https://vault.kerberos.io/api/storage/tus/","size":1591581}

machinery/src/capture/gortsplib.go

@@ -9,6 +9,7 @@ import "C"
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"image"
@@ -40,14 +41,34 @@ import (
 
 var tracer = otel.Tracer("github.com/kerberos-io/agent/machinery/src/capture")
 
-const rtspsInsecureEnv = "AGENT_CAPTURE_IPCAMERA_RTSPS_INSECURE"
+const (
+	rtspsCAFileEnv   = "AGENT_CAPTURE_IPCAMERA_RTSPS_CA_FILE"
+	rtspsInsecureEnv = "AGENT_CAPTURE_IPCAMERA_RTSPS_INSECURE"
+)
 
-func rtspsTLSConfig() *tls.Config {
-	if os.Getenv(rtspsInsecureEnv) != "true" {
-		return nil
+func rtspsTLSConfig() (*tls.Config, error) {
+	if os.Getenv(rtspsInsecureEnv) == "true" {
+		return &tls.Config{InsecureSkipVerify: true}, nil // #nosec G402 -- explicit opt-in for cameras with self-signed certificates
 	}
 
-	return &tls.Config{InsecureSkipVerify: true} // #nosec G402 -- explicit opt-in for cameras with self-signed certificates
+	caFile := os.Getenv(rtspsCAFileEnv)
+	if caFile == "" {
+		return nil, nil
+	}
+
+	rootCAs, err := x509.SystemCertPool()
+	if err != nil {
+		return nil, fmt.Errorf("load system CA pool: %w", err)
+	}
+	caPEM, err := os.ReadFile(caFile)
+	if err != nil {
+		return nil, fmt.Errorf("read RTSPS CA file %q: %w", caFile, err)
+	}
+	if !rootCAs.AppendCertsFromPEM(caPEM) {
+		return nil, fmt.Errorf("RTSPS CA file %q contains no certificates", caFile)
+	}
+
+	return &tls.Config{RootCAs: rootCAs}, nil
 }
 
 // Implements the RTSPClient interface.
@@ -329,12 +350,17 @@ func (g *Golibrtsp) Connect(ctx context.Context, ctxOtel context.Context) (err e
 	_, span := tracer.Start(ctxOtel, "Connect")
 	defer span.End()
 
+	tlsConfig, err := rtspsTLSConfig()
+	if err != nil {
+		return fmt.Errorf("configure RTSPS TLS: %w", err)
+	}
+
 	protocol := gortsplib.ProtocolTCP
 	g.health = newStreamHealth()
 	g.Client = gortsplib.Client{
 		RequestBackChannels: false,
 		Protocol:            &protocol,
-		TLSConfig:           rtspsTLSConfig(),
+		TLSConfig:           tlsConfig,
 		// Route gortsplib's packet-loss / decode-error reporting through our
 		// structured logger with stream context (replaces its plain stdout
 		// logging). These hooks are what let us tell whether the camera is
@@ -607,12 +633,17 @@ func (g *Golibrtsp) ConnectBackChannel(ctx context.Context, ctxRunAgent context.
 	_, span := tracer.Start(ctxRunAgent, "ConnectBackChannel")
 	defer span.End()
 
+	tlsConfig, err := rtspsTLSConfig()
+	if err != nil {
+		return fmt.Errorf("configure RTSPS TLS: %w", err)
+	}
+
 	// Transport TCP
 	protocol := gortsplib.ProtocolTCP
 	g.Client = gortsplib.Client{
 		RequestBackChannels: true,
 		Protocol:            &protocol,
-		TLSConfig:           rtspsTLSConfig(),
+		TLSConfig:           tlsConfig,
 	}
 	// parse URL
 	u, err := base.ParseURL(g.Url)

machinery/src/capture/gortsplib_test.go

@@ -1,22 +1,79 @@
 package capture
 
-import "testing"
+import (
+	"bytes"
+	"encoding/pem"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+)
 
 func TestRTSPSTLSConfig(t *testing.T) {
 	t.Run("verifies certificates by default", func(t *testing.T) {
+		t.Setenv(rtspsCAFileEnv, "")
 		t.Setenv(rtspsInsecureEnv, "")
 
-		if got := rtspsTLSConfig(); got != nil {
+		got, err := rtspsTLSConfig()
+		if err != nil {
+			t.Fatalf("rtspsTLSConfig() error = %v", err)
+		}
+		if got != nil {
 			t.Fatalf("rtspsTLSConfig() = %#v, want nil", got)
 		}
 	})
 
 	t.Run("allows explicit insecure mode", func(t *testing.T) {
+		t.Setenv(rtspsCAFileEnv, "/missing/ignored-in-insecure-mode.pem")
 		t.Setenv(rtspsInsecureEnv, "true")
 
-		got := rtspsTLSConfig()
+		got, err := rtspsTLSConfig()
+		if err != nil {
+			t.Fatalf("rtspsTLSConfig() error = %v", err)
+		}
 		if got == nil || !got.InsecureSkipVerify {
 			t.Fatalf("rtspsTLSConfig() = %#v, want InsecureSkipVerify enabled", got)
 		}
 	})
+
+	t.Run("adds a camera CA to system roots", func(t *testing.T) {
+		t.Setenv(rtspsInsecureEnv, "")
+		server := httptest.NewTLSServer(nil)
+		defer server.Close()
+
+		certificate := server.Certificate()
+		caFile := filepath.Join(t.TempDir(), "camera-ca.pem")
+		caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate.Raw})
+		if err := os.WriteFile(caFile, caPEM, 0o600); err != nil {
+			t.Fatal(err)
+		}
+		t.Setenv(rtspsCAFileEnv, caFile)
+
+		got, err := rtspsTLSConfig()
+		if err != nil {
+			t.Fatalf("rtspsTLSConfig() error = %v", err)
+		}
+		if got == nil || got.RootCAs == nil {
+			t.Fatalf("rtspsTLSConfig() = %#v, want custom RootCAs", got)
+		}
+		for _, subject := range got.RootCAs.Subjects() {
+			if bytes.Equal(subject, certificate.RawSubject) {
+				return
+			}
+		}
+		t.Fatal("camera CA was not added to RootCAs")
+	})
+
+	t.Run("rejects an invalid camera CA file", func(t *testing.T) {
+		t.Setenv(rtspsInsecureEnv, "")
+		caFile := filepath.Join(t.TempDir(), "camera-ca.pem")
+		if err := os.WriteFile(caFile, []byte("not a certificate"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		t.Setenv(rtspsCAFileEnv, caFile)
+
+		if _, err := rtspsTLSConfig(); err == nil {
+			t.Fatal("rtspsTLSConfig() error = nil, want invalid CA error")
+		}
+	})
 }

machinery/src/cloud/cloud.go

@@ -229,6 +229,54 @@ func rawJSONOrEmptyArray(b []byte) json.RawMessage {
 	return json.RawMessage(b)
 }
 
+const heartbeatResponseBodyLogLimit = 4 * 1024
+
+func readHeartbeatResponseBody(response *http.Response) (string, bool, error) {
+	if response == nil || response.Body == nil {
+		return "", false, nil
+	}
+	defer response.Body.Close()
+
+	body, err := io.ReadAll(io.LimitReader(response.Body, heartbeatResponseBodyLogLimit+1))
+	if err != nil {
+		return "", false, err
+	}
+
+	truncated := len(body) > heartbeatResponseBodyLogLimit
+	if truncated {
+		body = body[:heartbeatResponseBodyLogLimit]
+	}
+
+	return strings.TrimSpace(string(body)), truncated, nil
+}
+
+func formatHeartbeatFailureLog(response *http.Response, requestErr error, responseBody string, responseBodyTruncated bool, responseBodyErr error, elapsed time.Duration) string {
+	details := make([]string, 0, 6)
+	if response != nil {
+		details = append(details, "status_code="+strconv.Itoa(response.StatusCode))
+		if response.Status != "" {
+			details = append(details, "status="+strconv.Quote(response.Status))
+		}
+	} else {
+		details = append(details, "status_code=none")
+	}
+	details = append(details, "duration="+elapsed.Round(time.Millisecond).String())
+	if requestErr != nil {
+		details = append(details, "request_error="+strconv.Quote(requestErr.Error()))
+	}
+	if responseBody != "" {
+		details = append(details, "response_body="+strconv.Quote(responseBody))
+	}
+	if responseBodyTruncated {
+		details = append(details, "response_body_truncated=true")
+	}
+	if responseBodyErr != nil {
+		details = append(details, "response_body_error="+strconv.Quote(responseBodyErr.Error()))
+	}
+
+	return "cloud.HandleHeartBeat(): heartbeat request to Kerberos Hub failed: " + strings.Join(details, ", ")
+}
+
 func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) {
 	log.Log.Debug("cloud.HandleHeartBeat(): started")
 
@@ -644,20 +692,26 @@ loop:
 
 				var jsonStr = []byte(object)
 				buffy := bytes.NewBuffer(jsonStr)
-				req, _ := http.NewRequest("POST", hubURI, buffy)
-				req.Header.Set("Content-Type", "application/json")
-				resp, err := client.Do(req)
-				if resp != nil {
-					resp.Body.Close()
+				requestStarted := time.Now()
+				req, requestErr := http.NewRequest("POST", hubURI, buffy)
+				var resp *http.Response
+				if requestErr == nil {
+					req.Header.Set("Content-Type", "application/json")
+					resp, requestErr = client.Do(req)
 				}
-				if err == nil && resp.StatusCode == 200 {
+				if requestErr == nil && resp != nil && resp.StatusCode == http.StatusOK {
+					if resp.Body != nil {
+						_, _ = io.Copy(io.Discard, resp.Body)
+						resp.Body.Close()
+					}
 					communication.CloudTimestamp.Store(time.Now().Unix())
 					log.Log.Info("cloud.HandleHeartBeat(): (200) Heartbeat received by Kerberos Hub.")
 				} else {
+					responseBody, responseBodyTruncated, responseBodyErr := readHeartbeatResponseBody(resp)
 					if communication.CloudTimestamp != nil && communication.CloudTimestamp.Load() != nil {
 						communication.CloudTimestamp.Store(int64(0))
 					}
-					log.Log.Error("cloud.HandleHeartBeat(): (400) Something went wrong while sending to Kerberos Hub.")
+					log.Log.Error(formatHeartbeatFailureLog(resp, requestErr, responseBody, responseBodyTruncated, responseBodyErr, time.Since(requestStarted)))
 				}
 			} else {
 				log.Log.Error("cloud.HandleHeartBeat(): Disabled as we do not have a public key defined.")

machinery/src/cloud/cloud_test.go

@@ -0,0 +1,51 @@
+package cloud
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestHeartbeatFailureLogIncludesHubResponse(t *testing.T) {
+	response := &http.Response{
+		StatusCode: http.StatusBadRequest,
+		Status:     "400 Bad Request",
+		Body:       io.NopCloser(strings.NewReader(`{"error":"invalid heartbeat"}`)),
+	}
+
+	responseBody, truncated, err := readHeartbeatResponseBody(response)
+	if err != nil {
+		t.Fatalf("readHeartbeatResponseBody() error = %v", err)
+	}
+	message := formatHeartbeatFailureLog(response, nil, responseBody, truncated, nil, 125*time.Millisecond)
+
+	for _, expected := range []string{
+		"status_code=400",
+		`status="400 Bad Request"`,
+		"duration=125ms",
+		`response_body="{\"error\":\"invalid heartbeat\"}"`,
+	} {
+		if !strings.Contains(message, expected) {
+			t.Errorf("formatHeartbeatFailureLog() = %q, want it to contain %q", message, expected)
+		}
+	}
+}
+
+func TestReadHeartbeatResponseBodyTruncatesLargeBody(t *testing.T) {
+	response := &http.Response{
+		Body: io.NopCloser(strings.NewReader(strings.Repeat("x", heartbeatResponseBodyLogLimit+1))),
+	}
+
+	body, truncated, err := readHeartbeatResponseBody(response)
+	if err != nil {
+		t.Fatalf("readHeartbeatResponseBody() error = %v", err)
+	}
+	if !truncated {
+		t.Fatal("readHeartbeatResponseBody() truncated = false, want true")
+	}
+	if len(body) != heartbeatResponseBodyLogLimit {
+		t.Fatalf("len(body) = %d, want %d", len(body), heartbeatResponseBodyLogLimit)
+	}
+}

machinery/src/cloud/livemoq_enabled.go

@@ -151,6 +151,13 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 	}
 	defer client.Close()
 
+	sessionCtx, cancelSessionWatch := context.WithCancel(ctx)
+	defer cancelSessionWatch()
+	sessionClosed := make(chan error, 1)
+	go func() {
+		sessionClosed <- client.Session().Closed(sessionCtx)
+	}()
+
 	broadcast, err := client.CreateBroadcast(config.broadcast)
 	if err != nil {
 		return fmt.Errorf("create broadcast: %w", err)
@@ -180,6 +187,12 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 	var lastDuplicateKeyframeWarning time.Time
 	idle := false
 	for {
+		select {
+		case err := <-sessionClosed:
+			return fmt.Errorf("relay session closed: %w", err)
+		default:
+		}
+
 		packet, err := cursor.ReadPacket()
 		if err != nil {
 			return fmt.Errorf("read packet: %w", err)