narwhal.py

from twisted.web.server import Site
from twisted.web.static import File
from twisted.internet import endpoints, reactor, task
from pri import PRI

import datetime
import os
import re
import socket
import time

import dateutil.parser
import msgpack
import numpy as np
# import orjson
import json
import redis
import tablib
import zstd
from klein import Klein
from klein import route

VERSION = "0.2"
DASH_LINE = "-----------------------------------------------------------------"

configuration = {
    "SERVER_NAME": "Narwhal",
    "REMOTE_REDIS_HOST": "localhost",
    "REMOTE_REDIS_PORT": 6379,
    "REDIS_MAIN_DB": 0,
    "SYSLOG_CACHE_PROCESS_INTERVAL": 2,
    "SYSLOG_CACHE_DB_NES": 1,
    "COMPRESSION_TYPE": 3,
    "NETWORK_INTERFACE": "0.0.0.0",
    "DASHBOARD_WEB_PORT": 3000,
    "ENABLE_HTTPS": "Y",
    "PRIVATE_KEY": "localhost.pem",
    "SEVERITY_TO_RETURN": "0 1 2 3",
    "DASHBOARD_SHOW_HOURS": 4,
    "DASHBOARD_DATA_REFRESH_SECONDS": 3,
    "ENDPOINT_SYSLOG_TRANSMISSION_INTERVAL_SECONDS": 1,
    "SEVERITY_PATTERN": "sev*"
}

LOG_MESSAGE_START = "Narwhal server started"
LOG_MESSAGE_STOP = "Narwhal server stopped"

# as per RFC 5424: https://tools.ietf.org/html/rfc5424
SEVERITY_CODE = [
    "emerg",
    "alert",
    "crit",
    "err",
    "warning",
    "notice",
    "info",
    "debug"
]

redis_main_db = ""
narwhal_log_facility = []
chart_severity = 0


def display_console_banner():

    base_url = "https://" + \
        configuration["NETWORK_INTERFACE"] + ":" + \
        str(configuration["DASHBOARD_WEB_PORT"])

    print(DASH_LINE)
    print(" Narwhal server v." + VERSION)
    print("Access dashboard at " + base_url)
    print(DASH_LINE)
    return


def load_env_variable(env_variable):
    global configuration
    env_variable_value = os.getenv(env_variable)
    if env_variable_value:
        print(env_variable_value)
        configuration[env_variable] = env_variable_value
    return


def load_configuration():
    global configuration

    try:
        load_env_variable("SERVER_NAME")
        load_env_variable("REMOTE_REDIS_HOST")
        load_env_variable("REMOTE_REDIS_PORT")
        load_env_variable("REDIS_MAIN_DB")
        load_env_variable("SYSLOG_CACHE_PROCESS_INTERVAL")
        load_env_variable("SYSLOG_CACHE_DB_NES")
        load_env_variable("COMPRESSION_TYPE")
        load_env_variable("NETWORK_INTERFACE")
        load_env_variable("DASHBOARD_WEB_PORT")
        load_env_variable("ENABLE_HTTPS")
        load_env_variable("PRIVATE_KEY")
        load_env_variable("SEVERITY_TO_RETURN")
        load_env_variable("DASHBOARD_SHOW_HOURS")
        load_env_variable("DASHBOARD_DATA_REFRESH_SECONDS")
        load_env_variable("ENDPOINT_SYSLOG_TRANSMISSION_INTERVAL_SECONDS")
        load_env_variable("SEVERITY_PATTERN")
    except Exception:
        print(DASH_LINE)
        print("ERROR: Enviroment variable is not available.")
        print(DASH_LINE)
        raise


def redis_connect(redis, database):
    global configuration
    redis_connection = redis.Redis(
        host=configuration["REMOTE_REDIS_HOST"],
        port=configuration["REMOTE_REDIS_PORT"],
        db=database)
    return redis_connection


def decode_syslog_pri(pos):
    line = int(pos)
    if line < 192:
        facility = PRI[line][0]
        severity = PRI[line][1]
    else:
        facility = 999
        severity = 99
    return facility, severity


