Merge pull request #336 from 4alvit/fix/snapshot-closed-queue

fix(capture): stop snapshot readers when the packet queue closes
2026-10-19 17:38:25 +00:00 · 2026-10-02 08:00:56 +02:00 · 2026-09-29 11:55:38 -07:00 · 2026-09-27 22:11:04 +02:00 · 2026-09-27 22:00:52 +02:00 · 2026-09-24 19:24:52 +02:00
26 changed files with 2014 additions and 264 deletions
@@ -341,9 +341,10 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_CAPTURE_PIXEL_CHANGE`                | If `CONTINUOUS` set to `false`, the number of pixel require to change before motion triggers.   | "150"                          |
 | `AGENT_CAPTURE_FRAGMENTED`                  | Set the format of the recorded MP4 to fragmented (suitable for HLS).                            | "false"                        |
 | `AGENT_CAPTURE_FRAGMENTED_DURATION`         | If `AGENT_CAPTURE_FRAGMENTED` set to `true`, define the duration (seconds) of a fragment.       | "8"                            |
-| `AGENT_MQTT_URI`                            | An MQTT broker endpoint that is used for bi-directional communication (live view, onvif, etc)   | "tcp://mqtt.kerberos.io:1883"  |
+| `AGENT_MQTT_URI`                            | MQTT broker endpoint for bi-directional communication. Accepts ActiveMQ `mqtt+ssl://` URLs.     | "tcp://mqtt.kerberos.io:1883"  |
 | `AGENT_MQTT_USERNAME`                       | Username of the MQTT broker.                                                                    | ""                             |
 | `AGENT_MQTT_PASSWORD`                       | Password of the MQTT broker.                                                                    | ""                             |
+| `AGENT_REMOTE_ACCESS_ENABLED`               | Allow encrypted Hub MQTT sessions to stream Agent logs and open an interactive shell. Enable only for trusted deployments. | "false"                        |
 | `AGENT_REALTIME_PROCESSING`                 | If `AGENT_REALTIME_PROCESSING` set to `true`, the agent will send key frames to the topic       | ""                             |
 | `AGENT_REALTIME_PROCESSING_TOPIC`           | The topic to which keyframes will be sent in base64 encoded format.                             | ""                             |
 | `AGENT_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn-fra1.kerberos.io:3478"|
@@ -364,11 +365,13 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_KERBEROSVAULT_SECRET_KEY`            | The secret key of a Kerberos Vault account.                                                     | ""                             |
 | `AGENT_KERBEROSVAULT_PROVIDER`              | A Kerberos Vault provider you have created (optional).                                          | ""                             |
 | `AGENT_KERBEROSVAULT_DIRECTORY`             | The directory, in the Kerberos vault, where the recordings will be stored.                      | ""                             |
+| `AGENT_KERBEROSVAULT_CUSTOM_HEADERS`        | JSON object of custom upload headers that Vault includes as integration metadata.               | `{"site_id":"site-1"}`         |
 | `AGENT_KERBEROSVAULT_SECONDARY_URI`         | The Kerberos Vault API url.                                                                     | "https://vault.domain.com/api" |
 | `AGENT_KERBEROSVAULT_SECONDARY_ACCESS_KEY`  | The access key of a secondary Kerberos Vault account.                                           | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_SECRET_KEY`  | The secret key of a secondary Kerberos Vault account.                                           | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_PROVIDER`    | A secondary Kerberos Vault provider you have created (optional).                                | ""                             |
 | `AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY`   | The directory, in the secondary Kerberos vault, where the recordings will be stored.            | ""                             |
+| `AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS` | JSON object of custom upload headers for the secondary Vault.                                | `{"site_id":"site-1"}`         |
 | `AGENT_DROPBOX_ACCESS_TOKEN`                | The Access Token from your Dropbox app, that is used to leverage the Dropbox SDK.               | ""                             |
 | `AGENT_DROPBOX_DIRECTORY`                   | The directory, in Dropbox, where the recordings will be stored.                                 | ""                             |
 | `AGENT_ENCRYPTION`                          | Enable 'true' or disable 'false' end-to-end encryption for MQTT messages.                       | "false"                        |
@@ -379,6 +382,13 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_SIGNING`                             | Enable 'true' or disable 'false' for signing recordings.                                        | "true"                         |
 | `AGENT_SIGNING_PRIVATE_KEY`                 | The private key (RSA) to sign the recordings fingerprint to validate origin.                    | "" - uses default one if empty |

+Remote console access is disabled unless `AGENT_REMOTE_ACCESS_ENABLED=true`.
+The Agent also rejects remote session messages unless Hub encryption or
+end-to-end MQTT encryption is configured and used. A remote shell runs inside
+the Agent process environment as the Agent operating-system user; it is not an
+SSH server and does not expose a new network port. Keep the feature disabled on
+deployments where Hub owners should not have operating-system access.
+
 ### Resumable upload chunk size

 Hub and Vault resumable uploads use `AGENT_TUS_CHUNK_SIZE_BYTES` as the maximum
@@ -400,6 +410,94 @@ Smaller chunks provide more frequent resumable checkpoints but create more HTTP
 requests. Larger chunks reduce request overhead but require more data to be
 retransmitted when a request fails.

+### Custom metadata headers for Kerberos Vault
+
+Use `AGENT_KERBEROSVAULT_CUSTOM_HEADERS` to attach your own metadata (for
+example a site, line, or equipment identifier) to every recording uploaded to
+Kerberos Vault. The value is a JSON object whose keys are header names and whose
+values are strings. The Agent sends these as HTTP headers on every upload, and
+Vault adds them to the `metadata` object of its integration messages (Kafka,
+MQTT, SQS, RabbitMQ, etc.):
+
+```json
+{
+  "timestamp": "2026-09-24T15:06:45Z",
+  "key": "agent1/1790262405_3-425_agent1_0-0-0-0_-1_10000.mp4",
+  "device": "agent1",
+  "metadata": {
+    "site_id": "site-1",
+    "line_id": "line-2",
+    "equipment_id": "equipment-3"
+  }
+}
+```
+
+The value must be valid JSON on a single line, so wrap it in single quotes in a
+shell:
+
+```bash
+docker run -p 80:80 --name mycamera \
+  -e AGENT_CLOUD=kstorage \
+  -e AGENT_KERBEROSVAULT_URI=https://vault.domain.com/api \
+  -e AGENT_KERBEROSVAULT_ACCESS_KEY=xxx \
+  -e AGENT_KERBEROSVAULT_SECRET_KEY=xxx \
+  -e AGENT_KERBEROSVAULT_DIRECTORY=agent1 \
+  -e AGENT_KERBEROSVAULT_CUSTOM_HEADERS='{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}' \
+  -d --restart=always kerberos/agent:latest
+```
+
+In `docker compose` or an `.env` file:
+
+```yaml
+services:
+  agent:
+    image: kerberos/agent:latest
+    environment:
+      AGENT_CLOUD: kstorage
+      AGENT_KERBEROSVAULT_CUSTOM_HEADERS: '{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}'
+```
+
+```dotenv
+AGENT_KERBEROSVAULT_CUSTOM_HEADERS={"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}
+```
+
+In Kubernetes, quote the value so YAML keeps it as a string:
+
+```yaml
+env:
+  - name: AGENT_KERBEROSVAULT_CUSTOM_HEADERS
+    value: '{"site_id":"site-1","line_id":"line-2","equipment_id":"equipment-3"}'
+```
+
+Use `AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS` for the secondary Vault. In
+ConfigMap deployments you can set shared defaults with the `GLOBAL_` prefix
+(for example `GLOBAL_AGENT_KERBEROSVAULT_CUSTOM_HEADERS`) and override them per
+agent with `AGENT_KERBEROSVAULT_CUSTOM_HEADERS`. You can also set the same JSON
+under **Settings → Persistence → Custom upload headers** in the Agent UI, which
+stores it as `kstorage.custom_headers` in `config.json`.
+
+The rules for this value are:
+
+- It must be a JSON object with string values. Numbers, `null`, arrays, and
+  nested objects are rejected.
+- It can contain up to 32 headers, with names up to 128 characters and values
+  up to 4096 bytes.
+- Header names must be valid HTTP tokens (letters, digits, and ``!#$%&'*+-.^_`|~``).
+  Keys are matched case-insensitively, so `site_id` and `SITE_ID` cannot both
+  be used.
+- Reserved names are rejected: anything starting with `X-Kerberos-`,
+  `Authorization`, `Cookie`, `Host`, `Content-Type`, `Content-Length`,
+  `Trace-Id`, tus protocol headers (`Tus-Resumable`, `Upload-*`), and
+  hop-by-hop headers.
+- Values cannot contain line breaks, tabs, or other control characters.
+
+If the value is invalid, the Agent logs an error and does not upload to that
+Vault until the configuration is fixed. The recording stays queued, so it will
+upload once the value is corrected. Header names reach Vault unchanged, but
+proxies may normalise their casing. If Vault redirects an upload to another
+host, the Agent removes the custom headers and Vault credentials before
+following the redirect.
+

 ## Encryption

