Merge pull request #12 from kerberos-io/feature/enhance-ptz-failure-logging

feature/enhance-ptz-failure-logging

.github/workflows/pr-build.yml

@@ -0,0 +1,51 @@
+name: Pull Request Build
+
+# Builds, vets and tests the Go library on every pull request.
+# This repository is a Go library (no Dockerfile), so unlike the Docker-oriented
+# uug-ai/workflows pr-build.yml it validates the module directly instead of
+# building and pushing a container image.
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    name: Build & Test (Go)
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24'
+          check-latest: true
+          cache-dependency-path: go.sum
+
+      - name: Build
+        run: go build -v ./...
+
+      - name: Vet (non-blocking)
+        continue-on-error: true
+        run: go vet ./...
+
+      - name: Gofmt check (non-blocking)
+        continue-on-error: true
+        run: |
+          unformatted="$(gofmt -l .)"
+          if [ -n "$unformatted" ]; then
+            echo "The following files are not gofmt-clean:"
+            echo "$unformatted"
+            exit 1
+          fi
+
+      - name: Test
+        # ws-discovery/TestDevicesFromProbeResponses is a network-dependent
+        # integration test that probes real ONVIF cameras on the LAN, so it
+        # cannot pass on hosted runners. Run it locally against real devices.
+        run: go test $(go list ./... | grep -v '/ws-discovery')

.github/workflows/release-bump.yml

@@ -0,0 +1,28 @@
+name: Bump release
+
+# Manually bump the semantic version: determines the next tag from the latest
+# v* tag, creates a GitHub release with generated notes and exposes the new tag.
+# Thin wrapper around the reusable uug-ai/workflows release-bump workflow.
+
+on:
+  workflow_dispatch:
+    inputs:
+      bump:
+        description: "Which part of the version to bump"
+        required: true
+        default: patch
+        type: choice
+        options:
+          - major
+          - minor
+          - patch
+
+permissions:
+  contents: write
+
+jobs:
+  bump-release:
+    uses: uug-ai/workflows/.github/workflows/release-bump.yml@main
+    with:
+      bump: ${{ github.event.inputs.bump }}
+    secrets: inherit

.github/workflows/release.yml

@@ -0,0 +1,36 @@
+name: Release
+
+# Validates the tagged code when a release is published (build + test).
+# For a Go library a "release" is simply a git tag consumers depend on, so this
+# workflow guards that a released tag builds and passes its tests rather than
+# publishing a container image.
+
+on:
+  release:
+    types: [published]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  validate:
+    name: Validate release (Go)
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24'
+          check-latest: true
+          cache-dependency-path: go.sum
+
+      - name: Build
+        run: go build -v ./...
+
+      - name: Test
+        # See pr-build.yml: the ws-discovery probe test needs real cameras.
+        run: go test $(go list ./... | grep -v '/ws-discovery')

.gitignore

@@ -1,3 +1,4 @@
+.idea
 *.iml
 .idea/
 .vscode/

Device.go

@@ -280,21 +280,45 @@ func (dev Device) callMethodDo(endpoint string, method interface{}) (*http.Respo
 	soap.AddRootNamespaces(Xlmns)
 	soap.AddAction()
 
-	//Auth Handling
-	if dev.params.Username != "" && dev.params.Password != "" {
-		soap.AddWSSecurity(dev.params.Username, dev.params.Password)
-	}
+	return dev.sendSOAP(endpoint, soap)
+}
 
-	servResp, err := networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
-	if err != nil {
-		// Close server response body to reuse the connection
-		if servResp != nil {
-			servResp.Body.Close()
+// sendSOAP dispatches an assembled SOAP message to the endpoint using the
+// authentication mechanism selected through DeviceParams.AuthMode.
+//
+// Behaviour per AuthMode:
+//   - NoAuth ("none"):            no authentication is added.
+//   - UsernameTokenAuth:          WS-Security UsernameToken only; never falls
+//     back to HTTP digest, which keeps cameras that authenticate exclusively
+//     through WS-Security working.
+//   - DigestAuth ("digest"):      HTTP digest only; no WS-Security header.
+//   - Both ("both") / unset (""): WS-Security credentials are added (when
+//     available) and HTTP digest is attempted only if the device answers with
+//     an authentication challenge (HTTP 401 Unauthorized). On that digest
+//     retry the WS-Security header is dropped so the credentials are not sent
+//     twice.
+func (dev Device) sendSOAP(endpoint string, soap gosoap.SoapMessage) (*http.Response, error) {
+	hasCredentials := dev.params.Username != "" || dev.params.Password != ""
+
+	switch dev.params.AuthMode {
+	case NoAuth:
+		return networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
+
+	case UsernameTokenAuth:
+		if hasCredentials {
+			soap.AddWSSecurity(dev.params.Username, dev.params.Password)
 		}
-		servResp, err = networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
-	}
+		return networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
 
-	return servResp, err
+	case DigestAuth:
+		return networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
+
+	default: // Both and the empty/unset default.
+		if hasCredentials {
+			soap.AddWSSecurity(dev.params.Username, dev.params.Password)
+		}
+		return networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
+	}
 }
 
 func (dev *Device) GetDeviceParams() DeviceParams {
@@ -333,28 +357,75 @@ func (dev *Device) GetEndpointByRequestStruct(requestStruct interface{}) (string
 	return resp, err
 }*/
 
-// CallMethod functions call an method, defined <method> struct with authentication data
+// SendSoap POSTs the given body wrapped in a SOAP envelope.
 func (dev Device) SendSoap(endpoint string, xmlRequestBody string) (*http.Response, error) {
+	return dev.SendSoapWithOptions(endpoint, xmlRequestBody)
+}
 
+// SendSoapWithHeader is SendSoap plus arbitrary inner-Header XML —
+// needed to echo WS-Addressing ReferenceParameters (with
+// wsa:IsReferenceParameter="true") back to vendors like AXIS that
+// identify pull-point subscriptions through them rather than the URL.
+//
+// xmlHeaderContent must be well-formed XML representing one or more
+// SOAP Header child elements (siblings are supported; the spec lets
+// each reference parameter be its own header block). Malformed or
+// element-free content errors before any request is made.
+//
+// SECURITY: do not pass content sourced from untrusted clients. The
+// API assumes the caller is authoritative for the envelope. Header
+// content forwards verbatim — among others, <wsse:Security> overrides
+// auth, <wsa:Action> overrides intent, <wsa:To>/<wsa:ReplyTo>/
+// <wsa:FaultTo> redirect responses, <wsa:MessageID> enables replay-
+// token forgery, and <wsu:Timestamp> bypasses freshness checks.
+func (dev Device) SendSoapWithHeader(endpoint, xmlRequestBody, xmlHeaderContent string) (*http.Response, error) {
+	return dev.SendSoapWithOptions(endpoint, xmlRequestBody, WithSOAPHeader(xmlHeaderContent))
+}
+
+// SendSoapOption tweaks a single SendSoapWithOptions call. New options
+// (per-call timeout, context, custom envelope namespaces, ...) should
+// be added as WithX constructors here rather than as new method
+// variants on Device.
+type SendSoapOption func(*soapConfig)
+
+type soapConfig struct {
+	headerContent string
+}
+
+// WithSOAPHeader adds inner-Header XML to the envelope. See
+// SendSoapWithHeader for the content contract.
+func WithSOAPHeader(headerContent string) SendSoapOption {
+	return func(c *soapConfig) { c.headerContent = headerContent }
+}
+
+// SendSoapWithOptions is the workhorse behind SendSoap and
+// SendSoapWithHeader; call it directly when you need to combine
+// options or pass options not surfaced by the convenience wrappers.
+func (dev Device) SendSoapWithOptions(endpoint, xmlRequestBody string, opts ...SendSoapOption) (*http.Response, error) {
+	var cfg soapConfig
+	for _, o := range opts {
+		o(&cfg)
+	}
 	soap := gosoap.NewEmptySOAP()
 	soap.AddStringBodyContent(xmlRequestBody)
 	soap.AddRootNamespaces(Xlmns)
 	soap.AddAction()
-
-	//Auth Handling
+	if cfg.headerContent != "" {
+		if err := soap.AddStringHeaderContents(cfg.headerContent); err != nil {
+			return nil, fmt.Errorf("add header content: %w", err)
+		}
+	}
 	if dev.params.Username != "" && dev.params.Password != "" {
 		soap.AddWSSecurity(dev.params.Username, dev.params.Password)
 	}
 
 	servResp, err := networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
 	if err != nil {
-		// Close server response body to reuse the connection
 		if servResp != nil {
 			servResp.Body.Close()
 		}
 		servResp, err = networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
 	}
-
 	return servResp, err
 }
 
@@ -388,7 +459,7 @@ func (dev *Device) CallOnvifFunction(serviceName, functionName string, data []by
 	}
 	xmlRequestBody := string(requestBody)
 
-	servResp, err := dev.SendSoap(endpoint, xmlRequestBody)
+	servResp, err := dev.SendSoapWithOptions(endpoint, xmlRequestBody)
 	if err != nil {
 		return nil, fmt.Errorf("fail to send the '%s' request for the web service '%s', %v", functionName, serviceName, err)
 	}

Device_test.go

@@ -1,9 +1,14 @@
 package onvif
 
 import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestDevice_SetDeviceInfoFromScopes(t *testing.T) {
@@ -22,3 +27,242 @@ func TestDevice_SetDeviceInfoFromScopes(t *testing.T) {
 	assert.Equal(t, device.info.Name, name)
 	assert.Equal(t, device.info.Model, hardware)
 }
+
+// TestDevice_SendSoapWithHeader_InjectsHeaderXML verifies that the
+// supplied header XML lands inside the SOAP <Header> element of the
+// outgoing request. AXIS-style WS-Addressing reference parameter
+// echoing depends on this — without it the camera returns
+// ter:InvalidArgs on every PullMessages.
+func TestDevice_SendSoapWithHeader_InjectsHeaderXML(t *testing.T) {
+	const headerXML = `<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event" wsa:IsReferenceParameter="true">297</dom0:SubscriptionId>`
+	const bodyXML = `<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"><tev:Timeout>PT5S</tev:Timeout><tev:MessageLimit>32</tev:MessageLimit></tev:PullMessages>`
+
+	var captured string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		captured = string(b)
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{
+		params: DeviceParams{
+			Xaddr:      strings.TrimPrefix(srv.URL, "http://"),
+			HttpClient: srv.Client(),
+		},
+	}
+	resp, err := dev.SendSoapWithHeader(srv.URL, bodyXML, headerXML)
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+
+	headerStart := strings.Index(captured, "Header>")
+	bodyStart := strings.Index(captured, "Body>")
+	require.NotEqual(t, -1, headerStart, "envelope must contain <Header>; got: %s", captured)
+	require.Greater(t, bodyStart, headerStart, "Body must follow Header in the envelope")
+
+	headerSlice := captured[headerStart:bodyStart]
+	assert.Contains(t, headerSlice, "SubscriptionId",
+		"injected header element must land inside SOAP <Header>")
+	assert.Contains(t, headerSlice, "297")
+
+	bodySlice := captured[bodyStart:]
+	assert.Contains(t, bodySlice, "PullMessages",
+		"body content must land inside SOAP <Body>")
+}
+
+// Per WS-Addressing 1.0 SOAP Binding §3.4 every reference parameter is a separate
+// SOAP Header block. Vendors that declare two ref params would silently
+// produce a header-less request if the implementation only accepts a
+// single top-level element.
+func TestDevice_SendSoapWithHeader_AcceptsMultipleTopLevelChildren(t *testing.T) {
+	const headerXML = `<a:Foo xmlns:a="ns/a">1</a:Foo><b:Bar xmlns:b="ns/b">2</b:Bar>`
+	const bodyXML = `<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"/>`
+
+	var captured string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		captured = string(b)
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{
+		Xaddr:      strings.TrimPrefix(srv.URL, "http://"),
+		HttpClient: srv.Client(),
+	}}
+	resp, err := dev.SendSoapWithHeader(srv.URL, bodyXML, headerXML)
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+
+	headerSlice := captured[strings.Index(captured, "Header>"):strings.Index(captured, "Body>")]
+	assert.Contains(t, headerSlice, "Foo")
+	assert.Contains(t, headerSlice, "Bar")
+}
+
+func TestDevice_SendSoapWithHeader_RejectsElementFreeContent(t *testing.T) {
+	var hits int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits++
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+	dev := Device{params: DeviceParams{HttpClient: srv.Client()}}
+
+	// Well-formed XML but contains no child elements — would otherwise
+	// parse, yield zero ChildElements, and send a header-less request.
+	_, err := dev.SendSoapWithHeader(srv.URL, "<body/>", "just text content")
+	require.Error(t, err)
+	assert.Equal(t, 0, hits,
+		"non-empty header content with no element children must fail fast")
+}
+
+// SendSoapWithOptions is the variadic shape that future per-call
+// options (timeout, context, ...) will hang off. SendSoap and
+// SendSoapWithHeader stay as thin convenience wrappers so existing
+// callers are not forced to migrate.
+func TestDevice_SendSoapWithOptions_WithHeaderMatchesSendSoapWithHeader(t *testing.T) {
+	const headerXML = `<dom0:SubscriptionId xmlns:dom0="urn:test">42</dom0:SubscriptionId>`
+	const bodyXML = `<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"/>`
+
+	var captured string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		captured = string(b)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{HttpClient: srv.Client()}}
+	resp, err := dev.SendSoapWithOptions(srv.URL, bodyXML, WithSOAPHeader(headerXML))
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+	assert.Contains(t, captured, "SubscriptionId")
+	assert.Contains(t, captured, "42")
+}
+
+func TestDevice_SendSoapWithOptions_NoOptsMatchesSendSoap(t *testing.T) {
+	var captured string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		captured = string(b)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{HttpClient: srv.Client()}}
+	resp, err := dev.SendSoapWithOptions(srv.URL, `<tev:Body xmlns:tev="x"/>`)
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+	assert.NotContains(t, captured, "IsReferenceParameter",
+		"no opts should produce a header-less envelope")
+}
+
+// Digest auth fallback path: the camera 401s the first POST and the
+// retry computes a digest. The ref-params header must survive the
+// retry — losing it would silently re-introduce the AXIS regression
+// on every authenticated camera.
+func TestDevice_SendSoapWithHeader_PreservesHeaderAcrossDigestRetry(t *testing.T) {
+	const headerXML = `<dom0:SubscriptionId xmlns:dom0="urn:vendor:axis" wsa:IsReferenceParameter="true">297</dom0:SubscriptionId>`
+	var capturedSecondBody string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") == "" {
+			w.Header().Set("WWW-Authenticate", `Digest realm="onvif", nonce="abc", qop="auth"`)
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		b, _ := io.ReadAll(r.Body)
+		capturedSecondBody = string(b)
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{
+		Xaddr:      strings.TrimPrefix(srv.URL, "http://"),
+		HttpClient: srv.Client(),
+		Username:   "admin",
+		Password:   "secret",
+	}}
+	resp, err := dev.SendSoapWithHeader(srv.URL, `<tev:PullMessages xmlns:tev="http://www.onvif.org/ver10/events/wsdl"/>`, headerXML)
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+	assert.Contains(t, capturedSecondBody, "SubscriptionId",
+		"digest retry must carry the same ref-params header as the first attempt")
+	assert.Contains(t, capturedSecondBody, "297")
+}
+
+func TestDevice_SendSoapWithHeader_PropagatesMalformedHeaderError(t *testing.T) {
+	var hits int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits++
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+	dev := Device{params: DeviceParams{HttpClient: srv.Client()}}
+
+	_, err := dev.SendSoapWithHeader(srv.URL, "<body/>", "<not-closed")
+	require.Error(t, err)
+	assert.Equal(t, 0, hits,
+		"malformed header XML must fail fast — no request should reach the camera with a missing header block")
+}
+
+// Digest retry: networking.SendSoapWithDigest strips the wsse:Security
+// element from the envelope before re-POSTing so credentials don't go
+// on the wire twice (once via WS-Security, once via the digest header).
+// Pin the behaviour from the Device layer.
+func TestDevice_SendSoapWithOptions_DigestRetryStripsWSSE(t *testing.T) {
+	var authedBody string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") == "" {
+			w.Header().Set("WWW-Authenticate", `Digest realm="onvif", nonce="abc", qop="auth"`)
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		b, _ := io.ReadAll(r.Body)
+		authedBody = string(b)
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{
+		HttpClient: srv.Client(),
+		Username:   "admin",
+		Password:   "secret",
+	}}
+	resp, err := dev.SendSoapWithOptions(srv.URL, "<tev:X xmlns:tev='x'/>")
+	require.NoError(t, err)
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+	require.NotEmpty(t, authedBody, "expected an authenticated POST after the 401 challenge")
+	assert.NotContains(t, authedBody, "UsernameToken",
+		"digest retry must strip wsse:Security/UsernameToken; otherwise credentials go on the wire twice")
+}
+
+// Last-write-wins on duplicate SendSoapOption — pin the behaviour so
+// the next maintainer adding an option doesn't accidentally introduce
+// a merge or first-wins semantic.
+func TestDevice_SendSoapWithOptions_DuplicateWithSOAPHeaderLastWins(t *testing.T) {
+	var captured string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		captured = string(b)
+	}))
+	t.Cleanup(srv.Close)
+
+	dev := Device{params: DeviceParams{HttpClient: srv.Client()}}
+	_, err := dev.SendSoapWithOptions(srv.URL, "<body/>",
+		WithSOAPHeader(`<a:First xmlns:a="x"/>`),
+		WithSOAPHeader(`<b:Second xmlns:b="y"/>`),
+	)
+	require.NoError(t, err)
+	assert.NotContains(t, captured, "First", "first WithSOAPHeader must be overwritten")
+	assert.Contains(t, captured, "Second")
+}