def narwhal_log(message):
    global narwhal_log_facility
    global configuration
    ip = socket.gethostbyname(socket.gethostname())
    narwhal_log_facility.append(
        [
            datetime.datetime.now().replace(microsecond=0).isoformat(),
            ip,
            5,
            16,
            ip,
            configuration['SERVER_NAME'],
            datetime.datetime.utcnow().isoformat(),
            message
        ])


def syslog_cache_processor(redis_syslog_cache, redis_main_db):

    global narwhal_log_facility

    date_field = ""
    redis_cache_data = np.array([])
    dt = []
    ip = []
    endpoint = []
    raw_message = []
    redis_data = []
    data = np.empty([8])
    data_to_redis = []

    cache_stored_keys = list(map(
        decode_bytes, (redis_syslog_cache.hkeys("raw_message_block"))))

    redis_syslog_cache_pipeline = redis_syslog_cache.pipeline()
    for key in cache_stored_keys:
        redis_syslog_cache_pipeline.hget("raw_message_block", key)

    for compressed_data in redis_syslog_cache_pipeline.execute():
        decompressed_data = zstd.decompress(compressed_data)
        redis_cache_data = msgpack.unpackb(decompressed_data, raw=False)
        for entry in redis_cache_data:
            ip += entry['ip']
            endpoint += entry['ep']
            raw_message += entry['ms']

    for key in cache_stored_keys:
        redis_syslog_cache_pipeline.hdel("raw_message_block", key)

    redis_syslog_cache_pipeline.execute()

    for index in range(len(raw_message)):
        if raw_message[index].startswith("<"):
            facility, severity = decode_syslog_pri(
                raw_message[index].split(">", 1)[0].strip("<"))

        timestamp = re.search(
            r"[A-Z][a-z]{2}\s{1,2}\d{1,2}(?:\s\d{4})?\s\d{2}[:]\d{2}[:]\d{2}(?:\.\d{1,6})?(?:\s[A-Z]{3})?",
            raw_message[index].split(">", 1)[1],
        ).group()

        right_part_of_the_raw_message = raw_message[index].split(
            ">", 1)[1].replace(timestamp, '')
        try:
            system, message = right_part_of_the_raw_message.split(
                ": ", 1)
        except ValueError:
            system = str(right_part_of_the_raw_message)
            message = ""

        new_date_field = timestamp[:16]
        if date_field == "":
            date_field = new_date_field

        if new_date_field != date_field:
            data = np.reshape(data, (-1, 8))
            redis_main_db_pipeline = redis_main_db.pipeline()

            for severity_key in range(8):
                for index in range(len(data)):
                    if (data[index][2] == str(severity_key)):
                        data_to_redis.append(data[index].tolist())

                if data_to_redis:
                    data_compressed = zstd.compress(
                        msgpack.packb(data_to_redis),
                        configuration['COMPRESSION_TYPE'])
                    compressed_records_count = len(data_to_redis)

                    redis_main_db_pipeline.hset(
                        severity_key, date_field, data_compressed)
                    redis_main_db_pipeline.hincrby(
                        severity_key, 'total', compressed_records_count)
                    keys_to_score = {date_field:  compressed_records_count}
                    redis_main_db_pipeline.zadd(
                        (str(severity_key) + "T"), keys_to_score)
                    data_to_redis = []

            redis_main_db_pipeline.execute()
            date_field = new_date_field
            data = np.empty([8])

        data_row = np.array([timestamp[:16],
                             endpoint[index],
                             severity,
                             facility,
                             ip[index],
                             system,
                             timestamp,
                             message])
        data = np.concatenate((data, data_row), axis=0)

    if narwhal_log_facility:
        for log_row in narwhal_log_facility:
            data = np.concatenate((data, log_row), axis=0)
        narwhal_log_facility.clear()

    calculate_statistic(redis_main_db)
    return


