Merge pull request #338 from EpicMouse123/chore/bump-onvif-v1.2.3

chore(deps): bump github.com/kerberos-io/onvif to v1.2.3

README.md

@@ -365,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"                        |
@@ -408,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
 

machinery/docs/docs.go

@@ -1296,6 +1296,9 @@ const docTemplate = `{
                     "description": "old way, remove this",
                     "type": "string"
                 },
+                "custom_headers": {
+                    "type": "string"
+                },
                 "directory": {
                     "type": "string"
                 },

machinery/docs/swagger.json

@@ -1289,6 +1289,9 @@
                     "description": "old way, remove this",
                     "type": "string"
                 },
+                "custom_headers": {
+                    "type": "string"
+                },
                 "directory": {
                     "type": "string"
                 },

machinery/go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.2.2
+	github.com/kerberos-io/onvif v1.2.3
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/moq-dev/moq-go v0.5.7
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646

machinery/go.sum

@@ -776,8 +776,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.2.2 h1:QnxITps7xvAVD2abWRsa3+p9QexjKESQQldUFMx/mYA=
-github.com/kerberos-io/onvif v1.2.2/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
+github.com/kerberos-io/onvif v1.2.3 h1:RDnDvLBd92VCgFOrlbGavzCVN9lQA/qs9novdNFIAOU=
+github.com/kerberos-io/onvif v1.2.3/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=

machinery/src/capture/main.go

@@ -1070,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
@@ -1113,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

machinery/src/capture/snapshot_test.go

@@ -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")
+					}
+				})
+			}
+		}
+	}
+}

machinery/src/cloud/cloud.go

@@ -541,13 +541,23 @@ func (state *heartbeatONVIFState) fetchEvents(device *goonvif.Device) ([]byte, b
 		}
 	}
 
-	if !initialStateSubscriptionCreated && state.loopPullPoint == "" {
+	// Fall back to querying the relay outputs and digital inputs directly when
+	// no subscription could be created, and also when the subscriptions
+	// produced no I/O events: some devices accept the subscription but reject
+	// the pulls, or never publish the initial I/O state.
+	noSubscription := !initialStateSubscriptionCreated && state.loopPullPoint == ""
+	if noSubscription || heartbeatEventsListEmpty(onvifEventsList) {
 		return state.fallbackEvents(device), operationFailed
 	}
 
 	return onvifEventsList, operationFailed
 }
 
+func heartbeatEventsListEmpty(eventsList []byte) bool {
+	trimmed := strings.TrimSpace(string(eventsList))
+	return trimmed == "" || trimmed == "[]" || trimmed == "null"
+}
+
 func getHeartbeatONVIFPayload(camera models.IPCamera, state *heartbeatONVIFState) heartbeatONVIFPayload {
 	if state == nil {
 		state = newHeartbeatONVIFState()

machinery/src/cloud/cloud_test.go

@@ -15,6 +15,7 @@ import (
 	goonvif "github.com/kerberos-io/onvif"
 	goonvifdevice "github.com/kerberos-io/onvif/device"
 	goonvifptz "github.com/kerberos-io/onvif/ptz"
+	goonvifxsd "github.com/kerberos-io/onvif/xsd/onvif"
 )
 
 func TestHeartbeatFailureLogOmitsHubResponseBody(t *testing.T) {
@@ -590,3 +591,60 @@ func countString(values []string, want string) int {
 	}
 	return count
 }
+
+// Some devices accept the subscription but reject the pulls, or never publish
+// the initial I/O state. The relay outputs and digital inputs must then be
+// queried directly instead of reporting no I/O at all.
+func TestHeartbeatONVIFPayloadFallsBackToDirectIOWhenSubscriptionsReturnNoEvents(t *testing.T) {
+	restoreHeartbeatONVIFStubs(t)
+
+	device := newTestONVIFDevice()
+	camera := models.IPCamera{
+		ONVIFXAddr:    "http://camera/onvif",
+		ONVIFUsername: "operator",
+		ONVIFPassword: "secret",
+	}
+
+	state := newHeartbeatONVIFState()
+	state.cameraConfiguration = camera
+	state.cameraKey = heartbeatONVIFCameraKey(camera)
+	state.device = device
+	state.staticLoaded = true
+	state.staticPayload.enabled = "true"
+
+	heartbeatCreatePullPointSubscription = func(*goonvif.Device) (string, error) {
+		return "pull-point", nil
+	}
+	heartbeatGetEventMessages = func(*goonvif.Device, string) ([]agentonvif.ONVIFEvents, error) {
+		return nil, errors.New("pull rejected")
+	}
+	heartbeatUnsubscribePullPoint = func(*goonvif.Device, string) error {
+		return nil
+	}
+	var relayCalls, inputCalls int
+	heartbeatGetRelayOutputs = func(*goonvif.Device) (goonvifdevice.GetRelayOutputsResponse, error) {
+		relayCalls++
+		return goonvifdevice.GetRelayOutputsResponse{RelayOutputs: []goonvifxsd.RelayOutput{{DeviceEntity: goonvifxsd.DeviceEntity{Token: "relay0"}}}}, nil
+	}
+	heartbeatGetDigitalInputs = func(*goonvif.Device) (goonvifdevice.GetDigitalInputsResponse, error) {
+		inputCalls++
+		return goonvifdevice.GetDigitalInputsResponse{DigitalInputs: []goonvifxsd.DigitalInput{{Token: "di0"}}}, nil
+	}
+
+	payload := getHeartbeatONVIFPayload(camera, state)
+
+	if relayCalls != 1 || inputCalls != 1 {
+		t.Fatalf("direct I/O queries: relay=%d inputs=%d, want 1 each", relayCalls, inputCalls)
+	}
+	var events []agentonvif.ONVIFEvents
+	if err := json.Unmarshal(payload.eventsList, &events); err != nil {
+		t.Fatalf("payload.eventsList is not valid JSON: %v", err)
+	}
+	got := map[string]string{}
+	for _, event := range events {
+		got[event.Key] = event.Type
+	}
+	if got["relay0"] != "output" || got["di0"] != "input" || len(got) != 2 {
+		t.Fatalf("payload.eventsList = %s, want relay0 output and di0 input", payload.eventsList)
+	}
+}

machinery/src/cloud/custom_headers.go

@@ -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
+}

machinery/src/cloud/custom_headers_test.go

@@ -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)
+	}
+}

machinery/src/cloud/kerberos_vault.go

@@ -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
 	}

machinery/src/cloud/tus_client.go

@@ -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)
 }

machinery/src/cloud/tus_client_test.go

@@ -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) {

machinery/src/config/main.go

@@ -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":

machinery/src/config/main_test.go

@@ -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
 }

machinery/src/models/config.go

@@ -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"`
 }

