kerberos-io/agent
1
0
Fork 0
mirror of https://github.com/kerberos-io/agent.git synced 2026-08-23 15:08:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #273 from sharedjourney/feature/onvif-event-stream

Browse Source 
Feature/onvif event stream
This commit is contained in:
Cédric Verstraeten
2026-07-28 11:25:07 +02:00
committed by GitHub
parent 94b26cf096 b26f0190c6
commit e2e1f8cfa8
6 changed files with 619 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
machinery/go.mod
View File

@@ -32,6 +32,7 @@ require (
github.com/pion/rtp v1.8.19
github.com/pion/webrtc/v4 v4.1.2
github.com/sirupsen/logrus v1.9.3
github.com/stretchr/testify v1.10.0
github.com/swaggo/files v1.0.1
github.com/swaggo/gin-swagger v1.6.0
github.com/swaggo/swag v1.16.4
@@ -58,6 +59,7 @@ require (
github.com/clbanning/mxj v1.8.4 // indirect
github.com/clbanning/mxj/v2 v2.7.0 // indirect
github.com/cloudwego/base64x v0.1.5 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/elastic/go-windows v1.0.2 // indirect
github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect
github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect
@@ -108,6 +110,7 @@ require (
github.com/pion/stun/v3 v3.0.0 // indirect
github.com/pion/transport/v3 v3.0.7 // indirect
github.com/pion/turn/v4 v4.0.0 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/procfs v0.15.1 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.12 // indirect

5
machinery/src/components/kerberos.go
View File

@@ -341,6 +341,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
communication.HandleONVIF = make(chan models.OnvifAction, 10)
go onvif.HandleONVIFActions(configuration, communication)
// Handle ONVIF event stream — opt-in via Capture.ONVIFMotion="true".
// Stops when the agent's shared context is cancelled. The function
// is a no-op if ONVIFMotion is not enabled.
go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication)
communication.HandleAudio = make(chan models.AudioDataPartial, 10)
if rtspBackChannelClient.HasBackChannel {
communication.HasBackChannel = true

3
machinery/src/config/main.go
View File

@@ -377,6 +377,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
case "AGENT_CAPTURE_MOTION":
configuration.Config.Capture.Motion = value
break
case "AGENT_CAPTURE_ONVIF_MOTION":
configuration.Config.Capture.ONVIFMotion = value
break
case "AGENT_CAPTURE_SNAPSHOTS":
configuration.Config.Capture.Snapshots = value
break

8
machinery/src/models/config.go
View File

@@ -75,6 +75,14 @@ type Capture struct {
Fragmented string `json:"fragmented,omitempty" bson:"fragmented,omitempty"`
FragmentedDuration int64 `json:"fragmentedduration,omitempty" bson:"fragmentedduration,omitempty"`
PixelChangeThreshold *int `json:"pixelChangeThreshold,omitempty"`
// ONVIFMotion routes the camera's ONVIF motion events into the
// agent's motion-triggered recording pipeline. When "true" the
// agent opens an event/stream against the configured ONVIF
// endpoint and forwards Motion+Active events to HandleMotion.
// Requires Capture.IPCamera.ONVIFXAddr / ONVIFUsername /
// ONVIFPassword to be set. Default empty (disabled) keeps the
// existing pixel-diff motion detection as the only source.
ONVIFMotion string `json:"onvif_motion,omitempty" bson:"onvif_motion,omitempty"`
}
// IPCamera configuration, such as the RTSP url of the IPCamera and the FPS.

225
machinery/src/onvif/events.go Normal file
View File

@@ -0,0 +1,225 @@
package onvif
import (
"context"
"errors"
"strconv"
"strings"
"time"
"github.com/kerberos-io/agent/machinery/src/log"
"github.com/kerberos-io/agent/machinery/src/models"
"github.com/kerberos-io/onvif/event/stream"
)
// The library handles in-stream reconnect; these guards cover the
// initial-connect path the library cannot see.
const (
initialBackoff = time.Second
maxBackoff = 5 * time.Minute
)
// HandleONVIFEventStream opens an event/stream against the configured
// ONVIF camera and routes Motion events into communication.HandleMotion.
//
// Behind the Capture.ONVIFMotion flag; the goroutine returns
// immediately when not enabled. The flag is read once at start, so
// toggling at runtime requires an agent restart. On transient
// construction failure (camera not yet ready at boot, brief network
// blip, credential reload) the goroutine retries with exponential
// backoff. Exits when ctx is cancelled.
func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) {
log.Log.Debug("onvif.HandleONVIFEventStream(): started")
defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished")
if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) {
return
}
if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" {
log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do")
return
}
backoff := initialBackoff
for {
if ctx.Err() != nil {
return
}
recoverable := runStreamOnce(ctx, configuration, communication)
if !recoverable {
return
}
if !sleepCtx(ctx, backoff) {
return
}
backoff *= 2
if backoff > maxBackoff {
backoff = maxBackoff
}
}
}
// runStreamOnce returns true when the caller should retry construction
// (transient failure), false on clean ctx-driven shutdown.
func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) {
camera := configuration.Config.Capture.IPCamera
device, _, err := ConnectToOnvifDevice(&camera)
if err != nil {
log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error())
return true
}
deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr)
s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID})
if err != nil {
log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error())
return true
}
defer func() {
if err := s.Close(); err != nil {
log.Log.Debug("onvif.HandleONVIFEventStream(): close: " + err.Error())
}
}()
log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID)
// recovering = the first successful event after an error streak
// logs a recovery line so on-call operators see the clear-of-
// condition for the ERROR they were paged on.
var recovering bool
for {
select {
case <-ctx.Done():
return false
case ev, ok := <-s.Events():
if !ok {
return false
}
if recovering {
log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID)
recovering = false
}
dispatchEvent(ctx, ev, configuration, communication)
case e, ok := <-s.Errors():
if !ok {
return false
}
recovering = true
logStreamError(e)
}
}
}
// dispatchEvent routes motion-active events to HandleMotion.
//
// The ctx pre-check + ctx-in-select guards a shutdown race: the agent
// 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=" + topic)
return
}
if ev.State != stream.StateActive {
return
}
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" {
return
}
if ctx.Err() != nil {
return
}
dataToPass := models.MotionDataPartial{
Timestamp: time.Now().Unix(),
NumberOfChanges: 0, // ONVIF does not quantify motion area.
}
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.
func logStreamError(e error) {
var recreate stream.ErrRecreateFailed
var pull stream.ErrPullFailed
var renew stream.ErrRenewFailed
switch {
case errors.As(e, &recreate):
log.Log.Error("onvif.HandleONVIFEventStream(): subscription recreate failed (camera may be offline): " + recreate.Err.Error())
case errors.As(e, &renew):
log.Log.Debug("onvif.HandleONVIFEventStream(): renew failed (will recover via pull/recreate): " + renew.Err.Error())
case errors.As(e, &pull):
log.Log.Debug("onvif.HandleONVIFEventStream(): pull failed (will retry): " + pull.Err.Error())
default:
log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error())
}
}
func isONVIFMotionEnabled(v string) bool {
return strings.EqualFold(strings.TrimSpace(v), "true")
}
// resolveDeviceID falls back from operator-supplied name to ONVIF
// endpoint to a constant placeholder so log lines always have
// something to grep.
func resolveDeviceID(configName, xaddr string) string {
if n := strings.TrimSpace(configName); n != "" {
return n
}
if x := strings.TrimSpace(xaddr); x != "" {
return x
}
return "unknown"
}
// sleepCtx returns false if ctx was cancelled, true if d elapsed.
func sleepCtx(ctx context.Context, d time.Duration) bool {
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return false
case <-t.C:
return true
}
}