def available_severity_keys(redis_connection):

    severity_keys = []
    redis_pipeline = redis_connection.pipeline()
    for index in range(7):
        redis_pipeline.exists(str(index))
    sev_key = 0
    for item in redis_pipeline.execute():
        if item:
            severity_keys.append(str(sev_key))
        sev_key += 1
    return severity_keys


def calculate_statistic(redis_connection):
    messages_by_severity = [0, 0, 0, 0, 0, 0, 0, 0]
    severity_keys = available_severity_keys(redis_connection)
    for severity_key in severity_keys:
        messages_by_severity[int(severity_key)] = int((redis_connection.hget(
            severity_key, 'total')).decode('UTF-8'))
    print(" Messages by severity (1-8):" + str(messages_by_severity) +
          " Total messages: " + str(sum(messages_by_severity)),
          end="\r", flush=True)
    return


def decode_bytes(item):
    return item.decode('UTF-8')


def truncate_timestamp(item):
    return item.decode('UTF-8')[:16]


def truncate_timestamp_for_chart(item):
    return (item[:15] + "0")


def prepare_chart_data(item):
    global chart_severity
    return({"x": item[0].decode('UTF-8'), "y": chart_severity, "z": item[1].decode('UTF-8')})


def prepare_timeline(item):
    return(item[0].decode('UTF-8'))


def respond_to_dashboard_data_request(redis_connection):

    global chart_severity

    dashboard = {}
    dashboard['ChartData0'] = []
    dashboard['ChartData1'] = []
    dashboard['ChartData2'] = []
    dashboard['ChartData3'] = []
    dashboard['ChartData4'] = []
    dashboard['ChartData5'] = []
    dashboard['ChartData6'] = []
    dashboard['ChartData7'] = []
    timeline = []
    events = []
    messages_by_severity = [0, 0, 0, 0, 0, 0, 0, 0]

    severity_keys = available_severity_keys(redis_connection)
    for severity_key in severity_keys:
        chart_severity = severity_key
        events = redis_connection.zrange(
            (str(severity_key)+"T"), 0, -1, withscores=True)
        events = np.atleast_2d(events)
        dashboard[('ChartData' + str(severity_key))
                  ] = list(map(prepare_chart_data, events))
        timeline += list(map(prepare_timeline, events))

        messages_by_severity[int(severity_key)] = int((redis_connection.hget(
            severity_key, "total")).decode('UTF-8'))

    # timeline = np.reshape(timeline, (-1, 2))
    timeline.sort()

    first_timestamp = timeline[0]
    last_timestamp = timeline[-1]

    dashboard["total_events"] = sum(messages_by_severity)
    dashboard["messages_per_second"] = 0
    dashboard["seconds_between_messages"] = 0

    # sev_key = []
    # for severity in range(len(SEVERITY_CODE)):
    # sev_key += SEVERITY_CODE[severity]

    delta = (dateutil.parser.parse(
        last_timestamp) -
        dateutil.parser.parse(first_timestamp))

    dashboard["messages_per_second"] = round(
        (dashboard["total_events"] / delta.total_seconds()), 2
    )
    dashboard["seconds_between_messages"] = round(
        (delta.total_seconds() / dashboard["total_events"]), 2
    )

    dashboard["logAlertCount"] = messages_by_severity[0] + \
        messages_by_severity[1] + \
        messages_by_severity[2] + messages_by_severity[3]
    dashboard["logWarningsCount"] = messages_by_severity[4]
    dashboard["logMessageCount"] = messages_by_severity[5] + \
        messages_by_severity[6] + messages_by_severity[7]

    redis_config = redis_connection.info("memory")

    dashboard["redis_used_memory_human"] = redis_config["used_memory_human"]
    dashboard["redis_used_memory"] = redis_config["used_memory"]
    dashboard["redis_total_system_memory_human"] = redis_config[
        "total_system_memory_human"
    ]
    dashboard["redis_total_system_memory"] = redis_config[
        "total_system_memory"
    ]
    return json.dumps(dashboard)