event/stream/doc.go

@@ -18,7 +18,8 @@
 //	}
 //
 // NewStream performs network I/O so auth and reachability failures
-// surface synchronously. Events and Errors close when the Stream stops;
+// surface synchronously, and rejects a client timeout that cannot
+// outlast PullTimeout with ErrInvalidOptions. Events and Errors close when the Stream stops;
 // Errors sends are non-blocking so a stalled consumer drops older
 // errors rather than blocking the pull loop. After a silent reconnect,
 // the next batch's events carry Event.AfterReconnect=true.

event/stream/main_test.go

@@ -0,0 +1,14 @@
+package stream
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+// Catches any pull, renew, or recreate goroutine that outlives its
+// Stream — a regression class that's silent in production until
+// goroutine count drifts up and triggers OOM.
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}

event/stream/pulltimeout.go

@@ -0,0 +1,51 @@
+package stream
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// ErrInvalidOptions marks a configuration that cannot succeed. Callers
+// retry the pull/renew/recreate errors; retrying this one never helps,
+// so it is a distinct sentinel they can short-circuit on.
+var ErrInvalidOptions = errors.New("stream: invalid options")
+
+// minClientHeadroom is how far http.Client.Timeout must exceed
+// PullTimeout. The client ceiling covers dial, TLS and the response
+// transfer on top of the poll it has to outlast, and starts before the
+// camera has parsed the request; on a cellular bearer that overhead
+// runs to hundreds of milliseconds.
+const minClientHeadroom = 5 * time.Second
+
+// validateClientTimeout rejects a client ceiling that cannot outlast
+// the PullMessages long-poll plus minClientHeadroom.
+//
+// Zero means unbounded and is accepted: it is the SDK's default when a
+// caller passes no client, so rejecting it would break every default
+// consumer. Note it is not risk-free — the caller interface documents
+// that ctx cannot interrupt an in-flight SOAP call, so only the client
+// timeout can unwedge a stalled camera.
+func validateClientTimeout(clientTimeout, pullTimeout time.Duration) error {
+	if clientTimeout == 0 || clientTimeout >= pullTimeout+minClientHeadroom {
+		return nil
+	}
+	return fmt.Errorf(
+		"%w: http.Client.Timeout (%s) must exceed PullTimeout (%s) by at least %s; PullMessages is a long-poll and the client would abort every quiet pull",
+		ErrInvalidOptions, clientTimeout, pullTimeout, minClientHeadroom)
+}
+
+// clientTimeoutOf reports the device's HTTP client ceiling, or 0 when
+// the SDK is using its own default (unbounded) client.
+func clientTimeoutOf(dev *onvif.Device) time.Duration {
+	if dev == nil {
+		return 0
+	}
+	c := dev.GetDeviceParams().HttpClient
+	if c == nil {
+		return 0
+	}
+	return c.Timeout
+}

event/stream/pulltimeout_test.go

@@ -0,0 +1,68 @@
+package stream
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestValidateClientTimeout — PullMessages is a long-poll: the camera
+// holds the connection open for PullTimeout waiting for an event. An
+// http.Client.Timeout covers the whole exchange and starts before the
+// camera has parsed the request, so a client ceiling at or below
+// PullTimeout loses the race on every quiet interval and the pull can
+// only ever fail. This shipped once (both were 5s) and presented as a
+// slow camera rather than a misconfiguration.
+//
+// Strict inequality is not enough: the client also has to cover dial,
+// TLS and the response transfer, which on a cellular bearer runs to
+// hundreds of milliseconds. Hence a real headroom floor.
+func TestValidateClientTimeout(t *testing.T) {
+	tests := []struct {
+		name    string
+		client  time.Duration
+		pull    time.Duration
+		wantErr bool
+	}{
+		{"unbounded client is the caller's risk, not an error", 0, 30 * time.Second, false},
+		{"comfortable headroom", 40 * time.Second, 30 * time.Second, false},
+		{"exactly the minimum headroom", 30*time.Second + minClientHeadroom, 30 * time.Second, false},
+		{"a hair under the minimum headroom", 30*time.Second + minClientHeadroom - time.Millisecond, 30 * time.Second, true},
+		{"strictly greater but no headroom", 30*time.Second + time.Millisecond, 30 * time.Second, true},
+		{"equal timeouts always lose", 5 * time.Second, 5 * time.Second, true},
+		{"client below pull", 4 * time.Second, 30 * time.Second, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateClientTimeout(tt.client, tt.pull)
+			if tt.wantErr {
+				require.Error(t, err, "client=%s pull=%s must be rejected", tt.client, tt.pull)
+				assert.ErrorIs(t, err, ErrInvalidOptions,
+					"callers need a sentinel to tell a permanent misconfiguration from a transient failure")
+				assert.Contains(t, err.Error(), "PullTimeout",
+					"the error must name the option the caller has to change")
+				return
+			}
+			assert.NoError(t, err, "client=%s pull=%s must be accepted", tt.client, tt.pull)
+		})
+	}
+}
+
+// TestErrInvalidOptions_IsDistinctFromStreamErrors — the pull/renew/
+// recreate errors are transient and callers retry them. A bad Options
+// never becomes valid by retrying, so it must not be mistaken for one.
+func TestErrInvalidOptions_IsDistinctFromStreamErrors(t *testing.T) {
+	err := validateClientTimeout(5*time.Second, 5*time.Second)
+	require.Error(t, err)
+
+	var pull ErrPullFailed
+	var renew ErrRenewFailed
+	var recreate ErrRecreateFailed
+	assert.False(t, errors.As(err, &pull))
+	assert.False(t, errors.As(err, &renew))
+	assert.False(t, errors.As(err, &recreate))
+}

event/stream/reconnect.go