375
machinery/src/onvif/events_test.go Normal file
View File

@@ -0,0 +1,375 @@
package onvif
import (
"bytes"
"context"
"strings"
"testing"
"time"
"github.com/kerberos-io/agent/machinery/src/models"
"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 {
return &models.Configuration{
Name: name,
Config: models.Config{
Capture: models.Capture{
Recording: recording,
ONVIFMotion: onvifMotion,
},
},
}
}
func makeCommunication(buffer int) *models.Communication {
return &models.Communication{
HandleMotion: make(chan models.MotionDataPartial, buffer),
}
}
// --- dispatchEvent ---------------------------------------------------
func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) {
cfg := makeConfig("true", "true", "cam-1")
comm := makeCommunication(1)
ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
select {
case m := <-comm.HandleMotion:
assert.NotZero(t, m.Timestamp)
case <-time.After(time.Second):
t.Fatal("expected motion data on HandleMotion")
}
}
func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) {
cfg := makeConfig("true", "true", "cam-1")
comm := makeCommunication(1)
ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
select {
case <-comm.HandleMotion:
t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)")
case <-time.After(100 * time.Millisecond):
}
}
func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) {
cfg := makeConfig("true", "true", "cam-1")
comm := makeCommunication(1)
ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
select {
case <-comm.HandleMotion:
t.Fatal("non-motion kinds must not reach HandleMotion")
case <-time.After(100 * time.Millisecond):
}
}
// captureDebugLog redirects logrus to a buffer at debug level for the
// duration of a test and returns what was written. It mutates package
// globals, so callers must not run in parallel.
func captureDebugLog(t *testing.T) *bytes.Buffer {
t.Helper()
var buf bytes.Buffer
prevOut, prevLevel := logrus.StandardLogger().Out, logrus.GetLevel()
logrus.SetOutput(&buf)
logrus.SetLevel(logrus.DebugLevel)
t.Cleanup(func() {
logrus.SetOutput(prevOut)
logrus.SetLevel(prevLevel)
})
return &buf
}
// TestDispatchEvent_LogsTheTriggeringTopic — a dispatched event is what
// actually starts a recording, so its topic is the one an operator needs
// when a camera records for the wrong reason (or the right reason and
// nobody can prove which). Rejected events were already logged; without
// this the triggering topic is only knowable by elimination.
func TestDispatchEvent_LogsTheTriggeringTopic(t *testing.T) {
buf := captureDebugLog(t)
cfg := makeConfig("true", "true", "cam-1")
comm := makeCommunication(1)
ev := stream.Event{
Kind: stream.KindMotion,
State: stream.StateActive,
Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
assert.Contains(t, buf.String(), "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
"the dispatched event's topic must appear in the log")
assert.Contains(t, buf.String(), "Motion",
"the dispatched event's Kind must appear in the log")
}
// TestDispatchEvent_PropertyOperation — a camera replays the current
// state of every property topic as Initialized whenever a pull-point
// subscription is created. If that counts as a trigger, every
// reconnect restarts a recording for any motion property that happens
// to be active, and a flapping subscription manufactures motion out of
// nothing. Only reject Initialized specifically: PropertyOperation is
// optional per WS-Notification and absent on many non-property events,
// which decode reports as PropertyUnknown.
func TestDispatchEvent_PropertyOperation(t *testing.T) {
tests := []struct {
name string
op stream.PropertyOperation
wantSend bool
}{
{"changed is a real transition", stream.PropertyChanged, true},
{"absent attribute still counts", stream.PropertyUnknown, true},
{"initialized is a subscription state replay", stream.PropertyInitialized, false},
}
for _, tt := range tests {
t.Run(tt.name, 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,
Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
select {
case <-comm.HandleMotion:
if !tt.wantSend {
t.Fatalf("%v must not trigger a recording", tt.op)
}
case <-time.After(100 * time.Millisecond):
if tt.wantSend {
t.Fatalf("%v must trigger a recording", tt.op)
}
}
})
}
}
func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) {
cfg := makeConfig("false", "true", "cam-1")
comm := makeCommunication(1)
ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dispatchEvent(ctx, ev, cfg, comm)
select {
case <-comm.HandleMotion:
t.Fatal("Recording=false must gate the send (matches computervision behaviour)")
case <-time.After(100 * time.Millisecond):
}
}
func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) {
cfg := makeConfig("true", "true", "cam-1")
// Pre-fill the buffer so the next send would block.
comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
comm.HandleMotion <- models.MotionDataPartial{}
ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan struct{})
go func() {
dispatchEvent(ctx, ev, cfg, comm)
close(done)
}()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("dispatchEvent must drop when HandleMotion is full, not block")
}
}
func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) {
// Regression for the shutdown race: between cancel() and
// close(HandleMotion) the agent leaves a 3s window. If dispatchEvent
// runs in that window AFTER the channel is closed, a non-protected
// send would panic. The ctx pre-check must short-circuit before the
// send is attempted.
cfg := makeConfig("true", "true", "cam-1")
comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
close(comm.HandleMotion)
ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
ctx, cancel := context.WithCancel(context.Background())
cancel() // already cancelled, matching the shutdown sequence
assert.NotPanics(t, func() {
dispatchEvent(ctx, ev, cfg, comm)
})
}
// --- isONVIFMotionEnabled --------------------------------------------
func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) {
tests := []struct {
in string
want bool
}{
{"true", true},
{"True", true},
{"TRUE", true},
{" true", true},
{"true ", true},
{" true ", true},
{"false", false},
{"False", false},
{"", false},
{"yes", false},
{"1", false},
}
for _, tc := range tests {
t.Run(tc.in, func(t *testing.T) {
assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in))
})
}
}
// --- resolveDeviceID -------------------------------------------------
func TestResolveDeviceID_FallbackChain(t *testing.T) {
tests := []struct {
name string
cfgName string
xaddr string
want string
}{
{"name_set", "front-door", "192.168.1.10", "front-door"},
{"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"},
{"name_whitespace_only_xaddr_set", " ", "192.168.1.10", "192.168.1.10"},
{"both_empty", "", "", "unknown"},
{"name_with_trailing_whitespace", "cam-2 ", "192.168.1.10", "cam-2"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr))
})
}
}
// 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")
}