@@ -1296,6 +1296,9 @@ const docTemplate = `{
                    "description": "old way, remove this",
                    "type": "string"
                },
+                "custom_headers": {
+                    "type": "string"
+                },
                "directory": {
                    "type": "string"
                },
@@ -1289,6 +1289,9 @@
                    "description": "old way, remove this",
                    "type": "string"
                },
+                "custom_headers": {
+                    "type": "string"
+                },
                "directory": {
                    "type": "string"
                },
@@ -11,6 +11,7 @@ require (
 	github.com/bluenviron/gortsplib/v5 v5.6.3
 	github.com/bluenviron/mediacommon v1.14.0
 	github.com/cedricve/go-onvif v0.0.0-20200222191200-567e8ce298f6
+	github.com/creack/pty v1.1.24
 	github.com/dromara/carbon/v2 v2.6.8
 	github.com/dropbox/dropbox-sdk-go-unofficial/v6 v6.0.5
 	github.com/eclipse/paho.mqtt.golang v1.5.0
@@ -456,6 +456,8 @@ github.com/cncf/xds/go v0.0.0-20240905190251-b4127c9b8d78/go.mod h1:W+zGtBO5Y1Ig
 github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
+github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -53,12 +53,13 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration
 	}
 }

-func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4) models.RecordingUploadMetadata {
+func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4, encrypted bool) models.RecordingUploadMetadata {
 	metadata := models.RecordingUploadMetadata{
 		FileName:  filepath.Base(name),
 		DeviceKey: deviceKey,
 		Timestamp: timestamp,
 		Duration:  mp4Video.VideoTotalDuration,
+		Encrypted: encrypted,
 	}
 	value := mp4Video.AverageFPS()
 	if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) {
@@ -481,6 +482,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Info("capture.main.HandleRecordStream(continuous): no video data recorded, not renaming file.")
 					}

+					encrypted := false
 					// Check if we need to encrypt the recording.
 					if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 						// reopen file into memory 'fullName'
@@ -493,6 +495,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 								err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 								if err != nil {
 									log.Error("capture.main.HandleRecordStream(continuous): error writing file: " + err.Error())
+								} else {
+									encrypted = true
 								}
 							} else {
 								log.Error("capture.main.HandleRecordStream(continuous): error encrypting file: " + err.Error())
@@ -502,7 +506,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}

-					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video, encrypted))

 					recordingStatus = "idle"

@@ -638,6 +642,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Info("capture.main.HandleRecordStream(continuous): no video data recorded, not renaming file.")
 					}

+					encrypted := false
 					// Check if we need to encrypt the recording.
 					if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 						// reopen file into memory 'fullName'
@@ -650,6 +655,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 								err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 								if err != nil {
 									log.Error("capture.main.HandleRecordStream(motiondetection): error writing file: " + err.Error())
+								} else {
+									encrypted = true
 								}
 							} else {
 								log.Error("capture.main.HandleRecordStream(motiondetection): error encrypting file: " + err.Error())
@@ -659,7 +666,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}

-					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video, encrypted))

 					recordingStatus = "idle"

@@ -905,6 +912,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					log.Info("capture.main.HandleRecordStream(motiondetection): no video data recorded, not renaming file.")
 				}

+				encrypted := false
 				// Check if we need to encrypt the recording.
 				if config.Encryption != nil && config.Encryption.Enabled == "true" && config.Encryption.Recordings == "true" && config.Encryption.SymmetricKey != "" {
 					// reopen file into memory 'fullName'
@@ -917,6 +925,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 							err := os.WriteFile(fullName, []byte(encryptedContents), 0644)
 							if err != nil {
 								log.Error("capture.main.HandleRecordStream(motiondetection): error writing file: " + err.Error())
+							} else {
+								encrypted = true
 							}
 						} else {
 							log.Error("capture.main.HandleRecordStream(motiondetection): error encrypting file: " + err.Error())
@@ -926,7 +936,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}

-				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video))
+				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video, encrypted))

 				// Clean up the recording directory if necessary.
 				CleanupRecordingDirectory(configDirectory, configuration)
@@ -1060,6 +1070,10 @@ func Base64Image(captureDevice *Capture, communication *models.Communication, co
 					count++
 					continue
 				}
+			} else {
+				// A stopped run closes this queue permanently. Retrying its
+				// cursor would spin on EOF even after the HTTP client leaves.
+				break
 			}
 		} else {
 			break
@@ -1103,6 +1117,10 @@ func JpegImage(captureDevice *Capture, communication *models.Communication) imag
 				} else {
 					break
 				}
+			} else {
+				// A stopped run closes this queue permanently. Retrying its
+				// cursor would spin on EOF even after the HTTP client leaves.
+				break
 			}
 		} else {
 			break
@@ -42,7 +42,7 @@ func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
 	}

 	mp4Video := &video.MP4{VideoTotalDuration: 20452, SampleCount: 613}
-	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video)
+	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video, true)
 	queueRecordingForUpload(configDirectory, metadata)

 	got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
@@ -54,7 +54,7 @@ func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
 		t.Fatalf("decode upload marker: %v", err)
 	}
 	expectedFPS := mp4Video.AverageFPS()
-	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || math.Abs(stored.FPS-expectedFPS) > 1e-9 {
+	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || math.Abs(stored.FPS-expectedFPS) > 1e-9 || !stored.Encrypted {
 		t.Fatalf("upload marker = %+v", stored)
 	}
 	if stored.FPS == math.Floor(stored.FPS) {
@@ -0,0 +1,68 @@
+package capture
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/packets"
+)
+
+func TestSnapshotReturnsAfterQueueClosure(t *testing.T) {
+	for _, stream := range []string{"main", "sub"} {
+		for _, format := range []string{"jpeg", "base64"} {
+			for _, alreadyClosed := range []bool{true, false} {
+				name := stream + "/" + format + "/closed_during_read"
+				if alreadyClosed {
+					name = stream + "/" + format + "/already_closed"
+				}
+				t.Run(name, func(t *testing.T) {
+					communication := &models.Communication{}
+					run := models.NewAgentRun(context.Background(), communication, false)
+					queue := packets.NewQueue()
+					captureDevice := &Capture{}
+					if stream == "main" {
+						captureDevice.SetMainClient("rtsp://unused.invalid/main")
+						run.SetMainQueue(queue)
+					} else {
+						captureDevice.SetSubClient("rtsp://unused.invalid/sub")
+						run.SetSubQueue(queue)
+					}
+					if err := run.Activate(); err != nil {
+						t.Fatal(err)
+					}
+					defer queue.Close()
+					if alreadyClosed {
+						queue.Close()
+					}
+					done := make(chan bool, 1)
+					go func() {
+						if format == "jpeg" {
+							image := JpegImage(captureDevice, communication)
+							done <- image.Rect.Empty()
+						} else {
+							done <- Base64Image(captureDevice, communication, &models.Configuration{}) == ""
+						}
+					}()
+					if !alreadyClosed {
+						select {
+						case <-done:
+							t.Fatal("snapshot returned before a frame or queue closure")
+						case <-time.After(10 * time.Millisecond):
+						}
+						queue.Close()
+					}
+					select {
+					case empty := <-done:
+						if !empty {
+							t.Fatal("closed empty queue produced a snapshot")
+						}
+					case <-time.After(time.Second):
+						t.Fatal("snapshot did not return after its packet queue closed")
+					}
+				})
+			}
+		}
+	}
+}
@@ -0,0 +1,195 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+)
+
+const (
+	customHeadersManifestHeader = "X-Kerberos-Storage-Custom-Headers"
+	maxCustomHeaders            = 32
+	maxCustomHeaderNameLength   = 128
+	maxCustomHeaderValueLength  = 4096
+)
+
+var disallowedCustomHeaders = map[string]struct{}{
+	"authorization":       {},
+	"connection":          {},
+	"content-length":      {},
+	"content-type":        {},
+	"cookie":              {},
+	"expect":              {},
+	"host":                {},
+	"keep-alive":          {},
+	"proxy-authorization": {},
+	"proxy-connection":    {},
+	"te":                  {},
+	"trailer":             {},
+	"trace-id":            {},
+	"transfer-encoding":   {},
+	"tus-resumable":       {},
+	"upgrade":             {},
+}
+
+func parseCustomVaultHeaders(value string) (map[string]string, error) {
+	if strings.TrimSpace(value) == "" {
+		return nil, nil
+	}
+
+	// Decode into pointers so JSON null values can be told apart from strings.
+	var rawHeaders map[string]*string
+	if err := json.Unmarshal([]byte(value), &rawHeaders); err != nil {
+		return nil, fmt.Errorf("custom Vault headers must be a JSON object of string values: %w", err)
+	}
+	if rawHeaders == nil {
+		return nil, errors.New("custom Vault headers must be a JSON object of string values")
+	}
+	if len(rawHeaders) > maxCustomHeaders {
+		return nil, fmt.Errorf("custom Vault headers exceed the limit of %d", maxCustomHeaders)
+	}
+	headers := make(map[string]string, len(rawHeaders))
+	seen := make(map[string]struct{}, len(rawHeaders))
+	for name, rawValue := range rawHeaders {
+		if rawValue == nil {
+			return nil, fmt.Errorf("custom Vault header %q must be a string, not null", name)
+		}
+		value := *rawValue
+		headers[name] = value
+		if err := validateCustomVaultHeader(name, value); err != nil {
+			return nil, err
+		}
+		lowerName := strings.ToLower(name)
+		if _, duplicate := seen[lowerName]; duplicate {
+			return nil, fmt.Errorf("custom Vault header %q is duplicated with different casing", name)
+		}
+		seen[lowerName] = struct{}{}
+	}
+	return headers, nil
+}
+
+func validateCustomVaultHeader(name, value string) error {
+	if name == "" || len(name) > maxCustomHeaderNameLength || !isHTTPToken(name) {
+		return fmt.Errorf("invalid custom Vault header name %q", name)
+	}
+	lowerName := strings.ToLower(name)
+	if strings.HasPrefix(lowerName, "x-kerberos-") {
+		return fmt.Errorf("custom Vault header %q uses the reserved X-Kerberos namespace", name)
+	}
+	if strings.HasPrefix(lowerName, "upload-") {
+		return fmt.Errorf("custom Vault header %q uses the reserved tus Upload namespace", name)
+	}
+	if _, disallowed := disallowedCustomHeaders[lowerName]; disallowed {
+		return fmt.Errorf("custom Vault header %q is reserved", name)
+	}
+	if len(value) > maxCustomHeaderValueLength {
+		return fmt.Errorf("custom Vault header %q exceeds the %d byte value limit", name, maxCustomHeaderValueLength)
+	}
+	for _, character := range value {
+		if character < 0x20 || character == 0x7f {
+			return fmt.Errorf("custom Vault header %q contains an invalid value", name)
+		}
+	}
+	return nil
+}
+
+func isHTTPToken(value string) bool {
+	if value == "" {
+		return false
+	}
+	for _, character := range value {
+		if (character >= 'a' && character <= 'z') ||
+			(character >= 'A' && character <= 'Z') ||
+			(character >= '0' && character <= '9') ||
+			strings.ContainsRune("!#$%&'*+-.^_`|~", character) {
+			continue
+		}
+		return false
+	}
+	return true
+}
+
+func setCustomVaultHeaders(headers http.Header, customHeaders map[string]string) error {
+	if len(customHeaders) == 0 {
+		return nil
+	}
+
+	names := make([]string, 0, len(customHeaders))
+	for name, value := range customHeaders {
+		if err := validateCustomVaultHeader(name, value); err != nil {
+			return err
+		}
+		names = append(names, name)
+		// Assign directly so the configured name is not canonicalised by Set.
+		headers[name] = []string{value}
+	}
+	sort.Strings(names)
+	manifest, err := json.Marshal(names)
+	if err != nil {
+		return err
+	}
+	headers.Set(customHeadersManifestHeader, base64.RawURLEncoding.EncodeToString(manifest))
+	return nil
+}
+
+const maxVaultRedirects = 10
+
+// stripVaultHeadersOnCrossHostRedirect removes Vault credentials and the
+// customer-supplied metadata (custom headers, their manifest and the tus
+// Upload-Metadata that embeds them) before following a redirect to another host.
+func stripVaultHeadersOnCrossHostRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) >= maxVaultRedirects {
+		return fmt.Errorf("stopped after %d redirects", maxVaultRedirects)
+	}
+	if len(via) == 0 || req.URL.Host == via[0].URL.Host {
+		return nil
+	}
+
+	customNames := customVaultHeaderNamesFromManifest(req.Header)
+	for name := range req.Header {
+		canonicalName := http.CanonicalHeaderKey(name)
+		_, custom := customNames[strings.ToLower(name)]
+		if custom || strings.HasPrefix(canonicalName, "X-Kerberos-") || canonicalName == "Upload-Metadata" {
+			delete(req.Header, name)
+		}
+	}
+	return nil
+}
+
+func customVaultHeaderNamesFromManifest(headers http.Header) map[string]struct{} {
+	encoded := headers.Get(customHeadersManifestHeader)
+	if encoded == "" {
+		return nil
+	}
+	manifest, err := base64.RawURLEncoding.DecodeString(encoded)
+	if err != nil {
+		return nil
+	}
+	var names []string
+	if err := json.Unmarshal(manifest, &names); err != nil {
+		return nil
+	}
+	lowerNames := make(map[string]struct{}, len(names))
+	for _, name := range names {
+		lowerNames[strings.ToLower(name)] = struct{}{}
+	}
+	return lowerNames
+}
+
+func encodeCustomVaultHeaders(customHeaders map[string]string) (string, error) {
+	if len(customHeaders) == 0 {
+		return "", nil
+	}
+	encoded, err := json.Marshal(customHeaders)
+	if err != nil {
+		return "", err
+	}
+	if len(encoded) == 0 {
+		return "", errors.New("custom Vault headers could not be encoded")
+	}
+	return string(encoded), nil
+}
@@ -0,0 +1,100 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+	"reflect"
+	"testing"
+)
+
+func TestCustomVaultHeadersAreValidatedAndAdded(t *testing.T) {
+	customHeaders, err := parseCustomVaultHeaders(`{"site_id":"site-1","line_id":"line-2"}`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	headers := http.Header{}
+	if err := setCustomVaultHeaders(headers, customHeaders); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(headers["site_id"], []string{"site-1"}) || !reflect.DeepEqual(headers["line_id"], []string{"line-2"}) {
+		t.Fatalf("custom header names were not preserved: %#v", headers)
+	}
+	if _, canonicalised := headers["Site_id"]; canonicalised {
+		t.Fatalf("custom header name was canonicalised: %#v", headers)
+	}
+
+	encodedManifest, err := base64.RawURLEncoding.DecodeString(headers.Get(customHeadersManifestHeader))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var names []string
+	if err := json.Unmarshal(encodedManifest, &names); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(names, []string{"line_id", "site_id"}) {
+		t.Fatalf("manifest = %#v", names)
+	}
+}
+
+func TestCustomVaultHeadersRejectReservedAndInvalidValues(t *testing.T) {
+	for _, value := range []string{
+		`{"X-Kerberos-Storage-AccessKey":"secret"}`,
+		`{"Authorization":"secret"}`,
+		`{"site_id":"one\nInjected: value"}`,
+		`{"bad header":"value"}`,
+		`{"site_id":1}`,
+		`{"Content-Type":"text/plain"}`,
+		`{"site_id":"one","SITE_ID":"two"}`,
+		`{"site_id":null}`,
+		`{"site_id":"one\ttwo"}`,
+		`{"Upload-Checksum":"sha1 abc"}`,
+		`{"upload-concat":"final;/files/a"}`,
+		`{"Upload-Defer-Length":"1"}`,
+		`null`,
+	} {
+		if _, err := parseCustomVaultHeaders(value); err == nil {
+			t.Fatalf("parseCustomVaultHeaders(%q) succeeded", value)
+		}
+	}
+}
+
+func TestVaultRedirectStripsCustomHeadersOnHostChange(t *testing.T) {
+	newRedirect := func(t *testing.T, host string) (*http.Request, []*http.Request) {
+		t.Helper()
+		original, err := http.NewRequest(http.MethodPost, "https://vault.example.com/api/storage", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		redirected, err := http.NewRequest(http.MethodPost, "https://"+host+"/api/storage", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		redirected.Header.Set("X-Kerberos-Storage-AccessKey", "secret")
+		redirected.Header.Set("Upload-Metadata", "custom_headers abc")
+		redirected.Header.Set("Tus-Resumable", "1.0.0")
+		if err := setCustomVaultHeaders(redirected.Header, map[string]string{"site_id": "site-1", "Line-Id": "line-2"}); err != nil {
+			t.Fatal(err)
+		}
+		return redirected, []*http.Request{original}
+	}
+
+	crossHost, via := newRedirect(t, "other.example.com")
+	if err := stripVaultHeadersOnCrossHostRedirect(crossHost, via); err != nil {
+		t.Fatal(err)
+	}
+	if len(crossHost.Header) != 1 || crossHost.Header.Get("Tus-Resumable") != "1.0.0" {
+		t.Fatalf("cross-host redirect headers = %#v", crossHost.Header)
+	}
+
+	sameHost, via := newRedirect(t, "vault.example.com")
+	if err := stripVaultHeadersOnCrossHostRedirect(sameHost, via); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(sameHost.Header["site_id"], []string{"site-1"}) ||
+		sameHost.Header.Get("X-Kerberos-Storage-AccessKey") != "secret" ||
+		sameHost.Header.Get(customHeadersManifestHeader) == "" {
+		t.Fatalf("same-host redirect headers = %#v", sameHost.Header)
+	}
+}
@@ -178,6 +178,10 @@ func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, s
 // is sent as the body of a POST to {URI}/storage. Kept for backwards
 // compatibility with vault deployments that do not support resumable uploads.
 func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, label string) (bool, bool, string, error) {
+	customHeaders, err := parseCustomVaultHeaders(vault.CustomHeaders)
+	if err != nil {
+		return false, false, "", err
+	}
 	fullname := "data/recordings/" + fileName

 	file, err := os.Open(fullname)
@@ -203,6 +207,9 @@ func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, la
 	}
 	req.Header.Set("Content-Type", "video/mp4")
 	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+	if err := setCustomVaultHeaders(req.Header, customHeaders); err != nil {
+		return false, false, "", err
+	}
 	setQueuedRecordingMetadataHeaders(req.Header, fileName)

 	client := newVaultHTTPClient(0)
@@ -267,7 +274,7 @@ func newVaultHTTPClient(timeout time.Duration) *http.Client {
 		transport.TLSClientConfig.InsecureSkipVerify = true
 	}

-	client := &http.Client{Transport: transport}
+	client := &http.Client{Transport: transport, CheckRedirect: stripVaultHeadersOnCrossHostRedirect}
 	if timeout > 0 {
 		client.Timeout = timeout
 	}
@@ -15,6 +15,7 @@ import (
 const recordingFPSHeader = "X-Kerberos-Storage-Fps"
 const recordingDurationHeader = "X-Kerberos-Storage-Duration"
 const recordingTimestampHeader = "X-Kerberos-Storage-Timestamp"
+const recordingEncryptedHeader = "X-Kerberos-Storage-Encrypted"

 // queuedRecordingFPS reads the FPS snapshot written into the upload marker
 // when the recording was finalized. Historical empty markers intentionally
@@ -84,5 +85,8 @@ func setQueuedRecordingMetadataHeaders(header http.Header, fileName string) {
 		if metadata.Timestamp > 0 {
 			header.Set(recordingTimestampHeader, strconv.FormatInt(metadata.Timestamp, 10))
 		}
+		if metadata.Encrypted {
+			header.Set(recordingEncryptedHeader, "true")
+		}
 	}
 }
@@ -141,19 +141,6 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	size := info.Size()

 	client := newVaultHTTPClient(0)
-	client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
-		if len(via) == 0 {
-			return nil
-		}
-		if req.URL.Host != via[0].URL.Host {
-			for k := range req.Header {
-				if strings.HasPrefix(http.CanonicalHeaderKey(k), "X-Kerberos-") {
-					req.Header.Del(k)
-				}
-			}
-		}
-		return nil
-	}

 	sidecar := tusSidecarPath(fileName, slot)
 	uploadURL := loadTusResumeState(sidecar, baseURL)
@@ -311,6 +298,10 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 // X-Kerberos-Storage-* headers on every request and routing (directory/provider)
 // is additionally carried in the tus Upload-Metadata.
 func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) {
+	customHeaders, err := parseCustomVaultHeaders(vault.CustomHeaders)
+	if err != nil {
+		return false, false, true, "", err
+	}
 	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
 	metadataValues := map[string]string{
 		"filename":  fileName,
@@ -321,10 +312,22 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName,
 		"cloudkey":  publicKey,
 		"fps":       queuedRecordingFPS(fileName),
 	}
+	if encodedHeaders, err := encodeCustomVaultHeaders(customHeaders); err != nil {
+		return false, false, true, "", err
+	} else if encodedHeaders != "" {
+		metadataValues["custom_headers"] = encodedHeaders
+	}
+	customRequestHeaders := http.Header{}
+	if err := setCustomVaultHeaders(customRequestHeaders, customHeaders); err != nil {
+		return false, false, true, "", err
+	}
 	addRecordingTusMetadata(metadataValues, fileName)
 	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
+		for name, values := range customRequestHeaders {
+			h[name] = append([]string(nil), values...)
+		}
 	}
 	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
 }