@@ -74,7 +74,7 @@ func (s *Stream) pullLoop(ctx context.Context) {
 // attemptRecreate returns (justRecreated, cont). cont is false only
 // when ctx cancelled during backoff so the caller exits the loop.
 func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) (justRecreated, cont bool) {
-	addr, err := createPullPoint(s.caller, s.opts)
+	ref, err := createPullPoint(s.caller, s.opts)
 	if err != nil {
 		s.surfaceError(ErrRecreateFailed{Err: err})
 		if !sleepCtx(ctx, jitter(*backoff)) {
@@ -86,7 +86,7 @@ func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *ti
 		}
 		return false, true
 	}
-	s.setPullPoint(addr)
+	s.setPullPoint(ref)
 	*failures = 0
 	*backoff = s.opts.RetryBackoff
 	return true, true

event/stream/reconnect_test.go

@@ -13,7 +13,9 @@ import (
 
 // createPullPointRespAlt mirrors the first fixture but returns a
 // different SubscriptionReference Address so a test can prove that
-// subsequent pulls hit the recreated endpoint.
+// subsequent pulls hit the recreated endpoint. Like createPullPointResp,
+// it intentionally omits <TerminationTime> so renew scheduling stays
+// driven by opts.
 const createPullPointRespAlt = `<?xml version="1.0" encoding="UTF-8"?>
 <env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
               xmlns:wsa="http://www.w3.org/2005/08/addressing"
@@ -23,8 +25,6 @@ const createPullPointRespAlt = `<?xml version="1.0" encoding="UTF-8"?>
       <tev:SubscriptionReference>
         <wsa:Address>http://camera.local/onvif/Events/PullSub_2</wsa:Address>
       </tev:SubscriptionReference>
-      <tev:CurrentTime>2026-05-21T10:30:10Z</tev:CurrentTime>
-      <tev:TerminationTime>2026-05-21T10:31:10Z</tev:TerminationTime>
     </tev:CreatePullPointSubscriptionResponse>
   </env:Body>
 </env:Envelope>`

event/stream/renew.go

@@ -10,49 +10,88 @@ import (
 	"github.com/kerberos-io/onvif/xsd"
 )
 
-// renewLoop surfaces renew failures and continues. A permanently
+// renewLoop sleeps until the next deadline (camera-granted termination
+// minus RenewMargin), renews, and repeats. On failure it backs off via
+// nextRenewIntervalAfterError so the loop doesn't busy-loop against
+// the 1s floor when the previous grant has just expired. A permanently
 // failing renew lets the subscription die at the camera; the pull
 // loop's reconnect path then recreates it — recreate is the only
 // reliable recovery once a subscription is GC'd.
 func (s *Stream) renewLoop(ctx context.Context) {
-	interval := s.opts.InitialTermination - s.opts.RenewMargin
-	if interval <= 0 {
-		// Pathological config (margin >= termination): renew at
-		// half termination so we still refresh.
-		interval = s.opts.InitialTermination / 2
-		if interval <= 0 {
-			interval = time.Second
-		}
-	}
-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
 	for {
-		select {
-		case <-ctx.Done():
+		ref, gen := s.snapshotPullPoint()
+		if !sleepCtx(ctx, nextRenewInterval(ref.GrantedTermination, s.opts, s.now())) {
 			return
-		case <-ticker.C:
-			if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil {
-				s.surfaceError(ErrRenewFailed{Err: err})
+		}
+		granted, err := renewPullPoint(s.caller, s.getPullPoint(), s.opts)
+		if err != nil {
+			s.surfaceError(ErrRenewFailed{Err: err})
+			if !sleepCtx(ctx, nextRenewIntervalAfterError(s.opts)) {
+				return
 			}
+			continue
+		}
+		if !granted.IsZero() {
+			s.updateGrantedTerminationIfGen(gen, granted)
 		}
 	}
 }
 
+// nextRenewInterval prefers the camera-granted termination so we never
+// schedule a renew past the actual expiry, with opts.InitialTermination
+// as the fallback when the camera didn't supply one.
+func nextRenewInterval(granted time.Time, opts Options, now time.Time) time.Duration {
+	var base time.Duration
+	if !granted.IsZero() {
+		base = granted.Sub(now)
+	} else {
+		base = opts.InitialTermination
+	}
+	d := base - opts.RenewMargin
+	if d <= 0 {
+		d = base / 2
+	}
+	if d <= 0 {
+		d = time.Second
+	}
+	return d
+}
+
+// nextRenewIntervalAfterError returns the post-failure sleep. The
+// grant is typically already in the past by the time renew has failed
+// once, so nextRenewInterval would floor to 1s and hammer the camera.
+// Recovery is the pull loop's reconnect path; we just need to not
+// accelerate retries past the configured RetryBackoff.
+func nextRenewIntervalAfterError(opts Options) time.Duration {
+	if opts.RetryBackoff > 0 {
+		return opts.RetryBackoff
+	}
+	return time.Second
+}
+
 // renewPullPoint sends Renew with an absolute UTC TerminationTime.
 // WS-BaseNotification §6.1.1 also allows xsd:duration but older
 // Hikvision, some Dahua and some Bosch firmwares reject the
-// relative form.
-func renewPullPoint(c caller, endpoint string, opts Options) error {
+// relative form. Returns the camera-granted TerminationTime parsed
+// from the response (zero on absence) so the caller can reschedule.
+func renewPullPoint(c caller, ref subscriptionRef, opts Options) (time.Time, error) {
 	absoluteEnd := time.Now().UTC().Add(opts.InitialTermination).Format("2006-01-02T15:04:05Z")
 	req := event.Renew{TerminationTime: xsd.String(absoluteEnd)}
 	body, err := xml.Marshal(req)
 	if err != nil {
-		return fmt.Errorf("marshal Renew: %w", err)
+		return time.Time{}, fmt.Errorf("marshal Renew: %w", err)
 	}
-	resp, err := c.SendSoap(endpoint, string(body))
+	headerXML, err := buildRefParamsHeader(ref.RefParamsXML)
 	if err != nil {
-		return err
+		return time.Time{}, fmt.Errorf("build ref params header: %w", err)
 	}
-	_, err = readClose(resp)
-	return err
+	resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML)
+	if err != nil {
+		return time.Time{}, enrichSOAPErr(resp, err)
+	}
+	respBody, err := readClose(resp)
+	if err != nil {
+		return time.Time{}, err
+	}
+	return extractTerminationTime(respBody), nil
 }

event/stream/renew_test.go

@@ -2,6 +2,7 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"testing"
 	"time"
@@ -168,3 +169,109 @@ func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) {
 	assert.NotContains(t, renewBody, "PT", "Renew should not send relative duration; some firmwares reject it")
 	assert.Regexp(t, `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z`, renewBody, "Renew should send absolute RFC3339 UTC")
 }
+
+// --- Wiring: renew surfaces SOAP fault detail -------------------------
+
+func TestRenewPullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueSendSoap(renewFaultBody, errors.New("400 Bad Request"))
+	_, err := renewPullPoint(fc, subscriptionRef{Address: "http://camera/sub"}, defaultOptions())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "renew-specific complaint",
+		"renewPullPoint must enrich transport errors with the camera's SOAP fault")
+}
+
+const renewFaultBody = `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Body><env:Fault>
+    <env:Code><env:Value>env:Sender</env:Value></env:Code>
+    <env:Reason><env:Text xml:lang="en">renew-specific complaint</env:Text></env:Reason>
+  </env:Fault></env:Body>
+</env:Envelope>`
+
+func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) {
+	ref := subscriptionRef{
+		Address:      "http://192.168.1.10/onvif/services",
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
+	}
+	fc := newFakeCaller()
+	_, err := renewPullPoint(fc, ref, defaultOptions())
+	require.NoError(t, err)
+	require.Len(t, fc.sendSoapHeaders, 1)
+	hdr := fc.sendSoapHeaders[0]
+	assert.Contains(t, hdr, "SubscriptionId")
+	assert.Contains(t, hdr, "297")
+	assert.Contains(t, hdr, `IsReferenceParameter="true"`)
+}
+
+// Regression: when GrantedTermination has just passed (renew failed
+// at or after the deadline), nextRenewInterval floors to one second
+// and the loop hammers the camera at 1 Hz until reconnect. Original
+// ticker design retried at the configured cadence regardless. After
+// a failure the loop must use a backoff decoupled from the stale
+// grant.
+func TestNextRenewIntervalAfterError_BacksOffAtLeastRetryBackoff(t *testing.T) {
+	opts := Options{RetryBackoff: time.Second}
+	got := nextRenewIntervalAfterError(opts)
+	assert.GreaterOrEqual(t, got, opts.RetryBackoff,
+		"failure path must back off at least RetryBackoff, not 1s floor on stale grant")
+}
+
+func TestNextRenewIntervalAfterError_DefaultsToOneSecondWhenRetryBackoffZero(t *testing.T) {
+	got := nextRenewIntervalAfterError(Options{})
+	assert.Equal(t, time.Second, got,
+		"zero RetryBackoff must yield the safety floor, not a tight 0-duration sleep")
+}
+
+// Lost-update race: renew snapshots the ref, the SOAP call returns,
+// and meanwhile attemptRecreate replaced pullPoint with a fresh
+// subscription. If renew blindly writes the OLD subscription's
+// granted time onto the NEW subscription, the new schedule is wrong.
+// Update must be conditioned on "same subscription as when I read."
+func TestUpdateGrantedTermination_DropsWriteWhenSubscriptionRotated(t *testing.T) {
+	// s.now is intentionally nil — this test does not exercise any
+	// time-dependent path; only the gen-counter accessors.
+	s := &Stream{}
+	original := subscriptionRef{Address: "http://camera/sub-A"}
+	s.setPullPoint(original)
+	gen := s.pullPointGen()
+
+	// Simulate recreate happening between snapshot and write.
+	s.setPullPoint(subscriptionRef{Address: "http://camera/sub-B"})
+
+	// Old generation's renew result must NOT overwrite sub-B's grant.
+	bogus := time.Date(1999, 1, 1, 0, 0, 0, 0, time.UTC)
+	s.updateGrantedTerminationIfGen(gen, bogus)
+
+	assert.True(t, s.getPullPoint().GrantedTermination.IsZero(),
+		"stale renew result must be discarded after a subscription rotation")
+}
+
+func TestUpdateGrantedTermination_AppliesWhenGenMatches(t *testing.T) {
+	// s.now is intentionally nil — gen-counter path only.
+	s := &Stream{}
+	s.setPullPoint(subscriptionRef{Address: "http://camera/sub"})
+	gen := s.pullPointGen()
+	t1 := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	s.updateGrantedTerminationIfGen(gen, t1)
+	assert.Equal(t, t1, s.getPullPoint().GrantedTermination)
+}
+
+// renewLoop must capture (ref, gen) atomically. Two separate
+// getPullPoint() + pullPointGen() reads leave a window in which a
+// concurrent setPullPoint advances gen between the two reads — the
+// renew then runs against ref-N but believes its captured gen is N+1,
+// and updateGrantedTerminationIfGen("succeeds") writing the old
+// subscription's grant onto the new one. Single snapshot closes it.
+func TestSnapshotPullPoint_AtomicReadOfRefAndGen(t *testing.T) {
+	// s.now is intentionally nil — gen-counter path only.
+	s := &Stream{}
+	s.setPullPoint(subscriptionRef{Address: "A"}) // gen=1
+	ref, gen := s.snapshotPullPoint()
+	assert.Equal(t, "A", ref.Address)
+	assert.Equal(t, uint64(1), gen)
+
+	s.setPullPoint(subscriptionRef{Address: "B"}) // gen=2
+	ref, gen = s.snapshotPullPoint()
+	assert.Equal(t, "B", ref.Address)
+	assert.Equal(t, uint64(2), gen)
+}

event/stream/soap.go

@@ -12,17 +12,24 @@ import (
 	"strings"
 	"time"
 
+	"github.com/beevik/etree"
 	"github.com/kerberos-io/onvif/event"
 	"github.com/kerberos-io/onvif/xsd"
 )
 
-// maxResponseBytes caps SOAP response buffering. ONVIF PullMessages
-// bodies are normally <100KB even with dense analytics payloads;
-// 10 MiB is comfortably above legitimate traffic while keeping a
-// hostile or buggy camera from OOMing the process.
+// maxResponseBytes caps SOAP response buffering on success paths.
+// ONVIF PullMessages bodies are normally <100KB even with dense
+// analytics payloads; 10 MiB is well above legitimate traffic while
+// keeping a hostile or buggy camera from OOMing the process.
 const maxResponseBytes = 10 << 20
 
-func createPullPoint(c caller, opts Options) (string, error) {
+// maxErrorBodyBytes caps the body read by enrichSOAPErr. The pull
+// retry loop runs every RetryBackoff (~1s) so an unbounded read on
+// the error path would churn 10 MiB/s per wedged camera. Fault bodies
+// are always small.
+const maxErrorBodyBytes = 64 << 10
+
+func createPullPoint(c caller, opts Options) (subscriptionRef, error) {
 	term := xsd.String(durationToXSD(opts.InitialTermination))
 	req := event.CreatePullPointSubscription{InitialTerminationTime: &term}
 	if opts.RawTopicFilter != "" {
@@ -35,26 +42,53 @@ func createPullPoint(c caller, opts Options) (string, error) {
 	}
 	resp, err := c.CallMethod(req)
 	if err != nil {
-		return "", err
+		return subscriptionRef{}, enrichSOAPErr(resp, err)
 	}
 	body, err := readClose(resp)
 	if err != nil {
-		return "", err
+		return subscriptionRef{}, err
 	}
 	var decoded event.CreatePullPointSubscriptionResponse
 	if err := unmarshalNode(body, "CreatePullPointSubscriptionResponse", &decoded); err != nil {
-		return "", err
+		return subscriptionRef{}, err
 	}
 	addr := string(decoded.SubscriptionReference.Address)
 	if addr == "" {
-		return "", errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address")
+		return subscriptionRef{}, errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address")
 	}
-	return addr, nil
+	return subscriptionRef{
+		Address:            addr,
+		RefParamsXML:       extractReferenceParameters(body),
+		GrantedTermination: extractTerminationTime(body),
+	}, nil
 }
 
+// extractTerminationTime parses the absolute UTC instant the camera
+// granted as the subscription expiry. Returns zero on absence or parse
+// failure — callers fall back to opts.InitialTermination.
+func extractTerminationTime(body string) time.Time {
+	m := terminationTimeRE.FindStringSubmatch(body)
+	if len(m) < 2 {
+		return time.Time{}
+	}
+	t, err := time.Parse(time.RFC3339, strings.TrimSpace(m[1]))
+	if err != nil {
+		return time.Time{}
+	}
+	return t
+}
+
+// terminationTimeRE matches the first <*:TerminationTime> in the body.
+// Only safe on responses that contain exactly one — currently
+// CreatePullPointSubscriptionResponse and RenewResponse via
+// extractTerminationTime. PullMessagesResponse also has a
+// TerminationTime element; do not call extractTerminationTime on pull
+// bodies.
+var terminationTimeRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?TerminationTime\b[^>]*>(.*?)</(?:[^:>\s]+:)?TerminationTime>`)
+
 // pullMessages returns an empty slice (no error) when the camera had
 // nothing within PullTimeout.
-func pullMessages(c caller, endpoint string, opts Options) ([]event.NotificationMessage, error) {
+func pullMessages(c caller, ref subscriptionRef, opts Options) ([]event.NotificationMessage, error) {
 	req := event.PullMessages{
 		Timeout:      xsd.Duration(durationToXSD(opts.PullTimeout)),
 		MessageLimit: xsd.Int(opts.MessageLimit),
@@ -63,9 +97,13 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification
 	if err != nil {
 		return nil, fmt.Errorf("marshal PullMessages: %w", err)
 	}
-	resp, err := c.SendSoap(endpoint, string(body))
+	headerXML, err := buildRefParamsHeader(ref.RefParamsXML)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("build ref params header: %w", err)
+	}
+	resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML)
+	if err != nil {
+		return nil, enrichSOAPErr(resp, err)
 	}
 	respBody, err := readClose(resp)
 	if err != nil {
@@ -78,19 +116,23 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification
 	return decoded.NotificationMessage, nil
 }
 
-// unsubscribePullPoint is best-effort. Empty endpoint is a no-op
-// (construction failed before installing one).
-func unsubscribePullPoint(c caller, endpoint string) error {
-	if endpoint == "" {
+// unsubscribePullPoint is best-effort. Empty Address is a no-op
+// (construction failed before installing a subscription).
+func unsubscribePullPoint(c caller, ref subscriptionRef) error {
+	if ref.Address == "" {
 		return nil
 	}
 	body, err := xml.Marshal(event.Unsubscribe{})
 	if err != nil {
 		return fmt.Errorf("marshal Unsubscribe: %w", err)
 	}
-	resp, err := c.SendSoap(endpoint, string(body))
+	headerXML, err := buildRefParamsHeader(ref.RefParamsXML)
 	if err != nil {
-		return err
+		return fmt.Errorf("build ref params header: %w", err)
+	}
+	resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML)
+	if err != nil {
+		return enrichSOAPErr(resp, err)
 	}
 	_, err = readClose(resp)
 	return err
@@ -148,24 +190,166 @@ var (
 	soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)</(?:[^:>\s]+:)?faultstring>`)
 	// SOAP 1.2: <Fault>...<Reason><Text>reason</Text></Reason>...</Fault>
 	soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)</(?:[^:>\s]+:)?Text>`)
+	// SOAP 1.2 Subcode: <Code>...<Subcode><Value>ter:InvalidArgs</Value></Subcode>...
+	soap12SubcodeRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Subcode\b[^>]*>.*?<(?:[^:>\s]+:)?Value[^>]*>(.*?)</(?:[^:>\s]+:)?Value>`)
+
+	// WS-Security blocks may carry our Username/Password if the camera
+	// echoes the request in a fault; scrub before logging. The
+	// alternation handles the truncated case where the read cap fell
+	// between <Security> and </Security>: in that case nothing past
+	// the opening tag is safe to retain — the replacement re-emits a
+	// synthetic close tag, dropping the remainder of the body excerpt
+	// (max-redact preferred to max-context for log lines).
+	// wssePasswordRE is the belt-and-braces fallback for
+	// non-conformant cameras emitting Password / UsernameToken outside
+	// a Security wrapper; same truncation handling.
+	wsseSecurityRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Security\b[^>]*>(?:.*?</(?:[^:>\s]+:)?Security>|.*)`)
+	wssePasswordRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Password\b[^>]*>(?:.*?</(?:[^:>\s]+:)?Password>|.*)`)
 )
 
-// extractSOAPFault returns the reason text from a SOAP fault or empty
-// when the body is not a fault. Handles SOAP 1.1 (faultstring) and
-// SOAP 1.2 (Reason/Text) shapes.
+// extractSOAPFault returns the reason text from a SOAP fault, falling
+// back to the Subcode value when Reason/Text is empty (AXIS pattern).
+// Returns "" when the body is not a fault.
 func extractSOAPFault(body string) string {
 	if !strings.Contains(body, "Fault") {
 		return ""
 	}
 	if m := soap11FaultRE.FindStringSubmatch(body); len(m) > 1 {
-		return strings.TrimSpace(m[1])
+		if r := strings.TrimSpace(m[1]); r != "" {
+			return r
+		}
 	}
 	if m := soap12FaultRE.FindStringSubmatch(body); len(m) > 1 {
+		if r := strings.TrimSpace(m[1]); r != "" {
+			return r
+		}
+	}
+	return extractSOAPSubcode(body)
+}
+
+// Anchored to SubscriptionReference because other WS-Addressing
+// endpoint references in the same envelope (wsa:ReplyTo, wsa:FaultTo,
+// wsa:From) may also carry ReferenceParameters that are not ours.
+var (
+	subscriptionRefRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?SubscriptionReference\b[^>]*>(.*?)</(?:[^:>\s]+:)?SubscriptionReference>`)
+	refParamsRE       = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?ReferenceParameters\b[^>]*>(.*?)</(?:[^:>\s]+:)?ReferenceParameters>`)
+)
+
+// buildRefParamsHeader produces the SOAP <Header> inner XML from the
+// full <*:ReferenceParameters> element returned by
+// extractReferenceParameters. Each child element is re-emitted with
+// wsa:IsReferenceParameter="true" added and any xmlns:* declared on
+// the parent inherited onto it (so the standalone child stays valid).
+// Empty input yields empty output.
+func buildRefParamsHeader(refParamsXML string) (string, error) {
+	if strings.TrimSpace(refParamsXML) == "" {
+		return "", nil
+	}
+	doc := etree.NewDocument()
+	if err := doc.ReadFromString(refParamsXML); err != nil {
+		return "", fmt.Errorf("parse ref params: %w", err)
+	}
+	wrapper := doc.Root()
+	if wrapper == nil {
+		return "", errors.New("ref params has no root element")
+	}
+	if wrapper.Tag != "ReferenceParameters" {
+		return "", fmt.Errorf("ref params root must be <*:ReferenceParameters>, got <%s>", wrapper.Tag)
+	}
+	var out strings.Builder
+	for _, child := range wrapper.ChildElements() {
+		c := child.Copy()
+		inheritXmlns(c, wrapper)
+		c.CreateAttr("wsa:IsReferenceParameter", "true")
+		d := etree.NewDocument()
+		d.SetRoot(c)
+		s, err := d.WriteToString()
+		if err != nil {
+			return "", fmt.Errorf("serialise ref param child: %w", err)
+		}
+		out.WriteString(strings.TrimRight(s, "\n"))
+	}
+	return out.String(), nil
+}
+
+// inheritXmlns copies xmlns / xmlns:* declarations from src onto dst
+// when dst doesn't already declare them, so a child whose namespace
+// prefix was declared on an ancestor stays valid in isolation.
+func inheritXmlns(dst, src *etree.Element) {
+	for _, attr := range src.Attr {
+		isDefault := attr.Space == "" && attr.Key == "xmlns"
+		isPrefixed := attr.Space == "xmlns"
+		if !isDefault && !isPrefixed {
+			continue
+		}
+		key := attr.Key
+		if isPrefixed {
+			key = "xmlns:" + attr.Key
+		}
+		if dst.SelectAttr(key) != nil {
+			continue
+		}
+		dst.CreateAttr(key, attr.Value)
+	}
+}
+
+// extractReferenceParameters returns the verbatim inner XML so callers
+// can echo it (with wsa:IsReferenceParameter="true") into the SOAP
+// Header of subscription-scoped requests per WS-Addressing 1.0 SOAP Binding §3.4.
+// Without that echo, AXIS rejects PullMessages with ter:InvalidArgs.
+func extractReferenceParameters(body string) string {
+	sub := subscriptionRefRE.FindStringSubmatch(body)
+	if len(sub) < 2 {
+		return ""
+	}
+	return strings.TrimSpace(refParamsRE.FindString(sub[1]))
+}
+
+// extractSOAPSubcode is the fallback when Reason/Text is empty — AXIS
+// routinely sends an empty <Text/> alongside a populated Subcode
+// (e.g. "ter:InvalidArgs"), and that subcode is the only actionable
+// signal the operator gets.
+func extractSOAPSubcode(body string) string {
+	m := soap12SubcodeRE.FindStringSubmatch(body)
+	if len(m) > 1 {
 		return strings.TrimSpace(m[1])
 	}
 	return ""
 }
 
+// maxErrExcerpt caps the body excerpt appended to an enriched error so
+// a wedged camera streaming a multi-megabyte HTML error page can not
+// flood logs with every retry.
+const maxErrExcerpt = 512
+
+// enrichSOAPErr appends the camera's actual complaint (SOAP Fault
+// reason, then Subcode, then raw body excerpt) to a transport error so
+// operators see *why* the camera said 400 instead of just "400 Bad
+// Request". The original err is preserved via %w for errors.Is/As.
+func enrichSOAPErr(resp *http.Response, err error) error {
+	if err == nil {
+		return nil
+	}
+	if resp == nil || resp.Body == nil {
+		return err
+	}
+	defer resp.Body.Close()
+	b, readErr := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes))
+	if readErr != nil || len(b) == 0 {
+		return err
+	}
+	body := wsseSecurityRE.ReplaceAllString(string(b), "<Security>[REDACTED]</Security>")
+	body = wssePasswordRE.ReplaceAllString(body, "<Password>[REDACTED]</Password>")
+	if reason := extractSOAPFault(body); reason != "" {
+		return fmt.Errorf("SOAP fault: %s: %w", reason, err)
+	}
+	excerpt := strings.TrimSpace(body)
+	if len(excerpt) > maxErrExcerpt {
+		excerpt = excerpt[:maxErrExcerpt] + "...(truncated)"
+	}
+	return fmt.Errorf("response body: %s: %w", excerpt, err)
+}
+
 // durationToXSD formats a duration as xsd:duration PTnS. Second
 // precision is sufficient — ONVIF cameras do not honour sub-second
 // pull timeouts.

event/stream/soap_test.go

@@ -2,8 +2,12 @@ package stream
 
 import (
 	"context"
+	"errors"
+	"io"
+	"net/http"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -84,3 +88,634 @@ func testContext(t *testing.T) context.Context {
 	t.Cleanup(cancel)
 	return ctx
 }
+
+// --- Error enrichment from SOAP response bodies ----------------------
+
+func fakeResponse(body string) *http.Response {
+	return &http.Response{
+		StatusCode: http.StatusBadRequest,
+		Body:       io.NopCloser(strings.NewReader(body)),
+	}
+}
+
+func TestEnrichSOAPErr_NilErrReturnsNil(t *testing.T) {
+	assert.NoError(t, enrichSOAPErr(fakeResponse("anything"), nil))
+}
+
+func TestEnrichSOAPErr_NilRespPreservesOriginal(t *testing.T) {
+	orig := errors.New("transport boom")
+	got := enrichSOAPErr(nil, orig)
+	assert.ErrorIs(t, got, orig)
+	assert.Equal(t, orig.Error(), got.Error(), "no body, no extra context to add")
+}
+
+func TestEnrichSOAPErr_SOAP11FaultStringAppearsInError(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope/">
+  <env:Body><env:Fault><faultstring>not authorized</faultstring></env:Fault></env:Body>
+</env:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("400 Bad Request"))
+	require.Error(t, got)
+	assert.Contains(t, got.Error(), "400 Bad Request")
+	assert.Contains(t, got.Error(), "not authorized")
+}
+
+func TestEnrichSOAPErr_SOAP12ReasonAppearsInError(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Body><env:Fault>
+    <env:Code><env:Value>env:Sender</env:Value></env:Code>
+    <env:Reason><env:Text xml:lang="en">Subscription has expired</env:Text></env:Reason>
+  </env:Fault></env:Body>
+</env:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("400 Bad Request"))
+	require.Error(t, got)
+	assert.Contains(t, got.Error(), "Subscription has expired")
+}
+
+// Pins the AXIS case: a Fault with populated Subcode but an empty
+// <Reason><Text/></Reason>. Without subcode fallback, the only signal
+// the operator sees is "400 Bad Request".
+func TestEnrichSOAPErr_EmptyReasonFallsBackToSubcode(t *testing.T) {
+	body := `<SOAP-ENV:Envelope xmlns:SOAP-ENV="http://www.w3.org/2003/05/soap-envelope" xmlns:ter="http://www.onvif.org/ver10/error">
+  <SOAP-ENV:Body><SOAP-ENV:Fault>
+    <SOAP-ENV:Code>
+      <SOAP-ENV:Value>SOAP-ENV:Sender</SOAP-ENV:Value>
+      <SOAP-ENV:Subcode><SOAP-ENV:Value>ter:InvalidArgs</SOAP-ENV:Value></SOAP-ENV:Subcode>
+    </SOAP-ENV:Code>
+    <SOAP-ENV:Reason><SOAP-ENV:Text xml:lang="en"/></SOAP-ENV:Reason>
+  </SOAP-ENV:Fault></SOAP-ENV:Body>
+</SOAP-ENV:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("Post with digest error: 400: 400 Bad Request"))
+	require.Error(t, got)
+	assert.Contains(t, got.Error(), "ter:InvalidArgs",
+		"AXIS-style empty-Reason Faults must surface their Subcode")
+}
+
+func TestEnrichSOAPErr_NonFaultBodyIncludesExcerpt(t *testing.T) {
+	body := `<html><body>404 Not Found — /onvif/services missing</body></html>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("404 Not Found"))
+	require.Error(t, got)
+	assert.Contains(t, got.Error(), "/onvif/services missing")
+}
+
+func TestEnrichSOAPErr_LargeNonFaultBodyTruncated(t *testing.T) {
+	// A misbehaving camera could stream a multi-megabyte body. The
+	// helper must cap the excerpt so a wedged camera does not flood
+	// logs.
+	body := strings.Repeat("X", 8192)
+	got := enrichSOAPErr(fakeResponse(body), errors.New("500"))
+	require.Error(t, got)
+	assert.Less(t, len(got.Error()), 2048,
+		"enriched error must stay log-line sized even on huge bodies")
+}
+
+func TestEnrichSOAPErr_PreservesOriginalForErrorsIs(t *testing.T) {
+	// Callers wrap pull/renew/recreate errors with errors.As in
+	// logStreamError; enrichment must keep the original wrappable.
+	orig := errors.New("sentinel")
+	got := enrichSOAPErr(fakeResponse(`<env:Fault><faultstring>x</faultstring></env:Fault>`), orig)
+	assert.ErrorIs(t, got, orig)
+}
+
+// --- Subcode extraction ----------------------------------------------
+
+func TestExtractSOAPSubcode_Present(t *testing.T) {
+	body := `<SOAP-ENV:Code>
+  <SOAP-ENV:Value>SOAP-ENV:Sender</SOAP-ENV:Value>
+  <SOAP-ENV:Subcode><SOAP-ENV:Value>ter:InvalidArgs</SOAP-ENV:Value></SOAP-ENV:Subcode>
+</SOAP-ENV:Code>`
+	assert.Equal(t, "ter:InvalidArgs", extractSOAPSubcode(body))
+}
+
+func TestExtractSOAPSubcode_Absent(t *testing.T) {
+	assert.Empty(t, extractSOAPSubcode(`<env:Code><env:Value>env:Sender</env:Value></env:Code>`))
+}
+
+// --- Wiring: each SOAP call site routes errors through enrichSOAPErr -
+
+const faultBody = `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Body><env:Fault>
+    <env:Code><env:Value>env:Sender</env:Value></env:Code>
+    <env:Reason><env:Text xml:lang="en">camera-specific complaint</env:Text></env:Reason>
+  </env:Fault></env:Body>
+</env:Envelope>`
+
+func TestCreatePullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(faultBody, errors.New("400 Bad Request"))
+	_, err := createPullPoint(fc, defaultOptions())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "camera-specific complaint",
+		"createPullPoint must enrich transport errors with the camera's SOAP fault")
+}
+
+func TestPullMessages_EnrichesTransportErrWithFaultReason(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueSendSoap(faultBody, errors.New("400 Bad Request"))
+	_, err := pullMessages(fc, subscriptionRef{Address: "http://camera/sub"}, defaultOptions())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "camera-specific complaint",
+		"pullMessages must enrich transport errors with the camera's SOAP fault")
+}
+
+func TestUnsubscribePullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueSendSoap(faultBody, errors.New("400 Bad Request"))
+	err := unsubscribePullPoint(fc, subscriptionRef{Address: "http://camera/sub"})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "camera-specific complaint",
+		"unsubscribePullPoint must enrich transport errors with the camera's SOAP fault")
+}
+
+// --- ReferenceParameters extraction (WS-Addressing 1.0 SOAP Binding §3.4) ---------
+//
+// AXIS encodes the subscription identity in <wsa:ReferenceParameters>
+// inside CreatePullPointSubscriptionResponse rather than in the URL
+// itself. Subsequent PullMessages/Renew/Unsubscribe MUST echo those
+// elements verbatim into the SOAP Header, or the camera responds with
+// ter:InvalidArgs. The auto-generated event.ReferenceParametersType is
+// an empty struct (drops children), so we extract the raw inner XML.
+
+func TestExtractReferenceParameters_AXISShape(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa5="http://www.w3.org/2005/08/addressing"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference>
+      <wsa5:Address>http://192.168.1.10/onvif/services</wsa5:Address>
+      <wsa5:ReferenceParameters>
+        <dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>
+      </wsa5:ReferenceParameters>
+    </tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	got := extractReferenceParameters(body)
+	assert.Contains(t, got, "SubscriptionId")
+	assert.Contains(t, got, "297")
+	assert.Contains(t, got, `xmlns:dom0="http://www.axis.com/2009/event"`,
+		"namespace declaration on the SubscriptionId child must survive extraction")
+}
+
+func TestExtractReferenceParameters_AbsentReturnsEmpty(t *testing.T) {
+	// Geovision/Hikvision-style: Address only, no ReferenceParameters.
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa="http://www.w3.org/2005/08/addressing">
+  <env:Body><tev:CreatePullPointSubscriptionResponse xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+    <tev:SubscriptionReference>
+      <wsa:Address>http://camera/onvif/Events/Sub_1</wsa:Address>
+    </tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	assert.Empty(t, extractReferenceParameters(body))
+}
+
+func TestExtractReferenceParameters_EmptyBodyReturnsEmpty(t *testing.T) {
+	assert.Empty(t, extractReferenceParameters(""))
+}
+
+// --- createPullPoint returns both address and ref params -------------
+
+func TestCreatePullPoint_ReturnsRefParamsAlongsideAddress(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa5="http://www.w3.org/2005/08/addressing"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference>
+      <wsa5:Address>http://192.168.1.10/onvif/services</wsa5:Address>
+      <wsa5:ReferenceParameters>
+        <dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>
+      </wsa5:ReferenceParameters>
+    </tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	fc := newFakeCaller()
+	fc.queueCallMethod(body, nil)
+	ref, err := createPullPoint(fc, defaultOptions())
+	require.NoError(t, err)
+	assert.Equal(t, "http://192.168.1.10/onvif/services", ref.Address)
+	assert.Contains(t, ref.RefParamsXML, "SubscriptionId")
+	assert.Contains(t, ref.RefParamsXML, "297")
+}
+
+func TestCreatePullPoint_VendorWithoutRefParams_RefParamsEmpty(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	ref, err := createPullPoint(fc, defaultOptions())
+	require.NoError(t, err)
+	assert.NotEmpty(t, ref.Address)
+	assert.Empty(t, ref.RefParamsXML)
+}
+
+// --- Reference-parameter echoing in subscription-scoped calls --------
+//
+// WS-Addressing 1.0 SOAP Binding §3.4 requires each <wsa:ReferenceParameters> child
+// to be echoed as a SOAP Header block carrying wsa:IsReferenceParameter
+// ="true". AXIS rejects PullMessages with ter:InvalidArgs when this is
+// absent.
+
+func TestPullMessages_EchoesRefParamsWithIsReferenceParameterAttribute(t *testing.T) {
+	ref := subscriptionRef{
+		Address:      "http://192.168.1.10/onvif/services",
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
+	}
+	fc := newFakeCaller()
+	_, err := pullMessages(fc, ref, defaultOptions())
+	require.NoError(t, err)
+	require.Len(t, fc.sendSoapHeaders, 1)
+	hdr := fc.sendSoapHeaders[0]
+	assert.Contains(t, hdr, "SubscriptionId", "ref param element must be echoed")
+	assert.Contains(t, hdr, "297", "ref param value must be echoed")
+	assert.Contains(t, hdr, `IsReferenceParameter="true"`,
+		"WS-Addressing 1.0 SOAP Binding §3.4 requires the attribute on each echoed element")
+}
+
+func TestPullMessages_NoRefParams_HeaderEmpty(t *testing.T) {
+	ref := subscriptionRef{Address: "http://camera/sub", RefParamsXML: ""}
+	fc := newFakeCaller()
+	_, err := pullMessages(fc, ref, defaultOptions())
+	require.NoError(t, err)
+	require.Len(t, fc.sendSoapHeaders, 1)
+	assert.Empty(t, fc.sendSoapHeaders[0], "vendors without ref params get no extra header")
+}
+
+func TestPullMessages_PostsToAddressFromRef(t *testing.T) {
+	ref := subscriptionRef{Address: "http://camera/specific-sub-endpoint", RefParamsXML: ""}
+	fc := newFakeCaller()
+	_, err := pullMessages(fc, ref, defaultOptions())
+	require.NoError(t, err)
+	require.NotEmpty(t, fc.sendSoapCalls)
+	assert.Equal(t, "http://camera/specific-sub-endpoint", fc.sendSoapCalls[0][0])
+}
+
+// --- Building the header XML from raw ref params ----------------------
+
+func TestBuildRefParamsHeader_AddsIsReferenceParameter(t *testing.T) {
+	raw := `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+		`<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>` +
+		`</wsa:ReferenceParameters>`
+	got, err := buildRefParamsHeader(raw)
+	require.NoError(t, err)
+	assert.Contains(t, got, "SubscriptionId")
+	assert.Contains(t, got, "297")
+	assert.Contains(t, got, `xmlns:dom0="http://www.axis.com/2009/event"`,
+		"original namespace declaration must survive")
+	assert.Contains(t, got, `IsReferenceParameter="true"`)
+}
+
+func TestBuildRefParamsHeader_MultipleTopLevelChildren(t *testing.T) {
+	raw := `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+		`<a:Foo xmlns:a="ns/a">1</a:Foo><b:Bar xmlns:b="ns/b">2</b:Bar>` +
+		`</wsa:ReferenceParameters>`
+	got, err := buildRefParamsHeader(raw)
+	require.NoError(t, err)
+	assert.Equal(t, 2, strings.Count(got, `IsReferenceParameter="true"`),
+		"attribute must be added to every top-level child, not just the first")
+	assert.Contains(t, got, "Foo")
+	assert.Contains(t, got, "Bar")
+}
+
+func TestBuildRefParamsHeader_EmptyInputReturnsEmpty(t *testing.T) {
+	got, err := buildRefParamsHeader("")
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
+func TestUnsubscribePullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) {
+	ref := subscriptionRef{
+		Address:      "http://192.168.1.10/onvif/services",
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
+	}
+	fc := newFakeCaller()
+	require.NoError(t, unsubscribePullPoint(fc, ref))
+	require.Len(t, fc.sendSoapHeaders, 1)
+	hdr := fc.sendSoapHeaders[0]
+	assert.Contains(t, hdr, "SubscriptionId")
+	assert.Contains(t, hdr, `IsReferenceParameter="true"`)
+}
+
+func TestUnsubscribePullPoint_EmptyAddressIsNoOp(t *testing.T) {
+	fc := newFakeCaller()
+	require.NoError(t, unsubscribePullPoint(fc, subscriptionRef{}))
+	assert.Empty(t, fc.sendSoapCalls, "no SOAP call should happen when there is no subscription endpoint")
+}
+
+// End-to-end multi-child wiring through the production caller, not
+// just the unit-tested builder. Without the fix to addHeaderChildren
+// in Device.SendSoapWithHeader, the second child would silently
+// vanish from the wire envelope.
+func TestPullMessages_TwoRefParamsEachLandsOnTheWire(t *testing.T) {
+	ref := subscriptionRef{
+		Address: "http://camera/sub",
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+			`<a:Foo xmlns:a="ns/a">1</a:Foo>` +
+			`<b:Bar xmlns:b="ns/b">2</b:Bar>` +
+			`</wsa:ReferenceParameters>`,
+	}
+	fc := newFakeCaller()
+	_, err := pullMessages(fc, ref, defaultOptions())
+	require.NoError(t, err)
+	require.Len(t, fc.sendSoapHeaders, 1)
+	hdr := fc.sendSoapHeaders[0]
+	assert.Equal(t, 2, strings.Count(hdr, `IsReferenceParameter="true"`))
+	assert.Contains(t, hdr, "Foo")
+	assert.Contains(t, hdr, "Bar")
+}
+
+// A camera echoing our request in a fault response (some debug-mode
+// firmwares do) or a fault that includes the Security header verbatim
+// would otherwise leak the WS-Security Username/Password into operator
+// logs. The body excerpt must scrub the Security block before the
+// fault extractor and the excerpt fallback see it.
+func TestEnrichSOAPErr_RedactsWSSESecurityBlock(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Header><wsse:Security xmlns:wsse="x"><wsse:UsernameToken>
+    <wsse:Username>admin</wsse:Username>
+    <wsse:Password>hunter2</wsse:Password>
+  </wsse:UsernameToken></wsse:Security></env:Header>
+  <env:Body>plain text excerpt</env:Body>
+</env:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("400"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2", "Password must never reach logs")
+	assert.NotContains(t, got.Error(), "admin", "Username must never reach logs")
+	assert.Contains(t, got.Error(), "REDACTED", "redaction marker must remain visible")
+}
+
+// Same vendor pattern as the enrichSOAPErr case but reached via
+// unmarshalNode → extractSOAPFault on a 200 OK response carrying a
+// Fault. Diverging from enrichSOAPErr's fallback chain would mean
+// PullMessages reports "missing PullMessagesResponse element" instead
+// of the actionable ter:InvalidArgs.
+func TestExtractSOAPFault_FallsBackToSubcodeWhenReasonEmpty(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Body><env:Fault>
+    <env:Code>
+      <env:Value>env:Sender</env:Value>
+      <env:Subcode><env:Value>ter:InvalidArgs</env:Value></env:Subcode>
+    </env:Code>
+    <env:Reason><env:Text xml:lang="en"/></env:Reason>
+  </env:Fault></env:Body>
+</env:Envelope>`
+	assert.Equal(t, "ter:InvalidArgs", extractSOAPFault(body))
+}
+
+// WS-Addressing 1.0 Core §2.1 allows ReferenceParameters in any endpoint
+// reference (wsa:From, wsa:ReplyTo, wsa:FaultTo, ...). An unanchored
+// search would silently pick up the wrong one.
+func TestExtractReferenceParameters_AnchoredToSubscriptionReference(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa="http://www.w3.org/2005/08/addressing"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Header>
+    <wsa:ReplyTo>
+      <wsa:Address>http://anon</wsa:Address>
+      <wsa:ReferenceParameters>
+        <decoy:NotTheRealOne xmlns:decoy="urn:decoy">DO-NOT-PICK</decoy:NotTheRealOne>
+      </wsa:ReferenceParameters>
+    </wsa:ReplyTo>
+  </env:Header>
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference>
+      <wsa:Address>http://camera/sub</wsa:Address>
+      <wsa:ReferenceParameters>
+        <dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>
+      </wsa:ReferenceParameters>
+    </tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	got := extractReferenceParameters(body)
+	assert.Contains(t, got, "SubscriptionId")
+	assert.Contains(t, got, "297")
+	assert.NotContains(t, got, "DO-NOT-PICK",
+		"ref params from wsa:ReplyTo must not leak through — only SubscriptionReference's children belong on PullMessages")
+}
+
+// When a vendor declares the namespace prefix on the parent
+// <ReferenceParameters> element rather than the child (legal XML, just
+// different from AXIS's shape), naïve inner-only extraction strips the
+// declaration and produces children with orphaned prefixes that fail
+// to round-trip. Inheritance must propagate ancestor xmlns onto each
+// child before serialisation.
+func TestBuildRefParamsHeader_InheritsParentXmlns(t *testing.T) {
+	parentScopedXmlns := `<wsa:ReferenceParameters xmlns:dom0="urn:vendor:axis">` +
+		`<dom0:SubscriptionId>297</dom0:SubscriptionId>` +
+		`</wsa:ReferenceParameters>`
+	got, err := buildRefParamsHeader(parentScopedXmlns)
+	require.NoError(t, err)
+	assert.NotContains(t, got, "ReferenceParameters",
+		"the wrapping element must not appear in output — each param child is its own header block")
+	assert.Contains(t, got, "SubscriptionId")
+	assert.Contains(t, got, "297")
+	assert.Contains(t, got, `xmlns:dom0="urn:vendor:axis"`,
+		"the dom0 prefix is undeclared on the child itself — it must be inherited from the parent so the standalone child stays valid XML")
+	assert.Contains(t, got, `IsReferenceParameter="true"`)
+}
+
+// Pins the contract change: extractReferenceParameters returns the
+// full <ReferenceParameters> element (including its own attributes),
+// not just the inner content, so parent-scoped xmlns survives into
+// buildRefParamsHeader.
+func TestExtractReferenceParameters_IncludesParentElementForXmlnsPreservation(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa="http://www.w3.org/2005/08/addressing"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference>
+      <wsa:Address>http://camera</wsa:Address>
+      <wsa:ReferenceParameters xmlns:dom0="urn:vendor:axis">
+        <dom0:SubscriptionId>297</dom0:SubscriptionId>
+      </wsa:ReferenceParameters>
+    </tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	got := extractReferenceParameters(body)
+	assert.Contains(t, got, "ReferenceParameters",
+		"extractor must include the wrapping element so parent-scoped xmlns survives")
+	assert.Contains(t, got, `xmlns:dom0="urn:vendor:axis"`)
+	assert.Contains(t, got, "SubscriptionId")
+}
+
+// --- Camera-granted TerminationTime -----------------------------------
+//
+// Cameras may grant a shorter subscription than we ask for. Scheduling
+// the next renew from opts.InitialTermination instead of what the
+// camera actually granted leads to expired subscriptions and the
+// recreate-recovery path firing unnecessarily.
+
+func TestCreatePullPoint_CapturesGrantedTermination(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa="http://www.w3.org/2005/08/addressing"
+                      xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference><wsa:Address>http://camera/sub</wsa:Address></tev:SubscriptionReference>
+    <wsnt:CurrentTime>2026-05-27T13:19:11Z</wsnt:CurrentTime>
+    <wsnt:TerminationTime>2026-05-27T13:21:11Z</wsnt:TerminationTime>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	fc := newFakeCaller()
+	fc.queueCallMethod(body, nil)
+	ref, err := createPullPoint(fc, defaultOptions())
+	require.NoError(t, err)
+	expected, _ := time.Parse(time.RFC3339, "2026-05-27T13:21:11Z")
+	assert.Equal(t, expected, ref.GrantedTermination)
+}
+
+func TestCreatePullPoint_NoTerminationTimeYieldsZeroTime(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+                      xmlns:wsa="http://www.w3.org/2005/08/addressing"
+                      xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference><wsa:Address>http://camera/sub</wsa:Address></tev:SubscriptionReference>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	fc := newFakeCaller()
+	fc.queueCallMethod(body, nil)
+	ref, err := createPullPoint(fc, defaultOptions())
+	require.NoError(t, err)
+	assert.True(t, ref.GrantedTermination.IsZero(),
+		"absent TerminationTime must yield zero so renew falls back to opts")
+}
+
+func TestNextRenewInterval_UsesGrantedTerminationMinusMargin(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	granted := now.Add(60 * time.Second)
+	opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second}
+	assert.Equal(t, 50*time.Second, nextRenewInterval(granted, opts, now))
+}
+
+func TestNextRenewInterval_FallsBackToInitialTerminationWhenGrantedZero(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second}
+	assert.Equal(t, 50*time.Second, nextRenewInterval(time.Time{}, opts, now))
+}
+
+func TestNextRenewInterval_FloorsAtOneSecondIfAlreadyExpired(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	granted := now.Add(-1 * time.Second) // camera says we're already expired
+	opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second}
+	assert.Equal(t, time.Second, nextRenewInterval(granted, opts, now),
+		"never sleep zero or negative — recreate-recovery handles the truly-dead case")
+}
+
+func TestBuildRefParamsHeader_MalformedXMLReturnsError(t *testing.T) {
+	_, err := buildRefParamsHeader("<not-closed")
+	require.Error(t, err)
+}
+
+func TestBuildRefParamsHeader_WhitespaceOnlyReturnsEmpty(t *testing.T) {
+	got, err := buildRefParamsHeader("   \n\t ")
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
+// Worst case: the response body's <Security> block starts within the
+// 64 KiB error cap but its </Security> is past it. The non-greedy
+// regex needs a close tag — without one the redaction misses and
+// raw Username/Password reaches the excerpt. Verify the helper
+// strips from <Security to EOF when no close tag is present.
+func TestEnrichSOAPErr_RedactsSecurityBlockMissingCloseTag(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x"><env:Header>` +
+		`<wsse:Security xmlns:wsse="y">` +
+		`<wsse:Username>admin</wsse:Username>` +
+		`<wsse:Password>hunter2</wsse:Password>` +
+		// no </wsse:Security> — simulates a Security block truncated
+		// at the 64 KiB read cap.
+		strings.Repeat("padding ", 1000)
+	got := enrichSOAPErr(fakeResponse(body), errors.New("500"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2",
+		"truncated Security block must not leak Password to the excerpt")
+	assert.NotContains(t, got.Error(), "admin",
+		"truncated Security block must not leak Username to the excerpt")
+}
+
+// The wrapper-only contract means an element whose local name merely
+// ends in "ReferenceParameters" cannot be mistaken for the wrapper —
+// the root must be exactly <*:ReferenceParameters>. Anything else is
+// a contract violation by the caller and surfaces as an error.
+func TestBuildRefParamsHeader_RejectsNonReferenceParametersRoot(t *testing.T) {
+	raw := `<my:MyReferenceParameters xmlns:my="urn:vendor:my">X</my:MyReferenceParameters>`
+	_, err := buildRefParamsHeader(raw)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "ReferenceParameters")
+}
+
+// A non-conformant camera echoing UsernameToken/Password outside a
+// <Security> wrapper would still leak credentials through the body
+// excerpt. Belt-and-braces: redact Password elements directly too.
+func TestEnrichSOAPErr_RedactsBarePasswordElement(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x"><env:Body>` +
+		`<wsse:UsernameToken xmlns:wsse="y">` +
+		`<wsse:Username>admin</wsse:Username>` +
+		`<wsse:Password>hunter2</wsse:Password>` +
+		`</wsse:UsernameToken>` +
+		`</env:Body></env:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("400"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2",
+		"Password must be redacted regardless of whether it's wrapped in Security")
+}
+
+// --- Edge-case coverage flagged in review -----------------------------
+
+func TestExtractTerminationTime_MalformedDateYieldsZero(t *testing.T) {
+	body := `<env:Body><wsnt:TerminationTime>not-a-date</wsnt:TerminationTime></env:Body>`
+	assert.True(t, extractTerminationTime(body).IsZero(),
+		"unparseable datetime must not panic and must not return a garbage time — fall back to opts")
+}
+
+func TestNextRenewInterval_MarginEqualsBaseFallsToHalf(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	opts := Options{InitialTermination: 30 * time.Second, RenewMargin: 30 * time.Second}
+	got := nextRenewInterval(time.Time{}, opts, now)
+	assert.Equal(t, 15*time.Second, got,
+		"when margin == base, the helper must fall through to base/2 rather than the 1s floor")
+}
+
+func TestExtractReferenceParameters_EmptySubscriptionReferenceReturnsEmpty(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x" xmlns:tev="y">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference/>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	assert.Empty(t, extractReferenceParameters(body))
+}
+
+func TestEnrichSOAPErr_RedactsTruncatedPasswordOutsideSecurity(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x"><env:Body>` +
+		`<wsse:UsernameToken xmlns:wsse="y">` +
+		`<wsse:Username>admin</wsse:Username>` +
+		`<wsse:Password>hunter2` // truncated — no </Password>, no </UsernameToken>, no </Envelope>
+	got := enrichSOAPErr(fakeResponse(body), errors.New("500"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2",
+		"Password without a closing tag (truncated at cap) must still be redacted")
+}
+
+func TestEnrichSOAPErr_RedactsMultipleSecurityBlocks(t *testing.T) {
+	body := `<E>` +
+		`<wsse:Security xmlns:wsse="y"><wsse:Password>secret1</wsse:Password></wsse:Security>` +
+		`<wsse:Security xmlns:wsse="y"><wsse:Password>secret2</wsse:Password></wsse:Security>` +
+		`</E>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("500"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "secret1")
+	assert.NotContains(t, got.Error(), "secret2",
+		"ReplaceAllString must catch every Security block, not just the first")
+}
+
+func TestBuildRefParamsHeader_RejectsNonReferenceParametersRoot_Table(t *testing.T) {
+	cases := []struct {
+		name, raw string
+	}{
+		{"vendor suffix", `<my:MyReferenceParameters xmlns:my="urn:x">X</my:MyReferenceParameters>`},
+		{"multi-root", `<a:Foo xmlns:a="ns/a"/><b:Bar xmlns:b="ns/b"/>`},
+		{"unrelated element", `<not-the-wrapper>content</not-the-wrapper>`},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			_, err := buildRefParamsHeader(c.raw)
+			require.Error(t, err)
+		})
+	}
+}

event/stream/stream.go

@@ -39,7 +39,9 @@ type Options struct {
 	// server-side filtering is fragile across vendors and empty is
 	// required for AXIS.
 	RawTopicFilter string
-	// PullTimeout — zero means default (5s).
+	// PullTimeout — zero means default (5s). The device's
+	// http.Client.Timeout must exceed this by minClientHeadroom or
+	// NewStream returns ErrInvalidOptions.
 	PullTimeout time.Duration
 	// MessageLimit — zero means default (32). Busy AXIS cameras with
 	// many configured rules can burst beyond 10 per pull.
@@ -110,6 +112,22 @@ func (o Options) withDefaults() Options {
 	return d
 }
 
+// subscriptionRef holds the result of CreatePullPointSubscription.
+// AXIS encodes the subscription identity in RefParamsXML (a generic
+// /onvif/services Address plus a <wsa:ReferenceParameters> child);
+// other vendors put the identity in the Address itself, leaving
+// RefParamsXML empty. Subscription-scoped requests must echo a
+// non-empty RefParamsXML — see extractReferenceParameters.
+//
+// GrantedTermination is the absolute time the camera says the
+// subscription will expire if not renewed. May be less than
+// requested; renewLoop schedules from this rather than opts.
+type subscriptionRef struct {
+	Address            string
+	RefParamsXML       string
+	GrantedTermination time.Time
+}
+
 // caller is the *onvif.Device subset Stream depends on. Implementations
 // must:
 //
@@ -125,6 +143,7 @@ func (o Options) withDefaults() Options {
 type caller interface {
 	CallMethod(method any) (*http.Response, error)
 	SendSoap(endpoint, body string) (*http.Response, error)
+	SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error)
 }
 
 type deviceCaller struct{ dev *onvif.Device }
@@ -137,6 +156,10 @@ func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) {
 	return d.dev.SendSoap(endpoint, body)
 }
 
+func (d deviceCaller) SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error) {
+	return d.dev.SendSoapWithHeader(endpoint, body, headerXML)
+}
+
 // Stream owns a single ONVIF pull-point subscription. Safe for Close
 // from any goroutine while readers consume Events / Errors. Close is
 // idempotent.
@@ -144,8 +167,9 @@ type Stream struct {
 	caller caller
 	opts   Options
 
-	pullPointMu sync.Mutex
-	pullPoint   string
+	pullPointMu sync.Mutex // guards pullPoint and gen
+	pullPoint   subscriptionRef
+	gen         uint64 // bumped on every setPullPoint so renews detect mid-flight recreate
 
 	events chan Event
 	errors chan error
@@ -160,16 +184,50 @@ type Stream struct {
 	now func() time.Time
 }
 
-func (s *Stream) getPullPoint() string {
+func (s *Stream) getPullPoint() subscriptionRef {
 	s.pullPointMu.Lock()
 	defer s.pullPointMu.Unlock()
 	return s.pullPoint
 }
 
-func (s *Stream) setPullPoint(addr string) {
+func (s *Stream) setPullPoint(ref subscriptionRef) {
 	s.pullPointMu.Lock()
 	defer s.pullPointMu.Unlock()
-	s.pullPoint = addr
+	s.pullPoint = ref
+	s.gen++
+}
+
+// pullPointGen returns the current generation. Pair with
+// updateGrantedTerminationIfGen so a renew result issued against a
+// subscription that was rotated mid-flight (recreate path) is
+// discarded instead of overwriting the new subscription's grant.
+func (s *Stream) pullPointGen() uint64 {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	return s.gen
+}
+
+// snapshotPullPoint reads ref + gen under one lock so a concurrent
+// setPullPoint can't slip in between two separate accessor calls and
+// leave the caller with mismatched halves.
+func (s *Stream) snapshotPullPoint() (subscriptionRef, uint64) {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	return s.pullPoint, s.gen
+}
+
+// updateGrantedTerminationIfGen writes the granted time only when the
+// caller's snapshot is still current. Use setPullPoint to replace the
+// full ref; this updates GrantedTermination in place after a renew
+// response. Intentionally does not bump gen — that would defeat the
+// rotation-detection it implements.
+func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	if s.gen != gen {
+		return
+	}
+	s.pullPoint.GrantedTermination = t
 }
 
 // NewStream creates a Stream and performs CreatePullPointSubscription
@@ -178,12 +236,18 @@ func (s *Stream) setPullPoint(addr string) {
 //
 // The returned Stream stops when ctx is cancelled or Close is called.
 func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) {
+	// Checked before the subscription call: this config can only fail,
+	// so surfacing it here beats a stream that appears to work and
+	// silently survives on reconnects alone.
+	if err := validateClientTimeout(clientTimeoutOf(dev), opts.withDefaults().PullTimeout); err != nil {
+		return nil, err
+	}
 	return newStream(ctx, deviceCaller{dev: dev}, opts)
 }
 
 func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
 	opts = opts.withDefaults()
-	addr, err := createPullPoint(c, opts)
+	ref, err := createPullPoint(c, opts)
 	if err != nil {
 		return nil, fmt.Errorf("create pull point subscription: %w", err)
 	}
@@ -191,7 +255,7 @@ func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
 	s := &Stream{
 		caller:    c,
 		opts:      opts,
-		pullPoint: addr,
+		pullPoint: ref,
 		events:    make(chan Event, opts.BufferSize),
 		errors:    make(chan error, opts.BufferSize),
 		cancel:    cancel,

event/stream/stream_test.go

@@ -33,6 +33,7 @@ type fakeCaller struct {
 	defaultCall      fakeResp
 	callMethodCalls  []any
 	sendSoapCalls    [][2]string
+	sendSoapHeaders  []string
 	blockUnsubscribe chan struct{}
 	blockAllSendSoap chan struct{}
 }
@@ -71,10 +72,12 @@ func (f *fakeCaller) CallMethod(m any) (*http.Response, error) {
 		r = f.callMethodResps[0]
 		f.callMethodResps = f.callMethodResps[1:]
 	}
-	if r.err != nil {
+	// Mirror networking.SendSoap*: a 4xx/5xx returns body alongside
+	// err. Tests opt into that shape by queueing body + err together.
+	if r.err != nil && r.body == "" {
 		return nil, r.err
 	}
-	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
+	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, r.err
 }
 
 func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) {
@@ -96,10 +99,20 @@ func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) {
 		<-block
 	}
 
-	if r.err != nil {
+	if r.err != nil && r.body == "" {
 		return nil, r.err
 	}
-	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
+	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, r.err
+}
+
+// SendSoapWithHeader delegates body+endpoint recording to SendSoap so
+// existing assertions on sendSoapCalls keep working, and records the
+// header XML in a parallel slice for ref-params wiring tests.
+func (f *fakeCaller) SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error) {
+	f.mu.Lock()
+	f.sendSoapHeaders = append(f.sendSoapHeaders, headerXML)
+	f.mu.Unlock()
+	return f.SendSoap(endpoint, body)
 }
 
 func (f *fakeCaller) sendSoapCallCount() int {
@@ -112,6 +125,10 @@ func (f *fakeCaller) sendSoapCallCount() int {
 
 // createPullPointResp is the minimal SOAP envelope the lib's existing
 // xml.Decoder + getXMLNode path can extract a pull-point address from.
+// Intentionally omits <TerminationTime> so renewLoop falls back to
+// opts.InitialTermination — tests that drive renew timing depend on
+// that path. Tests that need the camera-granted termination capture
+// path use a dedicated fixture instead.
 const createPullPointResp = `<?xml version="1.0" encoding="UTF-8"?>
 <env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
               xmlns:wsa="http://www.w3.org/2005/08/addressing"
@@ -121,8 +138,6 @@ const createPullPointResp = `<?xml version="1.0" encoding="UTF-8"?>
       <tev:SubscriptionReference>
         <wsa:Address>http://camera.local/onvif/Events/PullSub_1</wsa:Address>
       </tev:SubscriptionReference>
-      <tev:CurrentTime>2026-05-21T10:30:00Z</tev:CurrentTime>
-      <tev:TerminationTime>2026-05-21T10:31:00Z</tev:TerminationTime>
     </tev:CreatePullPointSubscriptionResponse>
   </env:Body>
 </env:Envelope>`

event/stream/topics.go

@@ -79,10 +79,19 @@ var topicRules = []struct {
 	// https://developer.axis.com/vapix/network-video/event-and-action-services/
 	{"MotionRegionDetector/Motion", KindMotion},
 
-	// AXIS Guard suite — vendor analytics apps with Camera<N>Profile<ID>
-	// suffixes. Treated as motion so they can drive motion-triggered
-	// recording on cameras using these apps instead of basic VMD.
+	// tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_<N> — AXIS VMD 3, the
+	// firmware-builtin predecessor of the VMD 4 ACAP. It lives under
+	// RuleEngine, not CameraApplicationPlatform, so the VMD rule below
+	// does not cover it. Still shipping on deployed cameras.
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"RuleEngine/VMD3/", KindMotion},
+
+	// AXIS ACAP motion apps with Camera<N>Profile<ID> suffixes. VMD 4 is
+	// the stock app shipped on the camera; the Guard suite are the
+	// paid analytics products. Prefix-match because of the suffix.
+	// https://developer.axis.com/vapix/applications/vmd4
 	// https://developer.axis.com/vapix/applications/motion-guard
+	{"CameraApplicationPlatform/VMD/", KindMotion},
 	{"CameraApplicationPlatform/MotionGuard/", KindMotion},
 	{"CameraApplicationPlatform/FenceGuard/", KindMotion},
 	{"CameraApplicationPlatform/LoiteringGuard/", KindMotion},

event/stream/topics_test.go

@@ -25,6 +25,18 @@ func TestClassifyTopic(t *testing.T) {
 		{"axis_fence_guard", "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY", KindMotion},
 		{"axis_loitering_guard", "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1ProfileANY", KindMotion},
 
+		// AXIS VMD 4 — the stock motion app, and the one an installer
+		// reaches for before any Guard product. The profile suffix
+		// varies with the configured VMD profile.
+		{"axis_vmd4_profile_any", "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", KindMotion},
+		{"axis_vmd4_profile_numbered", "tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1", KindMotion},
+
+		// AXIS VMD 3 — the firmware-builtin predecessor, published under
+		// RuleEngine rather than CameraApplicationPlatform. Still
+		// shipping on deployed cameras.
+		{"axis_vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", KindMotion},
+		{"axis_vmd3_video_2", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_2", KindMotion},
+
 		// --- Tampering --------------------------------------------------
 
 		{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
@@ -88,6 +100,16 @@ func TestClassifyTopic(t *testing.T) {
 		{"relay_failure_not_digital_output", "tns1:Device/HardwareFailure/RelayFailure", KindUnknown},
 		{"digital_input_config_not_digital_input", "tns1:Device/IO/DigitalInputConfiguration", KindUnknown},
 		{"tamper_detector_log_not_tampering", "tns1:Device/Diagnostics/TamperDetectorLog", KindUnknown},
+
+		// The two AXIS VMD needles carry a trailing slash so they match a
+		// whole path segment. Without it, any sibling app or rule whose
+		// name merely starts with VMD / VMD3 would classify as motion and
+		// drive recording.
+		{"vmd_statistics_app_not_motion", "tnsaxis:CameraApplicationPlatform/VMDStatistics/Camera1", KindUnknown},
+		{"vmd3_config_rule_not_motion", "tns1:RuleEngine/tnsaxis:VMD3Config/Changed", KindUnknown},
+		// VMD3 is scoped to RuleEngine; the same name under another
+		// container is a different thing.
+		{"vmd3_outside_rule_engine_not_motion", "tnsaxis:Storage/VMD3/Status", KindUnknown},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {

examples/event/pullmessage/main.go

@@ -82,7 +82,7 @@ func main() {
 	if err != nil {
 		log.Fatalln(err)
 	}
-	res, err := dev.SendSoap(endPoint, string(requestBody))
+	res, err := dev.SendSoapWithOptions(endPoint, string(requestBody))
 	if err != nil {
 		log.Fatalln("fail to CallMethod:", err)
 	}

examples/event/renew/main.go

@@ -84,7 +84,7 @@ func main() {
 	if err != nil {
 		log.Fatalln(err)
 	}
-	res, err := dev.SendSoap(endPoint, string(requestBody))
+	res, err := dev.SendSoapWithOptions(endPoint, string(requestBody))
 	if err != nil {
 		log.Fatalln("fail to CallMethod:", err)
 	}

examples/event/unsubscribe/main.go

@@ -65,7 +65,7 @@ func main() {
 	if err != nil {
 		log.Fatalln(err)
 	}
-	res, err := dev.SendSoap(endPoint, string(requestBody))
+	res, err := dev.SendSoapWithOptions(endPoint, string(requestBody))
 	if err != nil {
 		log.Fatalln("fail to CallMethod:", err)
 	}

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/icholy/digest v0.1.23
 	github.com/juju/errors v1.0.0
 	github.com/stretchr/testify v1.8.4
+	go.uber.org/goleak v1.3.0
 	golang.org/x/net v0.19.0
 )
 

go.sum

@@ -74,6 +74,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.11 h1:BMaWp1Bb6fHwEtbplGBGJ498wD+LKlNSl25MjdZY4dU=
 github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
 golang.org/x/arch v0.3.0 h1:02VY4/ZcO/gBOH6PUaoiptASxtXU10jazRCP865E97k=
 golang.org/x/arch v0.3.0/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=

gosoap/soap-builder.go

@@ -2,6 +2,7 @@ package gosoap
 
 import (
 	"encoding/xml"
+	"errors"
 	"log"
 
 	"github.com/beevik/etree"
@@ -128,7 +129,13 @@ func (msg *SoapMessage) AddBodyContents(elements []*etree.Element) {
 	*msg = SoapMessage(res)
 }
 
-//AddStringHeaderContent for Envelope body
+// AddStringHeaderContent appends a single root element to the SOAP
+// Header. Use AddStringHeaderContents (plural) when the content
+// contains multiple sibling elements — for example WS-Addressing
+// reference parameters, which the spec requires as separate header
+// blocks. The two coexist for backwards compatibility: external
+// consumers of this library may rely on AddStringHeaderContent's
+// single-root constraint and the matching error on multi-root input.
 func (msg *SoapMessage) AddStringHeaderContent(data string) error {
 	doc := etree.NewDocument()
 
@@ -156,6 +163,44 @@ func (msg *SoapMessage) AddStringHeaderContent(data string) error {
 	return nil
 }
 
+// AddStringHeaderContents is the multi-root variant of
+// AddStringHeaderContent: it accepts any number of top-level sibling
+// elements (zero is an error) and appends each as its own SOAP Header
+// child. Needed because WS-Addressing 1.0 SOAP Binding §3.4 requires each
+// reference parameter to be a separate Header block, but a Go XML
+// document only has one root. Comments and text outside elements are
+// silently dropped.
+//
+// SECURITY: data forwards verbatim into the outbound envelope. Do not
+// pass content sourced from untrusted clients — see the same caveat
+// on onvif.Device.SendSoapWithOptions / SendSoapWithHeader.
+func (msg *SoapMessage) AddStringHeaderContents(data string) error {
+	in := etree.NewDocument()
+	if err := in.ReadFromString("<wrap>" + data + "</wrap>"); err != nil {
+		return err
+	}
+	wrap := in.SelectElement("wrap")
+	if wrap == nil {
+		return errors.New("AddStringHeaderContents: missing wrap root")
+	}
+	children := wrap.ChildElements()
+	if len(children) == 0 {
+		return errors.New("AddStringHeaderContents: no element children in content")
+	}
+
+	doc := etree.NewDocument()
+	if err := doc.ReadFromString(msg.String()); err != nil {
+		return err
+	}
+	header := doc.Root().SelectElement("Header")
+	for _, child := range children {
+		header.AddChild(child.Copy())
+	}
+	res, _ := doc.WriteToString()
+	*msg = SoapMessage(res)
+	return nil
+}
+
 //AddHeaderContent for Envelope body
 func (msg *SoapMessage) AddHeaderContent(element *etree.Element) {
 	doc := etree.NewDocument()

networking/digest_test.go

@@ -0,0 +1,147 @@
+package networking
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestSendSoapWithDigestReturnsServerError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for HTTP 500")
+	}
+}
+
+func TestSendSoapWithDigestReturnsServerErrorAfterAuthentication(t *testing.T) {
+	requestCount := 0
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestCount++
+		if requestCount == 1 {
+			writer.Header().Set("WWW-Authenticate", `Digest realm="AXIS", nonce="nonce", qop="auth", algorithm=MD5`)
+			writer.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		if !strings.HasPrefix(request.Header.Get("Authorization"), "Digest ") {
+			t.Error("authenticated retry is missing Digest Authorization header")
+		}
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for authenticated HTTP 500")
+	}
+	if requestCount != 2 {
+		t.Fatalf("request count = %d, want 2", requestCount)
+	}
+}
+
+func TestParseDigestChallenge(t *testing.T) {
+	challenge := `Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", algorithm=MD5`
+	parts := parseDigestChallenge(challenge)
+
+	cases := map[string]string{
+		"realm":     "testrealm@host.com",
+		"qop":       "auth,auth-int",
+		"nonce":     "dcd98b7102dd2f0e8b11d0f600bfb0c093",
+		"opaque":    "5ccc069c403ebaf9f0171e9517f40e41",
+		"algorithm": "MD5",
+	}
+	for key, want := range cases {
+		if got := parts[key]; got != want {
+			t.Errorf("parseDigestChallenge()[%q] = %q, want %q", key, got, want)
+		}
+	}
+}
+
+// TestMD5DigestVectors validates md5Hex and the response assembly order against
+// the canonical RFC 2617 section 3.5 worked example.
+func TestMD5DigestVectors(t *testing.T) {
+	ha1 := md5Hex("Mufasa:testrealm@host.com:Circle Of Life")
+	if want := "939e7578ed9e3c518a452acee763bce9"; ha1 != want {
+		t.Fatalf("HA1 = %q, want %q", ha1, want)
+	}
+
+	ha2 := md5Hex("GET:/dir/index.html")
+	if want := "39aff3a2bab6126f332b942af96d3366"; ha2 != want {
+		t.Fatalf("HA2 = %q, want %q", ha2, want)
+	}
+
+	response := md5Hex(strings.Join([]string{
+		ha1,
+		"dcd98b7102dd2f0e8b11d0f600bfb0c093", // nonce
+		"00000001",                           // nc
+		"0a4f113b",                           // cnonce
+		"auth",                               // qop
+		ha2,
+	}, ":"))
+	if want := "6629fae49393a05397450978507c4ef1"; response != want {
+		t.Fatalf("response = %q, want %q", response, want)
+	}
+}
+
+// TestNewDigestAuthorizationConsistency builds an Authorization header and then
+// recomputes the response from the emitted cnonce/nc to confirm the header is
+// internally consistent (correct field wiring and formula).
+func TestNewDigestAuthorizationConsistency(t *testing.T) {
+	const (
+		username = "admin"
+		password = "s3cret"
+		realm    = "IP Camera"
+		nonce    = "0cc175b9c0f1b6a831c399e269772661"
+	)
+	challenge := `Digest realm="` + realm + `", nonce="` + nonce + `", qop="auth", algorithm=MD5`
+
+	header := newDigestAuthorization(challenge, "POST", "http://192.168.1.10/onvif/ptz_service", username, password)
+	if header == "" {
+		t.Fatal("newDigestAuthorization returned empty header")
+	}
+	if !strings.HasPrefix(header, "Digest ") {
+		t.Fatalf("header does not start with Digest scheme: %q", header)
+	}
+
+	fields := parseDigestChallenge(header)
+	if fields["username"] != username {
+		t.Errorf("username = %q, want %q", fields["username"], username)
+	}
+	if fields["uri"] != "/onvif/ptz_service" {
+		t.Errorf("uri = %q, want %q", fields["uri"], "/onvif/ptz_service")
+	}
+	if fields["qop"] != "auth" {
+		t.Errorf("qop = %q, want %q", fields["qop"], "auth")
+	}
+
+	ha1 := md5Hex(username + ":" + realm + ":" + password)
+	ha2 := md5Hex("POST:/onvif/ptz_service")
+	want := md5Hex(strings.Join([]string{ha1, nonce, fields["nc"], fields["cnonce"], "auth", ha2}, ":"))
+	if fields["response"] != want {
+		t.Errorf("response = %q, want %q", fields["response"], want)
+	}
+}
+
+// TestNewDigestAuthorizationRejectsUnsupported ensures an empty header is
+// returned when required parameters are missing or the qop is unsupported.
+func TestNewDigestAuthorizationRejectsUnsupported(t *testing.T) {
+	if got := newDigestAuthorization(`Digest realm="r"`, "POST", "http://host/x", "u", "p"); got != "" {
+		t.Errorf("expected empty header when nonce missing, got %q", got)
+	}
+	if got := newDigestAuthorization(`Digest realm="r", nonce="n", qop="auth-int"`, "POST", "http://host/x", "u", "p"); got != "" {
+		t.Errorf("expected empty header for unsupported qop, got %q", got)
+	}
+}

networking/networking.go

@@ -2,90 +2,212 @@ package networking
 
 import (
 	"bytes"
+	"crypto/md5"
+	"crypto/rand"
+	"encoding/hex"
 	"fmt"
-	"io"
 	"net/http"
+	"net/url"
+	"regexp"
+	"strings"
 
 	"github.com/beevik/etree"
-	"github.com/icholy/digest"
 	"github.com/juju/errors"
 )
 
+const soapContentType = "application/soap+xml; charset=utf-8"
+
 // SendSoap send soap message
 func SendSoap(httpClient *http.Client, endpoint, message string) (*http.Response, error) {
-	resp, err := httpClient.Post(endpoint, "application/soap+xml; charset=utf-8", bytes.NewBufferString(message))
+	resp, err := httpClient.Post(endpoint, soapContentType, bytes.NewBufferString(message))
 	if err != nil {
 		return resp, errors.Annotate(err, "Post")
 	}
 
-	// if resp.StatusCode is 4xx,5xx, return error
-	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
-		return resp, errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
-	}
-
-	return resp, nil
+	return resp, responseError(resp)
 }
 
+func responseError(resp *http.Response) error {
+	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
+		return errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
+	}
+	return nil
+}
+
+// SendSoapWithDigest sends a soap message and, when the device answers with an
+// HTTP 401 digest challenge, transparently retries the request with the
+// computed HTTP digest Authorization header.
+//
+// Any wsse:Security header present in the message is stripped before sending:
+// when a device requires HTTP digest the credentials travel in the
+// Authorization header, so keeping the WS-Security UsernameToken in the body
+// would put the credentials on the wire twice. Other SOAP header blocks (for
+// example WS-Addressing reference parameters) are preserved so vendor-specific
+// routing keeps working across the retry.
 func SendSoapWithDigest(httpClient *http.Client, endpoint, message, username, password string) (*http.Response, error) {
-	doc := etree.NewDocument()
-	if err := doc.ReadFromString(message); err != nil {
-		return nil, err
+	if httpClient == nil {
+		httpClient = new(http.Client)
 	}
 
-	e := doc.FindElement("./Envelope/Header/Security")
-	if e != nil {
-		bodyTag := doc.Root().SelectElement("Header")
-		bodyTag.RemoveChild(e)
-		data, err := doc.WriteToString()
-		if err != nil {
-			return nil, err
-		}
-		message = data
-	}
+	// Avoid sending the credentials twice (WS-Security + digest) on the retry.
+	message = stripWSSecurityHeader(message)
 
-	req, err := http.NewRequest("POST", endpoint, bytes.NewBufferString(message))
+	resp, err := httpClient.Post(endpoint, soapContentType, bytes.NewBufferString(message))
 	if err != nil {
-		return nil, err
+		return resp, errors.Annotate(err, "Post")
 	}
-	req.Header.Set("Content-Type", "application/soap+xml; charset=utf-8")
-	resp, err := httpClient.Do(req)
+
+	// Only escalate to HTTP digest when the device explicitly asks for it.
+	if resp.StatusCode != http.StatusUnauthorized {
+		return resp, responseError(resp)
+	}
+
+	challenge := resp.Header.Get("WWW-Authenticate")
+	if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(challenge)), "digest") {
+		// Not a digest challenge (e.g. Basic) - nothing more we can do here.
+		return resp, responseError(resp)
+	}
+
+	authorization := newDigestAuthorization(challenge, http.MethodPost, endpoint, username, password)
+	if authorization == "" {
+		return resp, errors.New("unsupported digest challenge")
+	}
+
+	// Release the challenge response before issuing the authenticated retry.
+	resp.Body.Close()
+
+	req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewBufferString(message))
+	if err != nil {
+		return nil, errors.Annotate(err, "new digest request")
+	}
+	req.Header.Set("Content-Type", soapContentType)
+	req.Header.Set("Authorization", authorization)
+
+	resp, err = httpClient.Do(req)
 	if err != nil {
 		return resp, errors.Annotate(err, "Post with digest")
 	}
 
-	if resp.StatusCode != http.StatusUnauthorized {
-		return resp, err
-	}
-
-	wwwAuth := resp.Header.Get("WWW-Authenticate")
-	chal, err := digest.ParseChallenge(wwwAuth)
-	if err != nil {
-		return resp, fmt.Errorf("fail to parse challenge: %w", err)
-	}
-
-	cred, err := digest.Digest(chal, digest.Options{
-		Method:   "POST",
-		URI:      req.URL.RequestURI(),
-		Username: username,
-		Password: password,
-	})
-
-	if err != nil {
-		return resp, fmt.Errorf("fail to build digest: %w", err)
-	}
-
-	// Readout body to close the connection
-	resp.Body.Close()
-
-	req.Header.Add("Authorization", cred.String())
-	req.Body = io.NopCloser((bytes.NewBufferString(message)))
-	resp, err = httpClient.Do(req)
-	if err != nil {
-		return nil, errors.Annotate(err, "Post with digest")
-	}
-	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
-		return resp, errors.Errorf("Post with digest error: %d: %s", resp.StatusCode, resp.Status)
-	}
-
-	return resp, nil
+	return resp, responseError(resp)
+}
+
+// stripWSSecurityHeader removes the wsse:Security header block from a SOAP
+// envelope, leaving all other header blocks intact. The message is returned
+// unchanged if it cannot be parsed as XML or has no such header.
+func stripWSSecurityHeader(message string) string {
+	doc := etree.NewDocument()
+	if err := doc.ReadFromString(message); err != nil {
+		return message
+	}
+	security := doc.FindElement("./Envelope/Header/Security")
+	if security == nil {
+		return message
+	}
+	header := doc.Root().SelectElement("Header")
+	if header == nil {
+		return message
+	}
+	header.RemoveChild(security)
+	data, err := doc.WriteToString()
+	if err != nil {
+		return message
+	}
+	return data
+}
+
+var digestParamRe = regexp.MustCompile(`(\w+)=(?:"([^"]*)"|([^,]+))`)
+
+// parseDigestChallenge parses the parameters of a WWW-Authenticate: Digest header.
+func parseDigestChallenge(challenge string) map[string]string {
+	challenge = strings.TrimSpace(challenge)
+	if i := strings.IndexAny(challenge, " \t"); i >= 0 && strings.EqualFold(challenge[:i], "Digest") {
+		challenge = challenge[i+1:]
+	}
+
+	result := make(map[string]string)
+	for _, m := range digestParamRe.FindAllStringSubmatch(challenge, -1) {
+		value := m[2]
+		if value == "" {
+			value = m[3]
+		}
+		result[strings.ToLower(m[1])] = strings.TrimSpace(value)
+	}
+	return result
+}
+
+// newDigestAuthorization builds an RFC 2617 HTTP digest Authorization header
+// value. It supports the MD5 and MD5-sess algorithms and the "auth" qop, which
+// covers the vast majority of ONVIF devices. It returns an empty string when the
+// challenge is missing required parameters or requires an unsupported qop.
+func newDigestAuthorization(challenge, method, uri, username, password string) string {
+	parts := parseDigestChallenge(challenge)
+	realm := parts["realm"]
+	nonce := parts["nonce"]
+	if realm == "" || nonce == "" {
+		return ""
+	}
+	opaque := parts["opaque"]
+	algorithm := parts["algorithm"]
+
+	qop := ""
+	if rawQop, ok := parts["qop"]; ok {
+		for _, candidate := range strings.Split(rawQop, ",") {
+			if strings.TrimSpace(candidate) == "auth" {
+				qop = "auth"
+				break
+			}
+		}
+		// The server offered qop but none we support (e.g. auth-int only).
+		if qop == "" {
+			return ""
+		}
+	}
+
+	digestURI := uri
+	if u, err := url.Parse(uri); err == nil {
+		digestURI = u.RequestURI()
+	}
+
+	cnonce := randomCnonce()
+	const nc = "00000001"
+
+	ha1 := md5Hex(username + ":" + realm + ":" + password)
+	if strings.EqualFold(algorithm, "MD5-sess") {
+		ha1 = md5Hex(ha1 + ":" + nonce + ":" + cnonce)
+	}
+	ha2 := md5Hex(method + ":" + digestURI)
+
+	var response string
+	if qop == "auth" {
+		response = md5Hex(strings.Join([]string{ha1, nonce, nc, cnonce, qop, ha2}, ":"))
+	} else {
+		response = md5Hex(ha1 + ":" + nonce + ":" + ha2)
+	}
+
+	var b strings.Builder
+	fmt.Fprintf(&b, `Digest username="%s", realm="%s", nonce="%s", uri="%s", response="%s"`,
+		username, realm, nonce, digestURI, response)
+	if qop == "auth" {
+		fmt.Fprintf(&b, `, qop=auth, nc=%s, cnonce="%s"`, nc, cnonce)
+	}
+	if algorithm != "" {
+		fmt.Fprintf(&b, `, algorithm=%s`, algorithm)
+	}
+	if opaque != "" {
+		fmt.Fprintf(&b, `, opaque="%s"`, opaque)
+	}
+	return b.String()
+}
+
+func md5Hex(s string) string {
+	sum := md5.Sum([]byte(s))
+	return hex.EncodeToString(sum[:])
+}
+
+func randomCnonce() string {
+	b := make([]byte, 8)
+	if _, err := rand.Read(b); err != nil {
+		return "00000000"
+	}
+	return hex.EncodeToString(b)
 }

ptz/types.go

@@ -220,7 +220,7 @@ type GetPresetTours struct {
 }
 
 type GetPresetToursResponse struct {
-	PresetTour onvif.PresetTour
+	PresetTour []onvif.PresetTour
 }
 
 type GetPresetTour struct {

ptz/types_test.go

@@ -0,0 +1,25 @@
+package ptz
+
+import (
+	"encoding/xml"
+	"strings"
+	"testing"
+
+	"github.com/kerberos-io/onvif/xsd/onvif"
+)
+
+func TestContinuousMoveIncludesZeroPanTiltCoordinates(t *testing.T) {
+	request := ContinuousMove{
+		Velocity: onvif.PTZSpeedPanTilt{
+			PanTilt: onvif.Vector2D{X: 0.5, Y: 0},
+		},
+	}
+
+	encoded, err := xml.Marshal(request)
+	if err != nil {
+		t.Fatalf("xml.Marshal() error = %v", err)
+	}
+	if !strings.Contains(string(encoded), `x="0.5" y="0"`) {
+		t.Fatalf("ContinuousMove PanTilt = %s, want explicit x and y attributes", encoded)
+	}
+}

sdk/event/PullMessages_auto.go

@@ -0,0 +1,31 @@
+// Code generated : DO NOT EDIT.
+// Copyright (c) 2022 Jean-Francois SMIGIELSKI
+// Distributed under the MIT License
+
+package event
+
+import (
+	"context"
+
+	"github.com/juju/errors"
+	"github.com/kerberos-io/onvif"
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/sdk"
+)
+
+// Call_PullMessages forwards the call to dev.CallMethod() then parses the payload of the reply as a PullMessagesResponse.
+func Call_PullMessages(ctx context.Context, dev *onvif.Device, request event.PullMessages) (event.PullMessagesResponse, error) {
+	type Envelope struct {
+		Header struct{}
+		Body   struct {
+			PullMessagesResponse event.PullMessagesResponse
+		}
+	}
+	var reply Envelope
+	if httpReply, err := dev.CallMethod(request); err != nil {
+		return reply.Body.PullMessagesResponse, errors.Annotate(err, "call")
+	} else {
+		err = sdk.ReadAndParse(ctx, httpReply, &reply, "PullMessages")
+		return reply.Body.PullMessagesResponse, errors.Annotate(err, "reply")
+	}
+}

sdk/event/event.go

@@ -0,0 +1,8 @@
+package event
+
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event CreatePullPointSubscription
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event GetEventProperties
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event GetServiceCapabilities
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event Subscribe
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event Unsubscribe
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event PullMessages

sdk/sdk.go

@@ -0,0 +1,25 @@
+package sdk
+
+import (
+	"context"
+	"encoding/xml"
+	"io"
+	"net/http"
+
+	"github.com/juju/errors"
+)
+
+// ReadAndParse reads the body of the given HTTP reply and unmarshals it into
+// reply. The tag identifies the ONVIF action for diagnostic purposes.
+func ReadAndParse(ctx context.Context, httpReply *http.Response, reply interface{}, tag string) error {
+	// TODO(jfsmig): extract the deadline from ctx.Deadline() and apply it on the reply reading
+	b, err := io.ReadAll(httpReply.Body)
+	if err != nil {
+		return errors.Annotate(err, "read")
+	}
+
+	httpReply.Body.Close()
+
+	err = xml.Unmarshal(b, reply)
+	return errors.Annotate(err, "decode")
+}

xsd/onvif/onvif.go

@@ -666,13 +666,13 @@ type PTZSpeedPanTilt struct {
 }
 
 type Vector2D struct {
-	X     float64     `xml:"x,attr,omitempty"`
-	Y     float64     `xml:"y,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
+	Y     float64     `xml:"y,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
 type Vector1D struct {
-	X     float64     `xml:"x,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
@@ -1176,7 +1176,7 @@ type PresetTour struct {
 	Status            PTZPresetTourStatus            `xml:"Status"`
 	AutoStart         xsd.Boolean                    `xml:"AutoStart"`
 	StartingCondition PTZPresetTourStartingCondition `xml:"StartingCondition"`
-	TourSpot          PTZPresetTourSpot              `xml:"TourSpot"`
+	TourSpot          []PTZPresetTourSpot                    `xml:"TourSpot"`
 	Extension         PTZPresetTourExtension         `xml:"Extension"`
 }