diff --git a/Device.go b/Device.go index c7b0cce..9c89b14 100644 --- a/Device.go +++ b/Device.go @@ -333,31 +333,79 @@ func (dev *Device) GetEndpointByRequestStruct(requestStruct interface{}) (string return resp, err }*/ -// CallMethod functions call an method, defined 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, overrides +// auth, overrides intent, // +// redirect responses, enables replay- +// token forgery, and 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) } diff --git a/Device_test.go b/Device_test.go index f8bfe04..8645791 100644 --- a/Device_test.go +++ b/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
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 = `297` + const bodyXML = `PT5S32` + + 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
; 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
") + assert.Contains(t, headerSlice, "297") + + bodySlice := captured[bodyStart:] + assert.Contains(t, bodySlice, "PullMessages", + "body content must land inside SOAP ") +} + +// 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 = `12` + const bodyXML = `` + + 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, "", "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 = `42` + const bodyXML = `` + + 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, ``) + 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 = `297` + 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, ``, 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, "", "") + 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, "", + WithSOAPHeader(``), + WithSOAPHeader(``), + ) + require.NoError(t, err) + assert.NotContains(t, captured, "First", "first WithSOAPHeader must be overwritten") + assert.Contains(t, captured, "Second") +} diff --git a/event/stream/main_test.go b/event/stream/main_test.go new file mode 100644 index 0000000..d2867aa --- /dev/null +++ b/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) +} diff --git a/event/stream/reconnect.go b/event/stream/reconnect.go index 76bd791..2d0e413 100644 --- a/event/stream/reconnect.go +++ b/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 diff --git a/event/stream/reconnect_test.go b/event/stream/reconnect_test.go index 4dfbb5e..ea322ba 100644 --- a/event/stream/reconnect_test.go +++ b/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 so renew scheduling stays +// driven by opts. const createPullPointRespAlt = ` http://camera.local/onvif/Events/PullSub_2 - 2026-05-21T10:30:10Z - 2026-05-21T10:31:10Z ` diff --git a/event/stream/renew.go b/event/stream/renew.go index cd3f199..3e29edf 100644 --- a/event/stream/renew.go +++ b/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 } diff --git a/event/stream/renew_test.go b/event/stream/renew_test.go index cc758bb..c9a29e6 100644 --- a/event/stream/renew_test.go +++ b/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:Sender + renew-specific complaint + +` + +func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) { + ref := subscriptionRef{ + Address: "http://192.168.1.10/onvif/services", + RefParamsXML: `297`, + } + 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) +} diff --git a/event/stream/soap.go b/event/stream/soap.go index 879b22c..105fef6 100644 --- a/event/stream/soap.go +++ b/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: ...reason... soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)\s]+:)?Text>`) + // SOAP 1.2 Subcode: ...ter:InvalidArgs... + 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 and : 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
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 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), "[REDACTED]") + body = wssePasswordRE.ReplaceAllString(body, "[REDACTED]") + 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. diff --git a/event/stream/soap_test.go b/event/stream/soap_test.go index 7c5b6c2..ae8de38 100644 --- a/event/stream/soap_test.go +++ b/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 := ` + not authorized +` + 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:Sender + Subscription has expired + +` + 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 +// . Without subcode fallback, the only signal +// the operator sees is "400 Bad Request". +func TestEnrichSOAPErr_EmptyReasonFallsBackToSubcode(t *testing.T) { + body := ` + + + SOAP-ENV:Sender + ter:InvalidArgs + + + +` + 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 := `404 Not Found — /onvif/services missing` + 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(`x`), orig) + assert.ErrorIs(t, got, orig) +} + +// --- Subcode extraction ---------------------------------------------- + +func TestExtractSOAPSubcode_Present(t *testing.T) { + body := ` + SOAP-ENV:Sender + ter:InvalidArgs +` + assert.Equal(t, "ter:InvalidArgs", extractSOAPSubcode(body)) +} + +func TestExtractSOAPSubcode_Absent(t *testing.T) { + assert.Empty(t, extractSOAPSubcode(`env:Sender`)) +} + +// --- Wiring: each SOAP call site routes errors through enrichSOAPErr - + +const faultBody = ` + + env:Sender + camera-specific complaint + +` + +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 +// 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 := ` + + + http://192.168.1.10/onvif/services + + 297 + + + +` + 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 := ` + + + http://camera/onvif/Events/Sub_1 + + +` + 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 := ` + + + http://192.168.1.10/onvif/services + + 297 + + + +` + 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 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: `297`, + } + 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 := `` + + `297` + + `` + 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 := `` + + `12` + + `` + 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: `297`, + } + 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: `` + + `1` + + `2` + + ``, + } + 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 := ` + + admin + hunter2 + + plain text excerpt +` + 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:Sender + ter:InvalidArgs + + + +` + 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 := ` + + + http://anon + + DO-NOT-PICK + + + + + + http://camera/sub + + 297 + + + +` + 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 +// 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 := `` + + `297` + + `` + 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 element (including its own attributes), +// not just the inner content, so parent-scoped xmlns survives into +// buildRefParamsHeader. +func TestExtractReferenceParameters_IncludesParentElementForXmlnsPreservation(t *testing.T) { + body := ` + + + http://camera + + 297 + + + +` + 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 := ` + + http://camera/sub + 2026-05-27T13:19:11Z + 2026-05-27T13:21:11Z + +` + 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 := ` + + http://camera/sub + +` + 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(" block starts within the +// 64 KiB error cap but its 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 ` + + `` + + `admin` + + `hunter2` + + // no — 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 := `X` + _, err := buildRefParamsHeader(raw) + require.Error(t, err) + assert.Contains(t, err.Error(), "ReferenceParameters") +} + +// A non-conformant camera echoing UsernameToken/Password outside a +// wrapper would still leak credentials through the body +// excerpt. Belt-and-braces: redact Password elements directly too. +func TestEnrichSOAPErr_RedactsBarePasswordElement(t *testing.T) { + body := `` + + `` + + `admin` + + `hunter2` + + `` + + `` + 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 := `not-a-date` + 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 := ` + + + +` + assert.Empty(t, extractReferenceParameters(body)) +} + +func TestEnrichSOAPErr_RedactsTruncatedPasswordOutsideSecurity(t *testing.T) { + body := `` + + `` + + `admin` + + `hunter2` // truncated — no , no , no + 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 := `` + + `secret1` + + `secret2` + + `` + 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", `X`}, + {"multi-root", ``}, + {"unrelated element", `content`}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + _, err := buildRefParamsHeader(c.raw) + require.Error(t, err) + }) + } +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 9e39f06..9dac55b 100644 --- a/event/stream/stream.go +++ b/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 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, diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go index f5da324..f0c8df7 100644 --- a/event/stream/stream_test.go +++ b/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 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 = ` http://camera.local/onvif/Events/PullSub_1 - 2026-05-21T10:30:00Z - 2026-05-21T10:31:00Z ` diff --git a/examples/event/pullmessage/main.go b/examples/event/pullmessage/main.go index d5d1eb5..d10a7d4 100644 --- a/examples/event/pullmessage/main.go +++ b/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) } diff --git a/examples/event/renew/main.go b/examples/event/renew/main.go index 4d2e5c6..977306d 100644 --- a/examples/event/renew/main.go +++ b/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) } diff --git a/examples/event/unsubscribe/main.go b/examples/event/unsubscribe/main.go index 73d6f7b..09dea97 100644 --- a/examples/event/unsubscribe/main.go +++ b/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) } diff --git a/go.mod b/go.mod index f98d94f..c5eb141 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index b3fac64..5211fc0 100644 --- a/go.sum +++ b/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= diff --git a/gosoap/soap-builder.go b/gosoap/soap-builder.go index 2a31d8e..e1edb03 100644 --- a/gosoap/soap-builder.go +++ b/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("" + data + ""); 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()