Source code for tuoni.TuoniListener

from tuoni.TuoniExceptions import *


[docs]
class TuoniListener:
    """
    A class that provides data and functionality for a created listener.

    Attributes:
        listener_id (int): The unique identifier of the listener.
        name (str): The name of the listener.
        info (str): Information about the listener.
        status (str): The current status of the listener.
        plugin (str): The plugin associated with the listener.
        configuration (dict): The configuration settings for the listener.
    """

    def __init__(self, conf, c2):
        """
        Constructor for the listener class.

        Args:
            conf (dict): Data from the server.
            c2 (TuoniC2): The related server object that manages communication.
        """
        self._load_conf(conf)
        self.c2 = c2

    def _load_conf(self, conf):
        self.listener_id = conf["id"]
        self.name = conf["name"]
        self.info = conf["info"]
        self.status = conf["status"]
        self.plugin = conf["plugin"]
        self.configuration = conf["configuration"]
        self.configuration_files = conf.get("configurationFiles", [])



[docs]
    def stop(self):
        """
        Stop the listener.
        """
        if self.listener_id is None:
            raise ExceptionTuoniDeleted("")
        data = self.c2.request_put(f"/api/v1/listeners/{self.listener_id}/stop")
        self._load_conf(data)



[docs]
    def start(self):
        """
        Start the listener.
        """
        if self.listener_id is None:
            raise ExceptionTuoniDeleted("")
        data = self.c2.request_put(f"/api/v1/listeners/{self.listener_id}/start")
        self._load_conf(data)



[docs]
    def delete(self):
        """
        Delete the listener.
        """
        if self.listener_id is None:
            raise ExceptionTuoniDeleted("")
        self.c2.request_delete(f"/api/v1/listeners/{self.listener_id}")
        self.listener_id = None



[docs]
    def reload(self):
        """
        Reload the listener data from the C2 server.
        """
        if self.listener_id is None:
            raise ExceptionTuoniDeleted("")
        data = self.c2.request_get(f"/api/v1/listeners/{self.listener_id}")
        self._load_conf(data)



[docs]
    def update(self):
        """
        Update listener on the server.

        Sends the current ``name`` and ``configuration``. ``configurationFiles`` is
        only sent when explicitly set on this object, because the server treats a
        present (even empty) list as "replace all" — and ``ListenerResponse`` does
        not echo the field, so a naive round-trip would wipe server-side files.
        """
        if self.listener_id is None:
            raise ExceptionTuoniDeleted("")
        req = {
            "configuration": self.configuration,
            "name": self.name,
        }
        if self.configuration_files:
            req["configurationFiles"] = self.configuration_files
        self.c2.request_patch(f"/api/v1/listeners/{self.listener_id}", req)
        self.reload()