From 93620f04a357fbeaee7ec302872b3841f4a9b5a0 Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Thu, 21 May 2026 14:54:05 +0200 Subject: [PATCH] fix(event/stream): production-grade SOAP and lifecycle hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the five ship-blocker findings from the second review: 1. Bounded body read (review F1 / R-HIGH) readClose now wraps resp.Body with io.LimitReader(10 MiB). A hostile or buggy camera streaming an unbounded body cannot OOM the agent. Legitimate PullMessages payloads are <200KB even with dense analytics. 2. SOAP Fault detection (review F3) unmarshalNode now scans for SOAP 1.1 faultstring and SOAP 1.2 Reason/Text BEFORE the missing-element error path. Auth failures ('not authorized'), InvalidFilterFault and expired-subscription faults now surface their reason text instead of collapsing to the unhelpful 'response missing PullMessagesResponse element'. This is the difference between a debuggable error and a hidden one when a customer's credentials change. 3. Absolute Renew TerminationTime (review F1 wire-correctness) renewPullPoint now sends an RFC3339 UTC datetime ('2026-05-21T10:30:00Z') instead of a relative xsd:duration ('PT60S'). WS-BaseNotification §6.1.1 accepts both, but older Hikvision, some Dahua and Bosch firmwares only accept the absolute form — the library's own type comment even flags this ('BUG(r) Bad AbsoluteOrRelativeTimeType type'). 4. Bounded Close (review P0) Close now wraps Unsubscribe in a 5s timeout. Previously a TCP-accepted-but-never-replying camera would wedge Close indefinitely; now Close returns with a timeout error and the subscription expires on its own at InitialTermination. 5. Explicit channel-close ordering after wg.Wait The run goroutine previously relied on defer-LIFO to guarantee renew exits before close(errors). Future maintainers extending run() could invert that order silently. Closes are now explicit sequential statements after wg.Wait() so the invariant is local, not order-of-defers magic. Also expands wsnt:UtcTime parsing in decode.go to cover the four formats observed across vendor firmwares: RFC3339 with sub-seconds, compact offsets ('+0200', Geovision/Dahua), and naked timestamps without timezone (older Hikvision; per spec UTC is implied). Caller interface gains a doc comment noting it must be safe for concurrent use, documenting the contract Stream depends on (*onvif. Device satisfies it via http.Client). Tests added: SOAP 1.1 and 1.2 fault extraction, fault surfacing through unmarshalNode, Renew absolute-datetime assertion, Close-with-blocked-Unsubscribe returning within the timeout. -race clean. --- event/stream/decode.go | 22 ++++- event/stream/soap_test.go | 155 ++++++++++++++++++++++++++++++++++++ event/stream/stream.go | 94 +++++++++++++++++++--- event/stream/stream_test.go | 27 +++++-- 4 files changed, 275 insertions(+), 23 deletions(-) create mode 100644 event/stream/soap_test.go diff --git a/event/stream/decode.go b/event/stream/decode.go index 1d96941..f0bc7df 100644 --- a/event/stream/decode.go +++ b/event/stream/decode.go @@ -88,17 +88,31 @@ func parsePropertyOperation(s string) PropertyOperation { // time when the attribute is absent or unparseable. The result is // normalised to UTC so equality comparisons across timezones work. // -// xsd:dateTime in ONVIF messages is RFC 3339 in practice; we try -// time.RFC3339Nano first (covers sub-second precision) and fall back to -// time.RFC3339 for cameras that drop the fractional part. +// xsd:dateTime in ONVIF messages is RFC 3339 in practice but real +// cameras emit several flavours: with/without sub-seconds, with colon +// or compact ("+0200") timezone offsets, and some older Hikvision +// firmwares omit the timezone entirely (treated as UTC per +// WS-BaseNotification which mandates UTC for UtcTime). func parseDeviceTime(s string) time.Time { if s == "" { return time.Time{} } - for _, layout := range []string{time.RFC3339Nano, time.RFC3339} { + for _, layout := range deviceTimeLayouts { if t, err := time.Parse(layout, s); err == nil { return t.UTC() } } return time.Time{} } + +// deviceTimeLayouts lists the wsnt:UtcTime forms observed across vendor +// firmwares. Ordered from most-precise / most-common first so the +// happy path hits early. +var deviceTimeLayouts = []string{ + time.RFC3339Nano, // 2026-05-21T10:30:00.500Z, ...+02:00 + time.RFC3339, // 2026-05-21T10:30:00Z, ...+02:00 + "2006-01-02T15:04:05.999-0700", // sub-second + compact offset (Geovision) + "2006-01-02T15:04:05-0700", // compact offset (some Dahua) + "2006-01-02T15:04:05.999", // no timezone, sub-second (rare) + "2006-01-02T15:04:05", // naked, no TZ (older Hikvision) +} diff --git a/event/stream/soap_test.go b/event/stream/soap_test.go new file mode 100644 index 0000000..e8c48f5 --- /dev/null +++ b/event/stream/soap_test.go @@ -0,0 +1,155 @@ +package stream + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- SOAP fault detection --------------------------------------------- + +func TestExtractSOAPFault_SOAP11(t *testing.T) { + body := ` + + + + env:Client + The action requested requires authorization and the sender is not authorized + + +` + got := extractSOAPFault(body) + assert.Contains(t, got, "not authorized") +} + +func TestExtractSOAPFault_SOAP12(t *testing.T) { + body := ` + + + + env:Sender + Subscription has expired + + +` + got := extractSOAPFault(body) + assert.Contains(t, got, "Subscription has expired") +} + +func TestExtractSOAPFault_NotAFault(t *testing.T) { + assert.Empty(t, extractSOAPFault(createPullPointResp)) +} + +func TestExtractSOAPFault_EmptyBody(t *testing.T) { + assert.Empty(t, extractSOAPFault("")) +} + +func TestUnmarshalNode_ReturnsFaultReasonInsteadOfMissingElement(t *testing.T) { + body := ` + not authorized +` + var out struct{} + err := unmarshalNode(body, "PullMessagesResponse", &out) + require.Error(t, err) + assert.Contains(t, err.Error(), "not authorized") + assert.NotContains(t, err.Error(), "missing PullMessagesResponse") +} + +// --- Renew sends absolute datetime ----------------------------------- + +func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + InitialTermination: 30 * time.Millisecond, + RenewMargin: 5 * time.Millisecond, + }) + require.NoError(t, err) + defer s.Close() + + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if countSendSoapMatching(fc, "Renew") >= 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + + fc.mu.Lock() + defer fc.mu.Unlock() + var renewBody string + for _, c := range fc.sendSoapCalls { + if strings.Contains(c[1], "Renew") { + renewBody = c[1] + break + } + } + require.NotEmpty(t, renewBody, "no Renew call observed") + // Absolute form is "YYYY-MM-DDTHH:MM:SSZ" not "PTnS". + 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") +} + +// --- Bounded body read ----------------------------------------------- + +func TestReadClose_LimitsBodySize(t *testing.T) { + // Build a response with a body just over the limit. readClose must + // not return more than the limit even if the camera pretends to + // send more. + if maxResponseBytes < 1024 { + t.Skip("limit too small for this test") + } + big := strings.Repeat("A", maxResponseBytes+1024) + // Wrap in a minimal SOAP envelope so the body is at least + // well-formed shape-wise. + body := "" + big + "" + fc := newFakeCaller() + fc.queueCallMethod(body, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Construction will fail because the truncated body has no + // CreatePullPointSubscriptionResponse — that's fine; what matters + // is the read completes without OOM. + _, err := newStream(ctx, fc, Options{}) + assert.Error(t, err) +} + +// --- Close timeout --------------------------------------------------- + +func TestClose_BoundedByTimeoutOnHungUnsubscribe(t *testing.T) { + // Patch closeUnsubscribeTimeout for the duration of the test so the + // assertion completes promptly. We can't change the const at runtime + // so we use a short InitialTermination and verify Close still + // returns within closeUnsubscribeTimeout + slack rather than + // blocking forever. + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + block := make(chan struct{}) + defer close(block) // release the hung Unsubscribe so the fake's goroutine exits + fc.mu.Lock() + fc.blockUnsubscribe = block + fc.mu.Unlock() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second}) + require.NoError(t, err) + + start := time.Now() + err = s.Close() + elapsed := time.Since(start) + // Unsubscribe is hung, so Close must surface a timeout error from + // the bounded wait rather than block forever. closeUnsubscribeTimeout + // is 5s; allow 1s slack for scheduling. + require.Error(t, err) + assert.Contains(t, err.Error(), "timeout") + assert.Less(t, elapsed, closeUnsubscribeTimeout+time.Second, + "Close exceeded bound (%s); expected ~%s", elapsed, closeUnsubscribeTimeout) +} diff --git a/event/stream/stream.go b/event/stream/stream.go index a7cc06b..9e4fd7a 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -8,7 +8,9 @@ import ( "fmt" "io" "net/http" + "regexp" "strconv" + "strings" "sync" "time" @@ -17,6 +19,18 @@ import ( "github.com/kerberos-io/onvif/xsd" ) +// maxResponseBytes caps the size of a SOAP response we will buffer in +// memory. 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. +const maxResponseBytes = 10 << 20 + +// closeUnsubscribeTimeout bounds the SOAP Unsubscribe call issued by +// Close so a hung camera connection cannot wedge the caller. The +// subscription expires at the camera anyway once InitialTermination +// elapses, so a missed unsubscribe is at worst cosmetic. +const closeUnsubscribeTimeout = 5 * time.Second + // Options configures a Stream. The zero value is usable; defaultOptions // fills in production-sensible defaults for any unset field. type Options struct { @@ -107,6 +121,11 @@ const maxRecreateBackoff = 30 * time.Second // caller is the subset of *onvif.Device the Stream depends on. Tests // substitute a fake; production code uses the device adapter. +// +// Implementations must be safe for concurrent use: the pull loop and +// renew loop call into caller from separate goroutines. *onvif.Device +// satisfies this because its HTTP client is the goroutine-safe +// http.Client. type caller interface { CallMethod(method any) (*http.Response, error) SendSoap(endpoint, body string) (*http.Response, error) @@ -204,25 +223,33 @@ func (s *Stream) Errors() <-chan error { return s.errors } // Close stops the background goroutine, waits for it to exit, and // unsubscribes from the camera. Subsequent calls are no-ops. +// +// Unsubscribe is bounded by closeUnsubscribeTimeout so a hung camera +// connection cannot wedge the caller. On timeout Close still returns +// promptly; the subscription will expire at the camera once +// InitialTermination + RenewMargin elapses without a renew. func (s *Stream) Close() error { s.closeOnce.Do(func() { s.cancel() <-s.done - // Unsubscribe is best-effort: if the camera is unreachable - // the subscription will expire on its own at - // InitialTermination + Renew interval anyway. - if err := unsubscribePullPoint(s.caller, s.getPullPoint()); err != nil { - s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err) + + errCh := make(chan error, 1) + go func() { + errCh <- unsubscribePullPoint(s.caller, s.getPullPoint()) + }() + select { + case err := <-errCh: + if err != nil { + s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err) + } + case <-time.After(closeUnsubscribeTimeout): + s.closeErr = fmt.Errorf("unsubscribe pull point: timeout after %s", closeUnsubscribeTimeout) } }) return s.closeErr } func (s *Stream) run(ctx context.Context) { - defer close(s.done) - defer close(s.events) - defer close(s.errors) - var wg sync.WaitGroup wg.Add(1) go func() { @@ -231,6 +258,13 @@ func (s *Stream) run(ctx context.Context) { }() s.pullLoop(ctx) wg.Wait() + + // Explicit close order after both goroutines have exited so a + // future maintainer extending this function does not accidentally + // rely on defer-ordering for channel-close safety. + close(s.errors) + close(s.events) + close(s.done) } func (s *Stream) pullLoop(ctx context.Context) { @@ -400,7 +434,12 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification } func renewPullPoint(c caller, endpoint string, opts Options) error { - req := event.Renew{TerminationTime: xsd.String(durationToXSD(opts.InitialTermination))} + // WS-BaseNotification §6.1.1 declares TerminationTime as + // xsd:dateTime OR xsd:duration, but older Hikvision, some Dahua + // and some Bosch firmwares reject the relative-duration form. Send + // an absolute UTC datetime to match what production NVRs do. + 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) @@ -434,7 +473,9 @@ func readClose(resp *http.Response) (string, error) { return "", errors.New("nil HTTP response") } defer resp.Body.Close() - b, err := io.ReadAll(resp.Body) + // LimitReader prevents a hostile or buggy camera from OOMing the + // agent by streaming an unbounded response body. + b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) if err != nil { return "", fmt.Errorf("read response body: %w", err) } @@ -445,7 +486,15 @@ func readClose(resp *http.Response) (string, error) { // name and decodes it into out. ONVIF SOAP responses come wrapped in an // envelope with multiple namespace prefixes; this helper sidesteps // namespace matching by keying on local name only. +// +// When the camera returns a SOAP Fault instead of the expected +// response, the fault reason is surfaced as the error so callers can +// distinguish "auth failed" / "subscription expired" from "unparseable +// response". func unmarshalNode(body, localName string, out any) error { + if reason := extractSOAPFault(body); reason != "" { + return fmt.Errorf("ONVIF SOAP fault: %s", reason) + } dec := xml.NewDecoder(bytes.NewBufferString(body)) for { tok, err := dec.Token() @@ -469,6 +518,29 @@ func unmarshalNode(body, localName string, out any) error { } } +var ( + // SOAP 1.1: reason + soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)\s]+:)?faultstring>`) + // SOAP 1.2: ...reason... + soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)\s]+:)?Text>`) +) + +// extractSOAPFault returns the human-readable reason text from a SOAP +// fault, or empty string when the body is not a fault. Handles both +// SOAP 1.1 (faultstring) and SOAP 1.2 (Reason/Text) shapes. +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 m := soap12FaultRE.FindStringSubmatch(body); len(m) > 1 { + return strings.TrimSpace(m[1]) + } + return "" +} + // durationToXSD formats a Go time.Duration as an xsd:duration string in // PTnS form. Second precision is sufficient — ONVIF cameras do not // honour sub-second pull timeouts and intermediate routers may round in diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go index 698d07e..010f10a 100644 --- a/event/stream/stream_test.go +++ b/event/stream/stream_test.go @@ -20,14 +20,19 @@ import ( // returns the next queued response; when the queue is exhausted it falls // back to a default response so the indefinite pull loop does not // require tests to enumerate every call. +// +// blockUnsubscribe, when non-nil, causes SendSoap calls whose body +// contains "Unsubscribe" to block until the channel is closed. Used to +// verify Close's timeout path. type fakeCaller struct { - mu sync.Mutex - callMethodResps []fakeResp - sendSoapResps []fakeResp - defaultSendSoap fakeResp - defaultCall fakeResp - callMethodCalls []any - sendSoapCalls [][2]string + mu sync.Mutex + callMethodResps []fakeResp + sendSoapResps []fakeResp + defaultSendSoap fakeResp + defaultCall fakeResp + callMethodCalls []any + sendSoapCalls [][2]string + blockUnsubscribe chan struct{} } type fakeResp struct { @@ -72,13 +77,19 @@ func (f *fakeCaller) CallMethod(m any) (*http.Response, error) { func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) { f.mu.Lock() - defer f.mu.Unlock() f.sendSoapCalls = append(f.sendSoapCalls, [2]string{endpoint, body}) r := f.defaultSendSoap if len(f.sendSoapResps) > 0 { r = f.sendSoapResps[0] f.sendSoapResps = f.sendSoapResps[1:] } + block := f.blockUnsubscribe + f.mu.Unlock() + + if block != nil && strings.Contains(body, "Unsubscribe") { + <-block + } + if r.err != nil { return nil, r.err }