Syslog Sender

import demistomock as demisto
from CommonServerPython import *
from CommonServerUserPython import *

""" IMPORTS """

from contextlib import contextmanager
from logging.handlers import SysLogHandler
from logging import Logger, getLogger, INFO, DEBUG, WARNING, ERROR, CRITICAL
from socket import SOCK_STREAM
from collections.abc import Generator
from tempfile import NamedTemporaryFile
from rfc5424logging import Rfc5424SysLogHandler
import socket
import ssl

""" CONSTANTS """

PLAYGROUND_INVESTIGATION_TYPE = 9
INCIDENT_OPENED = "incidentOpened"
LOGGING_LEVEL_DICT = {"LOG_INFO": INFO, "LOG_DEBUG": DEBUG, "LOG_WARNING": WARNING, "LOG_ERR": ERROR, "LOG_CRIT": CRITICAL}
FACILITY_DICT = {
    "LOG_AUTH": SysLogHandler.LOG_AUTH,
    "LOG_AUTHPRIV": SysLogHandler.LOG_AUTHPRIV,
    "LOG_CRON": SysLogHandler.LOG_CRON,
    "LOG_DAEMON": SysLogHandler.LOG_DAEMON,
    "LOG_FTP": SysLogHandler.LOG_FTP,
    "LOG_KERN": SysLogHandler.LOG_KERN,
    "LOG_LPR": SysLogHandler.LOG_LPR,
    "LOG_MAIL": SysLogHandler.LOG_MAIL,
    "LOG_NEWS": SysLogHandler.LOG_NEWS,
    "LOG_SYSLOG": SysLogHandler.LOG_SYSLOG,
    "LOG_USER": SysLogHandler.LOG_USER,
    "LOG_UUCP": SysLogHandler.LOG_UUCP,
    "LOG_LOCAL0": SysLogHandler.LOG_LOCAL0,
    "LOG_LOCAL1": SysLogHandler.LOG_LOCAL1,
    "LOG_LOCAL2": SysLogHandler.LOG_LOCAL2,
    "LOG_LOCAL3": SysLogHandler.LOG_LOCAL3,
    "LOG_LOCAL4": SysLogHandler.LOG_LOCAL4,
    "LOG_LOCAL5": SysLogHandler.LOG_LOCAL5,
    "LOG_LOCAL6": SysLogHandler.LOG_LOCAL6,
    "LOG_LOCAL7": SysLogHandler.LOG_LOCAL7,
}
SEVERITY_DICT = {"Unknown": 0, "Low": 1, "Medium": 2, "High": 3, "Critical": 4}

TCP = "tcp"
UDP = "udp"
TLS = "tls"
PROTOCOLS = {TCP, UDP, TLS}
MAX_PORT = 65535
DEFAULT_TCP_SYSLOG_PORT = 514
DEFAULT_TLS_SYSLOG_PORT = 6514

"""SyslogHandlerTLS"""


class SyslogHandlerTLS(logging.Handler):
    def __init__(self, address: str, port: int, log_level: int, facility: int, cert_path: str, if_self_sign_cert: bool):
        """
        Initialize a handler.
        """
        logging.Handler.__init__(self)
        self.address = address
        self.port = port
        self.certfile = cert_path
        self.facility = facility
        self.level = log_level
        # Create a TCP socket
        ssl_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # Wrap the socket with SSL
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
        # In order to allow self signed certificate:
        if if_self_sign_cert:
            ssl_context.check_hostname = False
            ssl_context.verify_mode = ssl.CERT_NONE
        else:
            # Only load certificate for verification if not using self-signed cert
            ssl_context.load_verify_locations(self.certfile)
        ssl_sock = ssl_context.wrap_socket(ssl_sock, server_hostname=self.address)
        self.socket = ssl_sock
        try:
            self.socket.connect((self.address, self.port))
        except OSError as exc:
            if ssl_sock:
                ssl_sock.close()
            raise DemistoException(str(exc))

    def emit(self, record):
        """
        Emit a record.

        The record is formatted, and then sent to the syslog server. If
        exception information is present, it is NOT sent to the server.
        """
        ident = ""  # prepended to all messages
        try:
            msg = self.format(record)
            if ident:
                msg = ident + msg

            # Calculate the priority value
            priority = (self.facility << 3) | self.level
            # Construct the syslog message in RFC 5424 format
            syslog_message = "<{priority}>1 {timestamp} {hostname} {appname} {procid} {msgid} - {message}\n".format(
                priority=priority,
                timestamp=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                hostname=socket.gethostname(),
                appname=record.name,
                procid=os.getpid(),
                msgid="-",
                message=self.format(record),
            )
            # Connect to the syslog server
            self.socket.send(syslog_message.encode("utf-8"))

        except Exception as e:
            if self.socket:
                self.socket.close()
            demisto.error(str(e))