def respond_to_events_data_request(redis_connection, events_to_return, mode):

    global configuration
    dt = []
    endpoint = []
    severity = []
    facility = []
    ip = []
    system = []
    time = []
    event = []

    timeline = np.array([])
    timestamps = np.array([])

    if events_to_return == "alerts":
        severity_to_return = configuration['SEVERITY_TO_RETURN'].split()
    if events_to_return == "all":
        severity_to_return = list(available_severity_keys(redis_connection))

    redis_pipeline = redis_connection.pipeline()

    for severity_key in severity_to_return:
        if events_to_return == "alerts":
            severity_key = (str(severity_key))
        redis_pipeline.hgetall(severity_key)

    for redis_data in redis_pipeline.execute():
        if redis_data:
            redis_data_keys = redis_data.keys()
            for key in redis_data_keys:
                if key.decode() != "total":
                    data_block = np.reshape(
                        (msgpack.unpackb(zstd.decompress(redis_data[key]))),
                        (-1, 8))
                    for data in data_block:
                        dt.append(data[0].decode())
                        endpoint.append(data[1].decode())
                        severity.append(data[2].decode())
                        facility.append(data[3].decode())
                        ip.append(data[4].decode())
                        system.append(data[5].decode())
                        time.append(data[6].decode())
                        event.append(data[7].decode())

    if mode == "json":
        num = [str(n) for n in range(np.count_nonzero(dt))]
        return (json.dumps({
            "n": list(num),
            "dt": dt,
            "endpoint": endpoint,
            "severity": severity,
            "facility": facility,
            "ip": ip,
            "system": system,
            "timestamp": time,
            "event": event
        }))

    if mode == "csv":
        csv_dataset = tablib.Dataset()

        csv_dataset.append_col(list(dt), header='dt')
        csv_dataset.append_col(list(endpoint), header='endpoint')
        csv_dataset.append_col(list(severity), header='severity')
        csv_dataset.append_col(list(facility), header='facility')
        csv_dataset.append_col(list(ip), header='ip')
        csv_dataset.append_col(list(system), header='system')
        csv_dataset.append_col(list(time), header='timestamp')
        csv_dataset.append_col(list(event), header='event')

        return csv_dataset.export('csv')


def read_and_return_file(path):
    file = ""
    file_handler = open(path, "r")
    for readIndexLine in file_handler:
        file += readIndexLine
    file_handler.close()
    return file


nserv = Klein()


class NotFound(Exception):
    pass


def enable_cors(request):
    request.responseHeaders.addRawHeader("Access-Control-Allow-Origin", "*")
    request.responseHeaders.addRawHeader("Access-Control-Allow-Methods",
                                         "PUT, GET, POST, DELETE, OPTIONS")
    request.responseHeaders.addRawHeader(
        "Access-Control-Allow-Headers",
        "Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token",
    )
    return request


@nserv.route("/")
def home(request):
    return File("build/")

# @route('/')
# def home(request):
#     return File("/build/")


@nserv.route("/static/", branch=True)
def static(request):
    return File("build/static/")


@route("/src/")
def static(request):
    return File("/src/")


# @nserv.route("/manifest.json", branch=True)
# def static(request):
#     return File("/manifest.json")

# @nserv.route("/static/img/", branch=True)
# def img(request):
#     return File("..narwhal/build/static/img")

# @nserv.route("/public/", branch=True)
# def static(request):
#     return File("../narwhal/public")


@nserv.route("/dashboard", branch=True)
def return_dashboard_data(request):
    request = enable_cors(request)

    dashboard = respond_to_dashboard_data_request(redis_main_db)
    data = respond_to_events_data_request(redis_main_db, "alerts", "json")
    jsonMerged = {**json.loads(dashboard), **json.loads(data)}
    return json.dumps(jsonMerged)


@nserv.route("/server_data")
def server_data_req(request):
    request = enable_cors(request)
    # return respond_to_dashboard_data_request(redis_main_db)
    return respond_to_events_data_request(redis_main_db, "all", "json")


