diff --git a/machinery/src/onvif/events.go b/machinery/src/onvif/events.go index aff71d1..005694e 100644 --- a/machinery/src/onvif/events.go +++ b/machinery/src/onvif/events.go @@ -3,6 +3,7 @@ package onvif import ( "context" "errors" + "strconv" "strings" "time" @@ -116,18 +117,16 @@ func runStreamOnce(ctx context.Context, configuration *models.Configuration, com // closes HandleMotion shortly after cancelling ctx, and a stale event // reaching the send would otherwise panic on a closed channel. func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) { + topic := sanitiseTopic(ev.Topic) if ev.Kind != stream.KindMotion { - log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + ev.Topic) + log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + topic) return } if ev.State != stream.StateActive { return } - // A camera replays every property topic's current state as - // Initialized on each new subscription, so treating that as a - // trigger lets a flapping pull-point manufacture motion. - if ev.Operation == stream.PropertyInitialized { - log.Log.Debug("onvif.dispatchEvent(): subscription state replay, not a trigger: topic=" + ev.Topic) + if !isTransition(ev.Operation) { + log.Log.Debug("onvif.dispatchEvent(): " + ev.Operation.String() + " is not a transition, not a trigger: topic=" + topic) return } if configuration.Config.Capture.Recording == "false" { @@ -136,9 +135,6 @@ func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.C if ctx.Err() != nil { return } - // The topic that actually started a recording is the one on-call - // needs; the reject path below already names the ones that didn't. - log.Log.Debug("onvif.dispatchEvent(): recording trigger " + ev.Kind.String() + " topic=" + ev.Topic) dataToPass := models.MotionDataPartial{ Timestamp: time.Now().Unix(), @@ -147,11 +143,39 @@ func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.C select { case <-ctx.Done(): case communication.HandleMotion <- dataToPass: + // Logged on the send, not before it: this line records that a + // recording started, so a dropped event must not leave one. + log.Log.Debug("onvif.dispatchEvent(): recording trigger " + ev.Kind.String() + " topic=" + topic) default: log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event") } } +// isTransition reports whether an operation represents a state change. +// A camera replays every property's current state as Initialized on +// each new subscription and announces removals as Deleted; neither is +// motion starting. Absent (Unknown) counts — PropertyOperation is +// optional per WS-Notification and many non-property events omit it. +func isTransition(op stream.PropertyOperation) bool { + return op == stream.PropertyChanged || op == stream.PropertyUnknown +} + +// maxLoggedTopic bounds a topic in the log; the wire imposes no limit, +// and the reject path logs every event received. +const maxLoggedTopic = 256 + +// sanitiseTopic makes a camera-controlled topic safe to concatenate +// into a log line. logrus's coloured text formatter writes the message +// unquoted, so a raw newline would let a camera forge entries in the +// log being used to diagnose it. +func sanitiseTopic(topic string) string { + if len(topic) > maxLoggedTopic { + topic = topic[:maxLoggedTopic] + "…(truncated)" + } + quoted := strconv.Quote(topic) + return quoted[1 : len(quoted)-1] +} + // logStreamError logs at a level matching severity: recreate is loud // because it usually means the camera is offline; pull and renew are // debug because the library recovers from them automatically. diff --git a/machinery/src/onvif/events_test.go b/machinery/src/onvif/events_test.go index 597cc5b..2dba9b6 100644 --- a/machinery/src/onvif/events_test.go +++ b/machinery/src/onvif/events_test.go @@ -3,6 +3,7 @@ package onvif import ( "bytes" "context" + "strings" "testing" "time" @@ -10,6 +11,7 @@ import ( "github.com/kerberos-io/onvif/event/stream" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func makeConfig(recording, onvifMotion, name string) *models.Configuration { @@ -276,3 +278,98 @@ func TestResolveDeviceID_FallbackChain(t *testing.T) { }) } } + +// TestDispatchEvent_OnlyRealTransitionsTrigger — a camera replays every +// property topic's state on each new subscription (Initialized) and +// announces removals (Deleted). Neither is a motion transition, and a +// flapping pull-point would otherwise manufacture recordings out of +// replayed state. PropertyOperation is optional per WS-Notification, so +// absent (Unknown) still counts — many non-property events omit it. +func TestDispatchEvent_OnlyRealTransitionsTrigger(t *testing.T) { + tests := []struct { + op stream.PropertyOperation + wantSend bool + }{ + {stream.PropertyChanged, true}, + {stream.PropertyUnknown, true}, + {stream.PropertyInitialized, false}, + {stream.PropertyDeleted, false}, + } + + for _, tt := range tests { + t.Run(tt.op.String(), func(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Operation: tt.op} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + if tt.wantSend { + require.Len(t, comm.HandleMotion, 1, "%v must trigger a recording", tt.op) + return + } + require.Empty(t, comm.HandleMotion, "%v must not trigger a recording", tt.op) + }) + } +} + +// TestSanitiseTopic — ev.Topic is camera-controlled and reaches the log +// unmodified. logrus's coloured text formatter (the default) writes the +// message without quoting, so an embedded newline forges whole log +// lines: a compromised camera can fabricate ERROR entries or spoof +// another device's id, in the logs an operator is reading to diagnose +// that very camera. Length is also unbounded on the wire, and the +// reject path logs every event, so an oversized topic is a cheap way to +// evict a container's whole retained history. +func TestSanitiseTopic(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {"ordinary topic passes through", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1"}, + {"newline cannot forge a line", "a\nERRO[fake] boom", `a\nERRO[fake] boom`}, + {"carriage return", "a\rb", `a\rb`}, + {"tab", "a\tb", `a\tb`}, + {"NUL", "a\x00b", `a\x00b`}, + {"empty", "", ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := sanitiseTopic(tt.in) + assert.Equal(t, tt.want, got) + assert.NotContains(t, got, "\n", "no raw newline may survive") + assert.NotContains(t, got, "\r", "no raw carriage return may survive") + }) + } +} + +func TestSanitiseTopic_Truncates(t *testing.T) { + got := sanitiseTopic(strings.Repeat("x", maxLoggedTopic*2)) + assert.LessOrEqual(t, len(got), maxLoggedTopic+len("…(truncated)")) + assert.Contains(t, got, "truncated") +} + +// TestDispatchEvent_LogsTriggerOnlyWhenSent — the trigger line is the +// record that a recording started. Logging it before the send means a +// dropped event (full channel, or shutdown) leaves a line claiming a +// recording that never began. +func TestDispatchEvent_LogsTriggerOnlyWhenSent(t *testing.T) { + buf := captureDebugLog(t) + + cfg := makeConfig("true", "true", "cam-1") + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + comm.HandleMotion <- models.MotionDataPartial{} // full + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Topic: "tns1:VideoSource/MotionAlarm"} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + assert.NotContains(t, buf.String(), "recording trigger", + "a dropped event must not be logged as a trigger") + assert.Contains(t, buf.String(), "dropping", "the drop itself must still be logged") +}