Merge pull request #6 from sharedjourney/feat/event-stream-axis-compat

fix(event/stream): make `event/stream` work against AXIS cameras

Device.go

@@ -333,31 +333,79 @@ 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
 }
 
+
 func createHttpRequest(httpMethod string, endpoint string, soap string) (req *http.Request, err error) {
 	req, err = http.NewRequest(httpMethod, endpoint, bytes.NewBufferString(soap))
 	if err != nil {
@@ -388,7 +436,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/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/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

@@ -110,6 +110,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 +141,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 +154,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 +165,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 +182,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
@@ -183,7 +239,7 @@ func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, e
 
 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 +247,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>`

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()

ptz/types.go

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

xsd/onvif/onvif.go

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