diff --git a/machinery/src/cloud/Cloud.go b/machinery/src/cloud/Cloud.go deleted file mode 100644 index 5f628a1..0000000 --- a/machinery/src/cloud/Cloud.go +++ /dev/null @@ -1,1409 +0,0 @@ -package cloud - -import ( - "bytes" - "crypto/tls" - "encoding/base64" - "encoding/json" - "fmt" - "io" - "os" - "strings" - - "github.com/dromara/carbon/v2" - "github.com/elastic/go-sysinfo" - "github.com/gin-gonic/gin" - - mqtt "github.com/eclipse/paho.mqtt.golang" - - "net/http" - "strconv" - "time" - - "github.com/kerberos-io/agent/machinery/src/capture" - "github.com/kerberos-io/agent/machinery/src/encryption" - "github.com/kerberos-io/agent/machinery/src/log" - "github.com/kerberos-io/agent/machinery/src/models" - "github.com/kerberos-io/agent/machinery/src/onvif" - "github.com/kerberos-io/agent/machinery/src/packets" - "github.com/kerberos-io/agent/machinery/src/utils" - "github.com/kerberos-io/agent/machinery/src/webrtc" -) - -func PendingUpload(configDirectory string) { - ff, err := utils.ReadDirectory(configDirectory + "/data/cloud/") - if err == nil { - for _, f := range ff { - log.Log.Info(f.Name()) - } - } -} - -func HandleUpload(configDirectory string, configuration *models.Configuration, communication *models.Communication) { - - log.Log.Debug("HandleUpload: started") - - config := configuration.Config - watchDirectory := configDirectory + "/data/cloud/" - - if config.Offline == "true" { - log.Log.Debug("HandleUpload: stopping as Offline is enabled.") - } else { - - // Half a second delay between two uploads - delay := 500 * time.Millisecond - - loop: - for { - // This will check if we need to stop the thread, - // because of a reconfiguration. - select { - case <-communication.HandleUpload: - break loop - case <-time.After(2 * time.Second): - } - - ff, err := utils.ReadDirectory(watchDirectory) - if err != nil { - log.Log.Error("HandleUpload: " + err.Error()) - } else { - for _, f := range ff { - - // This will check if we need to stop the thread, - // because of a reconfiguration. - select { - case <-communication.HandleUpload: - break loop - default: - } - - fileName := f.Name() - uploaded := false - configured := false - err = nil - if config.Cloud == "s3" || config.Cloud == "kerberoshub" { - uploaded, configured, err = UploadKerberosHub(configuration, fileName) - } else if config.Cloud == "kstorage" || config.Cloud == "kerberosvault" { - uploaded, configured, err = UploadKerberosVault(configuration, fileName) - } else if config.Cloud == "dropbox" { - uploaded, configured, err = UploadDropbox(configuration, fileName) - } else if config.Cloud == "gdrive" { - // Todo: implement gdrive upload - } else if config.Cloud == "onedrive" { - // Todo: implement onedrive upload - } else if config.Cloud == "minio" { - // Todo: implement minio upload - } else if config.Cloud == "webdav" { - // Todo: implement webdav upload - } else if config.Cloud == "ftp" { - // Todo: implement ftp upload - } else if config.Cloud == "sftp" { - // Todo: implement sftp upload - } else if config.Cloud == "aws" { - // Todo: need to be updated, was previously used for hub. - uploaded, configured, err = UploadS3(configuration, fileName) - } else if config.Cloud == "azure" { - // Todo: implement azure upload - } else if config.Cloud == "google" { - // Todo: implement google upload - } - // And so on... (have a look here -> https://github.com/kerberos-io/agent/issues/95) - - // Check if the file is uploaded, if so, remove it. - if uploaded { - delay = 500 * time.Millisecond // reset - err := os.Remove(watchDirectory + fileName) - if err != nil { - log.Log.Error("HandleUpload: " + err.Error()) - } - - // Check if we need to remove the original recording - // removeAfterUpload is set to false by default - if config.RemoveAfterUpload != "false" { - err := os.Remove(configDirectory + "/data/recordings/" + fileName) - if err != nil { - log.Log.Error("HandleUpload: " + err.Error()) - } - } - } else if !configured { - err := os.Remove(watchDirectory + fileName) - if err != nil { - log.Log.Error("HandleUpload: " + err.Error()) - } - } else { - delay = 5 * time.Second // slow down - if err != nil { - log.Log.Error("HandleUpload: " + err.Error()) - } - } - - time.Sleep(delay) - } - } - } - } - - log.Log.Debug("HandleUpload: finished") -} - -func GetSystemInfo() (models.System, error) { - var usedMem uint64 = 0 - var totalMem uint64 = 0 - var freeMem uint64 = 0 - - var processUsedMem uint64 = 0 - - architecture := "" - cpuId := "" - KernelVersion := "" - agentVersion := "" - var MACs []string - var IPs []string - hostname := "" - bootTime := time.Time{} - - // Read agent version - version, err := os.Open("./version") - agentVersion = "unknown" - if err == nil { - defer version.Close() - agentVersionBytes, err := io.ReadAll(version) - agentVersion = string(agentVersionBytes) - if err != nil { - log.Log.Error(err.Error()) - } - } - - host, err := sysinfo.Host() - if err == nil { - cpuId = host.Info().UniqueID - architecture = host.Info().Architecture - KernelVersion = host.Info().KernelVersion - MACs = host.Info().MACs - IPs = host.Info().IPs - hostname = host.Info().Hostname - bootTime = host.Info().BootTime - memory, err := host.Memory() - if err == nil { - usedMem = memory.Used - totalMem = memory.Total - freeMem = memory.Free - } - } - - process, err := sysinfo.Self() - if err == nil { - memInfo, err := process.Memory() - if err == nil { - processUsedMem = memInfo.Resident - } - } - - system := models.System{ - Hostname: hostname, - CPUId: cpuId, - KernelVersion: KernelVersion, - Version: agentVersion, - MACs: MACs, - IPs: IPs, - BootTime: uint64(bootTime.Unix()), - Architecture: architecture, - UsedMemory: usedMem, - TotalMemory: totalMem, - FreeMemory: freeMem, - ProcessUsedMemory: processUsedMem, - } - - return system, nil -} - -func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) { - log.Log.Debug("cloud.HandleHeartBeat(): started") - - 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{} - } - - kerberosAgentVersion := utils.VERSION - - // Create a loop pull point address, which we will use to retrieve async events - // As you'll read below camera manufactures are having different implementations of events. - var pullPointAddressLoopState string - if configuration.Config.Capture.IPCamera.ONVIFXAddr != "" { - cameraConfiguration := configuration.Config.Capture.IPCamera - device, _, err := onvif.ConnectToOnvifDevice(&cameraConfiguration) - if err != nil { - pullPointAddressLoopState, err = onvif.CreatePullPointSubscription(device) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while creating pull point subscription: " + err.Error()) - } - } - } - -loop: - for { - // Configuration migh have changed, so we will reload it. - config := configuration.Config - - // We'll check ONVIF capabilitites anyhow.. Verify if we have PTZ, presets and inputs/outputs. - // For the inputs we will keep track of a the inputs and outputs state. - onvifEnabled := "false" - onvifZoom := "false" - onvifPanTilt := "false" - onvifPresets := "false" - var onvifPresetsList []byte - var onvifEventsList []byte - if config.Capture.IPCamera.ONVIFXAddr != "" { - cameraConfiguration := configuration.Config.Capture.IPCamera - device, _, err := onvif.ConnectToOnvifDevice(&cameraConfiguration) - if err == nil { - // We will try to retrieve the PTZ configurations from the device. - onvifEnabled = "true" - configurations, err := onvif.GetPTZConfigurationsFromDevice(device) - if err == nil { - _, 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("cloud.HandleHeartBeat(): error while marshalling presets: " + err.Error()) - onvifPresetsList = []byte("[]") - } - } else { - if err != nil { - log.Log.Debug("cloud.HandleHeartBeat(): error while getting presets: " + err.Error()) - } else { - log.Log.Debug("cloud.HandleHeartBeat(): no presets found.") - } - onvifPresetsList = []byte("[]") - } - } else { - log.Log.Debug("cloud.HandleHeartBeat(): error while getting PTZ configurations: " + err.Error()) - onvifPresetsList = []byte("[]") - } - - // We will also fetch some events, to know the status of the inputs and outputs. - // More event types might be added. - // -- We have two differen pull point subscriptions, one for the initials events and one for the loop. - // -- Some cameras do send recurrent events, others don't. - // a. For some older Hikvision models, events are send repeatedly (if input is high) with the strong state (set to false). - // - In this scenarion we are using a polling mechanism and set a timestamp to understand if the input is still active. - // b. For some newer Hikvision models, Avigilon, events are send only once (if state is set active). - // - In this scenario we are creating a new subscription to retrieve the initial (current) state of the inputs and outputs. - - // Get a new pull point address, to get the initiatal state of the inputs and outputs. - pullPointAddressInitialState, err := onvif.CreatePullPointSubscription(device) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while creating pull point subscription: " + err.Error()) - } - if pullPointAddressInitialState != "" { - log.Log.Debug("cloud.HandleHeartBeat(): Fetching events from pullPointAddressInitialState") - events, err := onvif.GetEventMessages(device, pullPointAddressInitialState) - log.Log.Debug("cloud.HandleHeartBeat(): Completed fetching events from pullPointAddressInitialState") - if err == nil && len(events) > 0 { - onvifEventsList, err = json.Marshal(events) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while marshalling events: " + err.Error()) - onvifEventsList = []byte("[]") - } - } else if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while getting events: " + err.Error()) - onvifEventsList = []byte("[]") - } else if len(events) == 0 { - log.Log.Debug("cloud.HandleHeartBeat(): no events found.") - onvifEventsList = []byte("[]") - } - onvif.UnsubscribePullPoint(device, pullPointAddressInitialState) - } - - // We do a second run an a long-living subscription to get the events asynchronously. - if pullPointAddressLoopState != "" { - log.Log.Debug("cloud.HandleHeartBeat(): Fetching events from pullPointAddressLoopState") - events, err := onvif.GetEventMessages(device, pullPointAddressLoopState) - log.Log.Debug("cloud.HandleHeartBeat(): Completed fetching events from pullPointAddressLoopState") - if err == nil && len(events) > 0 { - onvifEventsList, err = json.Marshal(events) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while marshalling events: " + err.Error()) - onvifEventsList = []byte("[]") - } - } else if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while getting events: " + err.Error()) - onvifEventsList = []byte("[]") - pullPointAddressLoopState, err = onvif.CreatePullPointSubscription(device) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while creating pull point subscription: " + err.Error()) - } - } else if len(events) == 0 { - log.Log.Debug("cloud.HandleHeartBeat(): no events found.") - onvifEventsList = []byte("[]") - } - } else { - log.Log.Debug("cloud.HandleHeartBeat(): no pull point address found.") - pullPointAddressLoopState, err = onvif.CreatePullPointSubscription(device) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while creating pull point subscription: " + err.Error()) - } - } - - // It also might be that events are not supported by the camera, in that case we will try to get the digital inputs and outputs. - // Through the `device` API, the `GetDigitalInputs` and `GetDigitalOutputs` functions are called. - // The disadvantage of this approach is that we don't have the state of the inputs and outputs (which is crazy..) - - if pullPointAddressInitialState == "" && pullPointAddressLoopState == "" { - var events []onvif.ONVIFEvents - outputs, err := onvif.GetRelayOutputs(device) - if err != nil { - log.Log.Debug("cloud.HandleHeartBeat(): error while getting relay outputs: " + err.Error()) - } else { - for _, output := range outputs.RelayOutputs { - event := onvif.ONVIFEvents{ - Key: string(output.Token), - Value: "false", - Type: "output", - Timestamp: time.Now().Unix(), - } - events = append(events, event) - } - } - - inputs, err := onvif.GetDigitalInputs(device) - if err != nil { - log.Log.Debug("cloud.HandleHeartBeat(): error while getting digital inputs: " + err.Error()) - } else { - for _, input := range inputs.DigitalInputs { - event := onvif.ONVIFEvents{ - Key: string(input.Token), - Value: "false", - Type: "input", - Timestamp: time.Now().Unix(), - } - events = append(events, event) - } - } - - // Marshal the events - onvifEventsList, err = json.Marshal(events) - if err != nil { - log.Log.Error("cloud.HandleHeartBeat(): error while marshalling events: " + err.Error()) - onvifEventsList = []byte("[]") - } - } - } else { - log.Log.Error("cloud.HandleHeartBeat(): error while connecting to ONVIF device: " + err.Error()) - onvifPresetsList = []byte("[]") - onvifEventsList = []byte("[]") - } - } else { - log.Log.Debug("cloud.HandleHeartBeat(): ONVIF is not enabled.") - onvifPresetsList = []byte("[]") - onvifEventsList = []byte("[]") - } - - // We'll capture some more metrics, and send it to Hub, if not in offline mode ofcourse ;) ;) - if config.Offline == "true" { - log.Log.Debug("cloud.HandleHeartBeat(): stopping as Offline is enabled.") - } else { - - hubURI := config.HeartbeatURI - key := "" - username := "" - vaultURI := "" - - if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" { - username = config.S3.Username - key = config.S3.Publickey - } else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" { - key = config.KStorage.CloudKey - username = config.KStorage.Directory - } - - // This is the new way ;) - if config.HubURI != "" { - hubURI = config.HubURI + "/devices/heartbeat" - } - if config.HubKey != "" { - key = config.HubKey - } - - // 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() - - // 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 - } - - // Congert to string - macs, _ := json.Marshal(system.MACs) - ips, _ := json.Marshal(system.IPs) - cameraConnected := "true" - if !communication.CameraConnected { - cameraConnected = "false" - } - - hasBackChannel := "false" - if communication.HasBackChannel { - hasBackChannel = "true" - } - - hub_encryption := "false" - if config.HubEncryption == "true" { - hub_encryption = "true" - } - - e2e_encryption := "false" - if config.Encryption != nil && config.Encryption.Enabled == "true" { - e2e_encryption = "true" - } - - // 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 need a hub URI and hub public key before we will send a heartbeat - if hubURI != "" && key != "" { - - var object = fmt.Sprintf(`{ - "key" : "%s", - "version" : "%s", - "hub_encryption": "%s", - "e2e_encryption": "%s", - "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, - "onvif_events_list": %s, - "cameraConnected": "%s", - "hasBackChannel": "%s", - "numberoffiles" : "33", - "timestamp" : 1564747908, - "cameratype" : "IPCamera", - "docker" : true, - "kios" : false, - "raspberrypi" : false - }`, config.Key, kerberosAgentVersion, hub_encryption, e2e_encryption, 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, onvifEventsList, cameraConnected, hasBackChannel) - - // Get the private key to encrypt the data using symmetric encryption: AES. - privateKey := config.HubPrivateKey - if hub_encryption == "true" && privateKey != "" { - // Encrypt the data using AES. - encrypted, err := encryption.AesEncrypt([]byte(object), privateKey) - if err != nil { - encrypted = []byte("") - log.Log.Error("cloud.HandleHeartBeat(): error while encrypting data: " + err.Error()) - } - - // Base64 encode the encrypted data. - encryptedBase64 := base64.StdEncoding.EncodeToString(encrypted) - object = fmt.Sprintf(`{ - "cloudpublicKey": "%s", - "encrypted" : %t, - "encryptedData" : "%s" - }`, config.HubKey, true, encryptedBase64) - } - - var jsonStr = []byte(object) - buffy := bytes.NewBuffer(jsonStr) - req, _ := http.NewRequest("POST", hubURI, 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 { - communication.CloudTimestamp.Store(time.Now().Unix()) - log.Log.Info("cloud.HandleHeartBeat(): (200) Heartbeat received by Kerberos Hub.") - } else { - if communication.CloudTimestamp != nil && communication.CloudTimestamp.Load() != nil { - communication.CloudTimestamp.Store(int64(0)) - } - log.Log.Error("cloud.HandleHeartBeat(): (400) Something went wrong while sending to Kerberos Hub.") - } - } else { - log.Log.Error("cloud.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 - accessKey := config.KStorage.AccessKey - secretAccessKey := config.KStorage.SecretAccessKey - if vaultURI != "" && accessKey != "" && secretAccessKey != "" { - - var object = fmt.Sprintf(`{ - "key" : "%s", - "version" : "%s", - "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, kerberosAgentVersion, 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("cloud.HandleHeartBeat(): (200) Heartbeat received by Kerberos Vault.") - } else { - log.Log.Error("cloud.HandleHeartBeat(): (400) Something went wrong while sending to Kerberos Vault.") - } - } - } - - // This will check if we need to stop the thread, - // because of a reconfiguration. - select { - case <-communication.HandleHeartBeat: - break loop - case <-time.After(10 * time.Second): - } - } - - if pullPointAddressLoopState != "" { - cameraConfiguration := configuration.Config.Capture.IPCamera - device, _, err := onvif.ConnectToOnvifDevice(&cameraConfiguration) - if err != nil { - onvif.UnsubscribePullPoint(device, pullPointAddressLoopState) - } - } - - log.Log.Debug("cloud.HandleHeartBeat(): finished") -} - -func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) { - - log.Log.Debug("cloud.HandleLiveStreamSD(): started") - - config := configuration.Config - - // If offline made is enabled, we will stop the thread. - if config.Offline == "true" { - log.Log.Debug("cloud.HandleLiveStreamSD(): stopping as Offline is enabled.") - } else { - - // Check if we need to enable the live stream - if config.Capture.Liveview != "false" { - - deviceId := config.Key - hubKey := "" - if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" { - hubKey = config.S3.Publickey - } else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" { - hubKey = config.KStorage.CloudKey - } - // This is the new way ;) - if config.HubKey != "" { - hubKey = config.HubKey - } - - lastLivestreamRequest := int64(0) - - var cursorError error - var pkt packets.Packet - - for cursorError == nil { - pkt, cursorError = livestreamCursor.ReadPacket() - if len(pkt.Data) == 0 || !pkt.IsKeyFrame { - continue - } - now := time.Now().Unix() - select { - case <-communication.HandleLiveSD: - lastLivestreamRequest = now - default: - } - if now-lastLivestreamRequest > 3 { - continue - } - log.Log.Info("cloud.HandleLiveStreamSD(): Sending base64 encoded images to MQTT.") - img, err := rtspClient.DecodePacket(pkt) - if err == nil { - imageResized, _ := utils.ResizeImage(&img, uint(config.Capture.IPCamera.BaseWidth), uint(config.Capture.IPCamera.BaseHeight)) - bytes, _ := utils.ImageToBytes(imageResized) - - chunking := config.Capture.LiveviewChunking - - if chunking == "true" { - - // Split encoded image into chunks of 2kb - // This is to prevent the MQTT message to be too large. - // By default, bytes are not encoded to base64 here; you are splitting the raw JPEG/PNG bytes. - // However, in MQTT and web contexts, binary data may not be handled well, so base64 is often used. - // To avoid base64 encoding, just send the raw []byte chunks as you do here. - // If you want to avoid base64, make sure the receiver can handle binary payloads. - - chunkSize := 25 * 1024 // 25KB chunks - var chunks [][]byte - for i := 0; i < len(bytes); i += chunkSize { - end := i + chunkSize - if end > len(bytes) { - end = len(bytes) - } - chunk := bytes[i:end] - chunks = append(chunks, chunk) - } - - log.Log.Infof("cloud.HandleLiveStreamSD(): Sending %d chunks of size %d bytes.", len(chunks), chunkSize) - - timestamp := time.Now().Unix() - for i, chunk := range chunks { - valueMap := make(map[string]interface{}) - valueMap["id"] = timestamp - valueMap["chunk"] = chunk - valueMap["chunkIndex"] = i - valueMap["chunkSize"] = chunkSize - valueMap["chunkCount"] = len(chunks) - message := models.Message{ - Payload: models.Payload{ - Version: "v1.0.0", - Action: "receive-sd-stream", - DeviceId: deviceId, - Value: valueMap, - }, - } - payload, err := models.PackageMQTTMessage(configuration, message) - if err == nil { - mqttClient.Publish("kerberos/hub/"+hubKey+"/"+deviceId, 1, false, payload) - log.Log.Infof("cloud.HandleLiveStreamSD(): sent chunk %d/%d to MQTT topic kerberos/hub/%s/%s", i+1, len(chunks), hubKey, deviceId) - time.Sleep(33 * time.Millisecond) // Sleep to avoid flooding the MQTT broker with messages - } else { - log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload)) - } - } - } else { - - valueMap := make(map[string]interface{}) - valueMap["image"] = bytes - message := models.Message{ - Payload: models.Payload{ - Action: "receive-sd-stream", - DeviceId: configuration.Config.Key, - Value: valueMap, - }, - } - payload, err := models.PackageMQTTMessage(configuration, message) - if err == nil { - mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload) - } else { - log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload)) - } - - } - } - time.Sleep(1000 * time.Millisecond) // Sleep to avoid flooding the MQTT broker with messages - } - - } else { - log.Log.Debug("cloud.HandleLiveStreamSD(): stopping as Liveview is disabled.") - } - } - - log.Log.Debug("cloud.HandleLiveStreamSD(): finished") -} - -func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) { - - config := configuration.Config - - if config.Offline == "true" { - log.Log.Debug("cloud.HandleLiveStreamHD(): stopping as Offline is enabled.") - } else { - - // Check if we need to enable the live stream - if config.Capture.Liveview != "false" { - - // Create per-peer broadcasters instead of shared tracks. - // Each viewer gets its own track with independent, non-blocking writes - // so a slow/congested peer cannot stall the others. - streams, _ := rtspClient.GetStreams() - videoBroadcaster := webrtc.NewVideoBroadcaster(streams) - audioBroadcaster := webrtc.NewAudioBroadcaster(streams) - - if videoBroadcaster == nil && audioBroadcaster == nil { - log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters") - return - } - - go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, rtspClient) - - if config.Capture.ForwardWebRTC == "true" { - - } else { - log.Log.Info("cloud.HandleLiveStreamHD(): Waiting for peer connections.") - for handshake := range communication.HandleLiveHDHandshake { - log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection.") - go webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, handshake) - } - } - - } else { - log.Log.Debug("cloud.HandleLiveStreamHD(): stopping as Liveview is disabled.") - } - } -} - -func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) { - - log.Log.Debug("cloud.RealtimeProcessing(): started") - - config := configuration.Config - - // If offline made is enabled, we will stop the thread. - if config.Offline == "true" { - log.Log.Debug("cloud.RealtimeProcessing(): stopping as Offline is enabled.") - } else { - - // Check if we need to enable the realtime processing - if config.RealtimeProcessing == "true" { - - hubKey := "" - if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" { - hubKey = config.S3.Publickey - } else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" { - hubKey = config.KStorage.CloudKey - } - // This is the new way ;) - if config.HubKey != "" { - hubKey = config.HubKey - } - - // We will publish the keyframes to the MQTT topic. - realtimeProcessingTopic := "kerberos/keyframes/" + hubKey - if config.RealtimeProcessingTopic != "" { - realtimeProcessingTopic = config.RealtimeProcessingTopic - } - - var cursorError error - var pkt packets.Packet - - for cursorError == nil { - pkt, cursorError = processingCursor.ReadPacket() - if len(pkt.Data) == 0 || !pkt.IsKeyFrame { - continue - } - - log.Log.Info("cloud.RealtimeProcessing(): Sending base64 encoded images to MQTT.") - img, err := rtspClient.DecodePacket(pkt) - if err == nil { - imageResized, _ := utils.ResizeImage(&img, uint(config.Capture.IPCamera.BaseWidth), uint(config.Capture.IPCamera.BaseHeight)) - bytes, _ := utils.ImageToBytes(imageResized) - encoded := base64.StdEncoding.EncodeToString(bytes) - - valueMap := make(map[string]interface{}) - valueMap["image"] = encoded - message := models.Message{ - Payload: models.Payload{ - Action: "receive-keyframe", - DeviceId: configuration.Config.Key, - Value: valueMap, - }, - } - payload, err := models.PackageMQTTMessage(configuration, message) - if err == nil { - mqttClient.Publish(realtimeProcessingTopic, 0, false, payload) - } else { - log.Log.Info("cloud.RealtimeProcessing(): something went wrong while sending acknowledge config to hub: " + string(payload)) - } - } - } - - } else { - log.Log.Debug("cloud.RealtimeProcessing(): stopping as Liveview is disabled.") - } - } - - log.Log.Debug("cloud.HandleLiveStreamSD(): finished") -} - -// VerifyHub godoc -// @Router /api/hub/verify [post] -// @ID verify-hub -// @Security Bearer -// @securityDefinitions.apikey Bearer -// @in header -// @name Authorization -// @Tags persistence -// @Param config body models.Config true "Config" -// @Summary Will verify the hub connectivity. -// @Description Will verify the hub connectivity. -// @Success 200 {object} models.APIResponse -func VerifyHub(c *gin.Context) { - - var config models.Config - err := c.BindJSON(&config) - - if err == nil { - hubURI := config.HubURI - publicKey := config.HubKey - privateKey := config.HubPrivateKey - - req, err := http.NewRequest("POST", hubURI+"/subscription/verify", nil) - if err == nil { - req.Header.Set("X-Kerberos-Hub-PublicKey", publicKey) - req.Header.Set("X-Kerberos-Hub-PrivateKey", privateKey) - 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 err == nil { - body, err := io.ReadAll(resp.Body) - defer resp.Body.Close() - if err == nil { - if resp.StatusCode == 200 { - c.JSON(200, body) - } else { - c.JSON(400, models.APIResponse{ - Data: "cloud.VerifyHub(): something went wrong while reaching the Kerberos Hub API: " + string(body), - }) - } - } else { - c.JSON(400, models.APIResponse{ - Data: "cloud.VerifyHub(): something went wrong while ready the response body: " + err.Error(), - }) - } - } else { - c.JSON(400, models.APIResponse{ - Data: "cloud.VerifyHub(): something went wrong while reaching to the Kerberos Hub API: " + hubURI, - }) - } - } else { - c.JSON(400, models.APIResponse{ - Data: "cloud.VerifyHub(): something went wrong while creating the HTTP request: " + err.Error(), - }) - } - } else { - c.JSON(400, models.APIResponse{ - Data: "cloud.VerifyHub(): something went wrong while receiving the config " + err.Error(), - }) - } -} - -// VerifyPersistence godoc -// @Router /api/persistence/verify [post] -// @ID verify-persistence -// @Security Bearer -// @securityDefinitions.apikey Bearer -// @in header -// @name Authorization -// @Tags persistence -// @Param config body models.Config true "Config" -// @Summary Will verify the persistence. -// @Description Will verify the persistence. -// @Success 200 {object} models.APIResponse -func VerifyPersistence(c *gin.Context, configDirectory string) { - - var config models.Config - err := c.BindJSON(&config) - if err != nil || config.Cloud != "" { - - if config.Cloud == "dropbox" { - VerifyDropbox(config, c) - } else if config.Cloud == "s3" || config.Cloud == "kerberoshub" { - - if config.HubURI == "" || - config.HubKey == "" || - config.HubPrivateKey == "" || - config.S3.Region == "" { - msg := "cloud.VerifyPersistence(kerberoshub): Kerberos Hub not properly configured." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } else { - - // Open test-480p.mp4 - file, err := os.Open(configDirectory + "/data/test-480p.mp4") - if err != nil { - msg := "cloud.VerifyPersistence(kerberoshub): error reading test-480p.mp4: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - defer file.Close() - - req, err := http.NewRequest("POST", config.HubURI+"/storage/upload", file) - if err != nil { - msg := "cloud.VerifyPersistence(kerberoshub): error reading Kerberos Hub HEAD request, " + config.HubURI + "/storage: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - - timestamp := time.Now().Unix() - fileName := strconv.FormatInt(timestamp, 10) + - "_6-967003_" + config.Name + "_200-200-400-400_24_769.mp4" - req.Header.Set("X-Kerberos-Storage-FileName", fileName) - req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera") - req.Header.Set("X-Kerberos-Storage-Device", config.Key) - req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey) - req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey) - req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region) - - 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 { - defer resp.Body.Close() - } - - if err == nil && resp != nil { - if resp.StatusCode == 200 { - msg := "cloud.VerifyPersistence(kerberoshub): Upload allowed using the credentials provided (" + config.HubKey + ", " + config.HubPrivateKey + ")" - log.Log.Info(msg) - c.JSON(200, models.APIResponse{ - Data: msg, - }) - } else { - msg := "cloud.VerifyPersistence(kerberoshub): Upload NOT allowed using the credentials provided (" + config.HubKey + ", " + config.HubPrivateKey + ")" - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberoshub): Error creating Kerberos Hub request" - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } - - } else if config.Cloud == "kstorage" || config.Cloud == "kerberosvault" { - - uri := config.KStorage.URI - accessKey := config.KStorage.AccessKey - secretAccessKey := config.KStorage.SecretAccessKey - directory := config.KStorage.Directory - provider := config.KStorage.Provider - - if err == nil && uri != "" && accessKey != "" && secretAccessKey != "" { - - 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{} - } - - req, err := http.NewRequest("POST", uri+"/ping", nil) - if err == nil { - req.Header.Add("X-Kerberos-Storage-AccessKey", accessKey) - req.Header.Add("X-Kerberos-Storage-SecretAccessKey", secretAccessKey) - resp, err := client.Do(req) - - if err == nil { - body, err := io.ReadAll(resp.Body) - defer resp.Body.Close() - if err == nil && resp.StatusCode == http.StatusOK { - - if provider != "" || directory != "" { - - // Generate a random name. - timestamp := time.Now().Unix() - fileName := strconv.FormatInt(timestamp, 10) + - "_6-967003_" + config.Name + "_200-200-400-400_24_769.mp4" - - // Open test-480p.mp4 - file, err := os.Open(configDirectory + "/data/test-480p.mp4") - if err != nil { - msg := "cloud.VerifyPersistence(kerberosvault): error reading test-480p.mp4: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - defer file.Close() - - req, err := http.NewRequest("POST", uri+"/storage", file) - if err == nil { - - req.Header.Set("Content-Type", "video/mp4") - req.Header.Set("X-Kerberos-Storage-CloudKey", config.HubKey) - req.Header.Set("X-Kerberos-Storage-AccessKey", accessKey) - req.Header.Set("X-Kerberos-Storage-SecretAccessKey", secretAccessKey) - req.Header.Set("X-Kerberos-Storage-Provider", provider) - req.Header.Set("X-Kerberos-Storage-FileName", fileName) - req.Header.Set("X-Kerberos-Storage-Device", config.Key) - req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera") - req.Header.Set("X-Kerberos-Storage-Directory", directory) - - 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 err == nil { - if resp != nil { - body, err := io.ReadAll(resp.Body) - defer resp.Body.Close() - if err == nil { - if resp.StatusCode == 200 { - msg := "cloud.VerifyPersistence(kerberosvault): Upload allowed using the credentials provided (" + accessKey + ", " + secretAccessKey + ")" - log.Log.Info(msg) - c.JSON(200, models.APIResponse{ - Data: body, - }) - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Something went wrong while verifying your persistence settings. Make sure your provider is the same as the storage provider in your Kerberos Vault, and the relevant storage provider is configured properly." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Upload of fake recording failed: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Something went wrong while creating /storage POST request." + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Provider and/or directory is missing from the request." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Something went wrong while verifying storage credentials: " + string(body) - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Something went wrong while verifying storage credentials:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): Something went wrong while verifying storage credentials:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifyPersistence(kerberosvault): please fill-in the required Kerberos Vault credentials." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } - } else { - msg := "cloud.VerifyPersistence(): No persistence was specified, so do not know what to verify:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } -} - -// VerifySecondaryPersistence godoc -// @Router /api/persistence/secondary/verify [post] -// @ID verify-persistence -// @Security Bearer -// @securityDefinitions.apikey Bearer -// @in header -// @name Authorization -// @Tags persistence -// @Param config body models.Config true "Config" -// @Summary Will verify the secondary persistence. -// @Description Will verify the secondary persistence. -// @Success 200 {object} models.APIResponse -func VerifySecondaryPersistence(c *gin.Context, configDirectory string) { - - var config models.Config - err := c.BindJSON(&config) - if err != nil || config.Cloud != "" { - - if config.Cloud == "kstorage" || config.Cloud == "kerberosvault" { - - if config.KStorageSecondary == nil { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): please fill-in the required Kerberos Vault credentials." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - - } else { - - uri := config.KStorageSecondary.URI - accessKey := config.KStorageSecondary.AccessKey - secretAccessKey := config.KStorageSecondary.SecretAccessKey - directory := config.KStorageSecondary.Directory - provider := config.KStorageSecondary.Provider - - if err == nil && uri != "" && accessKey != "" && secretAccessKey != "" { - - 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{} - } - - req, err := http.NewRequest("POST", uri+"/ping", nil) - if err == nil { - req.Header.Add("X-Kerberos-Storage-AccessKey", accessKey) - req.Header.Add("X-Kerberos-Storage-SecretAccessKey", secretAccessKey) - resp, err := client.Do(req) - - if err == nil { - body, err := io.ReadAll(resp.Body) - defer resp.Body.Close() - if err == nil && resp.StatusCode == http.StatusOK { - - if provider != "" || directory != "" { - - // Generate a random name. - timestamp := time.Now().Unix() - fileName := strconv.FormatInt(timestamp, 10) + - "_6-967003_" + config.Name + "_200-200-400-400_24_769.mp4" - - // Open test-480p.mp4 - file, err := os.Open(configDirectory + "/data/test-480p.mp4") - if err != nil { - msg := "cloud.VerifyPersistence(kerberosvault): error reading test-480p.mp4: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - defer file.Close() - - req, err := http.NewRequest("POST", uri+"/storage", file) - if err == nil { - - req.Header.Set("Content-Type", "video/mp4") - req.Header.Set("X-Kerberos-Storage-CloudKey", config.HubKey) - req.Header.Set("X-Kerberos-Storage-AccessKey", accessKey) - req.Header.Set("X-Kerberos-Storage-SecretAccessKey", secretAccessKey) - req.Header.Set("X-Kerberos-Storage-Provider", provider) - req.Header.Set("X-Kerberos-Storage-FileName", fileName) - req.Header.Set("X-Kerberos-Storage-Device", config.Key) - req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera") - req.Header.Set("X-Kerberos-Storage-Directory", directory) - - 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 err == nil { - if resp != nil { - body, err := io.ReadAll(resp.Body) - defer resp.Body.Close() - if err == nil { - if resp.StatusCode == 200 { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Upload allowed using the credentials provided (" + accessKey + ", " + secretAccessKey + ")" - log.Log.Info(msg) - c.JSON(200, models.APIResponse{ - Data: body, - }) - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Something went wrong while verifying your persistence settings. Make sure your provider is the same as the storage provider in your Kerberos Vault, and the relevant storage provider is configured properly." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Upload of fake recording failed: " + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Something went wrong while creating /storage POST request." + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Provider and/or directory is missing from the request." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Something went wrong while verifying storage credentials: " + string(body) - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Something went wrong while verifying storage credentials:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): Something went wrong while verifying storage credentials:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } else { - msg := "cloud.VerifySecondaryPersistence(kerberosvault): please fill-in the required Kerberos Vault credentials." - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } - } - } - } else { - msg := "cloud.VerifySecondaryPersistence(): No persistence was specified, so do not know what to verify:" + err.Error() - log.Log.Error(msg) - c.JSON(400, models.APIResponse{ - Data: msg, - }) - } -} diff --git a/machinery/src/cloud/Dropbox.go b/machinery/src/cloud/Dropbox.go deleted file mode 100644 index a3d6ae7..0000000 --- a/machinery/src/cloud/Dropbox.go +++ /dev/null @@ -1,135 +0,0 @@ -// Package cloud contains the Dropbox implementation of the Cloud interface. -// It uses the Dropbox SDK to upload files to Dropbox. -package cloud - -import ( - "bytes" - "errors" - "os" - - "github.com/dropbox/dropbox-sdk-go-unofficial/v6/dropbox" - "github.com/dropbox/dropbox-sdk-go-unofficial/v6/dropbox/files" - "github.com/dropbox/dropbox-sdk-go-unofficial/v6/dropbox/users" - "github.com/gin-gonic/gin" - "github.com/kerberos-io/agent/machinery/src/log" - "github.com/kerberos-io/agent/machinery/src/models" -) - -// UploadDropbox uploads the file to your Dropbox account using the access token and directory. -func UploadDropbox(configuration *models.Configuration, fileName string) (bool, bool, error) { - - config := configuration.Config - token := config.Dropbox.AccessToken - directory := config.Dropbox.Directory - if directory != "" { - // Check if trailing slash if not we'll add one. - if directory[len(directory)-1:] != "/" { - directory = directory + "/" - } - } - - if token == "" { - err := "UploadDropbox: Dropbox not properly configured" - log.Log.Info(err) - return false, true, errors.New(err) - } - - // Upload to Dropbox - log.Log.Info("UploadDropbox: Uploading to Dropbox") - log.Log.Info("UploadDropbox: Upload started for " + fileName) - fullname := "data/recordings/" + fileName - - dConfig := dropbox.Config{ - Token: token, - LogLevel: dropbox.LogInfo, // if needed, set the desired logging level. Default is off - } - - file, err := os.OpenFile(fullname, os.O_RDWR, 0755) - if file != nil { - defer file.Close() - } - - if err == nil { - // Upload the file - dbf := files.New(dConfig) - res, err := dbf.Upload(&files.UploadArg{ - CommitInfo: files.CommitInfo{ - Path: "/" + directory + fileName, - Mode: &files.WriteMode{ - Tagged: dropbox.Tagged{ - Tag: "overwrite", - }, - }, - }, - }, file) - - if err != nil { - log.Log.Error("UploadDropbox: Error uploading file: " + err.Error()) - return false, false, err - } - - log.Log.Info("UploadDropbox: File uploaded successfully, " + res.Name) - return true, true, nil - } - - log.Log.Error("UploadDropbox: Error opening file: " + err.Error()) - return false, true, err -} - -// VerifyDropbox verifies if the Dropbox token is valid and it is able to upload a file. -func VerifyDropbox(config models.Config, c *gin.Context) { - - token := config.Dropbox.AccessToken - directory := config.Dropbox.Directory - if directory != "" { - // Check if trailing slash if not we'll add one. - if directory[len(directory)-1:] != "/" { - directory = directory + "/" - } - } - - if token != "" { - dConfig := dropbox.Config{ - Token: token, - LogLevel: dropbox.LogInfo, // if needed, set the desired logging level. Default is off - } - dbx := users.New(dConfig) - _, err := dbx.GetCurrentAccount() - if err != nil { - c.JSON(400, models.APIResponse{ - Data: "Something went wrong while reaching the Dropbox API: " + err.Error(), - }) - } else { - - // Upload the file - content := TestFile - file := bytes.NewReader(content) - - dbf := files.New(dConfig) - _, err := dbf.Upload(&files.UploadArg{ - CommitInfo: files.CommitInfo{ - Path: "/" + directory + "kerbers-agent-test.mp4", - Mode: &files.WriteMode{ - Tagged: dropbox.Tagged{ - Tag: "overwrite", - }, - }, - }, - }, file) - - if err != nil { - c.JSON(400, models.APIResponse{ - Data: "Something went wrong while reaching the Dropbox API: " + err.Error(), - }) - } else { - c.JSON(200, models.APIResponse{ - Data: "Dropbox is working fine.", - }) - } - } - } else { - c.JSON(400, models.APIResponse{ - Data: "Dropbox token is not set.", - }) - } -} diff --git a/machinery/src/cloud/S3.go b/machinery/src/cloud/S3.go deleted file mode 100644 index 40070d7..0000000 --- a/machinery/src/cloud/S3.go +++ /dev/null @@ -1,137 +0,0 @@ -package cloud - -import ( - "crypto/tls" - "errors" - "net/http" - "net/url" - "os" - "strconv" - "strings" - - "github.com/kerberos-io/agent/machinery/src/log" - "github.com/kerberos-io/agent/machinery/src/models" - "github.com/minio/minio-go/v6" -) - -func UploadS3(configuration *models.Configuration, fileName string) (bool, bool, error) { - - config := configuration.Config - - // timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token - // 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4 - // - Timestamp - // - Size + - + microseconds - // - device - // - Region - // - Number of changes - // - Token - - if config.S3 == nil { - errorMessage := "UploadS3: Uploading Failed, as no settings found" - log.Log.Error(errorMessage) - return false, false, errors.New(errorMessage) - } - - // Legacy support, should get rid of it! - aws_access_key_id := config.S3.Publickey - aws_secret_access_key := config.S3.Secretkey - aws_region := config.S3.Region - - // This is the new way ;) - if config.HubKey != "" { - aws_access_key_id = config.HubKey - } - if config.HubPrivateKey != "" { - aws_secret_access_key = config.HubPrivateKey - } - - // Check if we have some credentials otherwise we abort the request. - if aws_access_key_id == "" || aws_secret_access_key == "" { - errorMessage := "UploadS3: Uploading Failed, as no credentials found" - log.Log.Error(errorMessage) - return false, false, errors.New(errorMessage) - } - - s3Client, err := minio.NewWithRegion("s3.amazonaws.com", aws_access_key_id, aws_secret_access_key, true, aws_region) - if err != nil { - errorMessage := "UploadS3: " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - - // Check if we need to use the proxy. - if config.S3.ProxyURI != "" { - var transport http.RoundTripper = &http.Transport{ - Proxy: func(*http.Request) (*url.URL, error) { - return url.Parse(config.S3.ProxyURI) - }, - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - } - s3Client.SetCustomTransport(transport) - } - - fileParts := strings.Split(fileName, "_") - if len(fileParts) == 1 { - errorMessage := "UploadS3: " + fileName + " is not a valid name." - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - - deviceKey := config.Key - startRecording, _ := strconv.ParseInt(fileParts[0], 10, 64) - devicename := fileParts[2] - coordinates := fileParts[3] - //numberOfChanges := fileParts[4] - token, _ := strconv.Atoi(fileParts[5]) - - log.Log.Info("UploadS3: Upload started for " + fileName) - fullname := "data/recordings/" + fileName - - file, err := os.OpenFile(fullname, os.O_RDWR, 0755) - if file != nil { - defer file.Close() - } - - if err != nil { - errorMessage := "UploadS3: " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - - fileInfo, err := file.Stat() - if err != nil { - errorMessage := "UploadS3: " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - - n, err := s3Client.PutObject(config.S3.Bucket, - config.S3.Username+"/"+fileName, - file, - fileInfo.Size(), - minio.PutObjectOptions{ - ContentType: "video/mp4", - StorageClass: "ONEZONE_IA", - UserMetadata: map[string]string{ - "event-timestamp": strconv.FormatInt(startRecording, 10), - "event-microseconds": deviceKey, - "event-instancename": devicename, - "event-regioncoordinates": coordinates, - "event-numberofchanges": deviceKey, - "event-token": strconv.Itoa(token), - "productid": deviceKey, - "publickey": aws_access_key_id, - "uploadtime": "now", - }, - }) - - if err != nil { - errorMessage := "UploadS3: Uploading Failed, " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } else { - log.Log.Info("UploadS3: Upload Finished, file has been uploaded to bucket: " + strconv.FormatInt(n, 10)) - return true, true, nil - } -} diff --git a/machinery/src/components/Kerberos.go b/machinery/src/components/Kerberos.go deleted file mode 100644 index da0486e..0000000 --- a/machinery/src/components/Kerberos.go +++ /dev/null @@ -1,845 +0,0 @@ -package components - -import ( - "context" - "fmt" - "os" - "strconv" - "sync/atomic" - "time" - - mqtt "github.com/eclipse/paho.mqtt.golang" - "github.com/gin-gonic/gin" - "go.opentelemetry.io/otel" - - "github.com/kerberos-io/agent/machinery/src/capture" - "github.com/kerberos-io/agent/machinery/src/cloud" - "github.com/kerberos-io/agent/machinery/src/computervision" - configService "github.com/kerberos-io/agent/machinery/src/config" - "github.com/kerberos-io/agent/machinery/src/log" - "github.com/kerberos-io/agent/machinery/src/models" - "github.com/kerberos-io/agent/machinery/src/onvif" - "github.com/kerberos-io/agent/machinery/src/packets" - routers "github.com/kerberos-io/agent/machinery/src/routers/mqtt" - "github.com/kerberos-io/agent/machinery/src/utils" - "github.com/kerberos-io/agent/machinery/src/webrtc" - "github.com/tevino/abool" -) - -var tracer = otel.Tracer("github.com/kerberos-io/agent/machinery/src/components") - -func Bootstrap(ctx context.Context, configDirectory string, configuration *models.Configuration, communication *models.Communication, captureDevice *capture.Capture) { - - log.Log.Debug("components.Kerberos.Bootstrap(): bootstrapping the kerberos agent.") - - bootstrapContext := context.Background() - _, span := tracer.Start(bootstrapContext, "Bootstrap") - - // We will keep track of the Kerberos Agent up time - // This is send to Kerberos Hub in a heartbeat. - uptimeStart := time.Now() - - // Initiate the packet counter, this is being used to detect - // if a camera is going blocky, or got disconnected. - var packageCounter atomic.Value - packageCounter.Store(int64(0)) - communication.PackageCounter = &packageCounter - - var packageCounterSub atomic.Value - packageCounterSub.Store(int64(0)) - communication.PackageCounterSub = &packageCounterSub - - // This is used when the last packet was received (timestamp), - // this metric is used to determine if the camera is still online/connected. - var lastPacketTimer atomic.Value - packageCounter.Store(int64(0)) - communication.LastPacketTimer = &lastPacketTimer - - var lastPacketTimerSub atomic.Value - packageCounterSub.Store(int64(0)) - communication.LastPacketTimerSub = &lastPacketTimerSub - - // This is used to understand if we have a working Kerberos Hub connection - // cloudTimestamp will be updated when successfully sending heartbeats. - var cloudTimestamp atomic.Value - cloudTimestamp.Store(int64(0)) - communication.CloudTimestamp = &cloudTimestamp - - communication.HandleStream = make(chan string, 1) - communication.HandleSubStream = make(chan string, 1) - communication.HandleUpload = make(chan string, 1) - communication.HandleHeartBeat = make(chan string, 1) - communication.HandleLiveSD = make(chan int64, 1) - communication.HandleLiveHDKeepalive = make(chan string, 1) - communication.HandleLiveHDPeers = make(chan string, 1) - communication.IsConfiguring = abool.New() - - cameraSettings := &models.Camera{} - - // Before starting the agent, we have a control goroutine, that might - // do several checks to see if the agent is still operational. - go ControlAgent(communication) - - // Handle heartbeats - go cloud.HandleHeartBeat(configuration, communication, uptimeStart) - - // We'll create a MQTT handler, which will be used to communicate with Kerberos Hub. - // Configure a MQTT client which helps for a bi-directional communication - mqttClient := routers.ConfigureMQTT(configDirectory, configuration, communication) - - span.End() - - // Run the agent and fire up all the other - // goroutines which do image capture, motion detection, onvif, etc. - for { - - // This will blocking until receiving a signal to be restarted, reconfigured, stopped, etc. - status := RunAgent(configDirectory, configuration, communication, mqttClient, uptimeStart, cameraSettings, captureDevice) - - if status == "stop" { - log.Log.Info("components.Kerberos.Bootstrap(): shutting down the agent in 3 seconds.") - time.Sleep(time.Second * 3) - os.Exit(0) - } - - if status == "not started" { - // We will re open the configuration, might have changed :O! - configService.OpenConfig(configDirectory, configuration) - // We will override the configuration with the environment variables - configService.OverrideWithEnvironmentVariables(configuration) - } - - // Reset the MQTT client, might have provided new information, so we need to reconnect. - if routers.HasMQTTClientModified(configuration) { - routers.DisconnectMQTT(mqttClient, &configuration.Config) - mqttClient = routers.ConfigureMQTT(configDirectory, configuration, communication) - } - - // We will create a new cancelable context, which will be used to cancel and restart. - // This is used to restart the agent when the configuration is updated. - ctx, cancel := context.WithCancel(context.Background()) - communication.Context = &ctx - communication.CancelContext = &cancel - } -} - -func RunAgent(configDirectory string, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, uptimeStart time.Time, cameraSettings *models.Camera, captureDevice *capture.Capture) string { - - ctx := context.Background() - ctxRunAgent, span := tracer.Start(ctx, "RunAgent") - - log.Log.Info("components.Kerberos.RunAgent(): Creating camera and processing threads.") - config := configuration.Config - - status := "not started" - - // Currently only support H264 encoded cameras, this will change. - // Establishing the camera connection without backchannel if no substream - rtspUrl := config.Capture.IPCamera.RTSP - rtspClient := captureDevice.SetMainClient(rtspUrl) - if rtspUrl != "" { - err := rtspClient.Connect(ctx, ctxRunAgent) - if err != nil { - log.Log.Error("components.Kerberos.RunAgent(): error connecting to RTSP stream: " + err.Error()) - rtspClient.Close(ctxRunAgent) - rtspClient = nil - time.Sleep(time.Second * 3) - return status - } - } else { - log.Log.Error("components.Kerberos.RunAgent(): no rtsp url found in config, please provide one.") - rtspClient = nil - time.Sleep(time.Second * 3) - return status - } - - log.Log.Info("components.Kerberos.RunAgent(): opened RTSP stream: " + rtspUrl) - - // Get the video streams from the RTSP server. - videoStreams, err := rtspClient.GetVideoStreams() - if err != nil || len(videoStreams) == 0 { - log.Log.Error("components.Kerberos.RunAgent(): no video stream found, might be the wrong codec (we only support H264 for the moment)") - rtspClient.Close(ctxRunAgent) - time.Sleep(time.Second * 3) - return status - } - - // Get the video stream from the RTSP server. - videoStream := videoStreams[0] - - // Get some information from the video stream. - width := videoStream.Width - height := videoStream.Height - - // Set config values as well - configuration.Config.Capture.IPCamera.Width = width - configuration.Config.Capture.IPCamera.Height = height - - // Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub). - baseWidth := config.Capture.IPCamera.BaseWidth - baseHeight := config.Capture.IPCamera.BaseHeight - // If the liveview height is not set, we will calculate it based on the width and aspect ratio of the camera. - if baseWidth > 0 && baseHeight == 0 { - widthAspectRatio := float64(baseWidth) / float64(width) - configuration.Config.Capture.IPCamera.BaseHeight = int(float64(height) * widthAspectRatio) - } else if baseHeight > 0 && baseWidth > 0 { - configuration.Config.Capture.IPCamera.BaseHeight = baseHeight - configuration.Config.Capture.IPCamera.BaseWidth = baseWidth - } else { - configuration.Config.Capture.IPCamera.BaseHeight = height - configuration.Config.Capture.IPCamera.BaseWidth = width - } - - // Set the SPS and PPS values in the configuration. - configuration.Config.Capture.IPCamera.SPSNALUs = [][]byte{videoStream.SPS} - configuration.Config.Capture.IPCamera.PPSNALUs = [][]byte{videoStream.PPS} - configuration.Config.Capture.IPCamera.VPSNALUs = [][]byte{videoStream.VPS} - - // Define queues for the main and sub stream. - var queue *packets.Queue - var subQueue *packets.Queue - - // Create a packet queue, which is filled by the HandleStream routing - // and consumed by all other routines: motion, livestream, etc. - if config.Capture.PreRecording <= 0 { - config.Capture.PreRecording = 1 - log.Log.Warning("components.Kerberos.RunAgent(): Prerecording value not found in config or invalid value! Found: " + strconv.FormatInt(config.Capture.PreRecording, 10)) - } - - // We might have a secondary rtsp url, so we might need to use that for livestreaming let us check first! - subStreamEnabled := false - subRtspUrl := config.Capture.IPCamera.SubRTSP - var videoSubStreams []packets.Stream - - if subRtspUrl != "" && subRtspUrl != rtspUrl { - // For the sub stream we will not enable backchannel. - subStreamEnabled = true - rtspSubClient := captureDevice.SetSubClient(subRtspUrl) - captureDevice.RTSPSubClient = rtspSubClient - - err := rtspSubClient.Connect(ctx, ctxRunAgent) - if err != nil { - log.Log.Error("components.Kerberos.RunAgent(): error connecting to RTSP sub stream: " + err.Error()) - time.Sleep(time.Second * 3) - return status - } - log.Log.Info("components.Kerberos.RunAgent(): opened RTSP sub stream: " + subRtspUrl) - - // Get the video streams from the RTSP server. - videoSubStreams, err = rtspSubClient.GetVideoStreams() - if err != nil || len(videoSubStreams) == 0 { - log.Log.Error("components.Kerberos.RunAgent(): no video sub stream found, might be the wrong codec (we only support H264 for the moment)") - rtspSubClient.Close(ctxRunAgent) - time.Sleep(time.Second * 3) - return status - } - - // Get the video stream from the RTSP server. - videoSubStream := videoSubStreams[0] - - width := videoSubStream.Width - height := videoSubStream.Height - - // Set config values as well - configuration.Config.Capture.IPCamera.SubWidth = width - configuration.Config.Capture.IPCamera.SubHeight = height - - // If we have a substream, we need to set the width and height of the substream. (so we will override above information) - // Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub). - baseWidth := config.Capture.IPCamera.BaseWidth - baseHeight := config.Capture.IPCamera.BaseHeight - // If the liveview height is not set, we will calculate it based on the width and aspect ratio of the camera. - if baseWidth > 0 && baseHeight == 0 { - widthAspectRatio := float64(baseWidth) / float64(width) - configuration.Config.Capture.IPCamera.BaseHeight = int(float64(height) * widthAspectRatio) - } else if baseHeight > 0 && baseWidth > 0 { - configuration.Config.Capture.IPCamera.BaseHeight = baseHeight - configuration.Config.Capture.IPCamera.BaseWidth = baseWidth - } else { - configuration.Config.Capture.IPCamera.BaseHeight = height - configuration.Config.Capture.IPCamera.BaseWidth = width - } - } - - // We are creating a queue to store the RTSP frames in, these frames will be - // processed by the different consumers: motion detection, recording, etc. - queue = packets.NewQueue() - communication.Queue = queue - - // Set the maximum GOP count, this is used to determine the pre-recording time. - log.Log.Info("components.Kerberos.RunAgent(): SetMaxGopCount was set with: " + strconv.Itoa(int(config.Capture.PreRecording)+1)) - queue.SetMaxGopCount(1) // We will adjust this later on, when we have the GOP size. - queue.WriteHeader(videoStreams) - go rtspClient.Start(ctx, "main", queue, configuration, communication) - - // Main stream is connected and ready to go. - communication.MainStreamConnected = true - - // Try to create backchannel - rtspBackChannelClient := captureDevice.SetBackChannelClient(rtspUrl) - err = rtspBackChannelClient.ConnectBackChannel(ctx, ctxRunAgent) - if err == nil { - log.Log.Info("components.Kerberos.RunAgent(): opened RTSP backchannel stream: " + rtspUrl) - go rtspBackChannelClient.StartBackChannel(ctx, ctxRunAgent) - } - - rtspSubClient := captureDevice.RTSPSubClient - if subStreamEnabled && rtspSubClient != nil { - subQueue = packets.NewQueue() - communication.SubQueue = subQueue - subQueue.SetMaxGopCount(1) // GOP time frame is set to 1 for motion detection and livestreaming. - subQueue.WriteHeader(videoSubStreams) - go rtspSubClient.Start(ctx, "sub", subQueue, configuration, communication) - - // Sub stream is connected and ready to go. - communication.SubStreamConnected = true - } - - // Handle livestream SD (low resolution over MQTT) - if subStreamEnabled { - livestreamCursor := subQueue.Latest() - go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, rtspSubClient) - } else { - livestreamCursor := queue.Latest() - go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, rtspClient) - } - - // Handle livestream HD (high resolution over WEBRTC) - communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100) - if subStreamEnabled { - livestreamHDCursor := subQueue.Latest() - go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspSubClient) - } else { - livestreamHDCursor := queue.Latest() - go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspClient) - } - - // Handle recording, will write an mp4 to disk. - go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient) - - // Handle processing of motion - communication.HandleMotion = make(chan models.MotionDataPartial, 10) - if subStreamEnabled { - motionCursor := subQueue.Latest() - go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, rtspSubClient) - } else { - motionCursor := queue.Latest() - go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, rtspClient) - } - - // Handle realtime processing if enabled. - if subStreamEnabled { - realtimeProcessingCursor := subQueue.Latest() - go cloud.HandleRealtimeProcessing(realtimeProcessingCursor, configuration, communication, mqttClient, rtspClient) - } else { - realtimeProcessingCursor := queue.Latest() - go cloud.HandleRealtimeProcessing(realtimeProcessingCursor, configuration, communication, mqttClient, rtspClient) - } - - // Handle Upload to cloud provider (Kerberos Hub, Kerberos Vault and others) - go cloud.HandleUpload(configDirectory, configuration, communication) - - // Handle ONVIF actions - communication.HandleONVIF = make(chan models.OnvifAction, 10) - go onvif.HandleONVIFActions(configuration, communication) - - communication.HandleAudio = make(chan models.AudioDataPartial, 10) - if rtspBackChannelClient.HasBackChannel { - communication.HasBackChannel = true - go WriteAudioToBackchannel(communication, rtspBackChannelClient) - } - - // If we reach this point, we have a working RTSP connection. - communication.CameraConnected = true - - // Otel end span - span.End() - - // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - // This will go into a blocking state, once this channel is triggered - // the agent will cleanup and restart. - - status = <-communication.HandleBootstrap - - // If we reach this point, we are stopping the stream. - communication.CameraConnected = false - communication.MainStreamConnected = false - communication.SubStreamConnected = false - - // Cancel the main context, this will stop all the other goroutines. - (*communication.CancelContext)() - - // We will re open the configuration, might have changed :O! - configService.OpenConfig(configDirectory, configuration) - - // We will override the configuration with the environment variables - configService.OverrideWithEnvironmentVariables(configuration) - - // Here we are cleaning up everything! - if configuration.Config.Offline != "true" { - select { - case communication.HandleUpload <- "stop": - log.Log.Info("components.Kerberos.RunAgent(): stopping upload") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.RunAgent(): stopping upload timed out") - } - } - - select { - case communication.HandleStream <- "stop": - log.Log.Info("components.Kerberos.RunAgent(): stopping stream") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.RunAgent(): stopping stream timed out") - } - // We use the steam channel to stop both main and sub stream. - //if subStreamEnabled { - // communication.HandleSubStream <- "stop" - //} - - time.Sleep(time.Second * 3) - - err = rtspClient.Close(ctxRunAgent) - if err != nil { - log.Log.Error("components.Kerberos.RunAgent(): error closing RTSP stream: " + err.Error()) - time.Sleep(time.Second * 3) - return status - } - - queue.Close() - queue = nil - communication.Queue = nil - - if subStreamEnabled { - err = rtspSubClient.Close(ctxRunAgent) - if err != nil { - log.Log.Error("components.Kerberos.RunAgent(): error closing RTSP sub stream: " + err.Error()) - time.Sleep(time.Second * 3) - return status - } - subQueue.Close() - subQueue = nil - communication.SubQueue = nil - } - - err = rtspBackChannelClient.Close(ctxRunAgent) - if err != nil { - log.Log.Error("components.Kerberos.RunAgent(): error closing RTSP backchannel stream: " + err.Error()) - } - - time.Sleep(time.Second * 3) - - close(communication.HandleLiveHDHandshake) - communication.HandleLiveHDHandshake = nil - - close(communication.HandleMotion) - communication.HandleMotion = nil - - close(communication.HandleAudio) - communication.HandleAudio = nil - - close(communication.HandleONVIF) - communication.HandleONVIF = nil - - // Waiting for some seconds to make sure everything is properly closed. - log.Log.Info("components.Kerberos.RunAgent(): waiting 3 seconds to make sure everything is properly closed.") - time.Sleep(time.Second * 3) - - return status -} - -// packetAgeString returns a human readable age (e.g. "12s") since the last -// packet timestamp stored in the given atomic.Value, or "unknown" when no -// packet has been received yet. Used to add context to watchdog restart logs. -func packetAgeString(timer *atomic.Value) string { - if timer == nil { - return "unknown" - } - - // atomic.Value panics on Load() if it was never initialized via Store(). - var v any - func() { - defer func() { - if recover() != nil { - v = nil - } - }() - v = timer.Load() - }() - - last, ok := v.(int64) - if !ok || last == 0 { - return "unknown" - } - - age := time.Now().Unix() - last - if age < 0 { - age = 0 - } - return strconv.FormatInt(age, 10) + "s" -} - -// ControlAgent will check if the camera is still connected, if not it will restart the agent. -// In the other thread we are keeping track of the number of packets received, and particular the keyframe packets. -// Once we are not receiving any packets anymore, we will restart the agent. -func ControlAgent(communication *models.Communication) { - log.Log.Debug("components.Kerberos.ControlAgent(): started") - packageCounter := communication.PackageCounter - packageSubCounter := communication.PackageCounterSub - go func() { - // A channel to check the camera activity - var previousPacket int64 = 0 - var previousPacketSub int64 = 0 - var occurence = 0 - var occurenceSub = 0 - for { - - // If camera is connected, we'll check if we are still receiving packets. - if communication.CameraConnected { - - // First we'll check the main stream. - packetsR := packageCounter.Load().(int64) - if packetsR == previousPacket { - // If we are already reconfiguring, - // we dont need to check if the stream is blocking. - if !communication.IsConfiguring.IsSet() { - occurence = occurence + 1 - } - } else { - occurence = 0 - } - - log.Log.Info("components.Kerberos.ControlAgent(): Number of packets read from mainstream: " + strconv.FormatInt(packetsR, 10)) - - // After 15 seconds without activity this is thrown.. - if occurence == 3 { - log.Log.Info(fmt.Sprintf("components.Kerberos.ControlAgent(): Restarting machinery because of blocking mainstream. (stalledKeyframeCounter=%d, lastPacket=%s ago, isConfiguring=%t)", - packetsR, packetAgeString(communication.LastPacketTimer), communication.IsConfiguring.IsSet())) - select { - case communication.HandleBootstrap <- "restart": - log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream.") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream timed out") - } - occurence = 0 - } - - // Now we'll check the sub stream. - packetsSubR := packageSubCounter.Load().(int64) - if communication.SubStreamConnected { - if packetsSubR == previousPacketSub { - // If we are already reconfiguring, - // we dont need to check if the stream is blocking. - if !communication.IsConfiguring.IsSet() { - occurenceSub = occurenceSub + 1 - } - } else { - occurenceSub = 0 - } - - log.Log.Info("components.Kerberos.ControlAgent(): Number of packets read from substream: " + strconv.FormatInt(packetsSubR, 10)) - - // After 15 seconds without activity this is thrown.. - if occurenceSub == 3 { - log.Log.Info(fmt.Sprintf("components.Kerberos.ControlAgent(): substream stalled (stalledKeyframeCounter=%d, lastPacket=%s ago, isConfiguring=%t)", - packetsSubR, packetAgeString(communication.LastPacketTimerSub), communication.IsConfiguring.IsSet())) - select { - case communication.HandleBootstrap <- "restart": - log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream.") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.ControlAgent(): Restarting machinery because of blocking substream timed out") - } - occurenceSub = 0 - } - } - - previousPacket = packageCounter.Load().(int64) - previousPacketSub = packageSubCounter.Load().(int64) - } - - time.Sleep(5 * time.Second) - } - }() - log.Log.Debug("components.Kerberos.ControlAgent(): finished") -} - -// GetDashboard godoc -// @Router /api/dashboard [get] -// @ID dashboard -// @Tags general -// @Summary Get all information showed on the dashboard. -// @Description Get all information showed on the dashboard. -// @Success 200 -func GetDashboard(c *gin.Context, configDirectory string, configuration *models.Configuration, communication *models.Communication) { - - // Check if camera is online. - cameraIsOnline := communication.CameraConnected - - // If an agent is properly setup with Kerberos Hub, we will send - // a ping to Kerberos Hub every 15seconds. On receiving a positive response - // it will update the CloudTimestamp value. - cloudIsOnline := false - if communication.CloudTimestamp != nil && communication.CloudTimestamp.Load() != nil { - timestamp := communication.CloudTimestamp.Load().(int64) - if timestamp > 0 { - cloudIsOnline = true - } - } - - // The total number of recordings stored in the directory. - recordingDirectory := configDirectory + "/data/recordings" - numberOfRecordings := utils.NumberOfMP4sInDirectory(recordingDirectory) - activeWebRTCReaders := webrtc.GetActivePeerConnectionCount() - pendingWebRTCHandshakes := 0 - if communication.HandleLiveHDHandshake != nil { - pendingWebRTCHandshakes = len(communication.HandleLiveHDHandshake) - } - - // All days stored in this agent. - days := []string{} - latestEvents := []models.Media{} - files, err := utils.ReadDirectory(recordingDirectory) - if err == nil { - events := utils.GetSortedDirectory(files) - - // Get All days - days = utils.GetDays(events, recordingDirectory, configuration) - - // Get all latest events - var eventFilter models.EventFilter - eventFilter.NumberOfElements = 5 - latestEvents = utils.GetMediaFormatted(events, recordingDirectory, configuration, eventFilter) // will get 5 latest recordings. - } - - c.JSON(200, gin.H{ - "offlineMode": configuration.Config.Offline, - "cameraOnline": cameraIsOnline, - "cloudOnline": cloudIsOnline, - "numberOfRecordings": numberOfRecordings, - "webrtcReaders": activeWebRTCReaders, - "webrtcPending": pendingWebRTCHandshakes, - "days": days, - "latestEvents": latestEvents, - }) -} - -// GetLatestEvents godoc -// @Router /api/latest-events [post] -// @ID latest-events -// @Tags general -// @Param eventFilter body models.EventFilter true "Event filter" -// @Summary Get the latest recordings (events) from the recordings directory. -// @Description Get the latest recordings (events) from the recordings directory. -// @Success 200 -func GetLatestEvents(c *gin.Context, configDirectory string, configuration *models.Configuration, communication *models.Communication) { - var eventFilter models.EventFilter - err := c.BindJSON(&eventFilter) - if err == nil { - // Default to 10 if no limit is set. - if eventFilter.NumberOfElements == 0 { - eventFilter.NumberOfElements = 10 - } - recordingDirectory := configDirectory + "/data/recordings" - files, err := utils.ReadDirectory(recordingDirectory) - if err == nil { - events := utils.GetSortedDirectory(files) - // We will get all recordings from the directory (as defined by the filter). - fileObjects := utils.GetMediaFormatted(events, recordingDirectory, configuration, eventFilter) - c.JSON(200, gin.H{ - "events": fileObjects, - }) - } else { - c.JSON(400, gin.H{ - "data": "Something went wrong: " + err.Error(), - }) - } - } else { - c.JSON(400, gin.H{ - "data": "Something went wrong: " + err.Error(), - }) - } -} - -// GetDays godoc -// @Router /api/days [get] -// @ID days -// @Tags general -// @Summary Get all days stored in the recordings directory. -// @Description Get all days stored in the recordings directory. -// @Success 200 -func GetDays(c *gin.Context, configDirectory string, configuration *models.Configuration, communication *models.Communication) { - recordingDirectory := configDirectory + "/data/recordings" - files, err := utils.ReadDirectory(recordingDirectory) - if err == nil { - events := utils.GetSortedDirectory(files) - days := utils.GetDays(events, recordingDirectory, configuration) - c.JSON(200, gin.H{ - "events": days, - }) - } else { - c.JSON(400, gin.H{ - "data": "Something went wrong: " + err.Error(), - }) - } -} - -// StopAgent godoc -// @Router /api/camera/stop [post] -// @ID camera-stop -// @Tags camera -// @Summary Stop the agent. -// @Description Stop the agent. -// @Success 200 {object} models.APIResponse -func StopAgent(c *gin.Context, communication *models.Communication) { - log.Log.Info("components.Kerberos.StopAgent(): sending signal to stop agent, this will os.Exit(0).") - select { - case communication.HandleBootstrap <- "stop": - log.Log.Info("components.Kerberos.StopAgent(): Stopping machinery.") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.StopAgent(): Stopping machinery timed out") - } - c.JSON(200, gin.H{ - "stopped": true, - }) -} - -// RestartAgent godoc -// @Router /api/camera/restart [post] -// @ID camera-restart -// @Tags camera -// @Summary Restart the agent. -// @Description Restart the agent. -// @Success 200 {object} models.APIResponse -func RestartAgent(c *gin.Context, communication *models.Communication) { - log.Log.Info("components.Kerberos.RestartAgent(): sending signal to restart agent.") - select { - case communication.HandleBootstrap <- "restart": - log.Log.Info("components.Kerberos.RestartAgent(): Restarting machinery.") - case <-time.After(1 * time.Second): - log.Log.Info("components.Kerberos.RestartAgent(): Restarting machinery timed out") - } - c.JSON(200, gin.H{ - "restarted": true, - }) -} - -// MakeRecording godoc -// @Router /api/camera/record [post] -// @ID camera-record -// @Tags camera -// @Summary Make a recording. -// @Description Make a recording. -// @Success 200 {object} models.APIResponse -func MakeRecording(c *gin.Context, communication *models.Communication) { - log.Log.Info("components.Kerberos.MakeRecording(): sending signal to start recording.") - dataToPass := models.MotionDataPartial{ - Timestamp: time.Now().Unix(), - NumberOfChanges: 100000000, // hack set the number of changes to a high number to force recording - } - communication.HandleMotion <- dataToPass //Save data to the channel - c.JSON(200, gin.H{ - "recording": true, - }) -} - -// GetSnapshotBase64 godoc -// @Router /api/camera/snapshot/base64 [get] -// @ID snapshot-base64 -// @Tags camera -// @Summary Get a snapshot from the camera in base64. -// @Description Get a snapshot from the camera in base64. -// @Success 200 -func GetSnapshotBase64(c *gin.Context, captureDevice *capture.Capture, configuration *models.Configuration, communication *models.Communication) { - // We'll try to get a snapshot from the camera. - base64Image := capture.Base64Image(captureDevice, communication, configuration) - if base64Image != "" { - communication.Image = base64Image - } - - c.JSON(200, gin.H{ - "base64": communication.Image, - }) -} - -// GetSnapshotJpeg godoc -// @Router /api/camera/snapshot/jpeg [get] -// @ID snapshot-jpeg -// @Tags camera -// @Summary Get a snapshot from the camera in jpeg format. -// @Description Get a snapshot from the camera in jpeg format. -// @Success 200 -func GetSnapshotRaw(c *gin.Context, captureDevice *capture.Capture, configuration *models.Configuration, communication *models.Communication) { - // We'll try to get a snapshot from the camera. - image := capture.JpegImage(captureDevice, communication) - - // encode image to jpeg - imageResized, _ := utils.ResizeImage(&image, uint(configuration.Config.Capture.IPCamera.BaseWidth), uint(configuration.Config.Capture.IPCamera.BaseHeight)) - bytes, _ := utils.ImageToBytes(imageResized) - - // Return image/jpeg - c.Data(200, "image/jpeg", bytes) -} - -// GetConfig godoc -// @Router /api/config [get] -// @ID config -// @Tags config -// @Summary Get the current configuration. -// @Description Get the current configuration. -// @Success 200 -func GetConfig(c *gin.Context, captureDevice *capture.Capture, configuration *models.Configuration, communication *models.Communication) { - // We'll try to get a fresh snapshot from the camera. Capturing a snapshot - // reads a keyframe from the live stream, which blocks until one arrives. - // When the camera is offline or the stream is stalled (no packets being - // received) this would block the /config endpoint indefinitely, making the - // agent appear unreachable even though its HTTP server is healthy. We - // therefore bound the snapshot fetch with a short timeout and fall back to - // the last cached snapshot, so /config always responds promptly. - snapshot := make(chan string, 1) - go func() { - snapshot <- capture.Base64Image(captureDevice, communication, configuration) - }() - select { - case base64Image := <-snapshot: - if base64Image != "" { - communication.Image = base64Image - } - case <-time.After(2 * time.Second): - log.Log.Info("components.Kerberos.GetConfig(): snapshot timed out (stream stalled or camera offline), returning configuration with the last cached snapshot.") - } - - c.JSON(200, gin.H{ - "config": configuration.Config, - "custom": configuration.CustomConfig, - "global": configuration.GlobalConfig, - "snapshot": communication.Image, - }) -} - -// UpdateConfig godoc -// @Router /api/config [post] -// @ID config -// @Tags config -// @Param config body models.Config true "Configuration" -// @Summary Update the current configuration. -// @Description Update the current configuration. -// @Success 200 -func UpdateConfig(c *gin.Context, configDirectory string, configuration *models.Configuration, communication *models.Communication) { - var config models.Config - err := c.BindJSON(&config) - if err == nil { - err := configService.SaveConfig(configDirectory, config, configuration, communication) - if err == nil { - c.JSON(200, gin.H{ - "data": "☄ Reconfiguring", - }) - } else { - c.JSON(200, gin.H{ - "data": "☄ Reconfiguring", - }) - } - } else { - c.JSON(400, gin.H{ - "data": "Something went wrong: " + err.Error(), - }) - } -} diff --git a/machinery/src/models/Camera.go b/machinery/src/models/Camera.go deleted file mode 100644 index 5f62827..0000000 --- a/machinery/src/models/Camera.go +++ /dev/null @@ -1,15 +0,0 @@ -package models - -import "github.com/kerberos-io/joy4/av" - -type Camera struct { - Width int - Height int - Num int - Denum int - Framerate float64 - RTSP string - SubRTSP string - Codec av.CodecType - Initialized bool -} diff --git a/machinery/src/models/Communication.go b/machinery/src/models/Communication.go deleted file mode 100644 index afb7b25..0000000 --- a/machinery/src/models/Communication.go +++ /dev/null @@ -1,52 +0,0 @@ -package models - -import ( - "context" - "sync/atomic" - - "github.com/kerberos-io/agent/machinery/src/packets" - "github.com/tevino/abool" -) - -type LiveHDSignalingCallbacks struct { - SendAnswer func(sessionID string, sdp string) error - SendCandidate func(sessionID string, candidate string) error - SendError func(sessionID string, message string) error -} - -type LiveHDHandshake struct { - Payload RequestHDStreamPayload - Signaling *LiveHDSignalingCallbacks -} - -// The communication struct that is managing -// all the communication between the different goroutines. -type Communication struct { - Context *context.Context - CancelContext *context.CancelFunc - PackageCounter *atomic.Value - LastPacketTimer *atomic.Value - PackageCounterSub *atomic.Value - LastPacketTimerSub *atomic.Value - CloudTimestamp *atomic.Value - HandleBootstrap chan string - HandleStream chan string - HandleSubStream chan string - HandleMotion chan MotionDataPartial - HandleAudio chan AudioDataPartial - HandleUpload chan string - HandleHeartBeat chan string - HandleLiveSD chan int64 - HandleLiveHDKeepalive chan string - HandleLiveHDHandshake chan LiveHDHandshake - HandleLiveHDPeers chan string - HandleONVIF chan OnvifAction - IsConfiguring *abool.AtomicBool - Queue *packets.Queue - SubQueue *packets.Queue - Image string - CameraConnected bool - MainStreamConnected bool - SubStreamConnected bool - HasBackChannel bool -} diff --git a/machinery/src/models/Config.go b/machinery/src/models/Config.go deleted file mode 100644 index 3759120..0000000 --- a/machinery/src/models/Config.go +++ /dev/null @@ -1,198 +0,0 @@ -package models - -// A struct which contains the global, local and merged config. -type Configuration struct { - Name string - Port string - Config Config - CustomConfig Config - GlobalConfig Config -} - -// Config is the highlevel struct which contains all the configuration of -// your Kerberos Open Source instance. -type Config struct { - Type string `json:"type"` - Key string `json:"key"` - Name string `json:"name"` - FriendlyName string `json:"friendly_name"` - Time string `json:"time" bson:"time"` - Offline string `json:"offline"` - AutoClean string `json:"auto_clean"` - RemoveAfterUpload string `json:"remove_after_upload"` - MaxDirectorySize int64 `json:"max_directory_size"` - Timezone string `json:"timezone"` - Capture Capture `json:"capture"` - Timetable []*Timetable `json:"timetable"` - Region *Region `json:"region"` - Cloud string `json:"cloud" bson:"cloud"` - S3 *S3 `json:"s3,omitempty" bson:"s3,omitempty"` - KStorage *KStorage `json:"kstorage,omitempty" bson:"kstorage,omitempty"` - KStorageSecondary *KStorage `json:"kstorage_secondary,omitempty" bson:"kstorage_secondary,omitempty"` - Dropbox *Dropbox `json:"dropbox,omitempty" bson:"dropbox,omitempty"` - MQTTURI string `json:"mqtturi" bson:"mqtturi,omitempty"` - MQTTUsername string `json:"mqtt_username" bson:"mqtt_username"` - MQTTPassword string `json:"mqtt_password" bson:"mqtt_password"` - STUNURI string `json:"stunuri" bson:"stunuri"` - ForceTurn string `json:"turn_force" bson:"turn_force"` - TURNURI string `json:"turnuri" bson:"turnuri"` - TURNUsername string `json:"turn_username" bson:"turn_username"` - TURNPassword string `json:"turn_password" bson:"turn_password"` - HeartbeatURI string `json:"heartbeaturi" bson:"heartbeaturi"` /*obsolete*/ - HubEncryption string `json:"hub_encryption" bson:"hub_encryption"` - HubURI string `json:"hub_uri" bson:"hub_uri"` - HubKey string `json:"hub_key" bson:"hub_key"` - HubPrivateKey string `json:"hub_private_key" bson:"hub_private_key"` - HubSite string `json:"hub_site" bson:"hub_site"` - ConditionURI string `json:"condition_uri" bson:"condition_uri"` - Encryption *Encryption `json:"encryption,omitempty" bson:"encryption,omitempty"` - Signing *Signing `json:"signing,omitempty" bson:"signing,omitempty"` - RealtimeProcessing string `json:"realtimeprocessing,omitempty" bson:"realtimeprocessing,omitempty"` - RealtimeProcessingTopic string `json:"realtimeprocessing_topic" bson:"realtimeprocessing_topic"` -} - -// Capture defines which camera type (Id) you are using (IP, USB or Raspberry Pi camera), -// and also contains recording specific parameters. -type Capture struct { - Name string `json:"name"` - IPCamera IPCamera `json:"ipcamera"` - USBCamera USBCamera `json:"usbcamera"` - RaspiCamera RaspiCamera `json:"raspicamera"` - Recording string `json:"recording,omitempty"` - Snapshots string `json:"snapshots,omitempty"` - Motion string `json:"motion,omitempty"` - Liveview string `json:"liveview,omitempty"` - LiveviewChunking string `json:"liveview_chunking,omitempty" bson:"liveview_chunking,omitempty"` - Continuous string `json:"continuous,omitempty"` - PostRecording int64 `json:"postrecording"` - PreRecording int64 `json:"prerecording"` - GopSize int `json:"gopsize,omitempty" bson:"gopsize,omitempty"` // GOP size in seconds, used for pre-recording - MaxLengthRecording int64 `json:"maxlengthrecording"` - TranscodingWebRTC string `json:"transcodingwebrtc"` - TranscodingResolution int64 `json:"transcodingresolution"` - ForwardWebRTC string `json:"forwardwebrtc"` - Fragmented string `json:"fragmented,omitempty" bson:"fragmented,omitempty"` - FragmentedDuration int64 `json:"fragmentedduration,omitempty" bson:"fragmentedduration,omitempty"` - PixelChangeThreshold int `json:"pixelChangeThreshold,omitempty"` -} - -// IPCamera configuration, such as the RTSP url of the IPCamera and the FPS. -// Also includes ONVIF integration -type IPCamera struct { - RTSP string `json:"rtsp"` - Width int `json:"width"` - Height int `json:"height"` - FPS string `json:"fps"` - - SubRTSP string `json:"sub_rtsp"` - SubWidth int `json:"sub_width"` - SubHeight int `json:"sub_height"` - - BaseWidth int `json:"base_width"` - BaseHeight int `json:"base_height"` - - SubFPS string `json:"sub_fps"` - ONVIF string `json:"onvif,omitempty" bson:"onvif"` - ONVIFXAddr string `json:"onvif_xaddr" bson:"onvif_xaddr"` - ONVIFUsername string `json:"onvif_username" bson:"onvif_username"` - ONVIFPassword string `json:"onvif_password" bson:"onvif_password"` - SPSNALUs [][]byte `json:"sps_nalus,omitempty" bson:"sps_nalus,omitempty"` - PPSNALUs [][]byte `json:"pps_nalus,omitempty" bson:"pps_nalus,omitempty"` - VPSNALUs [][]byte `json:"vps_nalus,omitempty" bson:"vps_nalus,omitempty"` - SampleRate int `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"` - Channels int `json:"channels,omitempty" bson:"channels,omitempty"` -} - -// USBCamera configuration, such as the device path (/dev/video*) -type USBCamera struct { - Device string `json:"device"` -} - -// RaspiCamera configuration, such as the device path (/dev/video*) -type RaspiCamera struct { - Device string `json:"device"` -} - -// Region specifies the type (Id) of Region Of Interest (ROI), you -// would like to use. -type Region struct { - Name string `json:"name"` - Rectangle Rectangle `json:"rectangle"` - Polygon []Polygon `json:"polygon"` -} - -// Rectangle is defined by a starting point, left top (x1,y1) and end point (x2,y2). -type Rectangle struct { - X1 int `json:"x1"` - Y1 int `json:"y1"` - X2 int `json:"x2"` - Y2 int `json:"y2"` -} - -// Polygon is a sequence of coordinates (x,y). The ID specifies an unique identifier, -// as multiple polygons can be defined. -type Polygon struct { - ID string `json:"id"` - Coordinates []Coordinate `json:"coordinates"` -} - -// Coordinate belongs to a Polygon. -type Coordinate struct { - X float64 `json:"x"` - Y float64 `json:"y"` -} - -// Timetable allows you to set a Time Of Intterest (TOI), which limits recording or -// detection to a predefined time interval. Two tracks can be set, which allows you -// to give some flexibility. -type Timetable struct { - Start1 int `json:"start1"` - End1 int `json:"end1"` - Start2 int `json:"start2"` - End2 int `json:"end2"` -} - -// S3 integration -type S3 struct { - Proxy string `json:"proxy,omitempty" bson:"proxy,omitempty"` - ProxyURI string `json:"proxyuri,omitempty" bson:"proxyuri,omitempty"` - Bucket string `json:"bucket,omitempty" bson:"bucket,omitempty"` - Region string `json:"region,omitempty" bson:"region,omitempty"` - Username string `json:"username,omitempty" bson:"username,omitempty"` - Publickey string `json:"publickey,omitempty" bson:"publickey,omitempty"` - Secretkey string `json:"secretkey,omitempty" bson:"secretkey,omitempty"` -} - -// KStorage contains the credentials of the Kerberos Storage/Kerberos Cloud instance. -// By defining KStorage you can make your recordings available in the cloud, at a centrel place. -type KStorage struct { - URI string `json:"uri,omitempty" bson:"uri,omitempty"` - CloudKey string `json:"cloud_key,omitempty" bson:"cloud_key,omitempty"` /* old way, remove this */ - AccessKey string `json:"access_key,omitempty" bson:"access_key,omitempty"` - SecretAccessKey string `json:"secret_access_key,omitempty" bson:"secret_access_key,omitempty"` - Provider string `json:"provider,omitempty" bson:"provider,omitempty"` - Directory string `json:"directory,omitempty" bson:"directory,omitempty"` - MaxRetries int `json:"max_retries,omitempty" bson:"max_retries,omitempty"` - Timeout int `json:"timeout,omitempty" bson:"timeout,omitempty"` -} - -// Dropbox integration -type Dropbox struct { - AccessToken string `json:"access_token,omitempty" bson:"access_token,omitempty"` - Directory string `json:"directory,omitempty" bson:"directory,omitempty"` -} - -// Encryption -type Encryption struct { - Enabled string `json:"enabled" bson:"enabled"` - Recordings string `json:"recordings" bson:"recordings"` - Fingerprint string `json:"fingerprint" bson:"fingerprint"` - PrivateKey string `json:"private_key" bson:"private_key"` - SymmetricKey string `json:"symmetric_key" bson:"symmetric_key"` -} - -// Signing -type Signing struct { - Enabled string `json:"enabled" bson:"enabled"` - PrivateKey string `json:"private_key" bson:"private_key"` -} diff --git a/machinery/src/models/MQTT.go b/machinery/src/models/MQTT.go deleted file mode 100644 index ded066e..0000000 --- a/machinery/src/models/MQTT.go +++ /dev/null @@ -1,201 +0,0 @@ -package models - -import ( - "crypto/rsa" - "crypto/x509" - "encoding/base64" - "encoding/json" - "encoding/pem" - "io" - "strings" - "time" - - "github.com/gofrs/uuid" - "github.com/kerberos-io/agent/machinery/src/encryption" - "github.com/kerberos-io/agent/machinery/src/log" -) - -func PackageMQTTMessage(configuration *Configuration, msg Message) ([]byte, error) { - // Create a Version 4 UUID. - u2, err := uuid.NewV4() - if err != nil { - log.Log.Error("failed to generate UUID: " + err.Error()) - } - - // We'll generate an unique id, and encrypt / decrypt it using the private key if available. - msg.Mid = u2.String() - msg.DeviceId = msg.Payload.DeviceId - msg.Timestamp = time.Now().Unix() - - // Configuration - config := configuration.Config - - // Next to hiding the message, we can also encrypt it using your own private key. - // Which is not stored in a remote environment (hence you are the only one owning it). - msg.Encrypted = false - if config.Encryption != nil && config.Encryption.Enabled == "true" { - msg.Encrypted = true - } - msg.PublicKey = "" - msg.Fingerprint = "" - - if msg.Encrypted { - pload := msg.Payload - - // Pload to base64 - data, err := json.Marshal(pload) - if err != nil { - log.Log.Error("models.mqtt.PackageMQTTMessage(): failed to marshal payload: " + err.Error()) - } - - // Encrypt the value - privateKey := configuration.Config.Encryption.PrivateKey - r := strings.NewReader(privateKey) - pemBytes, _ := io.ReadAll(r) - block, _ := pem.Decode(pemBytes) - if block == nil { - log.Log.Error("models.mqtt.PackageMQTTMessage(): error decoding PEM block containing private key") - } else { - // Parse private key - b := block.Bytes - key, err := x509.ParsePKCS8PrivateKey(b) - if err != nil { - log.Log.Error("models.mqtt.PackageMQTTMessage(): error parsing private key: " + err.Error()) - } - - // Conver key to *rsa.PrivateKey - rsaKey, _ := key.(*rsa.PrivateKey) - - // Create a 16bit key random - if config.Encryption != nil && config.Encryption.SymmetricKey != "" { - k := config.Encryption.SymmetricKey - encryptedValue, err := encryption.AesEncrypt(data, k) - if err == nil { - - data := base64.StdEncoding.EncodeToString(encryptedValue) - // Sign the encrypted value - signature, err := encryption.SignWithPrivateKey([]byte(data), rsaKey) - if err == nil { - base64Signature := base64.StdEncoding.EncodeToString(signature) - msg.Payload.EncryptedValue = data - msg.Payload.Signature = base64Signature - msg.Payload.Value = make(map[string]interface{}) - } - } - } - } - } - - // We'll hide the message (by default in latest version) - // We will encrypt using the Kerberos Hub private key if set. - msg.Hidden = false - if config.HubEncryption == "true" && config.HubPrivateKey != "" { - msg.Hidden = true - } - - if msg.Hidden { - pload := msg.Payload - // Pload to base64 - data, err := json.Marshal(pload) - if err != nil { - msg.Hidden = false - } else { - k := config.HubPrivateKey - encryptedValue, err := encryption.AesEncrypt(data, k) - if err == nil { - data := base64.StdEncoding.EncodeToString(encryptedValue) - msg.Payload.HiddenValue = data - msg.Payload.EncryptedValue = "" - msg.Payload.Signature = "" - msg.Payload.Value = make(map[string]interface{}) - } - } - } - - payload, err := json.Marshal(msg) - return payload, err -} - -// The message structure which is used to send over -// and receive messages from the MQTT broker -type Message struct { - Mid string `json:"mid"` - DeviceId string `json:"device_id"` - Timestamp int64 `json:"timestamp"` - Encrypted bool `json:"encrypted"` - Hidden bool `json:"hidden"` - PublicKey string `json:"public_key"` - Fingerprint string `json:"fingerprint"` - Payload Payload `json:"payload"` -} - -// The payload structure which is used to send over -// and receive messages from the MQTT broker -type Payload struct { - Version string `json:"version"` // Version of the message, e.g. "1.0" - Action string `json:"action"` - DeviceId string `json:"device_id"` - Signature string `json:"signature"` - EncryptedValue string `json:"encrypted_value"` - HiddenValue string `json:"hidden_value"` - Value map[string]interface{} `json:"value"` -} - -// We received a audio input -type AudioPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp of the recording request. - Data []int16 `json:"data"` -} - -// We received a recording request, we'll send it to the motion handler. -type RecordPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp of the recording request. -} - -// We received a preset position request, we'll request it through onvif and send it back. -type PTZPositionPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp of the preset request. -} - -// We received a request config request, we'll fetch the current config and send it back. -type RequestConfigPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp of the preset request. -} - -// We received a update config request, we'll update the current config and send a confirmation back. -type UpdateConfigPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp of the preset request. - Config Config `json:"config"` -} - -// We received a request SD stream request -type RequestSDStreamPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp -} - -// We received a request HD stream request -type RequestHDStreamPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp - HubKey string `json:"hub_key"` // hub key - SessionID string `json:"session_id"` // session id - SessionDescription string `json:"session_description"` // session description -} - -// We received a receive HD candidates request -type ReceiveHDCandidatesPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp - SessionID string `json:"session_id"` // session id - Candidate string `json:"candidate"` // candidate -} - -type NavigatePTZPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp - DeviceId string `json:"device_id"` // device id - Action string `json:"action"` // action -} - -type TriggerRelay struct { - Timestamp int64 `json:"timestamp"` // timestamp - DeviceId string `json:"device_id"` // device id - Token string `json:"token"` // token -} diff --git a/machinery/src/models/Media.go b/machinery/src/models/Media.go deleted file mode 100644 index 8efb414..0000000 --- a/machinery/src/models/Media.go +++ /dev/null @@ -1,18 +0,0 @@ -package models - -type Media struct { - Key string `json:"key"` - Path string `json:"path"` - Day string `json:"day"` - ShortDay string `json:"short_day"` - Time string `json:"time"` - Timestamp string `json:"timestamp"` - CameraName string `json:"camera_name"` - CameraKey string `json:"camera_key"` -} - -type EventFilter struct { - TimestampOffsetStart int64 `json:"timestamp_offset_start"` - TimestampOffsetEnd int64 `json:"timestamp_offset_end"` - NumberOfElements int `json:"number_of_elements"` -} diff --git a/machinery/src/models/Onvif.go b/machinery/src/models/Onvif.go deleted file mode 100644 index 5ecb117..0000000 --- a/machinery/src/models/Onvif.go +++ /dev/null @@ -1,24 +0,0 @@ -package models - -type OnvifAction struct { - Action string `json:"action" bson:"action"` - Payload interface{} `json:"payload" bson:"payload"` -} - -type OnvifActionPTZ struct { - Left int `json:"left" bson:"left"` - Right int `json:"right" bson:"right"` - Up int `json:"up" bson:"up"` - Down int `json:"down" bson:"down"` - Center int `json:"center" bson:"center"` - Zoom float64 `json:"zoom" bson:"zoom"` - X float64 `json:"x" bson:"x"` - Y float64 `json:"y" bson:"y"` - Z float64 `json:"z" bson:"z"` - Preset string `json:"preset" bson:"preset"` -} - -type OnvifActionPreset struct { - Name string `json:"name" bson:"name"` - Token string `json:"token" bson:"token"` -} diff --git a/machinery/src/models/System.go b/machinery/src/models/System.go deleted file mode 100644 index e29a54b..0000000 --- a/machinery/src/models/System.go +++ /dev/null @@ -1,17 +0,0 @@ -package models - -type System struct { - CPUId string `json:"cpu_idle" bson:"cpu_idle"` - Hostname string `json:"hostname" bson:"hostname"` - Version string `json:"version" bson:"version"` - Release string `json:"release" bson:"release"` - BootTime uint64 `json:"boot_time" bson:"boot_time"` - KernelVersion string `json:"kernel_version" bson:"kernel_version"` - MACs []string `json:"macs" bson:"macs"` - IPs []string `json:"ips" bson:"ips"` - Architecture string `json:"architecture" bson:"architecture"` - UsedMemory uint64 `json:"used_memory" bson:"used_memory"` - TotalMemory uint64 `json:"total_memory" bson:"total_memory"` - FreeMemory uint64 `json:"free_memory" bson:"free_memory"` - ProcessUsedMemory uint64 `json:"process_used_memory" bson:"process_used_memory"` -} diff --git a/machinery/src/models/User.go b/machinery/src/models/User.go deleted file mode 100644 index 1fd776c..0000000 --- a/machinery/src/models/User.go +++ /dev/null @@ -1,22 +0,0 @@ -package models - -type User struct { - Installed bool `json:"installed" bson:"installed"` - Username string `json:"username" bson:"username"` - Password string `json:"password" bson:"password"` - Role string `json:"role" bson:"role"` - Language string `json:"language" bson:"language"` -} - -type Authentication struct { - Username string `json:"username" bson:"username"` - Password string `json:"password" bson:"password"` -} - -type Authorization struct { - Code int `json:"code" bson:"code"` - Token string `json:"token" bson:"token"` - Expire string `json:"expire" bson:"expire"` - Username string `json:"username" bson:"username"` - Role string `json:"role" bson:"role"` -} diff --git a/machinery/src/models/WebRTC.go b/machinery/src/models/WebRTC.go deleted file mode 100644 index 32e1ff3..0000000 --- a/machinery/src/models/WebRTC.go +++ /dev/null @@ -1,13 +0,0 @@ -package models - -type SDPPayload struct { - Cuuid string `json:"cuuid"` - Sdp string `json:"sdp"` - CloudKey string `json:"cloud_key"` -} - -type Candidate struct { - Cuuid string `json:"cuuid"` - CloudKey string `json:"cloud_key"` - Candidate string `json:"candidate"` -} diff --git a/machinery/src/routers/http/Cors.go b/machinery/src/routers/http/Cors.go deleted file mode 100644 index 1f861b3..0000000 --- a/machinery/src/routers/http/Cors.go +++ /dev/null @@ -1,19 +0,0 @@ -package http - -import ( - "github.com/gin-contrib/cors" - "github.com/gin-gonic/gin" - "time" -) - -func CORS() gin.HandlerFunc { - c := cors.New(cors.Config{ - AllowOrigins: []string{"*"}, - AllowMethods: []string{"GET", "POST", "PUT", "PATCH", "DELETE"}, - AllowHeaders: []string{"Origin", "Content-Type", "Authorization"}, - ExposeHeaders: []string{"Content-Length"}, - AllowCredentials: true, - MaxAge: 12 * time.Hour, - }) - return c -} diff --git a/machinery/src/routers/http/Server.go b/machinery/src/routers/http/Server.go deleted file mode 100644 index 4987401..0000000 --- a/machinery/src/routers/http/Server.go +++ /dev/null @@ -1,144 +0,0 @@ -package http - -import ( - "io" - "os" - "strconv" - - jwt "github.com/appleboy/gin-jwt/v2" - "github.com/gin-contrib/pprof" - "github.com/gin-gonic/contrib/static" - "github.com/gin-gonic/gin" - - //Swagger documentantion - "log" - - _ "github.com/kerberos-io/agent/machinery/docs" - "github.com/kerberos-io/agent/machinery/src/capture" - "github.com/kerberos-io/agent/machinery/src/encryption" - "github.com/kerberos-io/agent/machinery/src/models" - swaggerFiles "github.com/swaggo/files" - ginSwagger "github.com/swaggo/gin-swagger" -) - -// @title Swagger Kerberos Agent API -// @version 1.0 -// @description This is the API for using and configure Kerberos Agent. -// @termsOfService https://kerberos.io - -// @contact.name API Support -// @contact.url https://www.kerberos.io -// @contact.email support@kerberos.io - -// @license.name Apache 2.0 - Commons Clause -// @license.url http://www.apache.org/licenses/LICENSE-2.0.html - -// @BasePath / - -// @securityDefinitions.apikey Bearer -// @in header -// @name Authorization - -func StartServer(configDirectory string, configuration *models.Configuration, communication *models.Communication, captureDevice *capture.Capture) { - - // Set release mode - gin.SetMode(gin.ReleaseMode) - - // Initialize REST API - r := gin.Default() - - // Profiler - pprof.Register(r) - - // Setup CORS - r.Use(CORS()) - - // Add Swagger - r.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) - - // The JWT middlewareergreggre - middleWare := JWTMiddleWare() - authMiddleware, err := jwt.New(&middleWare) - if err != nil { - log.Fatal("JWT Error:" + err.Error()) - } - - // Add all routes - AddRoutes(r, authMiddleware, configDirectory, configuration, communication, captureDevice) - - // Update environment variables - environmentVariables := configDirectory + "/www/env.js" - if os.Getenv("AGENT_MODE") == "demo" { - demoEnvironmentVariables := configDirectory + "/www/env.demo.js" - // Move demo environment variables to environment variables - err := os.Rename(demoEnvironmentVariables, environmentVariables) - if err != nil { - log.Fatal(err) - } - } - - // Add static routes to UI - r.Use(static.Serve("/", static.LocalFile(configDirectory+"/www", true))) - r.Use(static.Serve("/dashboard", static.LocalFile(configDirectory+"/www", true))) - r.Use(static.Serve("/media", static.LocalFile(configDirectory+"/www", true))) - r.Use(static.Serve("/settings", static.LocalFile(configDirectory+"/www", true))) - r.Use(static.Serve("/login", static.LocalFile(configDirectory+"/www", true))) - r.Handle("GET", "/file/*filepath", func(c *gin.Context) { - Files(c, configDirectory, configuration) - }) - - // Run the api on port - err = r.Run(":" + configuration.Port) - if err != nil { - log.Fatal(err) - } -} - -func Files(c *gin.Context, configDirectory string, configuration *models.Configuration) { - - // Get File - filePath := configDirectory + "/data/recordings" + c.Param("filepath") - _, err := os.Open(filePath) - if err != nil { - c.JSON(404, gin.H{"error": "File not found"}) - return - } - - contents, err := os.ReadFile(filePath) - if err == nil { - - // Get symmetric key - symmetricKey := configuration.Config.Encryption.SymmetricKey - encryptedRecordings := configuration.Config.Encryption.Recordings - // Decrypt file - if encryptedRecordings == "true" && symmetricKey != "" { - - // Read file - if err != nil { - c.JSON(404, gin.H{"error": "File not found"}) - return - } - - // Decrypt file - contents, err = encryption.AesDecrypt(contents, symmetricKey) - if err != nil { - c.JSON(404, gin.H{"error": "File not found"}) - return - } - } - - // Get fileSize from contents - fileSize := len(contents) - - // Send file to gin - c.Header("Access-Control-Allow-Origin", "*") - c.Header("Content-Disposition", "attachment; filename="+filePath) - c.Header("Content-Type", "video/mp4") - c.Header("Content-Length", strconv.Itoa(fileSize)) - // Send contents to gin - io.WriteString(c.Writer, string(contents)) - } else { - c.JSON(404, gin.H{"error": "File not found"}) - return - } -}