@nserv.route("/server_events")
def server_events_req(request):
    request = enable_cors(request)
    return respond_to_events_data_request(redis_main_db, "alerts", "json")


@nserv.route("/csv_alerts", branch=True)
def export_csv_alerts(request):
    request.responseHeaders.addRawHeader(
        b"content-type", b"application/csv")
    request.responseHeaders.addRawHeader(b"content-disposition", b"attachment")
    content = respond_to_events_data_request(redis_main_db, "alerts", "csv")
    return content


@nserv.route("/csv_all", branch=True)
def export_csv_all(request):
    request.responseHeaders.addRawHeader(
        b"content-type", b"application/csv")
    request.responseHeaders.addRawHeader(b"content-disposition", b"attachment")
    content = respond_to_events_data_request(redis_main_db, "all", "csv")
    return content


@nserv.route("/json_alerts", branch=True)
def export_json_alerts(request):
    request.responseHeaders.addRawHeader(b"content-type", b"application/json")
    content = respond_to_events_data_request(redis_main_db, "alerts", "json")
    return content


@nserv.route("/json_all", branch=True)
def export_json_all(request):
    request.responseHeaders.addRawHeader(b"content-type", b"application/json")
    content = respond_to_events_data_request(redis_main_db, "all", "json")
    return content


@nserv.handle_errors(NotFound)
@nserv.handle_errors(FileNotFoundError)
def error(self, request):
    # request.setResponseCode(404)
    error_file = ""
    error_file_handler = open("build/error.html", "r")
    for readErrorLine in error_file_handler:
        error_file += readErrorLine
    error_file_handler.close()
    return error_file


def main_server_loop_failed(failure):
    print("ERROR - main server loop failed:")
    print(DASH_LINE)
    print(failure.getBriefTraceback())
    reactor.stop()


resource = nserv.resource

if __name__ == "__main__":
    try:
        load_configuration()

        try:

            display_console_banner()

            redis_syslog_cache = redis_connect(
                redis, configuration['SYSLOG_CACHE_DB_NES'])
            redis_main_db = redis_connect(redis,
                                          configuration["REDIS_MAIN_DB"])
            connected_to_redis = True
            print("Connected to Redis.")

        except redis.ConnectionError:

            connected_to_redis = False

            print(DASH_LINE)
            print("Error - check configuration or status of the Redis server.")
            print(DASH_LINE)

        if connected_to_redis:

            narwhal_log(LOG_MESSAGE_START)

            print("Processing cache...")
            start_time = time.time()
            syslog_cache_processor(redis_syslog_cache, redis_main_db)
            print("Processed cache on server start in %s seconds.             " %
                  round(time.time() - start_time))

            if configuration["ENABLE_HTTPS"] == "Y":
                endpoint_description = (
                    "ssl:" + str(configuration["DASHBOARD_WEB_PORT"]) +
                    ":interface=" + configuration["NETWORK_INTERFACE"] +
                    ":privateKey="+configuration["PRIVATE_KEY"])
            else:
                endpoint_description = ("tcp:port=" + str(configuration["DASHBOARD_WEB_PORT"]) +
                                        ":interface=" + configuration["NETWORK_INTERFACE"])

            endpoint = endpoints.serverFromString(reactor,
                                                  endpoint_description)
            endpoint.listen(Site(nserv.resource()))

            syslog_cache_processor_task = task.LoopingCall(
                syslog_cache_processor, redis_syslog_cache, redis_main_db)

            main_server_processor_loop = syslog_cache_processor_task.start(
                configuration["SYSLOG_CACHE_PROCESS_INTERVAL"])

            main_server_processor_loop.addErrback(main_server_loop_failed)

            # reactor.suggestThreadPoolSize(30)
            reactor.run()

    except (IOError, SystemExit):

        raise
    except KeyboardInterrupt:
        narwhal_log(LOG_MESSAGE_STOP)
        print("Shutting down server...")
        syslog_cache_processor(redis_syslog_cache, redis_main_db)
        reactor.callFromThread(reactor.stop)
        # reactor.stop
        print("done.")