@@ -361,6 +364,9 @@ func addRecordingTusMetadata(values map[string]string, fileName string) {
 	if metadata.Timestamp > 0 {
 		values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10)
 	}
+	if metadata.Encrypted {
+		values["encrypted"] = "true"
+	}
 }

 // tusCreate performs the tus "creation" request (POST). On success it returns
@@ -3,6 +3,7 @@ package cloud
 import (
 	"bytes"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net"
@@ -304,7 +305,9 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	withRecording(t, fileName, payload)
 	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":29.97}`)

-	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	vault := testVault(ts.URL)
+	vault.CustomHeaders = `{"site_id":"site-1","equipment_id":"equipment-3"}`
+	uploaded, responded, supported, _, err := uploadVaultResumable(vault, "pk", "dev", fileName, "test", "primary")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -328,6 +331,23 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	if got := metadata["timestamp"]; got != "1785934709414" {
 		t.Fatalf("POST metadata timestamp = %q, want %q", got, "1785934709414")
 	}
+	var customHeaders map[string]string
+	if err := json.Unmarshal([]byte(metadata["custom_headers"]), &customHeaders); err != nil {
+		t.Fatal(err)
+	}
+	if customHeaders["site_id"] != "site-1" || customHeaders["equipment_id"] != "equipment-3" {
+		t.Fatalf("POST custom header metadata = %#v", customHeaders)
+	}
+	for _, method := range []string{http.MethodPost, http.MethodHead, http.MethodPatch} {
+		for _, request := range srv.requestsForMethod(method) {
+			if request.header.Get("site_id") != "site-1" || request.header.Get("equipment_id") != "equipment-3" {
+				t.Fatalf("%s custom headers = %#v", method, request.header)
+			}
+			if request.header.Get(customHeadersManifestHeader) == "" {
+				t.Fatalf("%s is missing the custom header manifest", method)
+			}
+		}
+	}
 }

 func TestQueuedRecordingFPSValidation(t *testing.T) {
@@ -386,7 +406,7 @@ func TestQueuedRecordingFPSAllowsMissingHistoricalMarker(t *testing.T) {
 func TestQueuedRecordingMetadataHeaders(t *testing.T) {
 	fileName := "recording.mp4"
 	withRecording(t, fileName, []byte("recording"))
-	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25}`)
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25,"encrypted":true}`)

 	header := make(http.Header)
 	setQueuedRecordingMetadataHeaders(header, fileName)
@@ -399,6 +419,14 @@ func TestQueuedRecordingMetadataHeaders(t *testing.T) {
 	if got := header.Get(recordingTimestampHeader); got != "1785934709414" {
 		t.Fatalf("timestamp header = %q", got)
 	}
+	if got := header.Get(recordingEncryptedHeader); got != "true" {
+		t.Fatalf("encrypted header = %q", got)
+	}
+	metadata := map[string]string{}
+	addRecordingTusMetadata(metadata, fileName)
+	if got := metadata["encrypted"]; got != "true" {
+		t.Fatalf("encrypted TUS metadata = %q", got)
+	}
 }

 func TestQueuedRecordingFPSAllowsLegacyMarkerFileName(t *testing.T) {
@@ -315,6 +315,15 @@ func initConfigPointers(config *models.Config) {
 func applyAgentEnvVars(configuration *models.Configuration, prefix string, applyDefaults bool) {
 	environmentVariables := os.Environ()

+	if prefix == "" {
+		if deploymentName := os.Getenv("DEPLOYMENT_NAME"); deploymentName != "" {
+			configuration.Config.Name = deploymentName
+			if configuration.Config.FriendlyName == "" {
+				configuration.Config.FriendlyName = deploymentName
+			}
+		}
+	}
+
 	// Initialize the configuration for some new fields.
 	if configuration.Config.KStorageSecondary == nil {
 		configuration.Config.KStorageSecondary = &models.KStorage{}
@@ -603,6 +612,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_KERBEROSVAULT_DIRECTORY":
 				configuration.Config.KStorage.Directory = value
 				break
+			case "AGENT_KERBEROSVAULT_CUSTOM_HEADERS":
+				configuration.Config.KStorage.CustomHeaders = value
+				break

 			/* Retry policy and timeout */
 			case "AGENT_KERBEROSVAULT_MAX_RETRIES":
@@ -634,6 +646,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY":
 				configuration.Config.KStorageSecondary.Directory = value
 				break
+			case "AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS":
+				configuration.Config.KStorageSecondary.CustomHeaders = value
+				break

 			/* When storing in dropbox */
 			case "AGENT_DROPBOX_ACCESS_TOKEN":
@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"errors"
+	"os"
 	"testing"
 	"time"

@@ -11,6 +12,53 @@ import (
 	"go.mongodb.org/mongo-driver/mongo"
 )

+func TestApplyAgentEnvVarsDeploymentName(t *testing.T) {
+	tests := []struct {
+		name           string
+		prefix         string
+		deploymentName string
+		agentName      string
+		initialName    string
+		initialDisplay string
+		wantName       string
+		wantDisplay    string
+	}{
+		{name: "deployment fallback", deploymentName: "camera-1", wantName: "camera-1", wantDisplay: "camera-1"},
+		{name: "explicit display name", deploymentName: "camera-1", agentName: "Front Door", wantName: "camera-1", wantDisplay: "Front Door"},
+		{name: "existing display name", deploymentName: "camera-1", initialDisplay: "Front Door", wantName: "camera-1", wantDisplay: "Front Door"},
+		{name: "deployment overrides bundled identity", deploymentName: "camera-1", initialName: "default", wantName: "camera-1", wantDisplay: "camera-1"},
+		{name: "no deployment", initialName: "existing", initialDisplay: "Existing Camera", wantName: "existing", wantDisplay: "Existing Camera"},
+		{name: "standalone display name", agentName: "Front Door", wantDisplay: "Front Door"},
+		{name: "global layer excludes deployment", prefix: "GLOBAL_", deploymentName: "camera-1"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Setenv("DEPLOYMENT_NAME", test.deploymentName)
+			t.Setenv(test.prefix+"AGENT_NAME", test.agentName)
+			if test.agentName == "" {
+				if err := os.Unsetenv(test.prefix + "AGENT_NAME"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			configuration := &models.Configuration{Config: models.Config{
+				Name:         test.initialName,
+				FriendlyName: test.initialDisplay,
+			}}
+			initConfigPointers(&configuration.Config)
+
+			applyAgentEnvVars(configuration, test.prefix, false)
+
+			if got := configuration.Config.Name; got != test.wantName {
+				t.Errorf("Name = %q, want %q", got, test.wantName)
+			}
+			if got := configuration.Config.FriendlyName; got != test.wantDisplay {
+				t.Errorf("FriendlyName = %q, want %q", got, test.wantDisplay)
+			}
+		})
+	}
+}
+
 func TestApplyAgentEnvVarsPixelChangeThresholdDefault(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -40,6 +88,22 @@ func TestApplyAgentEnvVarsPixelChangeThresholdDefault(t *testing.T) {
 	}
 }

+func TestApplyAgentEnvVarsVaultCustomHeaders(t *testing.T) {
+	t.Setenv("AGENT_KERBEROSVAULT_CUSTOM_HEADERS", `{"site_id":"site-1"}`)
+	t.Setenv("AGENT_KERBEROSVAULT_SECONDARY_CUSTOM_HEADERS", `{"site_id":"site-2"}`)
+	configuration := &models.Configuration{}
+	initConfigPointers(&configuration.Config)
+
+	applyAgentEnvVars(configuration, "", false)
+
+	if got := configuration.Config.KStorage.CustomHeaders; got != `{"site_id":"site-1"}` {
+		t.Fatalf("primary custom headers = %q", got)
+	}
+	if got := configuration.Config.KStorageSecondary.CustomHeaders; got != `{"site_id":"site-2"}` {
+		t.Fatalf("secondary custom headers = %q", got)
+	}
+}
+
 func intPointer(value int) *int {
 	return &value
 }
@@ -187,6 +187,7 @@ type KStorage struct {
 	SecretAccessKey string `json:"secret_access_key,omitempty" bson:"secret_access_key,omitempty"`
 	Provider        string `json:"provider,omitempty" bson:"provider,omitempty"`
 	Directory       string `json:"directory,omitempty" bson:"directory,omitempty"`
+	CustomHeaders   string `json:"custom_headers,omitempty" bson:"custom_headers,omitempty"`
 	MaxRetries      int    `json:"max_retries,omitempty" bson:"max_retries,omitempty"`
 	Timeout         int    `json:"timeout,omitempty" bson:"timeout,omitempty"`
 }
@@ -324,3 +324,23 @@ type TriggerRelay struct {
 	DeviceId  string `json:"device_id"` // device id
 	Token     string `json:"token"`     // token
 }
+
+// RemoteSessionPayload controls an interactive shell or log stream over MQTT.
+// Data is base64 encoded so terminal control bytes remain valid JSON.
+type RemoteSessionPayload struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind,omitempty"`
+	Data      string `json:"data,omitempty"`
+	Rows      uint16 `json:"rows,omitempty"`
+	Columns   uint16 `json:"columns,omitempty"`
+	Tail      int    `json:"tail,omitempty"`
+}
+
+type RemoteSessionStatus struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind"`
+	State     string `json:"state"`
+	Error     string `json:"error,omitempty"`
+}
@@ -16,6 +16,7 @@ type RecordingUploadMetadata struct {
 	Timestamp int64   `json:"timestamp"` // Unix milliseconds.
 	Duration  uint64  `json:"duration"`  // Milliseconds.
 	FPS       float64 `json:"fps,omitempty"`
+	Encrypted bool    `json:"encrypted,omitempty"`
 }

 // RecordingUploadMetadataFileName returns the queue marker name associated
@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"encoding/xml"
 	"errors"
+	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"strconv"
 	"strings"
@@ -130,7 +132,7 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode

 					if onvifAction.Action == "absolute-move" {

-						// We will move the camera to zero position.
+						// Move the camera to the saved absolute position.
 						x := ptzAction.X
 						y := ptzAction.Y
 						z := ptzAction.Z
@@ -141,42 +143,13 @@ func HandleONVIFActions(configuration *models.Configuration, communication *mode
 						// Log functions
 						log.Debug("onvif.HandleONVIFActions(): functions: " + strings.Join(functions, ", "))

-						// Check if we need to use absolute or continuous move
-						/*canAbsoluteMove := false
-						canContinuousMove := false
-
-						if len(functions) > 0 {
-							for _, function := range functions {
-								if function == "AbsolutePanTiltMove" || function == "AbsoluteZoomMove" {
-									canAbsoluteMove = true
-								} else if function == "ContinuousPanTiltMove" || function == "ContinuousZoomMove" {
-									canContinuousMove = true
-								}
-							}
-						}*/
-
-						// Ideally we should be able to use the AbsolutePanTiltMove function, but it looks like
-						// the current detection through GetPTZFuntionsFromDevice is not working properly. Therefore we will fallback
-						// on the ContinuousPanTiltMove function which is more compatible with more cameras.
-						err = AbsolutePanTiltMoveFake(device, configurations, token, x, y, z)
+						err = MoveToPosition(device, configurations, token, x, y, z)
 						if err != nil {
-							log.Debug("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: " + err.Error())
+							log.Error("onvif.HandleONVIFActions() - MoveToPosition: " + err.Error())
 						} else {
-							log.Info("onvif.HandleONVIFActions() - AbsolutePanTitleMoveFake: successfully moved camera.")
+							log.Info("onvif.HandleONVIFActions() - MoveToPosition: successfully moved camera.")
 						}

-						/*if canAbsoluteMove {
-							err = AbsolutePanTiltMove(device, configurations, token, x, y, z)
-							if err != nil {
-								log.Error("HandleONVIFActions (AbsolutePanTitleMove): " + err.Error())
-							}
-						} else if canContinuousMove {
-							err = AbsolutePanTiltMoveFake(device, configurations, token, x, y, z)
-							if err != nil {
-								log.Error("HandleONVIFActions (AbsolutePanTitleMoveFake): " + err.Error())
-							}
-						}*/
-
 					} else if onvifAction.Action == "preset" {

 						// Execute the preset
@@ -471,16 +444,20 @@ func handleONVIFResponse(operation string, response *http.Response, requestErr e
 }

 func AbsolutePanTiltMove(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]

 	absolutePantiltVector := xsdonvif.Vector2D{
 		X:     pan,
 		Y:     tilt,
-		Space: configuration.PTZConfiguration[0].DefaultAbsolutePantTiltPositionSpace,
+		Space: ptzConfiguration.DefaultAbsolutePantTiltPositionSpace,
 	}

 	absoluteZoomVector := xsdonvif.Vector1D{
 		X:     zoom,
-		Space: configuration.PTZConfiguration[0].DefaultAbsoluteZoomPositionSpace,
+		Space: ptzConfiguration.DefaultAbsoluteZoomPositionSpace,
 	}

 	resp, err := device.CallMethod(ptz.AbsoluteMove{
@@ -493,240 +470,325 @@ func AbsolutePanTiltMove(device *onvif.Device, configuration ptz.GetConfiguratio
 	return handleONVIFResponse("absolute_move", resp, err)
 }

-// This function will simulate the AbsolutePanTiltMove function.
-// However the AboslutePanTiltMove function is not working on all cameras.
-// So we'll use the ContinuousMove function to simulate the AbsolutePanTiltMove function using the position polling.
-func AbsolutePanTiltMoveFake(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
-	position, err := GetPosition(device, token)
-	if position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01 && position.PanTilt.Y >= tilt-0.01 && position.PanTilt.Y <= tilt+0.01 && position.Zoom.X >= zoom-0.01 && position.Zoom.X <= zoom+0.01 {
-		log.Debug("AbsolutePanTiltMoveFake: already at position")
-	} else {
-
-		// The speed of panning, the higher the faster we'll pan the camera
-		// value is a range between 0 and 1.
-		speed := 0.6
-		wait := 100 * time.Millisecond
-
-		// We'll move quickly to the position (might be inaccurate)
-		err = ZoomOutCompletely(device, configuration, token)
-		err = PanUntilPosition(device, configuration, token, pan, zoom, speed, wait)
-		err = TiltUntilPosition(device, configuration, token, tilt, zoom, speed, wait)
-
-		// Now we'll move a bit slower to make sure we are ok (will be more accurate)
-		speed = 0.1
-		wait = 200 * time.Millisecond
-
-		err = PanUntilPosition(device, configuration, token, pan, zoom, speed, wait)
-		err = TiltUntilPosition(device, configuration, token, tilt, zoom, speed, wait)
-		err = ZoomUntilPosition(device, configuration, token, zoom, speed, wait)
-
-		return err
-	}
-	return err
+type ptzMoveClient interface {
+	GetPosition() (xsdonvif.PTZVector, error)
+	AbsoluteMove(ptzTarget) error
+	ContinuousMove(ptzVelocity) error
+	Stop() error
 }

-func ZoomOutCompletely(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken) error {
-	// Zoom out completely!!!
-	zoomOut := xsdonvif.Vector1D{
-		X:     -1,
-		Space: configuration.PTZConfiguration[0].DefaultContinuousZoomVelocitySpace,
+type ptzTarget struct {
+	Pan  float64
+	Tilt float64
+	Zoom float64
+}
+
+type ptzVelocity struct {
+	Pan  float64
+	Tilt float64
+	Zoom float64
+}
+
+type ptzMoveSupport struct {
+	Absolute   bool
+	Continuous bool
+}
+
+type ptzMoveOptions struct {
+	PanTolerance      float64
+	TiltTolerance     float64
+	ZoomTolerance     float64
+	NativeTimeout     time.Duration
+	ContinuousTimeout time.Duration
+	NoProgressTimeout time.Duration
+	PollInterval      time.Duration
+	ProgressEpsilon   float64
+	VelocityGain      float64
+	MinVelocity       float64
+	MaxVelocity       float64
+	VelocityChange    float64
+}
+
+var defaultPTZMoveOptions = ptzMoveOptions{
+	PanTolerance:      0.01,
+	TiltTolerance:     0.005,
+	ZoomTolerance:     0.005,
+	NativeTimeout:     8 * time.Second,
+	ContinuousTimeout: 15 * time.Second,
+	NoProgressTimeout: 3 * time.Second,
+	PollInterval:      150 * time.Millisecond,
+	ProgressEpsilon:   0.001,
+	VelocityGain:      1.5,
+	MinVelocity:       0.08,
+	MaxVelocity:       0.7,
+	VelocityChange:    0.05,
+}
+
+type devicePTZMoveClient struct {
+	device        *onvif.Device
+	configuration xsdonvif.PTZConfiguration
+	token         xsdonvif.ReferenceToken
+}
+
+func (client *devicePTZMoveClient) GetPosition() (xsdonvif.PTZVector, error) {
+	return GetPosition(client.device, client.token)
+}
+
+func (client *devicePTZMoveClient) AbsoluteMove(target ptzTarget) error {
+	configuration := ptz.GetConfigurationsResponse{
+		PTZConfiguration: []xsdonvif.PTZConfiguration{client.configuration},
 	}
-	_, err := device.CallMethod(ptz.ContinuousMove{
-		ProfileToken: &token,
-		Velocity: xsdonvif.PTZSpeedZoom{
-			Zoom: zoomOut,
+	return AbsolutePanTiltMove(client.device, configuration, client.token, target.Pan, target.Tilt, target.Zoom)
+}
+
+func (client *devicePTZMoveClient) ContinuousMove(velocity ptzVelocity) error {
+	panTilt := xsdonvif.Vector2D{
+		X:     velocity.Pan,
+		Y:     velocity.Tilt,
+		Space: client.configuration.DefaultContinuousPanTiltVelocitySpace,
+	}
+	zoom := xsdonvif.Vector1D{
+		X:     velocity.Zoom,
+		Space: client.configuration.DefaultContinuousZoomVelocitySpace,
+	}
+	resp, err := client.device.CallMethod(ptz.ContinuousMove{
+		ProfileToken: &client.token,
+		Velocity: ptz.Speed{
+			PanTilt: &panTilt,
+			Zoom:    &zoom,
 		},
 	})
-	if err != nil {
-		log.Error("ZoomOutCompletely: " + err.Error())
-	}
+	return handleONVIFResponse("continuous_move_to_position", resp, err)
+}

-	for {
-		position, _ := GetPosition(device, token)
-		if position.Zoom.X == 0 {
-			break
-		}
-		time.Sleep(250 * time.Millisecond)
-	}
-
-	_, err = device.CallMethod(ptz.Stop{
-		ProfileToken: token,
+func (client *devicePTZMoveClient) Stop() error {
+	resp, err := client.device.CallMethod(ptz.Stop{
+		ProfileToken: client.token,
+		PanTilt:      true,
 		Zoom:         true,
 	})
+	return handleONVIFResponse("move_to_position_stop", resp, err)
+}
+
+func MoveToPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]
+	client := &devicePTZMoveClient{
+		device:        device,
+		configuration: ptzConfiguration,
+		token:         token,
+	}
+	support := ptzMoveSupport{
+		Absolute: ptzConfiguration.DefaultAbsolutePantTiltPositionSpace != nil &&
+			ptzConfiguration.DefaultAbsoluteZoomPositionSpace != nil,
+		Continuous: ptzConfiguration.DefaultContinuousPanTiltVelocitySpace != nil &&
+			ptzConfiguration.DefaultContinuousZoomVelocitySpace != nil,
+	}
+	return moveToPosition(client, support, ptzTarget{Pan: pan, Tilt: tilt, Zoom: zoom}, defaultPTZMoveOptions)
+}
+
+// AbsolutePanTiltMoveFake preserves the continuous-move compatibility path for callers that explicitly request it.
+func AbsolutePanTiltMoveFake(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64, zoom float64) error {
+	if len(configuration.PTZConfiguration) == 0 {
+		return errors.New("camera returned no PTZ configuration")
+	}
+	ptzConfiguration := configuration.PTZConfiguration[0]
+	client := &devicePTZMoveClient{
+		device:        device,
+		configuration: ptzConfiguration,
+		token:         token,
+	}
+	support := ptzMoveSupport{
+		Continuous: ptzConfiguration.DefaultContinuousPanTiltVelocitySpace != nil &&
+			ptzConfiguration.DefaultContinuousZoomVelocitySpace != nil,
+	}
+	return moveToPosition(client, support, ptzTarget{Pan: pan, Tilt: tilt, Zoom: zoom}, defaultPTZMoveOptions)
+}
+
+func moveToPosition(client ptzMoveClient, support ptzMoveSupport, target ptzTarget, options ptzMoveOptions) error {
+	if err := validatePTZTarget(target); err != nil {
+		return err
+	}
+
+	position, err := client.GetPosition()
 	if err != nil {
-		log.Error("ZoomOutCompletely: " + err.Error())
+		return fmt.Errorf("read initial PTZ position: %w", err)
 	}
-	return err
+	_, _, reached, err := ptzTargetError(position, target, options)
+	if err != nil {
+		return err
+	}
+	if reached {
+		return nil
+	}
+
+	var nativeErr error
+	if support.Absolute {
+		if err := client.AbsoluteMove(target); err != nil {
+			nativeErr = fmt.Errorf("native absolute move: %w", err)
+		} else if err := waitForPTZTarget(client, target, options.NativeTimeout, options); err != nil {
+			nativeErr = fmt.Errorf("verify native absolute move: %w", err)
+		} else {
+			return nil
+		}
+
+		if err := client.Stop(); err != nil {
+			nativeErr = errors.Join(nativeErr, fmt.Errorf("stop native absolute move: %w", err))
+		}
+		log.WithError(nativeErr).Warn("Native ONVIF absolute move failed; using continuous-move fallback")
+	}
+
+	if !support.Continuous {
+		if nativeErr != nil {
+			return errors.Join(nativeErr, errors.New("camera does not advertise complete continuous PTZ support"))
+		}
+		return errors.New("camera does not advertise a supported absolute or continuous PTZ movement method")
+	}
+
+	if err := moveContinuouslyToPTZTarget(client, target, options); err != nil {
+		if nativeErr != nil {
+			return errors.Join(nativeErr, fmt.Errorf("continuous-move fallback: %w", err))
+		}
+		return fmt.Errorf("continuous move: %w", err)
+	}
+	return nil
 }

-func PanUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
-
-	if position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01 {
-
-	} else {
-
-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionX := speed
-		if position.PanTilt.X > pan {
-			directionX = speed * -1
-		}
-
-		panTiltVector := xsdonvif.Vector2D{
-			X:     directionX,
-			Y:     0,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousPanTiltVelocitySpace,
-		}
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity: xsdonvif.PTZSpeedPanTilt{
-				PanTilt: panTiltVector,
-			},
-		})
-		err = handleONVIFResponse("continuous_pan", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
-		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.PanTilt.X == -1 || position.PanTilt.X == 1 || (directionX > 0 && position.PanTilt.X >= pan) || (directionX < 0 && position.PanTilt.X <= pan) || (position.PanTilt.X >= pan-0.01 && position.PanTilt.X <= pan+0.01) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
-		}
-
-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
+func waitForPTZTarget(client ptzMoveClient, target ptzTarget, timeout time.Duration, options ptzMoveOptions) error {
+	deadline := time.Now().Add(timeout)
+	lastProgress := time.Now()
+	bestDistance := math.Inf(1)

+	for {
+		position, err := client.GetPosition()
 		if err != nil {
-			log.Error("ContinuousPanTiltMove (Pan): " + err.Error())
+			return fmt.Errorf("read PTZ position: %w", err)
 		}
+		_, distance, reached, err := ptzTargetError(position, target, options)
+		if err != nil {
+			return err
+		}
+		if reached {
+			return nil
+		}
+
+		now := time.Now()
+		if bestDistance-distance >= options.ProgressEpsilon {
+			bestDistance = distance
+			lastProgress = now
+		}
+		if !now.Before(deadline) {
+			return fmt.Errorf("target not reached within %s", timeout)
+		}
+		if now.Sub(lastProgress) >= options.NoProgressTimeout {
+			return fmt.Errorf("PTZ position made no progress for %s", options.NoProgressTimeout)
+		}
+		time.Sleep(options.PollInterval)
 	}
-	return err
 }

-func TiltUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, tilt float64, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
+func moveContinuouslyToPTZTarget(client ptzMoveClient, target ptzTarget, options ptzMoveOptions) (err error) {
+	defer func() {
+		if stopErr := client.Stop(); stopErr != nil {
+			err = errors.Join(err, fmt.Errorf("stop continuous move: %w", stopErr))
+		}
+	}()

-	if position.PanTilt.Y >= tilt-0.005 && position.PanTilt.Y <= tilt+0.005 {
+	deadline := time.Now().Add(options.ContinuousTimeout)
+	lastProgress := time.Now()
+	bestDistance := math.Inf(1)
+	var lastVelocity ptzVelocity
+	haveVelocity := false

-	} else {
-
-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionY := speed
-		if position.PanTilt.Y > tilt {
-			directionY = speed * -1
+	for {
+		position, positionErr := client.GetPosition()
+		if positionErr != nil {
+			return fmt.Errorf("read PTZ position: %w", positionErr)
+		}
+		delta, distance, reached, targetErr := ptzTargetError(position, target, options)
+		if targetErr != nil {
+			return targetErr
+		}
+		if reached {
+			return nil
 		}

-		panTiltVector := xsdonvif.Vector2D{
-			X:     0,
-			Y:     directionY,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousPanTiltVelocitySpace,
-		}
-
-		velocity := xsdonvif.PTZSpeedPanTilt{
-			PanTilt: panTiltVector,
-		}
-
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity:     velocity,
-		})
-		err = handleONVIFResponse("continuous_tilt", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
 		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.PanTilt.Y == -1 || position.PanTilt.Y == 1 || (directionY > 0 && position.PanTilt.Y >= tilt) || (directionY < 0 && position.PanTilt.Y <= tilt) || (position.PanTilt.Y >= tilt-0.005 && position.PanTilt.Y <= tilt+0.005) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
+		if bestDistance-distance >= options.ProgressEpsilon {
+			bestDistance = distance
+			lastProgress = now
+		}
+		if !now.Before(deadline) {
+			return fmt.Errorf("target not reached within %s", options.ContinuousTimeout)
+		}
+		if now.Sub(lastProgress) >= options.NoProgressTimeout {
+			return fmt.Errorf("PTZ position made no progress for %s", options.NoProgressTimeout)
 		}

-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
-
-		if err != nil {
-			log.Error("ContinuousPanTiltMove (Tilt): " + err.Error())
+		velocity := ptzVelocity{
+			Pan:  proportionalPTZVelocity(delta.Pan, options.PanTolerance, options),
+			Tilt: proportionalPTZVelocity(delta.Tilt, options.TiltTolerance, options),
+			Zoom: proportionalPTZVelocity(delta.Zoom, options.ZoomTolerance, options),
 		}
+		if !haveVelocity || ptzVelocityChanged(lastVelocity, velocity, options.VelocityChange) {
+			if moveErr := client.ContinuousMove(velocity); moveErr != nil {
+				return fmt.Errorf("set continuous PTZ velocity: %w", moveErr)
+			}
+			lastVelocity = velocity
+			haveVelocity = true
+		}
+		time.Sleep(options.PollInterval)
 	}
-	return err
 }

-func ZoomUntilPosition(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, zoom float64, speed float64, wait time.Duration) error {
-	position, err := GetPosition(device, token)
+func ptzTargetError(position xsdonvif.PTZVector, target ptzTarget, options ptzMoveOptions) (ptzVelocity, float64, bool, error) {
+	if position.PanTilt == nil || position.Zoom == nil {
+		return ptzVelocity{}, 0, false, errors.New("camera returned an incomplete PTZ position")
+	}
+	delta := ptzVelocity{
+		Pan:  target.Pan - position.PanTilt.X,
+		Tilt: target.Tilt - position.PanTilt.Y,
+		Zoom: target.Zoom - position.Zoom.X,
+	}
+	reached := math.Abs(delta.Pan) <= options.PanTolerance &&
+		math.Abs(delta.Tilt) <= options.TiltTolerance &&
+		math.Abs(delta.Zoom) <= options.ZoomTolerance
+	distance := math.Abs(delta.Pan) + math.Abs(delta.Tilt) + math.Abs(delta.Zoom)
+	return delta, distance, reached, nil
+}

-	if position.Zoom.X >= zoom-0.005 && position.Zoom.X <= zoom+0.005 {
+func proportionalPTZVelocity(delta float64, tolerance float64, options ptzMoveOptions) float64 {
+	if math.Abs(delta) <= tolerance {
+		return 0
+	}
+	speed := math.Abs(delta) * options.VelocityGain
+	speed = math.Max(options.MinVelocity, math.Min(speed, options.MaxVelocity))
+	return math.Copysign(speed, delta)
+}

-	} else {
+func ptzVelocityChanged(previous ptzVelocity, next ptzVelocity, threshold float64) bool {
+	return math.Abs(previous.Pan-next.Pan) >= threshold ||
+		math.Abs(previous.Tilt-next.Tilt) >= threshold ||
+		math.Abs(previous.Zoom-next.Zoom) >= threshold
+}

-		// We'll need to determine if we need to move CW or CCW.
-		// Check the current position and compare it with the desired position.
-		directionZ := speed
-		if position.Zoom.X > zoom {
-			directionZ = speed * -1
-		}
-
-		zoomVector := xsdonvif.Vector1D{
-			X:     directionZ,
-			Space: configuration.PTZConfiguration[0].DefaultContinuousZoomVelocitySpace,
-		}
-		resp, err := device.CallMethod(ptz.ContinuousMove{
-			ProfileToken: &token,
-			Velocity: xsdonvif.PTZSpeedZoom{
-				Zoom: zoomVector,
-			},
-		})
-		err = handleONVIFResponse("continuous_zoom_to_position", resp, err)
-
-		// While moving we'll check if we reached the desired position.
-		// or if we overshot the desired position.
-
-		// Break after 3seconds
-		now := time.Now()
-		for {
-			position, _ := GetPosition(device, token)
-			if position.Zoom.X == -1 || position.Zoom.X == 1 || (directionZ > 0 && position.Zoom.X >= zoom) || (directionZ < 0 && position.Zoom.X <= zoom) || (position.Zoom.X >= zoom-0.005 && position.Zoom.X <= zoom+0.005) {
-				break
-			}
-			if time.Since(now) > 3*time.Second {
-				break
-			}
-			time.Sleep(wait)
-		}
-
-		_, err = device.CallMethod(ptz.Stop{
-			ProfileToken: token,
-			PanTilt:      true,
-			Zoom:         true,
-		})
-
-		if err != nil {
-			log.Error("ContinuousPanTiltMove (Zoom): " + err.Error())
+func validatePTZTarget(target ptzTarget) error {
+	values := []struct {
+		name  string
+		value float64
+	}{
+		{name: "pan", value: target.Pan},
+		{name: "tilt", value: target.Tilt},
+		{name: "zoom", value: target.Zoom},
+	}
+	for _, value := range values {
+		if math.IsNaN(value.value) || math.IsInf(value.value, 0) {
+			return fmt.Errorf("invalid PTZ %s target", value.name)
 		}
 	}
-	return err
+	return nil
 }

 func ContinuousPanTilt(device *onvif.Device, configuration ptz.GetConfigurationsResponse, token xsdonvif.ReferenceToken, pan float64, tilt float64) error {
@@ -895,6 +957,10 @@ func GetPTZFunctionsFromDevice(configurations ptz.GetConfigurationsResponse) ([]
 	canZoom := false
 	canPanTilt := false

+	if len(configurations.PTZConfiguration) == 0 {
+		return functions, canZoom, canPanTilt
+	}
+
 	if configurations.PTZConfiguration[0].DefaultAbsolutePantTiltPositionSpace != nil {
 		functions = append(functions, "AbsolutePanTiltMove")
 		canPanTilt = true
@@ -0,0 +1,264 @@
+package onvif
+
+import (
+	"encoding/xml"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/onvif/ptz"
+	xsdonvif "github.com/kerberos-io/onvif/xsd/onvif"
+)
+
+type fakePTZMoveClient struct {
+	current              ptzTarget
+	target               ptzTarget
+	lastVelocity         ptzVelocity
+	absoluteErr          error
+	continuousErr        error
+	stopErr              error
+	ignoreAbsolute       bool
+	stalled              bool
+	incompletePosition   bool
+	positionErr          error
+	positionErrAtCall    int
+	positionCalls        int
+	absoluteCalls        int
+	continuousVelocities []ptzVelocity
+	stopCalls            int
+	moving               bool
+}
+
+func (client *fakePTZMoveClient) GetPosition() (xsdonvif.PTZVector, error) {
+	client.positionCalls++
+	if client.positionErr != nil && client.positionCalls == client.positionErrAtCall {
+		return xsdonvif.PTZVector{}, client.positionErr
+	}
+	if client.moving && !client.stalled {
+		client.current.Pan = moveFakeAxis(client.current.Pan, client.target.Pan, client.lastVelocity.Pan)
+		client.current.Tilt = moveFakeAxis(client.current.Tilt, client.target.Tilt, client.lastVelocity.Tilt)
+		client.current.Zoom = moveFakeAxis(client.current.Zoom, client.target.Zoom, client.lastVelocity.Zoom)
+	}
+
+	panTilt := &xsdonvif.Vector2D{X: client.current.Pan, Y: client.current.Tilt}
+	if client.incompletePosition {
+		return xsdonvif.PTZVector{PanTilt: panTilt}, nil
+	}
+	return xsdonvif.PTZVector{
+		PanTilt: panTilt,
+		Zoom:    &xsdonvif.Vector1D{X: client.current.Zoom},
+	}, nil
+}
+
+func (client *fakePTZMoveClient) AbsoluteMove(target ptzTarget) error {
+	client.absoluteCalls++
+	client.target = target
+	if client.absoluteErr != nil {
+		return client.absoluteErr
+	}
+	if !client.ignoreAbsolute {
+		client.current = target
+	}
+	return nil
+}
+
+func (client *fakePTZMoveClient) ContinuousMove(velocity ptzVelocity) error {
+	if client.continuousErr != nil {
+		return client.continuousErr
+	}
+	client.lastVelocity = velocity
+	client.continuousVelocities = append(client.continuousVelocities, velocity)
+	client.moving = true
+	return nil
+}
+
+func (client *fakePTZMoveClient) Stop() error {
+	client.stopCalls++
+	client.moving = false
+	return client.stopErr
+}
+
+func moveFakeAxis(current float64, target float64, velocity float64) float64 {
+	step := velocity * 0.5
+	if step == 0 {
+		return current
+	}
+	next := current + step
+	if (velocity > 0 && next >= target) || (velocity < 0 && next <= target) {
+		return target
+	}
+	return next
+}
+
+func testPTZMoveOptions() ptzMoveOptions {
+	options := defaultPTZMoveOptions
+	options.NativeTimeout = 50 * time.Millisecond
+	options.ContinuousTimeout = 100 * time.Millisecond
+	options.NoProgressTimeout = 10 * time.Millisecond
+	options.PollInterval = time.Millisecond
+	options.ProgressEpsilon = 0.0001
+	options.VelocityChange = 0.01
+	return options
+}
+
+func TestMoveToPositionUsesVerifiedNativeMove(t *testing.T) {
+	target := ptzTarget{Pan: 0.5, Tilt: -0.25, Zoom: 0.4}
+	client := &fakePTZMoveClient{}
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, testPTZMoveOptions())
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if client.absoluteCalls != 1 {
+		t.Fatalf("absolute calls = %d, want 1", client.absoluteCalls)
+	}
+	if len(client.continuousVelocities) != 0 {
+		t.Fatalf("continuous calls = %d, want 0", len(client.continuousVelocities))
+	}
+}
+
+func TestMoveToPositionFallsBackToSimultaneousContinuousMove(t *testing.T) {
+	target := ptzTarget{Pan: 0.5, Tilt: -0.25, Zoom: 0.4}
+	client := &fakePTZMoveClient{
+		target:      target,
+		absoluteErr: errors.New("absolute move rejected"),
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, testPTZMoveOptions())
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if len(client.continuousVelocities) == 0 {
+		t.Fatal("continuous fallback was not used")
+	}
+	first := client.continuousVelocities[0]
+	if first.Pan == 0 || first.Tilt == 0 || first.Zoom == 0 {
+		t.Fatalf("first fallback velocity = %+v, want all axes moving", first)
+	}
+	if client.stopCalls != 2 {
+		t.Fatalf("stop calls = %d, want native cleanup and fallback cleanup", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionFallsBackWhenNativeMoveMakesNoProgress(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:         target,
+		ignoreAbsolute: true,
+	}
+	options := testPTZMoveOptions()
+	options.NoProgressTimeout = 3 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Absolute: true, Continuous: true}, target, options)
+	if err != nil {
+		t.Fatalf("moveToPosition() error = %v", err)
+	}
+	if len(client.continuousVelocities) == 0 {
+		t.Fatal("continuous fallback was not used")
+	}
+}
+
+func TestMoveToPositionStopsAfterContinuousStatusError(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:            target,
+		positionErr:       errors.New("status unavailable"),
+		positionErrAtCall: 3,
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "status unavailable") {
+		t.Fatalf("moveToPosition() error = %v, want status error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionStopsWhenContinuousMoveIsRejected(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:        target,
+		continuousErr: errors.New("continuous move rejected"),
+	}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "continuous move rejected") {
+		t.Fatalf("moveToPosition() error = %v, want movement error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionReturnsNoProgressErrorAndStops(t *testing.T) {
+	target := ptzTarget{Pan: 0.3, Tilt: 0.2, Zoom: 0.1}
+	client := &fakePTZMoveClient{
+		target:  target,
+		stalled: true,
+	}
+	options := testPTZMoveOptions()
+	options.NoProgressTimeout = 3 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, options)
+	if err == nil || !strings.Contains(err.Error(), "made no progress") {
+		t.Fatalf("moveToPosition() error = %v, want no-progress error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionReturnsDeadlineErrorAndStops(t *testing.T) {
+	target := ptzTarget{Pan: 100, Tilt: 100, Zoom: 100}
+	client := &fakePTZMoveClient{target: target}
+	options := testPTZMoveOptions()
+	options.ContinuousTimeout = 3 * time.Millisecond
+	options.NoProgressTimeout = 50 * time.Millisecond
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, target, options)
+	if err == nil || !strings.Contains(err.Error(), "target not reached within") {
+		t.Fatalf("moveToPosition() error = %v, want deadline error", err)
+	}
+	if client.stopCalls != 1 {
+		t.Fatalf("stop calls = %d, want 1", client.stopCalls)
+	}
+}
+
+func TestMoveToPositionRejectsIncompletePosition(t *testing.T) {
+	client := &fakePTZMoveClient{incompletePosition: true}
+
+	err := moveToPosition(client, ptzMoveSupport{Continuous: true}, ptzTarget{Pan: 0.1}, testPTZMoveOptions())
+	if err == nil || !strings.Contains(err.Error(), "incomplete PTZ position") {
+		t.Fatalf("moveToPosition() error = %v, want incomplete-position error", err)
+	}
+}
+
+func TestCombinedContinuousMoveVelocityIncludesEveryAxis(t *testing.T) {
+	request := ptz.ContinuousMove{
+		Velocity: ptz.Speed{
+			PanTilt: &xsdonvif.Vector2D{X: 0.5, Y: -0.25},
+			Zoom:    &xsdonvif.Vector1D{X: 0.2},
+		},
+	}
+
+	encoded, err := xml.Marshal(request)
+	if err != nil {
+		t.Fatalf("xml.Marshal() error = %v", err)
+	}
+	xmlBody := string(encoded)
+	if !strings.Contains(xmlBody, `PanTilt x="0.5" y="-0.25"`) {
+		t.Fatalf("ContinuousMove XML = %s, want pan and tilt", xmlBody)
+	}
+	if !strings.Contains(xmlBody, `Zoom x="0.2"`) {
+		t.Fatalf("ContinuousMove XML = %s, want zoom", xmlBody)
+	}
+}
+
+func TestGetPTZFunctionsHandlesMissingConfiguration(t *testing.T) {
+	functions, canZoom, canPanTilt := GetPTZFunctionsFromDevice(ptz.GetConfigurationsResponse{})
+	if len(functions) != 0 || canZoom || canPanTilt {
+		t.Fatalf("GetPTZFunctionsFromDevice() = %v, %t, %t; want empty capabilities", functions, canZoom, canPanTilt)
+	}
+}
@@ -1,7 +1,9 @@
 package mqtt

 import (
+	"context"
 	"crypto/rsa"
+	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/json"
@@ -9,13 +11,14 @@ import (
 	"fmt"
 	"io/ioutil"
 	"math/rand"
+	"net"
+	"net/url"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
 	"time"

-	"context"
-
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/kerberos-io/agent/machinery/src/capture"
 	configService "github.com/kerberos-io/agent/machinery/src/config"
@@ -24,6 +27,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/onvif"
 	"github.com/kerberos-io/agent/machinery/src/webrtc"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/net/proxy"
 )

 // We'll cache the MQTT settings to know if we need to reinitialize the MQTT client connection.
@@ -34,6 +38,177 @@ var PREV_MQTTPassword string
 var PREV_HubKey string
 var PREV_AgentKey string

+type pahoErrorLogger struct{}
+
+func (pahoErrorLogger) Println(values ...interface{}) {
+	log.WithFields(log.Fields{
+		"component": "routers/mqtt",
+		"event":     "paho_error",
+	}).Error(strings.TrimSpace(fmt.Sprintln(values...)))
+}
+
+func (pahoErrorLogger) Printf(format string, values ...interface{}) {
+	log.WithFields(log.Fields{
+		"component": "routers/mqtt",
+		"event":     "paho_error",
+	}).Errorf(strings.TrimSpace(format), values...)
+}
+
+func init() {
+	mqtt.ERROR = pahoErrorLogger{}
+}
+
+func enableMQTTConnectionDiagnostics(options *mqtt.ClientOptions, brokerURL string) {
+	if !strings.Contains(brokerURL, "://") {
+		options.SetCustomOpenConnectionFn(openMQTTConnection)
+		return
+	}
+
+	parsedURL, err := url.Parse(brokerURL)
+	if err != nil {
+		return
+	}
+
+	switch strings.ToLower(parsedURL.Scheme) {
+	case "", "mqtt", "tcp", "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
+		options.SetCustomOpenConnectionFn(openMQTTConnection)
+	}
+}
+
+func openMQTTConnection(uri *url.URL, options mqtt.ClientOptions) (net.Conn, error) {
+	host := uri.Hostname()
+	fields := log.Fields{
+		"component": "routers/mqtt",
+		"host":      host,
+		"port":      uri.Port(),
+		"scheme":    uri.Scheme,
+	}
+	logMQTTDNSResolution(host, options.ConnectTimeout, fields)
+
+	connectionStartedAt := time.Now()
+	dialer := options.Dialer
+	if dialer == nil {
+		dialer = &net.Dialer{Timeout: options.ConnectTimeout}
+	}
+
+	proxyConfigured := os.Getenv("all_proxy") != ""
+	proxyMode := "direct"
+	if proxyConfigured {
+		proxyMode = "socks"
+	}
+	fields["proxy_mode"] = proxyMode
+	log.WithFields(fields).Info("Opening MQTT TCP connection")
+
+	var (
+		connection net.Conn
+		err        error
+	)
+	if proxyConfigured {
+		connection, err = proxy.FromEnvironment().Dial("tcp", uri.Host)
+	} else {
+		connection, err = dialer.Dial("tcp", uri.Host)
+	}
+	fields["duration_ms"] = time.Since(connectionStartedAt).Milliseconds()
+	if err != nil {
+		logMQTTNetworkError("MQTT TCP connection failed", err, fields)
+		return nil, err
+	}
+
+	fields["local_address"] = connection.LocalAddr().String()
+	fields["remote_address"] = connection.RemoteAddr().String()
+	log.WithFields(fields).Info("MQTT TCP connection established")
+
+	if !isSecureMQTTScheme(uri.Scheme) {
+		return connection, nil
+	}
+
+	tlsConfig := options.TLSConfig
+	if tlsConfig == nil {
+		tlsConfig = &tls.Config{}
+	} else {
+		tlsConfig = tlsConfig.Clone()
+	}
+	if tlsConfig.ServerName == "" {
+		tlsConfig.ServerName = host
+	}
+
+	tlsConnection := tls.Client(connection, tlsConfig)
+	tlsStartedAt := time.Now()
+	if options.ConnectTimeout > 0 {
+		_ = tlsConnection.SetDeadline(connectionStartedAt.Add(options.ConnectTimeout))
+	}
+	if err = tlsConnection.Handshake(); err != nil {
+		_ = connection.Close()
+		fields["duration_ms"] = time.Since(tlsStartedAt).Milliseconds()
+		fields["server_name"] = tlsConfig.ServerName
+		logMQTTNetworkError("MQTT TLS handshake failed", err, fields)
+		return nil, err
+	}
+	_ = tlsConnection.SetDeadline(time.Time{})
+
+	state := tlsConnection.ConnectionState()
+	fields["cipher_suite"] = tls.CipherSuiteName(state.CipherSuite)
+	fields["duration_ms"] = time.Since(tlsStartedAt).Milliseconds()
+	fields["server_name"] = tlsConfig.ServerName
+	fields["tls_version"] = tls.VersionName(state.Version)
+	log.WithFields(fields).Info("MQTT TLS handshake established")
+	return tlsConnection, nil
+}
+
+func logMQTTDNSResolution(host string, timeout time.Duration, fields log.Fields) {
+	if host == "" || net.ParseIP(host) != nil {
+		return
+	}
+
+	lookupTimeout := timeout
+	if lookupTimeout <= 0 || lookupTimeout > 5*time.Second {
+		lookupTimeout = 5 * time.Second
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), lookupTimeout)
+	defer cancel()
+
+	startedAt := time.Now()
+	addresses, err := net.DefaultResolver.LookupHost(ctx, host)
+	dnsFields := cloneLogFields(fields)
+	dnsFields["duration_ms"] = time.Since(startedAt).Milliseconds()
+	if err != nil {
+		logMQTTNetworkError("MQTT broker DNS resolution failed", err, dnsFields)
+		return
+	}
+
+	dnsFields["resolved_addresses"] = addresses
+	log.WithFields(dnsFields).Info("MQTT broker DNS resolved")
+}
+
+func logMQTTNetworkError(message string, err error, fields log.Fields) {
+	errorFields := cloneLogFields(fields)
+	if networkError, ok := err.(net.Error); ok {
+		errorFields["network_timeout"] = networkError.Timeout()
+	}
+	if operationError, ok := err.(*net.OpError); ok {
+		errorFields["network"] = operationError.Net
+		errorFields["operation"] = operationError.Op
+	}
+	log.WithError(err).WithFields(errorFields).Error(message)
+}
+
+func cloneLogFields(fields log.Fields) log.Fields {
+	cloned := make(log.Fields, len(fields))
+	for key, value := range fields {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func isSecureMQTTScheme(scheme string) bool {
+	switch strings.ToLower(scheme) {
+	case "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
+		return true
+	default:
+		return false
+	}
+}
+
 func HasMQTTClientModified(configuration *models.Configuration) bool {
 	MTTURI := configuration.Config.MQTTURI
 	MTTUsername := configuration.Config.MQTTUsername
@@ -59,6 +234,7 @@ func HasMQTTClientModified(configuration *models.Configuration) bool {
 // - kerberos/{hubkey}/device/{devicekey}/motion: a motion signal

 func ConfigureMQTT(configDirectory string, configuration *models.Configuration, communication *models.Communication) mqtt.Client {
+	installRemoteAccessHook()

 	config := configuration.Config

@@ -116,6 +292,7 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 		// Some extra options to make sure the connection behaves
 		// properly. More information here: github.com/eclipse/paho.mqtt.golang.
 		//opts.SetCleanSession(true)
+		enableMQTTConnectionDiagnostics(opts, mqttURL)
 		opts.SetCleanSession(false)
 		opts.SetResumeSubs(true)
 		opts.SetStore(mqtt.NewMemoryStore())
@@ -203,7 +380,7 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 				"component":  "routers/mqtt",
 				"event":      "initial_connection_timeout",
 				"timeout_ms": (30 * time.Second).Milliseconds(),
-			}).Error("Timed out establishing initial MQTT connection")
+			}).Warn("Initial MQTT connection is still retrying")
 		}
 		return mqc
 	}
@@ -273,6 +450,7 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 			// We will receive all messages from our hub, so we'll need to filter to the relevant device.
 			if message.Mid != "" && message.Timestamp != 0 && message.DeviceId == configuration.Config.Key {
 				var payload models.Payload
+				remoteAuthenticated := false

 				// Messages might be hidden, if so we'll need to decrypt them using the Kerberos Hub private key.
 				if message.Hidden && configuration.Config.HubEncryption == "true" {
@@ -289,8 +467,10 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 								log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 								return
 							}
-							json.Unmarshal(visibleValue, &payload)
-							message.Payload = payload
+							if err := json.Unmarshal(visibleValue, &payload); err == nil {
+								message.Payload = payload
+								remoteAuthenticated = true
+							}
 						} else {
 							log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, no private key provided.")
 						}
@@ -338,7 +518,9 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 											log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 											return
 										}
-										json.Unmarshal(decryptedValue, &payload)
+										if err := json.Unmarshal(decryptedValue, &payload); err == nil {
+											remoteAuthenticated = true
+										}
 									} else {
 										log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, assymetric keys do not match.")
 										return
@@ -396,6 +578,14 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 					go HandleReceiveHDCandidates(mqttClient, hubKey, payload, configuration, communication)
 				case "trigger-relay":
 					go HandleTriggerRelay(mqttClient, hubKey, payload, configuration, communication)
+				case "remote-session-open":
+					go HandleRemoteSessionOpen(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-input":
+					go HandleRemoteSessionInput(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-resize":
+					go HandleRemoteSessionResize(payload, remoteAuthenticated)
+				case "remote-session-close":
+					go HandleRemoteSessionClose(payload, remoteAuthenticated)
 				}

 			}
@@ -1,12 +1,95 @@
 package mqtt

 import (
+	"crypto/tls"
+	"errors"
+	"net/http/httptest"
+	"net/url"
+	"os"
 	"testing"
 	"time"

+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/kerberos-io/agent/machinery/src/models"
 )

+func TestEnableMQTTConnectionDiagnostics(t *testing.T) {
+	tests := []struct {
+		name      string
+		brokerURL string
+		want      bool
+	}{
+		{name: "ActiveMQ TLS", brokerURL: "mqtt+ssl://broker.example:8883", want: true},
+		{name: "TCP", brokerURL: "tcp://broker.example:1883", want: true},
+		{name: "default TCP", brokerURL: "broker.example:1883", want: true},
+		{name: "WebSocket", brokerURL: "wss://broker.example/mqtt", want: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			options := mqtt.NewClientOptions()
+			enableMQTTConnectionDiagnostics(options, test.brokerURL)
+			if got := options.CustomOpenConnectionFn != nil; got != test.want {
+				t.Fatalf("CustomOpenConnectionFn configured = %t, want %t", got, test.want)
+			}
+		})
+	}
+}
+
+func TestOpenMQTTConnectionReturnsTCPError(t *testing.T) {
+	options := *mqtt.NewClientOptions().SetConnectTimeout(100 * time.Millisecond)
+	brokerURL, err := url.Parse("tcp://127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connection, err := openMQTTConnection(brokerURL, options)
+	if connection != nil {
+		connection.Close()
+		t.Fatal("openMQTTConnection() returned a connection for an unavailable endpoint")
+	}
+	if err == nil {
+		t.Fatal("openMQTTConnection() returned no TCP error")
+	}
+}
+
+func TestOpenMQTTConnectionEstablishesActiveMQTLS(t *testing.T) {
+	server := httptest.NewTLSServer(nil)
+	defer server.Close()
+
+	serverURL, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	brokerURL, err := url.Parse("mqtt+ssl://" + serverURL.Host)
+	if err != nil {
+		t.Fatal(err)
+	}
+	options := *mqtt.NewClientOptions().
+		SetConnectTimeout(time.Second).
+		SetTLSConfig(&tls.Config{InsecureSkipVerify: true}) // #nosec G402 -- local test server
+
+	connection, err := openMQTTConnection(brokerURL, options)
+	if err != nil {
+		t.Fatalf("openMQTTConnection() error = %v", err)
+	}
+	defer connection.Close()
+	if _, ok := connection.(*tls.Conn); !ok {
+		t.Fatalf("openMQTTConnection() connection type = %T, want *tls.Conn", connection)
+	}
+}
+
+func TestIsSecureMQTTScheme(t *testing.T) {
+	for _, scheme := range []string{"ssl", "tls", "mqtts", "mqtt+ssl", "tcps"} {
+		if !isSecureMQTTScheme(scheme) {
+			t.Errorf("isSecureMQTTScheme(%q) = false, want true", scheme)
+		}
+	}
+	if isSecureMQTTScheme("tcp") {
+		t.Error("isSecureMQTTScheme(\"tcp\") = true, want false")
+	}
+}
+
 func TestConfigureMQTTRequiresHubKey(t *testing.T) {
 	configuration := &models.Configuration{Config: models.Config{Key: "agent-key"}}

@@ -53,3 +136,73 @@ func TestEnqueueLatestAudioDoesNotBlockNilChannel(t *testing.T) {
 		t.Fatal("enqueueLatestAudio() blocked on a nil channel")
 	}
 }
+
+func TestRemoteAccessRequiresExplicitOptIn(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+
+	_ = os.Unsetenv(remoteAccessEnvironment)
+	if remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = true without opt-in")
+	}
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+	if !remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = false after opt-in")
+	}
+}
+
+func TestNormalizeTerminalSize(t *testing.T) {
+	rows, columns := normalizeTerminalSize(0, 0)
+	if rows != 24 || columns != 80 {
+		t.Fatalf("normalizeTerminalSize(0, 0) = (%d, %d), want (24, 80)", rows, columns)
+	}
+
+	rows, columns = normalizeTerminalSize(500, 500)
+	if rows != 200 || columns != 400 {
+		t.Fatalf("normalizeTerminalSize(500, 500) = (%d, %d), want (200, 400)", rows, columns)
+	}
+}
+
+func TestDecodeRemotePayloadRejectsMissingSession(t *testing.T) {
+	_, err := decodeRemotePayload(models.Payload{Value: map[string]interface{}{
+		"kind": "shell",
+	}})
+	if err == nil {
+		t.Fatal("decodeRemotePayload() accepted a missing session id")
+	}
+}
+
+func TestRemoteSessionOpenRejectsUnprovenEncryption(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+
+	// The listener passes false when an envelope merely claims to be hidden but
+	// no ciphertext was successfully decrypted. The remote handler must reject it.
+	if err := validateRemoteAccess(false); !errors.Is(err, errRemoteUnauthenticated) {
+		t.Fatalf("validateRemoteAccess(false) error = %v, want %v", err, errRemoteUnauthenticated)
+	}
+}
+
+func TestRemoteSessionReservationIsIdempotent(t *testing.T) {
+	manager := newRemoteAccessManager()
+	session := &remoteSession{id: "session-1", kind: "logs"}
+	if _, err := manager.reserve(session); err != nil {
+		t.Fatalf("first reserve() failed: %v", err)
+	}
+	if _, err := manager.reserve(session); !errors.Is(err, errRemoteSessionExists) {
+		t.Fatalf("duplicate reserve() error = %v, want %v", err, errRemoteSessionExists)
+	}
+}
@@ -0,0 +1,397 @@
+package mqtt
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/creack/pty"
+	paho "github.com/eclipse/paho.mqtt.golang"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	remoteAccessEnvironment = "AGENT_REMOTE_ACCESS_ENABLED"
+	remoteHistoryLimit      = 500
+	remoteSessionLimit      = 5
+	remoteOutputChunkSize   = 4096
+	remoteInputLimit        = 64 * 1024
+	remoteSessionLifetime   = time.Hour
+)
+
+type remoteSession struct {
+	id     string
+	kind   string
+	pty    *os.File
+	cancel context.CancelFunc
+	logs   chan string
+	done   chan struct{}
+	timer  *time.Timer
+}
+
+type remoteAccessManager struct {
+	mu       sync.Mutex
+	sessions map[string]*remoteSession
+	history  []string
+}
+
+var (
+	remoteAccess             = newRemoteAccessManager()
+	remoteHookOnce           sync.Once
+	errRemoteDisabled        = errors.New("remote access is disabled on this agent")
+	errRemoteUnauthenticated = errors.New("remote access requires encrypted MQTT")
+	errRemoteSessionExists   = errors.New("remote session already exists")
+	errRemoteSessionLimit    = errors.New("remote session limit reached")
+)
+
+func newRemoteAccessManager() *remoteAccessManager {
+	return &remoteAccessManager{sessions: make(map[string]*remoteSession)}
+}
+
+func installRemoteAccessHook() {
+	remoteHookOnce.Do(func() {
+		log.AddHook(remoteAccess)
+	})
+}
+
+func (manager *remoteAccessManager) Levels() []log.Level {
+	return log.AllLevels
+}
+
+func (manager *remoteAccessManager) Fire(entry *log.Entry) error {
+	line, err := json.Marshal(map[string]interface{}{
+		"timestamp": entry.Time.Format(time.RFC3339Nano),
+		"level":     entry.Level.String(),
+		"message":   entry.Message,
+		"fields":    entry.Data,
+	})
+	if err != nil {
+		return nil
+	}
+	encoded := base64.StdEncoding.EncodeToString(append(line, '\n'))
+
+	manager.mu.Lock()
+	manager.history = append(manager.history, encoded)
+	if len(manager.history) > remoteHistoryLimit {
+		manager.history = manager.history[len(manager.history)-remoteHistoryLimit:]
+	}
+	for _, session := range manager.sessions {
+		if session.kind != "logs" {
+			continue
+		}
+		select {
+		case session.logs <- encoded:
+		default:
+		}
+	}
+	manager.mu.Unlock()
+	return nil
+}
+
+func remoteAccessEnabled() bool {
+	enabled, err := strconv.ParseBool(strings.TrimSpace(os.Getenv(remoteAccessEnvironment)))
+	return err == nil && enabled
+}
+
+func validateRemoteAccess(authenticated bool) error {
+	if !authenticated {
+		return errRemoteUnauthenticated
+	}
+	if !remoteAccessEnabled() {
+		return errRemoteDisabled
+	}
+	return nil
+}
+
+func decodeRemotePayload(payload models.Payload) (models.RemoteSessionPayload, error) {
+	data, err := json.Marshal(payload.Value)
+	if err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	var request models.RemoteSessionPayload
+	if err := json.Unmarshal(data, &request); err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	if request.SessionID == "" || len(request.SessionID) > 128 {
+		return models.RemoteSessionPayload{}, errors.New("invalid remote session id")
+	}
+	return request, nil
+}
+
+func normalizeTerminalSize(rows uint16, columns uint16) (uint16, uint16) {
+	if rows < 5 {
+		rows = 24
+	}
+	if rows > 200 {
+		rows = 200
+	}
+	if columns < 20 {
+		columns = 80
+	}
+	if columns > 400 {
+		columns = 400
+	}
+	return rows, columns
+}
+
+func HandleRemoteSessionOpen(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	if accessErr := validateRemoteAccess(authenticated); accessErr != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", accessErr.Error())
+		return
+	}
+
+	switch request.Kind {
+	case "logs":
+		err = remoteAccess.openLogs(client, hubKey, configuration, request)
+	case "shell":
+		err = remoteAccess.openShell(client, hubKey, configuration, request)
+	default:
+		err = errors.New("unsupported remote session kind")
+	}
+	if err != nil {
+		if errors.Is(err, errRemoteSessionExists) {
+			publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "opened", "")
+			return
+		}
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", err.Error())
+	}
+}
+
+func HandleRemoteSessionInput(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil || len(request.Data) > remoteInputLimit*2 {
+		return
+	}
+	data, err := base64.StdEncoding.DecodeString(request.Data)
+	if err != nil || len(data) > remoteInputLimit {
+		return
+	}
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session == nil || session.kind != "shell" || session.pty == nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "remote session is not open")
+		return
+	}
+	if _, err := session.pty.Write(data); err != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "failed to write terminal input")
+	}
+}
+
+func HandleRemoteSessionResize(payload models.Payload, authenticated bool) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session != nil && session.kind == "shell" && session.pty != nil {
+		_ = pty.Setsize(session.pty, &pty.Winsize{Rows: rows, Cols: columns})
+	}
+}
+
+func HandleRemoteSessionClose(payload models.Payload, authenticated bool) {
+	if !authenticated {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err == nil {
+		remoteAccess.close(request.SessionID)
+	}
+}
+
+func (manager *remoteAccessManager) reserve(session *remoteSession) ([]string, error) {
+	manager.mu.Lock()
+	defer manager.mu.Unlock()
+	if _, exists := manager.sessions[session.id]; exists {
+		return nil, errRemoteSessionExists
+	}
+	if len(manager.sessions) >= remoteSessionLimit {
+		return nil, errRemoteSessionLimit
+	}
+	manager.sessions[session.id] = session
+	history := append([]string(nil), manager.history...)
+	return history, nil
+}
+
+func (manager *remoteAccessManager) expire(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession) {
+	session.timer = time.AfterFunc(remoteSessionLifetime, func() {
+		manager.close(session.id)
+		if session.kind == "logs" {
+			publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "session lifetime reached")
+		}
+	})
+}
+
+func (manager *remoteAccessManager) openLogs(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	session := &remoteSession{
+		id:   request.SessionID,
+		kind: "logs",
+		logs: make(chan string, 256),
+		done: make(chan struct{}),
+	}
+	history, err := manager.reserve(session)
+	if err != nil {
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	tail := request.Tail
+	if tail <= 0 || tail > remoteHistoryLimit {
+		tail = 200
+	}
+	if len(history) > tail {
+		history = history[len(history)-tail:]
+	}
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go func() {
+		for _, line := range history {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+		}
+		for {
+			select {
+			case line := <-session.logs:
+				publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+			case <-session.done:
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func (manager *remoteAccessManager) openShell(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	ctx, cancel := context.WithCancel(context.Background())
+	session := &remoteSession{
+		id:     request.SessionID,
+		kind:   "shell",
+		cancel: cancel,
+		done:   make(chan struct{}),
+	}
+	if _, err := manager.reserve(session); err != nil {
+		cancel()
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	command := exec.CommandContext(ctx, "/bin/sh")
+	command.Env = append(os.Environ(), "TERM=xterm-256color", "HISTFILE=/dev/null")
+	terminal, err := pty.StartWithSize(command, &pty.Winsize{Rows: rows, Cols: columns})
+	if err != nil {
+		manager.remove(session.id)
+		cancel()
+		return err
+	}
+	session.pty = terminal
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go manager.forwardShell(client, hubKey, configuration, session, command)
+	return nil
+}
+
+func (manager *remoteAccessManager) forwardShell(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession, command *exec.Cmd) {
+	buffer := make([]byte, remoteOutputChunkSize)
+	for {
+		count, err := session.pty.Read(buffer)
+		if count > 0 {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, base64.StdEncoding.EncodeToString(buffer[:count]))
+		}
+		if err != nil {
+			if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrClosed) {
+				publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "error", "terminal stream closed unexpectedly")
+			}
+			break
+		}
+	}
+	_ = command.Wait()
+	manager.remove(session.id)
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "")
+}
+
+func (manager *remoteAccessManager) close(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session == nil {
+		return
+	}
+	if session.cancel != nil {
+		session.cancel()
+	}
+	if session.pty != nil {
+		_ = session.pty.Close()
+	}
+	if session.logs != nil {
+		close(session.done)
+	}
+	if session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func (manager *remoteAccessManager) remove(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session != nil && session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func publishRemoteStatus(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, state string, errorMessage string) {
+	status := models.RemoteSessionStatus{
+		Timestamp: time.Now().Unix(),
+		SessionID: sessionID,
+		Kind:      kind,
+		State:     state,
+		Error:     errorMessage,
+	}
+	value, _ := json.Marshal(status)
+	var statusValue map[string]interface{}
+	_ = json.Unmarshal(value, &statusValue)
+	publishRemote(client, hubKey, configuration, "remote-session-status", statusValue, 1)
+}
+
+func publishRemoteOutput(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, data string) {
+	publishRemote(client, hubKey, configuration, "remote-session-output", map[string]interface{}{
+		"timestamp":  time.Now().Unix(),
+		"session_id": sessionID,
+		"kind":       kind,
+		"data":       data,
+	}, 0)
+}
+
+func publishRemote(client paho.Client, hubKey string, configuration *models.Configuration, action string, value map[string]interface{}, qos byte) {
+	message := models.Message{Payload: models.Payload{
+		Action:   action,
+		DeviceId: configuration.Config.Key,
+		Value:    value,
+	}}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		client.Publish("kerberos/hub/"+hubKey, qos, false, payload)
+	}
+}
@@ -223,6 +223,8 @@
      "kerberosvault_description_accesskey": "The access key of your Vault account.",
      "kerberosvault_secretkey": "Secret key",
      "kerberosvault_description_secretkey": "The secret key of your Vault account.",
+      "kerberosvault_customheaders": "Custom upload headers",
+      "kerberosvault_description_customheaders": "A JSON object of custom HTTP headers, for example {\"site_id\":\"site-1\",\"line_id\":\"line-2\"}. These values are included as metadata in Vault integration messages.",
      "kerberosvault_maxretries": "Max retries",
      "kerberosvault_description_maxretries": "The maximum number of retries to upload a recording.",
      "kerberosvault_timeout": "Timeout",
@@ -2542,6 +2542,28 @@ export class Settings extends React.Component {
                          )
                        }
                      />
+                      <Input
+                        noPadding
+                        label={t(
+                          'settings.persistence.kerberosvault_customheaders'
+                        )}
+                        placeholder={t(
+                          'settings.persistence.kerberosvault_description_customheaders'
+                        )}
+                        value={
+                          config.kstorage
+                            ? config.kstorage.custom_headers || ''
+                            : ''
+                        }
+                        onChange={(value) =>
+                          this.onUpdateField(
+                            'kstorage',
+                            'custom_headers',
+                            value,
+                            config.kstorage
+                          )
+                        }
+                      />
                      <Input
                        noPadding
                        label={t(
@@ -2815,6 +2837,28 @@ export class Settings extends React.Component {
                      )
                    }
                  />
+                  <Input
+                    noPadding
+                    label={t(
+                      'settings.persistence.kerberosvault_customheaders'
+                    )}
+                    placeholder={t(
+                      'settings.persistence.kerberosvault_description_customheaders'
+                    )}
+                    value={
+                      config.kstorage_secondary
+                        ? config.kstorage_secondary.custom_headers || ''
+                        : ''
+                    }
+                    onChange={(value) =>
+                      this.onUpdateField(
+                        'kstorage_secondary',
+                        'custom_headers',
+                        value,
+                        config.kstorage_secondary
+                      )
+                    }
+                  />
                </BlockBody>
                <BlockFooter>
                  <Button
Author	SHA1	Message	Date
Cédric Verstraeten	a91d06c3bc	Merge pull request #336 from 4alvit/fix/snapshot-closed-queue fix(capture): stop snapshot readers when the packet queue closes	2026-10-02 08:00:56 +02:00
alvit	022881d707	fix(capture): stop snapshot readers when the packet queue closes	2026-09-29 11:55:38 -07:00
Cédric Verstraeten	15088e5b70	Merge pull request #335 from kerberos-io/feature/improve-custom-ptz-preset feature/improve-custom-ptz-preset	2026-09-27 22:11:04 +02:00
Cédric Verstraeten	be933a4b7c	Rework ONVIF absolute move with verified fallback Replace the old AbsolutePanTiltMoveFake logic, which zoomed out and then moved one axis at a time with fixed speeds, with a new MoveToPosition implementation. The new flow: - Tries a native AbsoluteMove when the camera advertises absolute position spaces. - Polls the position to confirm the move, with overall and no-progress timeouts. - Falls back to a closed-loop ContinuousMove that drives pan, tilt and zoom at the same time. Velocity is proportional to the remaining error, clamped between min and max limits. - Always stops the camera after a failed or finished move and returns combined errors instead of dropping them. - Validates the target (NaN/Inf) and rejects incomplete position responses. AbsolutePanTiltMoveFake is kept as a continuous-only wrapper. AbsolutePanTiltMove and GetPTZFunctionsFromDevice now guard against an empty PTZConfiguration. Movement failures are now logged at error level. Adds unit tests that use a fake PTZ client to cover the native path, the fallback paths, timeouts, stop cleanup and XML encoding of combined ContinuousMove velocities.	2026-09-27 22:00:52 +02:00
Cédric Verstraeten	e242882f24	Merge pull request #334 from kerberos-io/feature/custom-headers-integrations feature/custom-headers-integrations	2026-09-24 19:24:52 +02:00
Cédric VerstraetenandCopilot	ebbde1e722	Harden custom Vault header validation and redirects - Reject JSON null values and every Upload-* header. - Reject tabs and all other control characters in values. - Preserve configured header name casing on the wire. - Strip custom headers, their manifest, Upload-Metadata and Vault credentials on cross-host redirects for legacy and tus uploads. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-09-24 16:28:55 +00:00
Cédric VerstraetenandCopilot	12db58439a	Document custom Vault header environment variables Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-09-24 16:18:26 +00:00
Cédric VerstraetenandCopilot	5dd3a85e50	Add custom metadata headers to Vault uploads Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-09-24 15:48:21 +00:00
Cédric Verstraeten	c9dc1f783d	Merge pull request #333 from kerberos-io/fix/deployment-name-to-friendly-name fix/deployment-name-to-friendly-name	2026-09-22 12:08:42 +02:00
cedricve	6866524274	feat: add deployment name fallback to friendly name in environment variable configuration	2026-09-22 10:04:03 +00:00
Cédric Verstraeten	c2a07672d9	Merge pull request #331 from kerberos-io/feature/add-encrypted-metadata-field feature/add-encrypted-metadata-field	2026-09-14 22:10:20 +02:00
cedricve	f64cdfd605	feat: add encrypted metadata field to recording uploads	2026-09-14 19:20:24 +00:00
Cédric Verstraeten	af5d728921	Merge pull request #330 from kerberos-io/feature/add-remote-control-ssh-logging feature/add-remote-control-ssh-logging	2026-09-14 19:27:45 +02:00
Cédric Verstraeten	b6c8855595	Merge branch 'master' into feature/add-remote-control-ssh-logging	2026-09-14 19:25:11 +02:00
Cédric Verstraeten	23ffb19b4d	Merge pull request #329 from kerberos-io/feature/support-mqtt-ssl feature/support-mqtt-ssl	2026-09-14 19:23:59 +02:00
cedricve	7849f34386	feat: add MQTT connection diagnostics and enhance testing for connection establishment	2026-09-14 17:01:26 +00:00
cedricve	50a77b4591	feat: enhance MQTT error logging and update README for MQTT URI	2026-09-14 14:58:27 +00:00
cedricve	e9ef597442	feat: add remote control SSH logging and session management over MQTT	2026-09-07 08:03:41 +00:00