""" Syslog Manager """


class SyslogManager:
    def __init__(
        self,
        address: str,
        port: int,
        protocol: str,
        logging_level: int,
        facility: int,
        cert_path: str | None,
        self_signed_certificate: bool,
    ):
        """
        Class for managing instances of a syslog logger.
        :param address: The IP address of the syslog server.
        :param port: The port of the syslog server.
        :param protocol: The messaging protocol (TCP / UDP / TLS).
        :param logging_level: The logging level.
        """
        self.address = address
        self.port = port
        self.protocol = protocol
        self.logging_level = logging_level
        self.facility = facility
        self.syslog_cert_path = cert_path
        self.self_signed_cert = self_signed_certificate

    @contextmanager  # type: ignore[misc, arg-type]
    def get_logger(self) -> Generator:
        """
        Get a new instance of a syslog logger.
        :return: syslog logger
        """
        if self.protocol == TLS and self.syslog_cert_path:
            demisto.debug("creating tls logger handler")
            handler = self.init_handler_tls(self.syslog_cert_path)
        else:
            demisto.debug("creating tcp/udp logger handler")
            handler = self._get_handler()
        syslog_logger = self._init_logger(handler)
        demisto.debug("logger was created ")
        try:
            yield syslog_logger
        finally:
            syslog_logger.removeHandler(handler)
            handler.close()

    def _get_handler(self) -> Rfc5424SysLogHandler:
        sock_kind = SOCK_STREAM if self.protocol == TCP else socket.SOCK_DGRAM
        return Rfc5424SysLogHandler(
            address=(self.address, self.port), facility=self.facility, socktype=sock_kind, utc_timestamp=True
        )

    def init_handler_tls(self, certfile: str):
        return SyslogHandlerTLS(
            address=self.address,
            port=self.port,
            cert_path=certfile,
            facility=self.facility,
            log_level=self.logging_level,
            if_self_sign_cert=self.self_signed_cert,
        )

    def _init_logger(self, handler: Rfc5424SysLogHandler | SyslogHandlerTLS) -> Logger:
        """
        Initialize a logger with a syslog handler.
        :param handler: A syslog handler
        :return: A syslog logger
        """
        syslog_logger = getLogger("SysLogLogger")
        syslog_logger.setLevel(self.logging_level)
        syslog_logger.addHandler(handler)
        return syslog_logger


""" HELPER FUNCTIONS """


def prepare_certificate_file(certificate: str) -> str:
    """
    Prepares the certificate file and key for ssl connection.
    Args:
        certificate (str): Certificate. For SSL connection.
    Returns:
        (str, str): certificate_path.
    """
    certificate_file = NamedTemporaryFile(delete=False)
    certificate_path = certificate_file.name
    certificate_file.write(bytes(certificate, "utf-8"))
    certificate_file.close()
    demisto.debug("Successfully preparing the certificate")
    return certificate_path


def init_manager(params: dict) -> SyslogManager:
    """
    Create a syslog manager instance according to provided parameters.
    :param params: Parameters for the syslog manager.
    :return: syslog manager
    """
    address = params.get("address")
    protocol = params.get("protocol", UDP).lower()
    facility = FACILITY_DICT.get(params.get("facility", "LOG_SYSLOG"), SysLogHandler.LOG_SYSLOG)
    logging_level = LOGGING_LEVEL_DICT.get(params.get("priority", "LOG_INFO"), INFO)
    certificate: Optional[str] = replace_spaces_in_credential(params.get("certificate", {}).get("password")) or params.get(
        "certificate", None
    )
    certificate_path: Optional[str] = None
    default_port: int = DEFAULT_TLS_SYSLOG_PORT if protocol == "tls" else DEFAULT_TCP_SYSLOG_PORT
    port = arg_to_number(params.get("port"), required=False) or default_port
    self_signed_certificate = params.get("self_signed_certificate", False)
    if not address:
        raise DemistoException("A address must be provided.")
    if port and (port < 0 or port > MAX_PORT):
        raise DemistoException(f"Given port: {port} is not valid and must be between 0-{MAX_PORT}")
    if protocol == "tls" and not certificate:
        raise DemistoException("A certificate must be provided in TLS protocol.")
    if certificate and protocol == "tls":
        certificate_path = prepare_certificate_file(certificate)
    return SyslogManager(address, port, protocol, logging_level, facility, certificate_path, self_signed_certificate)