machinery/src/onvif/main.go

@@ -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
@@ -1042,6 +1108,23 @@ func CreatePullPointSubscription(dev *onvif.Device) (string, error) {
 	return pullPointAdress, err
 }
 
+const (
+	wsAddressingNamespace = "http://www.w3.org/2005/08/addressing"
+	pullMessagesAction    = "http://www.onvif.org/ver10/events/wsdl/PullPointSubscription/PullMessagesRequest"
+	unsubscribeAction     = "http://docs.oasis-open.org/wsn/bw-2/SubscriptionManager/UnsubscribeRequest"
+)
+
+// subscriptionAddressingHeader builds the WS-Addressing Action and To headers
+// for a request sent to a pull-point subscription. Some devices identify the
+// subscription from wsa:To rather than the request URL, and reject requests
+// without it as NotAuthorized.
+func subscriptionAddressingHeader(subscriptionAddress string, action string) string {
+	var address bytes.Buffer
+	_ = xml.EscapeText(&address, []byte(subscriptionAddress))
+	return `<wsa:Action xmlns:wsa="` + wsAddressingNamespace + `">` + action + `</wsa:Action>` +
+		`<wsa:To xmlns:wsa="` + wsAddressingNamespace + `">` + address.String() + `</wsa:To>`
+}
+
 func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 
 	// Unsubscribe from the device
