cleanup comments + add ouputs

README.md

@@ -28,8 +28,8 @@ Kerberos Agent is an isolated and scalable video (surveillance) management agent
 
 ## :thinking: Prerequisites
 
-- An IP camera which supports a RTSP H264 encoded stream,
-  - (or) a USB camera, Raspberry Pi camera or other camera, that [you can transform to a valid RTSP H264 stream](https://github.com/kerberos-io/camera-to-rtsp).
+- An IP camera which supports a RTSP H264 or H265 encoded stream,
+  - (or) a USB camera, Raspberry Pi camera or other camera, that [you can transform to a valid RTSP H264 or H265 stream](https://github.com/kerberos-io/camera-to-rtsp).
 - Any hardware (ARMv6, ARMv7, ARM64, AMD) that can run a binary or container, for example: a Raspberry Pi, NVidia Jetson, Intel NUC, a VM, Bare metal machine or a full blown Kubernetes cluster.
 
 ## :video_camera: Is my camera working?
@@ -104,19 +104,21 @@ This repository contains everything you'll need to know about our core product,
 
 - Low memory and CPU usage.
 - Simplified and modern user interface.
-- Multi architecture (ARMv7, ARMv8, amd64, etc).
-- Multi camera support: IP Cameras (H264), USB cameras and Raspberry Pi Cameras [through a RTSP proxy](https://github.com/kerberos-io/camera-to-rtsp).
+- Multi architecture (ARMv7, ARMv8, amd64, etc).).
+- Multi stream, for example recording in H265, live streaming and motion detection in H264.
+- Multi camera support: IP Cameras (H264 and H265), USB cameras and Raspberry Pi Cameras [through a RTSP proxy](https://github.com/kerberos-io/camera-to-rtsp
 - Single camera per instance (e.g. one container per camera).
-- Primary and secondary stream setup (record full-res, stream low-res).
-- Low resolution streaming through MQTT and full resolution streaming through WebRTC.
-- End-to-end encryption through MQTT using RSA and AES.
-- Ability to specifiy conditions: offline mode, motion region, time table, continuous recording, etc.
-- Post- and pre-recording on motion detection.
+- Low resolution streaming through MQTT and high resolution streaming through WebRTC (only supports H264/PCM).
+- Backchannel audio from Kerberos Hub to IP camera (requires PCM ULAW codec)
+- Audio (AAC) and video (H264/H265) recording in MP4 container.
+- End-to-end encryption through MQTT using RSA and AES (livestreaming, ONVIF, remote configuration, etc)
+- Conditional recording: offline mode, motion region, time table, continuous recording, webhook condition etc.
+- Post- and pre-recording for motion detection.
 - Encryption at rest using AES-256-CBC.
-- Ability to create fragmented recordings, and streaming though HLS fMP4.
+- Ability to create fragmented recordings, and streaming through HLS fMP4.
 - [Deploy where you want](#how-to-run-and-deploy-a-kerberos-agent) with the tools you use: `docker`, `docker compose`, `ansible`, `terraform`, `kubernetes`, etc.
 - Cloud storage/persistance: Kerberos Hub, Kerberos Vault and Dropbox. [(WIP: Minio, Storj, Google Drive, FTP etc.)](https://github.com/kerberos-io/agent/issues/95)
-- WIP: Integrations (Webhooks, MQTT, Script, etc).
+- Outputs: trigger an integration (Webhooks, MQTT, Script, etc) when a specific event (motion detection or start recording ) occurs
 - REST API access and documentation through Swagger (trigger recording, update configuration, etc).
 - MIT License
 
@@ -192,6 +194,7 @@ Next to attaching the configuration file, it is also possible to override the co
 
 | Name                                    | Description                                                                                     | Default Value                  |
 | --------------------------------------- | ----------------------------------------------------------------------------------------------- | ------------------------------ |
+| `LOG_LEVEL`                             | Level for logging, could be "info", "warning", "debug", "error" or "fatal".                     | "info"                         |
 | `AGENT_MODE`                            | You can choose to run this in 'release' for production, and or 'demo' for showcasing.           | "release"                      |
 | `AGENT_TLS_INSECURE`                    | Specify if you want to use `InsecureSkipVerify` for the internal HTTP client.                   | "false"                        |
 | `AGENT_USERNAME`                        | The username used to authenticate against the Kerberos Agent login page.                        | "root"                         |

machinery/docs/docs.go

@@ -103,7 +103,7 @@ const docTemplate = `{
                         "in": "body",
                         "required": true,
                         "schema": {
-                            "$ref": "#/definitions/models.IPCamera"
+                            "$ref": "#/definitions/models.OnvifPreset"
                         }
                     }
                 ],

machinery/docs/swagger.json

@@ -95,7 +95,7 @@
                         "in": "body",
                         "required": true,
                         "schema": {
-                            "$ref": "#/definitions/models.IPCamera"
+                            "$ref": "#/definitions/models.OnvifPreset"
                         }
                     }
                 ],

machinery/docs/swagger.yaml

@@ -379,7 +379,7 @@ paths:
         name: cameraConfig
         required: true
         schema:
-          $ref: '#/definitions/models.IPCamera'
+          $ref: '#/definitions/models.OnvifPreset'
       responses:
         "200":
           description: OK

machinery/main.go

@@ -70,32 +70,38 @@ func main() {
 	flag.Parse()
 
 	timezone, _ := time.LoadLocation("CET")
-	log.Log.Init(configDirectory, timezone)
+
+	// Specify the level of loggin: "info", "warning", "debug", "error" or "fatal."
+	logLevel := os.Getenv("LOG_LEVEL")
+	if logLevel == "" {
+		logLevel = "info"
+	}
+	log.Log.Init(logLevel, configDirectory, timezone)
 
 	switch action {
 
 	case "version":
-		log.Log.Info("Main(): You are currrently running Kerberos Agent " + VERSION)
+		log.Log.Info("main.Main(): You are currrently running Kerberos Agent " + VERSION)
 
 	case "discover":
 		// Convert duration to int
 		timeout, err := time.ParseDuration(timeout + "ms")
 		if err != nil {
-			log.Log.Fatal("Main(): could not parse timeout: " + err.Error())
+			log.Log.Fatal("main.Main(): could not parse timeout: " + err.Error())
 			return
 		}
 		onvif.Discover(timeout)
 
 	case "decrypt":
-		log.Log.Info("Main(): Decrypting: " + flag.Arg(0) + " with key: " + flag.Arg(1))
+		log.Log.Info("main.Main(): Decrypting: " + flag.Arg(0) + " with key: " + flag.Arg(1))
 		symmetricKey := []byte(flag.Arg(1))
 
 		if symmetricKey == nil || len(symmetricKey) == 0 {
-			log.Log.Fatal("Main(): symmetric key should not be empty")
+			log.Log.Fatal("main.Main(): symmetric key should not be empty")
 			return
 		}
 		if len(symmetricKey) != 32 {
-			log.Log.Fatal("Main(): symmetric key should be 32 bytes")
+			log.Log.Fatal("main.Main(): symmetric key should be 32 bytes")
 			return
 		}
 
@@ -131,7 +137,7 @@ func main() {
 
 			// Set timezone
 			timezone, _ := time.LoadLocation(configuration.Config.Timezone)
-			log.Log.Init(configDirectory, timezone)
+			log.Log.Init(logLevel, configDirectory, timezone)
 
 			// Check if we have a device Key or not, if not
 			// we will generate one.
@@ -140,9 +146,9 @@ func main() {
 				configuration.Config.Key = key
 				err := configService.StoreConfig(configDirectory, configuration.Config)
 				if err == nil {
-					log.Log.Info("Main(): updated unique key for agent to: " + key)
+					log.Log.Info("main.Main(): updated unique key for agent to: " + key)
 				} else {
-					log.Log.Info("Main(): something went wrong while trying to store key: " + key)
+					log.Log.Info("main.Main(): something went wrong while trying to store key: " + key)
 				}
 			}
 
@@ -150,7 +156,8 @@ func main() {
 			// This is used to restart the agent when the configuration is updated.
 			ctx, cancel := context.WithCancel(context.Background())
 
-			// We create a capture object.
+			// We create a capture object, this will contain all the streaming clients.
+			// And allow us to extract media from within difference places in the agent.
 			capture := capture.Capture{
 				RTSPClient:    nil,
 				RTSPSubClient: nil,
@@ -169,6 +176,6 @@ func main() {
 			routers.StartWebserver(configDirectory, &configuration, &communication, &capture)
 		}
 	default:
-		log.Log.Error("Main(): Sorry I don't understand :(")
+		log.Log.Error("main.Main(): Sorry I don't understand :(")
 	}
 }

machinery/src/encryption/main.go

@@ -14,27 +14,15 @@ import (
 	"hash"
 )
 
-// DecryptWithPrivateKey decrypts data with private key
 func DecryptWithPrivateKey(ciphertext string, privateKey *rsa.PrivateKey) ([]byte, error) {
-
-	// decode our encrypted string into cipher bytes
 	cipheredValue, _ := base64.StdEncoding.DecodeString(ciphertext)
-
-	// decrypt the data
 	out, err := rsa.DecryptPKCS1v15(nil, privateKey, cipheredValue)
-
 	return out, err
 }
 
-// SignWithPrivateKey signs data with private key
 func SignWithPrivateKey(data []byte, privateKey *rsa.PrivateKey) ([]byte, error) {
-
-	// hash the data with sha256
 	hashed := sha256.Sum256(data)
-
-	// sign the data
 	signature, err := rsa.SignPKCS1v15(nil, privateKey, crypto.SHA256, hashed[:])
-
 	return signature, err
 }
 
@@ -59,16 +47,10 @@ func AesEncrypt(content []byte, password string) ([]byte, error) {
 	copy(cipherText[:8], []byte("Salted__"))
 	copy(cipherText[8:16], salt)
 	copy(cipherText[16:], cipherBytes)
-
-	//cipherText := base64.StdEncoding.EncodeToString(data)
 	return cipherText, nil
 }
 
 func AesDecrypt(cipherText []byte, password string) ([]byte, error) {
-	//data, err := base64.StdEncoding.DecodeString(cipherText)
-	//if err != nil {
-	//	return nil, err
-	//}
 	if string(cipherText[:8]) != "Salted__" {
 		return nil, errors.New("invalid crypto js aes encryption")
 	}
@@ -92,8 +74,6 @@ func AesDecrypt(cipherText []byte, password string) ([]byte, error) {
 	return result, nil
 }
 
-// https://stackoverflow.com/questions/27677236/encryption-in-javascript-and-decryption-with-php/27678978#27678978
-// https://github.com/brix/crypto-js/blob/8e6d15bf2e26d6ff0af5277df2604ca12b60a718/src/evpkdf.js#L55
 func EvpKDF(password []byte, salt []byte, keySize int, iterations int, hashAlgorithm string) ([]byte, error) {
 	var block []byte
 	var hasher hash.Hash
@@ -124,7 +104,6 @@ func EvpKDF(password []byte, salt []byte, keySize int, iterations int, hashAlgor
 }
 
 func DefaultEvpKDF(password []byte, salt []byte) (key []byte, iv []byte, err error) {
-	// https://github.com/brix/crypto-js/blob/8e6d15bf2e26d6ff0af5277df2604ca12b60a718/src/cipher-core.js#L775
 	keySize := 256 / 32
 	ivSize := 128 / 32
 	derivedKeyBytes, err := EvpKDF(password, salt, keySize+ivSize, 1, "md5")
@@ -134,7 +113,6 @@ func DefaultEvpKDF(password []byte, salt []byte) (key []byte, iv []byte, err err
 	return derivedKeyBytes[:keySize*4], derivedKeyBytes[keySize*4:], nil
 }
 
-// https://stackoverflow.com/questions/41579325/golang-how-do-i-decrypt-with-des-cbc-and-pkcs7
 func PKCS5UnPadding(src []byte) []byte {
 	length := len(src)
 	unpadding := int(src[length-1])

machinery/src/log/main.go

@@ -12,7 +12,6 @@ import (
 // The logging library being used everywhere.
 var Log = Logging{
 	Logger: "logrus",
-	Level:  "debug",
 }
 
 // -----------------
@@ -45,7 +44,7 @@ func ConfigureGoLogging(configDirectory string, timezone *time.Location) {
 // This a logrus
 // -> github.com/sirupsen/logrus
 
-func ConfigureLogrus(timezone *time.Location) {
+func ConfigureLogrus(level string, timezone *time.Location) {
 	// Log as JSON instead of the default ASCII formatter.
 	logrus.SetFormatter(LocalTimeZoneFormatter{
 		Timezone:  timezone,
@@ -57,7 +56,17 @@ func ConfigureLogrus(timezone *time.Location) {
 	logrus.SetOutput(os.Stdout)
 
 	// Only log the warning severity or above.
-	logrus.SetLevel(logrus.InfoLevel)
+	logLevel := logrus.InfoLevel
+	if level == "error" {
+		logLevel = logrus.ErrorLevel
+	} else if level == "debug" {
+		logLevel = logrus.DebugLevel
+	} else if level == "fatal" {
+		logLevel = logrus.FatalLevel
+	} else if level == "warning" {
+		logLevel = logrus.WarnLevel
+	}
+	logrus.SetLevel(logLevel)
 }
 
 type LocalTimeZoneFormatter struct {
@@ -72,15 +81,14 @@ func (u LocalTimeZoneFormatter) Format(e *logrus.Entry) ([]byte, error) {
 
 type Logging struct {
 	Logger string
-	Level  string
 }
 
-func (self *Logging) Init(configDirectory string, timezone *time.Location) {
+func (self *Logging) Init(level string, configDirectory string, timezone *time.Location) {
 	switch self.Logger {
 	case "go-logging":
 		ConfigureGoLogging(configDirectory, timezone)
 	case "logrus":
-		ConfigureLogrus(timezone)
+		ConfigureLogrus(level, timezone)
 	default:
 	}
 }

machinery/src/models/output.go

@@ -0,0 +1,15 @@
+package models
+
+import "time"
+
+// The OutputMessage contains the relevant information
+// to specify the type of triggers we want to execute.
+type OutputMessage struct {
+	Name      string
+	Outputs   []string
+	Trigger   string
+	Timestamp time.Time
+	File      string
+	CameraId  string
+	SiteId    string
+}

machinery/src/onvif/main.go

@@ -14,9 +14,9 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
-	"github.com/kerberos-io/onvif/media"
-
 	"github.com/kerberos-io/onvif"
+	"github.com/kerberos-io/onvif/device"
+	"github.com/kerberos-io/onvif/media"
 	"github.com/kerberos-io/onvif/ptz"
 	xsd "github.com/kerberos-io/onvif/xsd/onvif"
 )
@@ -718,9 +718,9 @@ func ContinuousZoom(device *onvif.Device, configuration ptz.GetConfigurationsRes
 	return err
 }
 
-func GetCapabilitiesFromDevice(device *onvif.Device) []string {
+func GetCapabilitiesFromDevice(dev *onvif.Device) []string {
 	var capabilities []string
-	services := device.GetServices()
+	services := dev.GetServices()
 	for key, _ := range services {
 		log.Log.Debug("GetCapabilitiesFromDevice: has key: " + key)
 		if key != "" {
@@ -731,6 +731,34 @@ func GetCapabilitiesFromDevice(device *onvif.Device) []string {
 			}
 		}
 	}
+
+	var getCapabilitiesResponse device.GetCapabilitiesResponse
+	resp, err := dev.CallMethod(device.GetCapabilities{
+		Category: "All",
+	})
+
+	var b []byte
+	if resp != nil {
+		b, err = io.ReadAll(resp.Body)
+		resp.Body.Close()
+	}
+	if err == nil {
+		stringBody := string(b)
+		decodedXML, et, err := getXMLNode(stringBody, "GetCapabilitiesResponse")
+		if err != nil {
+			log.Log.Error("GetCapabilitiesResponse: " + err.Error())
+			//return err
+		} else {
+			if err := decodedXML.DecodeElement(&getCapabilitiesResponse, et); err != nil {
+				log.Log.Error("GetCapabilitiesResponse: " + err.Error())
+				//	return err
+			}
+			//return err
+		}
+	} else {
+		log.Log.Error("GoToPresetFromDevice: " + err.Error())
+	}
+
 	return capabilities
 }
 
@@ -909,12 +937,12 @@ func VerifyOnvifConnection(c *gin.Context) {
 	}
 }
 
-func GetDigitalInputs(device *onvif.Device) (ptz.GetConfigurationsResponse, error) {
-	// We'll try to receive the PTZ configurations from the server
-	var configurations ptz.GetConfigurationsResponse
+func GetRelayOutputs(dev *onvif.Device) (device.GetRelayOutputsResponse, error) {
+	// We'll try to receive the relay outputs from the server
+	var relayoutputs device.GetRelayOutputsResponse
 
 	// Get the PTZ configurations from the device
-	resp, err := device.CallMethod(ptz.GetConfigurations{})
+	resp, err := dev.CallMethod(device.GetRelayOutputs{})
 	var b []byte
 	if resp != nil {
 		b, err = io.ReadAll(resp.Body)
@@ -924,19 +952,19 @@ func GetDigitalInputs(device *onvif.Device) (ptz.GetConfigurationsResponse, erro
 	if err == nil {
 		if err == nil {
 			stringBody := string(b)
-			decodedXML, et, err := getXMLNode(stringBody, "GetConfigurationsResponse")
+			decodedXML, et, err := getXMLNode(stringBody, "GetRelayOutputsResponse")
 			if err != nil {
-				log.Log.Debug("onvif.GetPTZConfigurationsFromDevice(): " + err.Error())
-				return configurations, err
+				log.Log.Error("onvif.main.GetRelayOutputs(): " + err.Error())
+				return relayoutputs, err
 			} else {
-				if err := decodedXML.DecodeElement(&configurations, et); err != nil {
-					log.Log.Debug("onvif.GetPTZConfigurationsFromDevice(): " + err.Error())
-					return configurations, err
+				if err := decodedXML.DecodeElement(&relayoutputs, et); err != nil {
+					log.Log.Debug("onvif.main.GetRelayOutputs(): " + err.Error())
+					return relayoutputs, err
 				}
 			}
 		}
 	}
-	return configurations, err
+	return relayoutputs, err
 }
 
 func getXMLNode(xmlBody string, nodeName string) (*xml.Decoder, *xml.StartElement, error) {

machinery/src/outputs/main.go

@@ -0,0 +1,59 @@
+package outputs
+
+import (
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+type Output interface {
+	// Triggers the integration
+	Trigger(message models.OutputMessage) error
+}
+
+func Execute(message *models.OutputMessage) (err error) {
+	err = nil
+
+	outputs := message.Outputs
+	for _, output := range outputs {
+		switch output {
+		case "slack":
+			slack := &SlackOutput{}
+			err := slack.Trigger(message)
+			if err == nil {
+				log.Log.Debug("outputs.main.Execute(slack): message was processed by output.")
+			} else {
+				log.Log.Error("outputs.main.Execute(slack): " + err.Error())
+			}
+			break
+		case "webhook":
+			webhook := &WebhookOutput{}
+			err := webhook.Trigger(message)
+			if err == nil {
+				log.Log.Debug("outputs.main.Execute(webhook): message was processed by output.")
+			} else {
+				log.Log.Error("outputs.main.Execute(webhook): " + err.Error())
+			}
+			break
+		case "onvif_relay":
+			onvif := &OnvifRelayOutput{}
+			err := onvif.Trigger(message)
+			if err == nil {
+				log.Log.Debug("outputs.main.Execute(onvif): message was processed by output.")
+			} else {
+				log.Log.Error("outputs.main.Execute(onvif): " + err.Error())
+			}
+			break
+		case "script":
+			script := &ScriptOutput{}
+			err := script.Trigger(message)
+			if err == nil {
+				log.Log.Debug("outputs.main.Execute(script): message was processed by output.")
+			} else {
+				log.Log.Error("outputs.main.Execute(script): " + err.Error())
+			}
+			break
+		}
+	}
+
+	return err
+}

machinery/src/outputs/onvif_relay.go

@@ -0,0 +1,12 @@
+package outputs
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+type OnvifRelayOutput struct {
+	Output
+}
+
+func (o *OnvifRelayOutput) Trigger(message *models.OutputMessage) (err error) {
+	err = nil
+	return err
+}

machinery/src/outputs/script.go

@@ -0,0 +1,12 @@
+package outputs
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+type ScriptOutput struct {
+	Output
+}
+
+func (scr *ScriptOutput) Trigger(message *models.OutputMessage) (err error) {
+	err = nil
+	return err
+}

machinery/src/outputs/slack.go

@@ -0,0 +1,12 @@
+package outputs
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+type SlackOutput struct {
+	Output
+}
+
+func (s *SlackOutput) Trigger(message *models.OutputMessage) (err error) {
+	err = nil
+	return err
+}

machinery/src/outputs/webhook.go

@@ -0,0 +1,12 @@
+package outputs
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+type WebhookOutput struct {
+	Output
+}
+
+func (w *WebhookOutput) Trigger(message *models.OutputMessage) (err error) {
+	err = nil
+	return err
+}

machinery/src/packets/packet.go

@@ -8,10 +8,7 @@ import (
 
 // Packet represents an RTP Packet
 type Packet struct {
-	// for Gortsplib library
-	Packet *rtp.Packet
-
-	// for JOY4 and Gortsplib library
+	Packet          *rtp.Packet
 	IsAudio         bool          // packet is audio
 	IsVideo         bool          // packet is video
 	IsKeyFrame      bool          // video packet is key frame

machinery/src/routers/http/methods.go

@@ -361,7 +361,7 @@ func GoToOnvifPreset(c *gin.Context) {
 // @in header
 // @name Authorization
 // @Tags camera
-// @Param cameraConfig body models.IPCamera true "Camera Config"
+// @Param cameraConfig body models.OnvifPreset true "Camera Config"
 // @Summary Will get the digital inputs from the ONVIF device.
 // @Description Will get the digital inputs from the ONVIF device.
 // @Success 200 {object} models.APIResponse
@@ -386,10 +386,10 @@ func DoGetDigitalInputs(c *gin.Context) {
 		cameraConfiguration := configuration.Config.Capture.IPCamera
 		device, err := onvif.ConnectToOnvifDevice(&cameraConfiguration)
 		if err == nil {
-			inputs, err := onvif.GetDigitalInputs(device)
+			outputs, err := onvif.GetRelayOutputs(device)
 			if err == nil {
 				c.JSON(200, gin.H{
-					"data": inputs,
+					"data": outputs,
 				})
 			} else {
 				c.JSON(400, gin.H{