404Forbiden

✏️ Editing: ipc_client.py

import json
import logging
import os

class IPCClient:
    def __init__(self, port, ipc_path=None):
        self.ipc_path = ipc_path
        if self.ipc_path is None:
            self.ipc_path = "/tmp/com.fortinet.fortimonitor"
        self.port = port
        if self.port is None:
            raise Exception("Missing port")
        self.log = logging.getLogger("ipc_client")

def send_receive(self, command, payload=None):
        import socket

if payload:
            command = "{}:{}".format(command, payload)
        command += "\n"
        # self.log.info('Send {}'.format(command))
        try:
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
                sock.settimeout(10.0)
                sock.connect(os.path.join(self.ipc_path, self.port))
                toSend = command.encode("utf-8")

sock.sendall(toSend)
                receivedBytes = bytes()
                while True:
                    r = sock.recv(1024)
                    if self.log.isEnabledFor(logging.DEBUG):
                        self.log.info("  received {} bytes".format(len(r)))
                    if 0 == len(r):
                        break
                    receivedBytes += r
                # self.log.info('Send {} bytes: received {} bytes'.format(len(toSend), len(receivedBytes)))
                received = receivedBytes.decode("utf-8")
                if "nack" == received:
                    return None
                return received
        except Exception as e:
            self.log.error("Send/recv failure: {}".format(e))
            return None

class DEMClient(IPCClient):
    def __init__(self, dem_port, ipc_path=None):
        super(DEMClient, self).__init__(dem_port, ipc_path)
        self.log = logging.getLogger("dem_client")

def get_dem_wifi_info(self):
        response = self.send_receive("wifi-info")
        try:
            if not response:
                raise ValueError("No payload recorded")
            return json.loads(response)
        except Exception as err:
            logging.warning(f"Unable to communicate with service {err}")
            return None

def get_dem_results(self):
        rv = {}
        response = self.send_receive("collect")
        if response is None:
            return rv

return json.loads(response)

def init_dem_schedules(self, schedules):
        try:
            """
            schedules = {
                'icmp_server_resources' : response.get('icmp_server_resources', []),
                'monitor_schedules' : response.get('monitor_schedules', []),
                'traceroute_checks' : response.get('traceroute_checks', [])
            }
            """
            self.send_receive("initSchedules", payload=json.dumps(schedules))
        except Exception as ex:
            self.log.error("init_schedules error: {}".format(str(ex)))

return

def update_schedules(self, newSchedules):
        _ = self.send_receive("update-schedules", payload=json.dumps(newSchedules))

Sid Gifari From Gifari Industries - BD Cyber Security Team