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 }