diff --git a/README.md b/README.md index b20ac8b..019adf9 100644 --- a/README.md +++ b/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" | diff --git a/machinery/docs/docs.go b/machinery/docs/docs.go index 4da1670..cbbb6e2 100644 --- a/machinery/docs/docs.go +++ b/machinery/docs/docs.go @@ -103,7 +103,7 @@ const docTemplate = `{ "in": "body", "required": true, "schema": { - "$ref": "#/definitions/models.IPCamera" + "$ref": "#/definitions/models.OnvifPreset" } } ], diff --git a/machinery/docs/swagger.json b/machinery/docs/swagger.json index cc96c14..c8327de 100644 --- a/machinery/docs/swagger.json +++ b/machinery/docs/swagger.json @@ -95,7 +95,7 @@ "in": "body", "required": true, "schema": { - "$ref": "#/definitions/models.IPCamera" + "$ref": "#/definitions/models.OnvifPreset" } } ], diff --git a/machinery/docs/swagger.yaml b/machinery/docs/swagger.yaml index 6033d7c..8dffd0c 100644 --- a/machinery/docs/swagger.yaml +++ b/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 diff --git a/machinery/main.go b/machinery/main.go index d450533..4339c99 100644 --- a/machinery/main.go +++ b/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 :(") } } diff --git a/machinery/src/encryption/main.go b/machinery/src/encryption/main.go index 3d4e8a1..9c1f995 100644 --- a/machinery/src/encryption/main.go +++ b/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]) diff --git a/machinery/src/log/main.go b/machinery/src/log/main.go index 11c562e..fd3f274 100644 --- a/machinery/src/log/main.go +++ b/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: } } diff --git a/machinery/src/models/output.go b/machinery/src/models/output.go new file mode 100644 index 0000000..a8a24ae --- /dev/null +++ b/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 +} diff --git a/machinery/src/onvif/main.go b/machinery/src/onvif/main.go index c790a38..68eda3f 100644 --- a/machinery/src/onvif/main.go +++ b/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) { diff --git a/machinery/src/outputs/main.go b/machinery/src/outputs/main.go new file mode 100644 index 0000000..a025423 --- /dev/null +++ b/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 +} diff --git a/machinery/src/outputs/onvif_relay.go b/machinery/src/outputs/onvif_relay.go new file mode 100644 index 0000000..63aceb3 --- /dev/null +++ b/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 +} diff --git a/machinery/src/outputs/script.go b/machinery/src/outputs/script.go new file mode 100644 index 0000000..e4f4c3d --- /dev/null +++ b/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 +} diff --git a/machinery/src/outputs/slack.go b/machinery/src/outputs/slack.go new file mode 100644 index 0000000..135daba --- /dev/null +++ b/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 +} diff --git a/machinery/src/outputs/webhook.go b/machinery/src/outputs/webhook.go new file mode 100644 index 0000000..04c8a0e --- /dev/null +++ b/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 +} diff --git a/machinery/src/packets/packet.go b/machinery/src/packets/packet.go index 62390ad..d385bbc 100644 --- a/machinery/src/packets/packet.go +++ b/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 diff --git a/machinery/src/routers/http/Methods.go b/machinery/src/routers/http/methods.go similarity index 98% rename from machinery/src/routers/http/Methods.go rename to machinery/src/routers/http/methods.go index f0faf81..f438390 100644 --- a/machinery/src/routers/http/Methods.go +++ b/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{ diff --git a/machinery/src/routers/http/Routes.go b/machinery/src/routers/http/routes.go similarity index 100% rename from machinery/src/routers/http/Routes.go rename to machinery/src/routers/http/routes.go