def send_log(manager: SyslogManager, message: str, log_level: str):
    """
    Use a syslog manager to get a logger and send a message to syslog.
    :param manager: The syslog manager
    :param message: The message to send
    :param log_level: The logging level
    """
    with manager.get_logger() as syslog_logger:  # type: Logger
        if log_level == "DEBUG":
            syslog_logger.debug(message)
        if log_level == "INFO":
            syslog_logger.info(message)
        if log_level == "WARNING":
            syslog_logger.warning(message)
        if log_level == "ERROR":
            syslog_logger.error(message)
        if log_level == "CRITICAL":
            syslog_logger.critical(message)


def mirror_investigation():
    """
    Update the integration context with a new or existing mirror.
    """
    mirror_type = demisto.args().get("type", "all")

    investigation = demisto.investigation()

    if investigation.get("type") == PLAYGROUND_INVESTIGATION_TYPE:
        return_error("Can not perform this action in the playground.")

    investigation_id = investigation.get("id")

    demisto.mirrorInvestigation(investigation_id, f"{mirror_type}:FromDemisto", False)

    demisto.results("Investigation mirrored to Syslog successfully.")


""" Syslog send command """


def syslog_send_notification(manager: SyslogManager, min_severity: int):
    """
    Send a message to syslog
    :param manager: Syslog manager
    :param min_severity: Minimum severity of incidents to send messages about
    """
    message = demisto.args().get("message", "")
    entry = demisto.args().get("entry")
    ignore_add_url = demisto.args().get("ignoreAddURL", False)
    log_level = demisto.args().get("level", "INFO")
    severity = demisto.args().get("severity")  # From server
    message_type = demisto.args().get("messageType", "")  # From server

    if severity:
        try:
            severity = int(severity)
        except Exception:
            severity = None

    if message_type == INCIDENT_OPENED and (severity is not None and severity < min_severity):
        return

    if not message:
        message = ""

    message = message.replace("\n", " ").replace("\r", " ").replace("`", "")
    investigation = demisto.investigation()
    if investigation:
        investigation_id = investigation.get("id")
        if entry:
            message = f"{entry}, {message}"
        message = f"{investigation_id}, {message}"

    if ignore_add_url and isinstance(ignore_add_url, str):
        ignore_add_url = argToBoolean(ignore_add_url)
    if not ignore_add_url:
        investigation = demisto.investigation()
        server_links = demisto.demistoUrls()
        if investigation:
            if investigation.get("type") != PLAYGROUND_INVESTIGATION_TYPE:
                link = server_links.get("warRoom")
                if link:
                    if entry:
                        link += "/" + entry
                    message += f" {link}"
            else:
                link = server_links.get("server", "")
                if link:
                    message += f" {link}#/home"

    if not message:
        raise ValueError("No message received")

    send_log(manager, message, log_level)

    demisto.results("Message sent to Syslog successfully.")


def syslog_send(manager):
    message = demisto.args().get("message", "")
    log_level = demisto.args().get("level", "INFO")
    send_log(manager, message, log_level)
    demisto.results("Message sent to Syslog successfully.")


""" MAIN """


def main():
    LOG(f"Command being called is {demisto.command()}")
    try:
        if demisto.command() == "test-module":
            syslog_manager = init_manager(demisto.params())
            with syslog_manager.get_logger() as syslog_logger:  # type: Logger
                syslog_logger.info("The connection was successfully established")
            demisto.results("ok")
        elif demisto.command() == "mirror-investigation":
            mirror_investigation()
        elif demisto.command() == "syslog-send":
            if "address" in demisto.args():
                # params provided in the command args
                syslog_manager = init_manager(demisto.args())
            else:
                syslog_manager = init_manager(demisto.params())
            syslog_send(syslog_manager)
        elif demisto.command() == "send-notification":
            min_severity = SEVERITY_DICT.get(demisto.params().get("severity", "Low"), 1)
            syslog_manager = init_manager(demisto.params())
            syslog_send_notification(syslog_manager, min_severity)
    except Exception as e:
        exception_msg = str(e)
        error_message = f"The following error was thrown: {exception_msg} "
        if "PEM lib (_ssl.c:4123)" in exception_msg:
            error_message += (
                "Potential causes could include: "
                "That the certificate is not in the correct format (e.g. it's not in PEM format)- "
                "Make sure to insert the Certificate was insert correctly. "
                "or, The certificate is expired or otherwise invalid"
            )
        elif "CERTIFICATE_VERIFY_FAILED" in exception_msg:
            error_message += (
                "If the certificate is self sign, make sure to check the Self Signed Certificate button."
                "Otherwise, The certificate is not trusted by the system or by the client trying to"
                " establish the connection"
            )
        elif "UnicodeError: label too long" in exception_msg:
            error_message += (
                "\nPotential cause could be too long URL label, which means  there is a part of the"
                " URL between two dots that is longer than 64 chars."
            )
        raise DemistoException(error_message)


if __name__ in ["__main__", "__builtin__", "builtins"]:
    main()