diff --git a/.vscode/launch.json b/.vscode/launch.json index d552cd71..fedc3c31 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -17,7 +17,7 @@ "8080" ], "envFile": "${workspaceFolder}/machinery/.env.local", - "buildFlags": "--tags dynamic", + "buildFlags": "--tags dynamic,moq", "env": { "GOWORK": "off" }, diff --git a/README-RTSPS-TLS.md b/README-RTSPS-TLS.md index aec391d0..242eb74d 100644 --- a/README-RTSPS-TLS.md +++ b/README-RTSPS-TLS.md @@ -377,7 +377,7 @@ For a process running directly in the same environment: AGENT_CAPTURE_IPCAMERA_RTSP="rtsps://:@10.0.30.11:9554/?inst=1" AGENT_CAPTURE_IPCAMERA_SUB_RTSP="rtsps://:@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://:@10.0.30.11:9554/?inst=1' \ -e 'AGENT_CAPTURE_IPCAMERA_SUB_RTSP=rtsps://:@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. diff --git a/machinery/data/tus/1786715908_3-737_encrypted_0-0-0-0_-1_25498.mp4.primary.json b/machinery/data/tus/1786715908_3-737_encrypted_0-0-0-0_-1_25498.mp4.primary.json new file mode 100644 index 00000000..53e323dc --- /dev/null +++ b/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} \ No newline at end of file diff --git a/machinery/src/capture/gortsplib.go b/machinery/src/capture/gortsplib.go index 3a6b43e2..163d6d1a 100644 --- a/machinery/src/capture/gortsplib.go +++ b/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, error) { + if os.Getenv(rtspsInsecureEnv) == "true" { + return &tls.Config{InsecureSkipVerify: true}, nil // #nosec G402 -- explicit opt-in for cameras with self-signed certificates + } -func rtspsTLSConfig() *tls.Config { - if os.Getenv(rtspsInsecureEnv) != "true" { - return nil + 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{InsecureSkipVerify: true} // #nosec G402 -- explicit opt-in for cameras with self-signed certificates + 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) diff --git a/machinery/src/capture/gortsplib_test.go b/machinery/src/capture/gortsplib_test.go index adfbfb21..9ff5d6e0 100644 --- a/machinery/src/capture/gortsplib_test.go +++ b/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") + } + }) } diff --git a/machinery/src/cloud/cloud.go b/machinery/src/cloud/cloud.go index 8000c023..cea35d97 100644 --- a/machinery/src/cloud/cloud.go +++ b/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.") diff --git a/machinery/src/cloud/cloud_test.go b/machinery/src/cloud/cloud_test.go new file mode 100644 index 00000000..149b6c47 --- /dev/null +++ b/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) + } +} diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go index 8aa371d8..7e536dde 100644 --- a/machinery/src/cloud/livemoq_enabled.go +++ b/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)