404Forbiden

✏️ Editing: dem_plugin.py

import agent_util
import ipc_client

from datetime import datetime
import json
import logging
import sys

class DEMPlugin(agent_util.Plugin):
    textkey = "dem"
    label = "Digital Experience"
    log = logging.getLogger(__name__)
    wifi_client = ipc_client.DEMClient(dem_port="demservice")

@classmethod
    def get_metadata(self, config):
        status = agent_util.SUPPORTED
        msg = None

dem_configured = True
        if config.get("enabled", "false").lower() != "true":
            status = agent_util.UNSUPPORTED
            msg = "DEM not configured"
            dem_configured = False

if dem_configured:
            info = self.wifi_client.get_dem_wifi_info()
            if not info:
                status = agent_util.UNSUPPORTED
                msg = "No wifi info found"

metadata = {
            "dem.agrCtlRSSI": {
                "label": "Wifi signal strength",
                "options": None,
                "status": status,
                "error_msg": msg,
                "unit": "dBm",
            },
            "dem.downloadspeed": {
                "label": "Download Speed",
                "options": None,
                "status": agent_util.SUPPORTED,
                "error_msg": None,
                "unit": "Mbit/s",
            },
            "dem.lastTxRate": {
                "label": "Last transmission rate",
                "options": None,
                "status": status,
                "error_msg": msg,
                "unit": "mbps/s",
            },
        }
        if "darwin" == sys.platform:
            metadata["dem.agrCtlNoise"] = {
                "label": "Noise",
                "options": None,
                "status": status,
                "error_msg": msg,
                "unit": "dBm",
            }
            metadata["dem.MCS"] = {
                "label": "MCS Index",
                "options": None,
                "status": status,
                "error_msg": msg,
                "unit": "index",
            }

speedtest_upload = agent_util.SUPPORTED
        speedtest_msg = None
        if (
            config.get("speedtest_mode", "speedtest").lower() != "iperf3"
            or not dem_configured
            or "darwin" != sys.platform
        ):
            speedtest_upload = agent_util.UNSUPPORTED
            speedtest_msg = "Upload speed not supported"

metadata["dem.uploadspeed"] = {
            "label": "Upload Speed",
            "options": None,
            "status": speedtest_upload,
            "error_msg": speedtest_msg,
            "unit": "Mbit/s",
        }

battery_status = agent_util.SUPPORTED
        battery_msg = None
        if not dem_configured:
            battery_status = agent_util.UNSUPPORTED
            battery_msg = "DEM not configured"
        else:
            battery_pct = self.get_battery_remaining()
            if battery_pct is None:
                battery_status = agent_util.UNSUPPORTED
                battery_msg = "Battery metric unavailable"
        metadata["dem.battery_percent_remaining"] = {
            "label": "Battery Percent Remaining",
            "options": None,
            "status": battery_status,
            "error_msg": battery_msg,
            "unit": "percent",
        }
        return metadata

def check(self, textkey, data, config):
        if config.get("enabled", "false").lower() != "true":
            return None
        if textkey in ("dem.downloadspeed", "dem.uploadspeed"):
            return self._measureNetworkSpeed(textkey, config)

if "dem.battery_percent_remaining" == textkey:
            return self.get_battery_remaining()

try:
            info = self.wifi_client.get_dem_wifi_info()
            if not info:
                raise Exception("No wifi info received")
            key = textkey[len("dem.") :]
            metric_value = info.get(key, None)
            if metric_value is None:
                raise Exception("Missing key {}".format(key))
            return float(metric_value)
        except Exception as e:
            self.log.error("Wifi metrics error: {}".format(str(e)))
            return None

def _run_iperf3_test(self, textkey, config):
        speedtest_bin = "/usr/local/FortiMonitor/agent/latest/bin/iperf3"
        try:
            start_url = config.get("iperf3_start_url", None)
            from iperf3 import Iperf3Runner

# Old configuration included an start http request. No longer needed.
            start_url = start_url.removeprefix("https://").removesuffix("/start_iperf")
            runner = Iperf3Runner(iperf3_bin=speedtest_bin, url=start_url)
            result = None
            if "dem.downloadspeed" == textkey:
                result = runner.get_download_speed()
            elif "dem.uploadspeed" == textkey:
                result = runner.get_upload_speed()
            return float(result / (1000 * 1000))
        except:
            self.log.exception("Iperf3 error:")

def _measureNetworkSpeed(self, textkey, config):
        if config.get("speedtest_mode", "speedtest").lower() == "iperf3":
            return self._run_iperf3_test(textkey, config)

if "dem.uploadspeed" == textkey:
            raise Exception("Service does not support upload speed")
        try:
            import speedtest

start = datetime.now()
            s = speedtest.Speedtest()
            s.get_best_server()
            k = s.download()
            rv = float(k / 1000000)
            self.log.info(
                "Download speed {} in {:.2f}".format(
                    rv, (datetime.now() - start).total_seconds()
                )
            )
            return rv
        except Exception as ste:
            self.log.error("Speedtest exception: {}".format(ste))
            return None

@classmethod
    def get_battery_remaining(self):
        if "darwin" == sys.platform:
            try:
                sp = agent_util.which("system_profiler")
                power_tk = "SPPowerDataType"
                output = agent_util.run_command([sp, "-json", power_tk])
                data = json.loads(output)
                for d in data[power_tk]:
                    if d.get("_name", "") == "spbattery_information":
                        mv = d.get("sppower_battery_charge_info", None)
                        if not mv:
                            return None
                        return float(mv["sppower_battery_state_of_charge"])

return None
            except Exception as e:
                self.log.exception(e)
                return None
        elif "linux" == sys.platform:
            return self.get_battery_percent_linux()
        else:
            return None

@classmethod
    def get_battery_percent_linux(self):
        try:
            proc_args = ["upower", "-i", "/org/freedesktop/UPower/devices/battery_BAT0"]
            output = agent_util.run_command(proc_args).strip().splitlines()
            seen_battery = False
            seen_present = False
            percentage = None
            for line in output:
                line = line.strip()
                if "battery" == line:
                    seen_battery = True
                elif "present:" in line:
                    seen_present = True
                elif "percentage:" in line:
                    percentage = line.split(":")[1].strip().rstrip("%")
                    break

if seen_battery and seen_present:
                return float(percentage)

return None

except Exception as e:
            self.log.exception(e)
            return None

Sid Gifari From Gifari Industries - BD Cyber Security Team