From d726ed8edb311ba58018a2992ad404436c154ecc Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Wed, 27 May 2026 18:07:56 +0200 Subject: [PATCH] fix(event/stream): address review findings on AXIS compat work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Critical - Device.SendSoapWithHeader now parses the supplied header content with etree and adds each top-level child as its own SOAP Header block. gosoap.AddStringHeaderContent only accepts a single root element; previously a multi-child ref-params header silently produced a header-less request because the parse error was discarded. Errors are now propagated. - enrichSOAPErr scrubs <*:Security> blocks from response bodies before fault extraction or excerpt slicing so a camera that echoes the WS-Security header in a fault response cannot leak Username/Password into operator logs. Important - extractSOAPFault falls back to the SOAP 1.2 Subcode (e.g. ter:InvalidArgs) when Reason/Text is empty — consistent with enrichSOAPErr and surfaces actionable detail on 200-OK fault bodies reached via unmarshalNode. - subscriptionRef captures the camera-granted TerminationTime from CreatePullPointSubscription and Renew responses. renewLoop schedules from it via the new nextRenewInterval helper so we never miss a renew when the camera grants less than requested. renew is now a sleep-loop driven by the latest granted time. - enrichSOAPErr reads at most 64 KiB from the body (vs. 10 MiB on success paths). Fault bodies are always small; the prior cap let a wedged camera churn 10 MiB/s through the retry loop. Suggestions - extractReferenceParameters anchors to so a wsa:ReplyTo / wsa:FaultTo that also carries ReferenceParameters elsewhere in the envelope cannot leak through and break PullMessages. - buildRefParamsHeader accepts either raw children or the full <*:ReferenceParameters> wrapper, and propagates ancestor xmlns:* onto each child so a vendor that declares the prefix on the parent (not the child itself, as AXIS does) still produces valid standalone children on the wire. - SendSoapWithHeader documents that xmlHeaderContent must be well-formed XML and that the caller is responsible for escaping any externally sourced data. - Error-message ordering is now context-first ("SOAP fault: X: ") per Go convention. - Dead headerEnd slicing removed from the SendSoapWithHeader test. Tests - End-to-end multi-child wiring through pullMessages. - Digest auth retry preserves the injected header. - Malformed-XML header content fast-fails before any request. - buildRefParamsHeader malformed / whitespace-only edge cases. - goleak.VerifyTestMain in event/stream catches any pull/renew goroutine that outlives its Stream. No new behavioural surface added to onvif core; SendSoap retains its signature, SendSoapWithHeader is the only new public method. --- Device.go | 42 ++++++- Device_test.go | 89 +++++++++++++- event/stream/main_test.go | 14 +++ event/stream/reconnect_test.go | 2 - event/stream/renew.go | 74 +++++++----- event/stream/renew_test.go | 5 +- event/stream/soap.go | 131 ++++++++++++++++----- event/stream/soap_test.go | 209 +++++++++++++++++++++++++++++++++ event/stream/stream.go | 15 ++- event/stream/stream_test.go | 2 - go.mod | 1 + go.sum | 2 + 12 files changed, 514 insertions(+), 72 deletions(-) create mode 100644 event/stream/main_test.go diff --git a/Device.go b/Device.go index a9749be..a4d2a1c 100644 --- a/Device.go +++ b/Device.go @@ -333,7 +333,7 @@ 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.SendSoapWithHeader(endpoint, xmlRequestBody, "") } @@ -342,13 +342,20 @@ func (dev Device) SendSoap(endpoint string, xmlRequestBody string) (*http.Respon // 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 zero or more +// SOAP Header child elements (sibling top-level elements are +// supported; the spec lets each reference parameter be its own header +// block). The caller is responsible for escaping any externally +// sourced data inside it. Malformed XML returns an error before any +// request is made. func (dev Device) SendSoapWithHeader(endpoint, xmlRequestBody, xmlHeaderContent string) (*http.Response, error) { soap := gosoap.NewEmptySOAP() soap.AddStringBodyContent(xmlRequestBody) soap.AddRootNamespaces(Xlmns) soap.AddAction() - if xmlHeaderContent != "" { - _ = soap.AddStringHeaderContent(xmlHeaderContent) + if err := addHeaderChildren(&soap, xmlHeaderContent); err != nil { + return nil, err } if dev.params.Username != "" && dev.params.Password != "" { soap.AddWSSecurity(dev.params.Username, dev.params.Password) @@ -364,6 +371,35 @@ func (dev Device) SendSoapWithHeader(endpoint, xmlRequestBody, xmlHeaderContent return servResp, err } +// addHeaderChildren wraps the fragment so etree can parse multi-root +// XML, then adds each top-level child as its own SOAP Header block. +// gosoap.AddStringHeaderContent only accepts a single root element. +func addHeaderChildren(soap *gosoap.SoapMessage, xmlHeaderContent string) error { + if xmlHeaderContent == "" { + return nil + } + doc := etree.NewDocument() + if err := doc.ReadFromString("" + xmlHeaderContent + ""); err != nil { + return fmt.Errorf("parse header content: %w", err) + } + wrap := doc.SelectElement("wrap") + if wrap == nil { + return errors.New("parse header content: missing wrap root") + } + for _, child := range wrap.ChildElements() { + d := etree.NewDocument() + d.SetRoot(child.Copy()) + s, err := d.WriteToString() + if err != nil { + return fmt.Errorf("serialise header child: %w", err) + } + if err := soap.AddStringHeaderContent(s); err != nil { + return fmt.Errorf("add header child: %w", err) + } + } + return nil +} + 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 { diff --git a/Device_test.go b/Device_test.go index 8577e85..b6c2d5a 100644 --- a/Device_test.go +++ b/Device_test.go @@ -58,16 +58,97 @@ func TestDevice_SendSoapWithHeader_InjectsHeaderXML(t *testing.T) { } headerStart := strings.Index(captured, "Header>") - headerEnd := strings.Index(captured, "") require.NotEqual(t, -1, headerStart, "envelope must contain
; got: %s", captured) - assert.Greater(t, headerEnd, headerStart, "envelope must close the header") + require.Greater(t, bodyStart, headerStart, "Body must follow Header in the envelope") - headerSlice := captured[headerStart:strings.Index(captured, "Body>")] + headerSlice := captured[headerStart:bodyStart] assert.Contains(t, headerSlice, "SubscriptionId", "injected header element must land inside SOAP
") assert.Contains(t, headerSlice, "297") - bodySlice := captured[strings.Index(captured, "Body>"):] + bodySlice := captured[bodyStart:] assert.Contains(t, bodySlice, "PullMessages", "body content must land inside SOAP ") } + +// Per WS-Addressing 1.0 §3.1 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") +} + +// 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, "", " 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 ef0280d..e8cee2f 100644 --- a/event/stream/renew.go +++ b/event/stream/renew.go @@ -10,53 +10,71 @@ import ( "github.com/kerberos-io/onvif/xsd" ) -// renewLoop surfaces renew failures and continues. 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. +// renewLoop sleeps until the next deadline (camera-granted termination +// minus RenewMargin), renews, and repeats. 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 := s.getPullPoint() + if !sleepCtx(ctx, nextRenewInterval(ref.GrantedTermination, s.opts, time.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}) + continue + } + if !granted.IsZero() { + s.updateGrantedTermination(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 +} + // 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, ref subscriptionRef, 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) } headerXML, err := buildRefParamsHeader(ref.RefParamsXML) if err != nil { - return fmt.Errorf("build ref params header: %w", err) + return time.Time{}, fmt.Errorf("build ref params header: %w", err) } resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML) if err != nil { - return enrichSOAPErr(resp, err) + return time.Time{}, enrichSOAPErr(resp, err) } - _, err = readClose(resp) - return 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 f9c0a12..55acf55 100644 --- a/event/stream/renew_test.go +++ b/event/stream/renew_test.go @@ -175,7 +175,7 @@ func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) { 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()) + _, 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") @@ -194,7 +194,8 @@ func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) { RefParamsXML: `297`, } fc := newFakeCaller() - require.NoError(t, renewPullPoint(fc, ref, defaultOptions())) + _, err := renewPullPoint(fc, ref, defaultOptions()) + require.NoError(t, err) require.Len(t, fc.sendSoapHeaders, 1) hdr := fc.sendSoapHeaders[0] assert.Contains(t, hdr, "SubscriptionId") diff --git a/event/stream/soap.go b/event/stream/soap.go index 17ab885..489882c 100644 --- a/event/stream/soap.go +++ b/event/stream/soap.go @@ -17,12 +17,18 @@ import ( "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 +// 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} @@ -50,9 +56,30 @@ func createPullPoint(c caller, opts Options) (subscriptionRef, error) { if addr == "" { return subscriptionRef{}, errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address") } - return subscriptionRef{Address: addr, RefParamsXML: extractReferenceParameters(body)}, 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 +} + +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, ref subscriptionRef, opts Options) ([]event.NotificationMessage, error) { @@ -159,31 +186,47 @@ var ( 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. + wsseSecurityRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Security\b[^>]*>.*?\s]+:)?Security>`) ) -// 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 { - return strings.TrimSpace(m[1]) + if r := strings.TrimSpace(m[1]); r != "" { + return r + } } - return "" + return extractSOAPSubcode(body) } -var refParamsRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?ReferenceParameters\b[^>]*>(.*?)\s]+:)?ReferenceParameters>`) +// 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 for a set -// of WS-Addressing ReferenceParameters: each top-level child element -// is re-emitted with wsa:IsReferenceParameter="true" added, as the -// spec requires. Empty input yields empty output (no-op for vendors -// that encode subscription identity in the URL). +// of WS-Addressing ReferenceParameters: each ref-param element is +// re-emitted with wsa:IsReferenceParameter="true" and any xmlns:* +// it inherited from the parent element. Input +// may be either the raw children or the full <*:ReferenceParameters> +// wrapper — extractReferenceParameters returns the wrapper so parent- +// scoped namespace declarations survive into the rebuild. func buildRefParamsHeader(rawXML string) (string, error) { if strings.TrimSpace(rawXML) == "" { return "", nil @@ -196,11 +239,23 @@ func buildRefParamsHeader(rawXML string) (string, error) { if wrap == nil { return "", errors.New("parse ref params: missing wrap root") } + + children := wrap.ChildElements() + var ambient *etree.Element + if len(children) == 1 && strings.HasSuffix(children[0].Tag, "ReferenceParameters") { + ambient = children[0] + children = ambient.ChildElements() + } + var out strings.Builder - for _, child := range wrap.ChildElements() { - child.CreateAttr("wsa:IsReferenceParameter", "true") + for _, child := range children { + c := child.Copy() + if ambient != nil { + inheritXmlns(c, ambient) + } + c.CreateAttr("wsa:IsReferenceParameter", "true") d := etree.NewDocument() - d.SetRoot(child.Copy()) + d.SetRoot(c) s, err := d.WriteToString() if err != nil { return "", fmt.Errorf("serialise ref param child: %w", err) @@ -210,16 +265,37 @@ func buildRefParamsHeader(rawXML string) (string, error) { 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 §3.1. // Without that echo, AXIS rejects PullMessages with ter:InvalidArgs. func extractReferenceParameters(body string) string { - m := refParamsRE.FindStringSubmatch(body) - if len(m) < 2 { + sub := subscriptionRefRE.FindStringSubmatch(body) + if len(sub) < 2 { return "" } - return strings.TrimSpace(m[1]) + return strings.TrimSpace(refParamsRE.FindString(sub[1])) } // extractSOAPSubcode is the fallback when Reason/Text is empty — AXIS @@ -251,22 +327,19 @@ func enrichSOAPErr(resp *http.Response, err error) error { return err } defer resp.Body.Close() - b, readErr := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + b, readErr := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) if readErr != nil || len(b) == 0 { return err } - body := string(b) + body := wsseSecurityRE.ReplaceAllString(string(b), "[REDACTED]") if reason := extractSOAPFault(body); reason != "" { - return fmt.Errorf("%w: SOAP fault: %s", err, reason) - } - if sub := extractSOAPSubcode(body); sub != "" { - return fmt.Errorf("%w: SOAP fault subcode: %s", err, sub) + return fmt.Errorf("SOAP fault: %s: %w", reason, err) } excerpt := strings.TrimSpace(body) if len(excerpt) > maxErrExcerpt { excerpt = excerpt[:maxErrExcerpt] + "...(truncated)" } - return fmt.Errorf("%w: response body: %s", err, excerpt) + return fmt.Errorf("response body: %s: %w", excerpt, err) } // durationToXSD formats a duration as xsd:duration PTnS. Second diff --git a/event/stream/soap_test.go b/event/stream/soap_test.go index 9e2b954..20edd4d 100644 --- a/event/stream/soap_test.go +++ b/event/stream/soap_test.go @@ -7,6 +7,7 @@ import ( "net/http" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -392,3 +393,211 @@ func TestUnsubscribePullPoint_EmptyAddressIsNoOp(t *testing.T) { 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 §3.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(" http://camera.local/onvif/Events/PullSub_1 - 2026-05-21T10:30:00Z - 2026-05-21T10:31:00Z ` diff --git a/go.mod b/go.mod index f98d94f..5554aaa 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( github.com/pmezard/go-difflib v1.0.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.11 // indirect + go.uber.org/goleak v1.3.0 // indirect golang.org/x/arch v0.3.0 // indirect golang.org/x/crypto v0.16.0 // indirect golang.org/x/sys v0.15.0 // indirect 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=