@@ -1055,7 +1138,7 @@ func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 		return err
 	}
 
-	res, err := dev.SendSoap(pullPointAddress, string(requestBody))
+	res, err := dev.SendSoapWithHeader(pullPointAddress, string(requestBody), subscriptionAddressingHeader(pullPointAddress, unsubscribeAction))
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
 			"component": "onvif",
@@ -1137,7 +1220,7 @@ func GetEventMessages(dev *onvif.Device, pullPointAddress string) ([]ONVIFEvents
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err
 			}
-			res, err := dev.SendSoap(string(subscriptionURI), string(requestBody))
+			res, err := dev.SendSoapWithHeader(string(subscriptionURI), string(requestBody), subscriptionAddressingHeader(subscriptionURI, pullMessagesAction))
 			if err != nil {
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err

machinery/src/onvif/move_test.go

@@ -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)
+	}
+}

machinery/src/onvif/subscription_test.go

@@ -0,0 +1,88 @@
+package onvif
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// newSubscriptionTestDevice starts a fake ONVIF device and returns a connected
+// device plus the request bodies it received on the subscription endpoint.
+func newSubscriptionTestDevice(t *testing.T, response string) (*onvif.Device, string, func() []string) {
+	t.Helper()
+
+	var mu sync.Mutex
+	var subscriptionBodies []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		if r.URL.Path == "/onvif/subscription" {
+			mu.Lock()
+			subscriptionBodies = append(subscriptionBodies, string(body))
+			mu.Unlock()
+			_, _ = w.Write([]byte(response))
+			return
+		}
+		_, _ = w.Write([]byte(`<Envelope><Body/></Envelope>`))
+	}))
+	t.Cleanup(srv.Close)
+
+	dev, err := onvif.NewDevice(onvif.DeviceParams{Xaddr: strings.TrimPrefix(srv.URL, "http://")})
+	if err != nil {
+		t.Fatalf("NewDevice() error = %v", err)
+	}
+	subscriptionAddress := srv.URL + "/onvif/subscription?subscription=7&idx=2"
+	return dev, subscriptionAddress, func() []string {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]string(nil), subscriptionBodies...)
+	}
+}
+
+func assertSubscriptionAddressing(t *testing.T, body string, action string) {
+	t.Helper()
+
+	for _, want := range []string{
+		`<wsa:Action xmlns:wsa="http://www.w3.org/2005/08/addressing">` + action + `</wsa:Action>`,
+		`<wsa:To xmlns:wsa="http://www.w3.org/2005/08/addressing">`,
+		`/onvif/subscription?subscription=7&amp;idx=2</wsa:To>`,
+	} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("request body does not contain %q:\n%s", want, body)
+		}
+	}
+}
+
+// Some devices identify a pull-point subscription by wsa:To and reject
+// PullMessages without it as NotAuthorized.
+func TestGetEventMessagesSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><PullMessagesResponse/></Body></Envelope>`)
+
+	if _, err := GetEventMessages(dev, subscriptionAddress); err != nil {
+		t.Fatalf("GetEventMessages() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], pullMessagesAction)
+}
+
+func TestUnsubscribePullPointSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><UnsubscribeResponse/></Body></Envelope>`)
+
+	if err := UnsubscribePullPoint(dev, subscriptionAddress); err != nil {
+		t.Fatalf("UnsubscribePullPoint() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], unsubscribeAction)
+}

ui/public/locales/en/translation.json

@@ -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",

ui/src/pages/Settings/Settings.jsx

@@ -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