Merge branch 'master' of https://github.com/kerberos-io/agent

doc(README): remove typo

Dockerfile

@@ -10,7 +10,7 @@ ENV GOSUMDB=off
 ##########################################
 # Installing some additional dependencies.
 
-RUN apt-get upgrade -y && apt-get update && apt-get install -y --no-install-recommends \
+RUN apt-get upgrade -y && apt-get update && apt-get install -y --fix-missing --no-install-recommends \
 	git build-essential cmake pkg-config unzip libgtk2.0-dev \
 	curl ca-certificates libcurl4-openssl-dev libssl-dev libjpeg62-turbo-dev && \
 	rm -rf /var/lib/apt/lists/*

README.md

@@ -29,7 +29,7 @@ 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 tranform to a valid RTSP H264 stream](https://github.com/kerberos-io/camera-to-rtsp).
+  - (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).
 - 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?

machinery/go.mod

@@ -2,7 +2,8 @@ module github.com/kerberos-io/agent/machinery
 
 go 1.19
 
-// replace github.com/kerberos-io/joy4 v1.0.57 => ../../../../github.com/kerberos-io/joy4
+//replace github.com/kerberos-io/joy4 v1.0.58 => ../../../../github.com/kerberos-io/joy4
+
 // replace github.com/kerberos-io/onvif v0.0.6 => ../../../../github.com/kerberos-io/onvif
 
 require (
@@ -20,11 +21,12 @@ require (
 	github.com/gin-contrib/pprof v1.4.0
 	github.com/gin-gonic/contrib v0.0.0-20221130124618-7e01895a63f2
 	github.com/gin-gonic/gin v1.8.2
+	github.com/gofrs/uuid v3.2.0+incompatible
 	github.com/golang-jwt/jwt/v4 v4.4.3
 	github.com/golang-module/carbon/v2 v2.2.3
 	github.com/gorilla/websocket v1.5.0
 	github.com/kellydunn/golang-geo v0.7.0
-	github.com/kerberos-io/joy4 v1.0.58
+	github.com/kerberos-io/joy4 v1.0.60
 	github.com/kerberos-io/onvif v0.0.7
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/nsmith5/mjpeg v0.0.0-20200913181537-54b8ada0e53e
@@ -72,7 +74,6 @@ require (
 	github.com/go-playground/validator/v10 v10.11.1 // indirect
 	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/goccy/go-json v0.10.0 // indirect
-	github.com/gofrs/uuid v3.2.0+incompatible // indirect
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.4 // indirect

machinery/go.sum

@@ -264,8 +264,8 @@ github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+MrarZYNNg=
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
-github.com/kerberos-io/joy4 v1.0.58 h1:R8EECSF+bG7o2yHC6cX/lF77Z+bDVGl6OioLZ3+5MN4=
-github.com/kerberos-io/joy4 v1.0.58/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
+github.com/kerberos-io/joy4 v1.0.60 h1:W9LMTHw+Lgz4J9/28xCvvVebhcAioup49NqxYVmrH38=
+github.com/kerberos-io/joy4 v1.0.60/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
 github.com/kerberos-io/onvif v0.0.7 h1:LIrXjTH7G2W9DN69xZeJSB0uS3W1+C3huFO8kTqx7/A=
 github.com/kerberos-io/onvif v0.0.7/go.mod h1:Hr2dJOH2LM5SpYKk17gYZ1CMjhGhUl+QlT5kwYogrW0=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=

machinery/src/cloud/Cloud.go

@@ -231,7 +231,7 @@ loop:
 			log.Log.Debug("HandleHeartBeat: stopping as Offline is enabled.")
 		} else {
 
-			url := config.HeartbeatURI
+			hubURI := config.HeartbeatURI
 			key := ""
 			username := ""
 			vaultURI := ""
@@ -247,98 +247,110 @@ loop:
 
 			// This is the new way ;)
 			if config.HubURI != "" {
-				url = config.HubURI + "/devices/heartbeat"
+				hubURI = config.HubURI + "/devices/heartbeat"
 			}
 			if config.HubKey != "" {
 				key = config.HubKey
 			}
 
-			if key != "" {
-				// Check if we have a friendly name or not.
-				name := config.Name
-				if config.FriendlyName != "" {
-					name = config.FriendlyName
-				}
+			// Check if we have a friendly name or not.
+			name := config.Name
+			if config.FriendlyName != "" {
+				name = config.FriendlyName
+			}
 
-				// Get some system information
-				// like the uptime, hostname, memory usage, etc.
-				system, _ := GetSystemInfo()
+			// Get some system information
+			// like the uptime, hostname, memory usage, etc.
+			system, _ := GetSystemInfo()
 
-				// We will formated the uptime to a human readable format
-				// this will be used on Kerberos Hub: Uptime -> 1 day and 2 hours.
-				uptimeFormatted := uptimeStart.Format("2006-01-02 15:04:05")
-				uptimeString := carbon.Parse(uptimeFormatted).DiffForHumans()
-				uptimeString = strings.ReplaceAll(uptimeString, "ago", "")
+			// Check if the agent is running inside a cluster (Kerberos Factory) or as
+			// an open source agent
+			isEnterprise := false
+			if os.Getenv("DEPLOYMENT") == "factory" || os.Getenv("MACHINERY_ENVIRONMENT") == "kubernetes" {
+				isEnterprise = true
+			}
 
-				// Do the same for boottime
-				bootTimeFormatted := time.Unix(int64(system.BootTime), 0).Format("2006-01-02 15:04:05")
-				boottimeString := carbon.Parse(bootTimeFormatted).DiffForHumans()
-				boottimeString = strings.ReplaceAll(boottimeString, "ago", "")
+			// Congert to string
+			macs, _ := json.Marshal(system.MACs)
+			ips, _ := json.Marshal(system.IPs)
+			cameraConnected := "true"
+			if !communication.CameraConnected {
+				cameraConnected = "false"
+			}
 
-				// We'll check which mode is enabled for the camera.
-				onvifEnabled := "false"
-				onvifZoom := "false"
-				onvifPanTilt := "false"
-				onvifPresets := "false"
-				var onvifPresetsList []byte
-				if config.Capture.IPCamera.ONVIFXAddr != "" {
-					cameraConfiguration := configuration.Config.Capture.IPCamera
-					device, err := onvif.ConnectToOnvifDevice(&cameraConfiguration)
+			// We will formated the uptime to a human readable format
+			// this will be used on Kerberos Hub: Uptime -> 1 day and 2 hours.
+			uptimeFormatted := uptimeStart.Format("2006-01-02 15:04:05")
+			uptimeString := carbon.Parse(uptimeFormatted).DiffForHumans()
+			uptimeString = strings.ReplaceAll(uptimeString, "ago", "")
+
+			// Do the same for boottime
+			bootTimeFormatted := time.Unix(int64(system.BootTime), 0).Format("2006-01-02 15:04:05")
+			boottimeString := carbon.Parse(bootTimeFormatted).DiffForHumans()
+			boottimeString = strings.ReplaceAll(boottimeString, "ago", "")
+
+			// We'll check which mode is enabled for the camera.
+			onvifEnabled := "false"
+			onvifZoom := "false"
+			onvifPanTilt := "false"
+			onvifPresets := "false"
+			var onvifPresetsList []byte
+			if config.Capture.IPCamera.ONVIFXAddr != "" {
+				cameraConfiguration := configuration.Config.Capture.IPCamera
+				device, err := onvif.ConnectToOnvifDevice(&cameraConfiguration)
+				if err == nil {
+					configurations, err := onvif.GetPTZConfigurationsFromDevice(device)
 					if err == nil {
-						configurations, err := onvif.GetPTZConfigurationsFromDevice(device)
-						if err == nil {
-							onvifEnabled = "true"
-							_, canZoom, canPanTilt := onvif.GetPTZFunctionsFromDevice(configurations)
-							if canZoom {
-								onvifZoom = "true"
-							}
-							if canPanTilt {
-								onvifPanTilt = "true"
-							}
-							// Try to read out presets
-							presets, err := onvif.GetPresetsFromDevice(device)
-							if err == nil && len(presets) > 0 {
-								onvifPresets = "true"
-								onvifPresetsList, err = json.Marshal(presets)
-								if err != nil {
-									log.Log.Error("HandleHeartBeat: error while marshalling presets: " + err.Error())
-									onvifPresetsList = []byte("[]")
-								}
-							} else {
-								if err != nil {
-									log.Log.Error("HandleHeartBeat: error while getting presets: " + err.Error())
-								} else {
-									log.Log.Debug("HandleHeartBeat: no presets found.")
-								}
+						onvifEnabled = "true"
+						_, canZoom, canPanTilt := onvif.GetPTZFunctionsFromDevice(configurations)
+						if canZoom {
+							onvifZoom = "true"
+						}
+						if canPanTilt {
+							onvifPanTilt = "true"
+						}
+						// Try to read out presets
+						presets, err := onvif.GetPresetsFromDevice(device)
+						if err == nil && len(presets) > 0 {
+							onvifPresets = "true"
+							onvifPresetsList, err = json.Marshal(presets)
+							if err != nil {
+								log.Log.Error("HandleHeartBeat: error while marshalling presets: " + err.Error())
 								onvifPresetsList = []byte("[]")
 							}
 						} else {
-							log.Log.Error("HandleHeartBeat: error while getting PTZ configurations: " + err.Error())
+							if err != nil {
+								log.Log.Error("HandleHeartBeat: error while getting presets: " + err.Error())
+							} else {
+								log.Log.Debug("HandleHeartBeat: no presets found.")
+							}
 							onvifPresetsList = []byte("[]")
 						}
 					} else {
-						log.Log.Error("HandleHeartBeat: error while connecting to ONVIF device: " + err.Error())
+						log.Log.Error("HandleHeartBeat: error while getting PTZ configurations: " + err.Error())
 						onvifPresetsList = []byte("[]")
 					}
 				} else {
-					log.Log.Debug("HandleHeartBeat: ONVIF is not enabled.")
+					log.Log.Error("HandleHeartBeat: error while connecting to ONVIF device: " + err.Error())
 					onvifPresetsList = []byte("[]")
 				}
+			} else {
+				log.Log.Debug("HandleHeartBeat: ONVIF is not enabled.")
+				onvifPresetsList = []byte("[]")
+			}
 
-				// Check if the agent is running inside a cluster (Kerberos Factory) or as
-				// an open source agent
-				isEnterprise := false
-				if os.Getenv("DEPLOYMENT") == "factory" || os.Getenv("MACHINERY_ENVIRONMENT") == "kubernetes" {
-					isEnterprise = true
+			var client *http.Client
+			if os.Getenv("AGENT_TLS_INSECURE") == "true" {
+				tr := &http.Transport{
+					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 				}
+				client = &http.Client{Transport: tr}
+			} else {
+				client = &http.Client{}
+			}
 
-				// Congert to string
-				macs, _ := json.Marshal(system.MACs)
-				ips, _ := json.Marshal(system.IPs)
-				cameraConnected := "true"
-				if communication.CameraConnected == false {
-					cameraConnected = "false"
-				}
+			// We need a hub URI and hub public key before we will send a heartbeat
+			if hubURI != "" && key != "" {
 
 				var object = fmt.Sprintf(`{
 						"key" : "%s",
@@ -380,19 +392,8 @@ loop:
 
 				var jsonStr = []byte(object)
 				buffy := bytes.NewBuffer(jsonStr)
-				req, _ := http.NewRequest("POST", url, buffy)
+				req, _ := http.NewRequest("POST", hubURI, buffy)
 				req.Header.Set("Content-Type", "application/json")
-
-				var client *http.Client
-				if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-					tr := &http.Transport{
-						TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-					}
-					client = &http.Client{Transport: tr}
-				} else {
-					client = &http.Client{}
-				}
-
 				resp, err := client.Do(req)
 				if resp != nil {
 					resp.Body.Close()
@@ -406,27 +407,68 @@ loop:
 					}
 					log.Log.Error("HandleHeartBeat: (400) Something went wrong while sending to Kerberos Hub.")
 				}
-
-				// If we have a Kerberos Vault connected, we will also send some analytics
-				// to that service.
-				vaultURI = config.KStorage.URI
-				if vaultURI != "" {
-					buffy = bytes.NewBuffer(jsonStr)
-					req, _ = http.NewRequest("POST", vaultURI+"/devices/heartbeat", buffy)
-					req.Header.Set("Content-Type", "application/json")
-					resp, err = client.Do(req)
-					if resp != nil {
-						resp.Body.Close()
-					}
-					if err == nil && resp.StatusCode == 200 {
-						log.Log.Info("HandleHeartBeat: (200) Heartbeat received by Kerberos Vault.")
-					} else {
-						log.Log.Error("HandleHeartBeat: (400) Something went wrong while sending to Kerberos Vault.")
-					}
-				}
 			} else {
 				log.Log.Error("HandleHeartBeat: Disabled as we do not have a public key defined.")
 			}
+
+			// If we have a Kerberos Vault connected, we will also send some analytics
+			// to that service.
+			vaultURI = config.KStorage.URI
+			if vaultURI != "" {
+
+				var object = fmt.Sprintf(`{
+					"key" : "%s",
+					"version" : "3.0.0",
+					"release" : "%s",
+					"cpuid" : "%s",
+					"clouduser" : "%s",
+					"cloudpublickey" : "%s",
+					"cameraname" : "%s",
+					"enterprise" : %t,
+					"hostname" : "%s",
+					"architecture" : "%s",
+					"totalMemory" : "%d",
+					"usedMemory" : "%d",
+					"freeMemory" : "%d",
+					"processMemory" : "%d",
+					"mac_list" : %s,
+					"ip_list" : %s,
+					"board" : "",
+					"disk1size" : "%s",
+					"disk3size" : "%s",
+					"diskvdasize" :  "%s",
+					"uptime" : "%s",
+					"boot_time" : "%s",
+					"siteID" : "%s",
+					"onvif" : "%s",
+					"onvif_zoom" : "%s",
+					"onvif_pantilt" : "%s",
+					"onvif_presets": "%s",
+					"onvif_presets_list": %s,
+					"cameraConnected": "%s",
+					"numberoffiles" : "33",
+					"timestamp" : 1564747908,
+					"cameratype" : "IPCamera",
+					"docker" : true,
+					"kios" : false,
+					"raspberrypi" : false
+				}`, config.Key, system.Version, system.CPUId, username, key, name, isEnterprise, system.Hostname, system.Architecture, system.TotalMemory, system.UsedMemory, system.FreeMemory, system.ProcessUsedMemory, macs, ips, "0", "0", "0", uptimeString, boottimeString, config.HubSite, onvifEnabled, onvifZoom, onvifPanTilt, onvifPresets, onvifPresetsList, cameraConnected)
+
+				var jsonStr = []byte(object)
+				buffy := bytes.NewBuffer(jsonStr)
+				req, _ := http.NewRequest("POST", vaultURI+"/devices/heartbeat", buffy)
+				req.Header.Set("Content-Type", "application/json")
+
+				resp, err := client.Do(req)
+				if resp != nil {
+					resp.Body.Close()
+				}
+				if err == nil && resp.StatusCode == 200 {
+					log.Log.Info("HandleHeartBeat: (200) Heartbeat received by Kerberos Vault.")
+				} else {
+					log.Log.Error("HandleHeartBeat: (400) Something went wrong while sending to Kerberos Vault.")
+				}
+			}
 		}
 
 		// This will check if we need to stop the thread,
@@ -554,10 +596,12 @@ func HandleLiveStreamHD(livestreamCursor *pubsub.QueueCursor, configuration *mod
 				for handshake := range communication.HandleLiveHDHandshake {
 					log.Log.Info("HandleLiveStreamHD: setting up a peer connection.")
 					key := config.Key + "/" + handshake.SessionID
+					webrtc.CandidatesMutex.Lock()
 					_, ok := webrtc.CandidateArrays[key]
 					if !ok {
 						webrtc.CandidateArrays[key] = make(chan string)
 					}
+					webrtc.CandidatesMutex.Unlock()
 					webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoTrack, audioTrack, handshake, webrtc.CandidateArrays[key])
 
 				}
@@ -778,7 +822,7 @@ func VerifyPersistence(c *gin.Context, configDirectory string) {
 								"_6-967003_" + config.Name + "_200-200-400-400_24_769.mp4"
 
 							// Open test-480p.mp4
-							file, err := os.Open(configDirectory + "/test-480p.mp4")
+							file, err := os.Open(configDirectory + "/data/test-480p.mp4")
 							if err != nil {
 								msg := "VerifyPersistence: error reading test-480p.mp4: " + err.Error()
 								log.Log.Error(msg)

machinery/src/config/main.go

@@ -162,6 +162,12 @@ func OpenConfig(configDirectory string, configuration *models.Configuration) {
 		conjungo.Merge(&s3, configuration.CustomConfig.S3, opts)
 		configuration.Config.S3 = &s3
 
+		// Merge Encryption settings
+		var encryption models.Encryption
+		conjungo.Merge(&encryption, configuration.GlobalConfig.Encryption, opts)
+		conjungo.Merge(&encryption, configuration.CustomConfig.Encryption, opts)
+		configuration.Config.Encryption = &encryption
+
 		// Merge timetable manually because it's an array
 		configuration.Config.Timetable = configuration.CustomConfig.Timetable
 
@@ -511,10 +517,12 @@ func SaveConfig(configDirectory string, config models.Config, configuration *mod
 func StoreConfig(configDirectory string, config models.Config) error {
 
 	// Encryption key can be set wrong.
-	encryptionPrivateKey := config.Encryption.PrivateKey
-	// Replace \\n by \n
-	encryptionPrivateKey = strings.ReplaceAll(encryptionPrivateKey, "\\n", "\n")
-	config.Encryption.PrivateKey = encryptionPrivateKey
+	if config.Encryption != nil {
+		encryptionPrivateKey := config.Encryption.PrivateKey
+		// Replace \\n by \n
+		encryptionPrivateKey = strings.ReplaceAll(encryptionPrivateKey, "\\n", "\n")
+		config.Encryption.PrivateKey = encryptionPrivateKey
+	}
 
 	// Save into database
 	if os.Getenv("DEPLOYMENT") == "factory" || os.Getenv("MACHINERY_ENVIRONMENT") == "kubernetes" {

machinery/src/routers/mqtt/main.go

@@ -457,7 +457,12 @@ func HandleReceiveHDCandidates(mqttClient mqtt.Client, hubKey string, payload mo
 
 	if receiveHDCandidatesPayload.Timestamp != 0 {
 		if communication.CameraConnected {
-			channel := webrtc.CandidateArrays[receiveHDCandidatesPayload.SessionID]
+			key := configuration.Config.Key + "/" + receiveHDCandidatesPayload.SessionID
+			channel := webrtc.CandidateArrays[key]
+			if channel == nil {
+				channel = make(chan string)
+				webrtc.CandidateArrays[key] = channel
+			}
 			log.Log.Info("HandleReceiveHDCandidates: " + receiveHDCandidatesPayload.Candidate)
 			channel <- receiveHDCandidatesPayload.Candidate
 		} else {

machinery/src/webrtc/main.go

@@ -125,6 +125,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 						Credential: w.TurnServersCredential,
 					},
 				},
+				//ICETransportPolicy: pionWebRTC.ICETransportPolicyRelay,
 			},
 		)
 
@@ -179,6 +180,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				panic(err)
 			}
 
+			// When an ICE candidate is available send to the other Pion instance
+			// the other Pion instance will add this candidate by calling AddICECandidate
+			var candidatesMux sync.Mutex
 			// When an ICE candidate is available send to the other peer using the signaling server (MQTT).
 			// The other peer will add this candidate by calling AddICECandidate
 			peerConnection.OnICECandidate(func(candidate *pionWebRTC.ICECandidate) {
@@ -186,6 +190,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 					return
 				}
 
+				candidatesMux.Lock()
+				defer candidatesMux.Unlock()
+
 				//  Create a config map
 				valueMap := make(map[string]interface{})
 				candateJSON := candidate.ToJSON()
@@ -209,7 +216,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				payload, err := models.PackageMQTTMessage(configuration, message)
 				if err == nil {
 					log.Log.Info("InitializeWebRTCConnection:" + string(candateBinary))
-					token := mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 					token.Wait()
 				} else {
 					log.Log.Info("HandleRequestConfig: something went wrong while sending acknowledge config to hub: " + string(payload))
@@ -235,7 +242,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				}
 				payload, err := models.PackageMQTTMessage(configuration, message)
 				if err == nil {
-					token := mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 					token.Wait()
 				} else {
 					log.Log.Info("HandleRequestConfig: something went wrong while sending acknowledge config to hub: " + string(payload))

ui/.eslintrc.json

@@ -25,6 +25,7 @@
     "jsx-a11y/media-has-caption": "off",
     "jsx-a11y/anchor-is-valid": "off",
     "jsx-a11y/click-events-have-key-events": "off",
+    "jsx-a11y/control-has-associated-label": "off",
     "jsx-a11y/no-noninteractive-element-interactions": "off",
     "jsx-a11y/no-static-element-interactions": "off",
     "jsx-a11y/label-has-associated-control": [