klein-projects/wazuh-notify
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Converted from fixed position parameters to named parameters.

Browse Source 
Using f-strings for log message with variables
This commit is contained in:
Rudi klein 2024-05-30 17:11:49 +02:00
parent 6c267f555f
commit 84fc2bc1a8
2 changed files with 119 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
wazuh-notify-python/wazuh-notify.py
View File

@ -9,69 +9,68 @@
import requests import requests
from requests import Response
from wazuh_notify_module import * from wazuh_notify_module import *
def main(): def main():
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging. # The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
# Load the TOML config.
config: dict = get_config() config: dict = get_config()
logger(0, config, me, him, "############ Processing event ###############################") # Write header line in logfile
logger(level=99, config=config, me=me, him=him, message="")
# Load the TOML config.
logger(level=0, config=config, me=me, him=him, message="############# [Processing event] #########################")
# Get the arguments used with running the script. # Get the arguments used with running the script.
arguments = get_arguments() arguments: dict = get_arguments()
# Check for test mode. Use test data if true. # Check for test mode. Use test data if true.
event_data = check_test_mode(config) event_data: dict = check_test_mode(config)
# Extract the 'alert' section of the (JSON) event. alert: dict = event_data["parameters"]["alert"]
alert = event_data["parameters"]["alert"] logger(level=2, config=config, me=me, him=him, message="Extracting data from the event")
logger(2, config, me, him, "Extracting data from the event")
# Check the config for any exclusion rules and abort when excluded. # Check the config for any exclusion rules and abort when excluded.
exclusions_check(config, alert) if not exclusions_check(config, alert):
logger(level=1, config=config, me=me, him=him, message="Event excluded, no notification sent. Exiting")
exit()
logger(level=2, config=config, me=me, him=him, message="Event NOT excluded, notification will be sent")
# Get the mapping from event threat level to priority, color and mention_flag. # Get the mapping from event threat level to priority, color and mention_flag.
priority, color, mention = threat_mapping(config, alert['rule']['level'], alert['rule']['firedtimes']) priority, color, mention = threat_mapping(config, alert['rule']['level'], alert['rule']['firedtimes'])
# If the target argument was used with the script, we'll use that instead of the configuration parameter.
config["targets"] = arguments['targets'] if arguments['targets'] != "" else config["targets"] config["targets"] = arguments['targets'] if arguments['targets'] != "" else config["targets"]
# Prepare the messaging platform specific notification and execute if configured.
# Discord notification handler # Discord notification handler
if "discord" in config["targets"]: if "discord" in config["targets"]:
payload_json, discord_url = handle_discord_notification(config=config, arguments=arguments, alert=alert, payload_json, discord_url = handle_discord_notification(config=config, arguments=arguments, alert=alert,
color=color, priority=priority, mention=mention) color=color, priority=priority, mention=mention)
# POST the notification through requests. discord_result: Response = requests.post(url=discord_url, json=payload_json)
discord_result = requests.post(url=discord_url, json=payload_json) logger(level=1, config=config, me=me, him=him, message="Discord notification constructed and sent: " +
logger(1, config, me, him, "Discord notification constructed and sent: " + str(discord_result)) str(discord_result))
# ntfy.sh notification handler # ntfy.sh notification handler
if "ntfy" in config["targets"]: if "ntfy" in config["targets"]:
payload_data, payload_headers, ntfy_url = handle_ntfy_notification(config=config, arguments=arguments, payload_data, payload_headers, ntfy_url = handle_ntfy_notification(config=config, arguments=arguments,
alert=alert, priority=priority) alert=alert, priority=priority)
# POST the notification through requests. ntfy_result: Response = requests.post(url=ntfy_url, data=payload_data, headers=payload_headers)
ntfy_result = requests.post(url=ntfy_url, data=payload_data, headers=payload_headers) logger(level=1, config=config, me=me, him=him, message="Ntfy notification constructed and sent: " +
logger(1, config, me, him, "Ntfy notification constructed and sent: " + str(ntfy_result)) str(ntfy_result))
# Slack notification handler # Slack notification handler
if "slack" in config["targets"]: if "slack" in config["targets"]:
payload_json, slack_url = handle_slack_notification(config=config, arguments=arguments, alert=alert, payload_json, slack_url = handle_slack_notification(config=config, arguments=arguments, alert=alert,
color=color, priority=priority, mention=mention) color=color, priority=priority, mention=mention)
slack_result: Response = requests.post(url=slack_url, headers={'Content-Type': 'application/json'},
slack_result = requests.post(url=slack_url, headers={'Content-Type': 'application/json'}, json=payload_json) json=payload_json)
logger(1, config, me, him, "Slack notification constructed and sent: " + str(slack_result)) logger(1, config, me, him, "Slack notification constructed and sent: " + str(slack_result))
# The end of processing logger(0, config, me, him, "############# [Event processed] #########################")
logger(0, config, me, him, "############ Event processed ################################")
exit(0) exit(0)
if "__main__" == __name__: if __name__ == "__main__":
main() main()

176
wazuh-notify-python/wazuh_notify_module.py
View File

@ -12,16 +12,15 @@ import tomli
from dotenv import load_dotenv from dotenv import load_dotenv
# Define paths: wazuh_path = wazuh root directory
# log_path = wazuh-notify.log path,
# config_path = wazuh-notify-config.yaml
############################################################################################## ##############################################################################################
# General process environment handlers # # General process environment handlers #
############################################################################################## ##############################################################################################
# Define paths: wazuh_path = wazuh root directory
# log_path = wazuh-notify.log,
# config_path = wazuh-notify-config.toml
def set_environment() -> tuple: def set_environment() -> tuple:
set_wazuh_path = os.path.abspath(os.path.join(__file__, "../..")) set_wazuh_path = os.path.abspath(os.path.join(__file__, "../.."))
# set_wazuh_path = os.path.abspath(os.path.join(__file__, "../../..")) # set_wazuh_path = os.path.abspath(os.path.join(__file__, "../../.."))
@ -31,11 +30,9 @@ def set_environment() -> tuple:
return set_wazuh_path, set_log_path, set_config_path return set_wazuh_path, set_log_path, set_config_path
# Set paths for use in this module
wazuh_path, log_path, config_path = set_environment() wazuh_path, log_path, config_path = set_environment()
# Set structured timestamps for notifications.
def set_time_format() -> tuple: def set_time_format() -> tuple:
now_message = time.strftime('%A, %d %b %Y %H:%M:%S') now_message = time.strftime('%A, %d %b %Y %H:%M:%S')
now_logging = time.strftime('%Y-%m-%d %H:%M:%S') now_logging = time.strftime('%Y-%m-%d %H:%M:%S')
@ -46,7 +43,7 @@ def set_time_format() -> tuple:
# Logger: print to console and/or log to file # Logger: print to console and/or log to file
def logger(level, config, me, him, message) -> None: def logger(level: int, config: dict, me: str, him: str, message: str) -> None:
_, now_logging, _, _ = set_time_format() _, now_logging, _, _ = set_time_format()
logger_wazuh_path, logger_log_path, _ = set_environment() logger_wazuh_path, logger_log_path, _ = set_environment()
@ -60,9 +57,25 @@ def logger(level, config, me, him, message) -> None:
print(log_line) print(log_line)
try: try:
# Compare the extended_logging level in the configuration to the log level of the message. # Compare the extended_logging level in the configuration to the log level of the message.
if config.get('python').get('extended_logging', 0) >= level: with open(logger_log_path, mode="a") as log_file:
with open(logger_log_path, mode="a") as log_file: if config.get('python').get('extended_logging', 0) >= level:
log_file.write(log_line + "\n") log_file.write(log_line)
else:
now_logging = 'Timestamp'
level = 'Lvl'
him = 'Origination function'
me = 'Executed function'
message = "Information"
log_line: str = f'\n{now_logging: <19} |{level: <3}| {me: <27} | {him: <27} | {message}'
log_file.write(log_line + '\n')
now_logging = '------------------'
level = '---'
him = '---------------------------'
me = '---------------------------'
message = "------------------------------------------------"
log_line: str = f'\n{now_logging: <19} |{level: <3}| {me: <27} | {him: <27} | {message}'
log_file.write(log_line + '\n')
except (FileNotFoundError, PermissionError, OSError): except (FileNotFoundError, PermissionError, OSError):
# Special message to console when logging to file fails and console logging might not be set. # Special message to console when logging to file fails and console logging might not be set.
log_line: str = f'{now_logging} | {level} | {me: <27} | {him: <17} | error opening log file: {logger_log_path}' log_line: str = f'{now_logging} | {level} | {me: <27} | {him: <17} | error opening log file: {logger_log_path}'
@ -77,36 +90,35 @@ def get_env() -> tuple:
# Write the configuration to a dictionary. # Write the configuration to a dictionary.
config: dict = get_config() config: dict = get_config()
logger(2, config, me, him, "Configuration retrieved to dictionary") logger(level=2, config=config, me=me, him=him, message="Configuration retrieved to dictionary")
# Check if the secrets .env file is available. # Check if the secrets .env file is available.
try: try:
dotenv_path = join(dirname(__file__), '.env') dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path) load_dotenv(dotenv_path)
if not os.path.isfile(dotenv_path): if not os.path.isfile(dotenv_path):
logger(0, config, me, him, dotenv_path + " not found") logger(level=0, config=config, me=me, him=him, message=f'%s not found' % dotenv_path)
raise Exception(dotenv_path, "file not found") raise Exception(dotenv_path, "file not found")
# Retrieve URLs from .env # Retrieve URLs from .env
discord_url = os.getenv("DISCORD_URL") discord_url = os.getenv("DISCORD_URL")
logger(2, config, me, him, "DISCORD_URL: " + discord_url) logger(level=2, config=config, me=me, him=him, message=f"DISCORD_URL: %s" % discord_url)
ntfy_url = os.getenv("NTFY_URL") ntfy_url = os.getenv("NTFY_URL")
logger(2, config, me, him, "NTFY_URL: " + ntfy_url) logger(level=2, config=config, me=me, him=him, message=f"NTFY_URL: %s" % ntfy_url)
slack_url = os.getenv("SLACK_URL") slack_url = os.getenv("SLACK_URL")
logger(2, config, me, him, "SLACK_URL: " + slack_url) logger(level=2, config=config, me=me, him=him, message=f"SLACK_URL: %s" % slack_url)
except Exception as err: except Exception as err:
# output error, and return with an error code # output error, and return with an error code
logger(0, config, me, him, 'Error reading ' + str(err)) logger(level=0, config=config, me=me, him=him, message=f'Error reading %s' % err)
exit(err) exit(err)
logger(2, config, me, him, dotenv_path + " loaded") logger(level=2, config=config, me=me, him=him, message=f'%s loaded' % dotenv_path)
return discord_url, ntfy_url, slack_url return discord_url, ntfy_url, slack_url
# Read and process configuration settings from wazuh-notify-config.yaml and create dictionary. # Read and process configuration settings from wazuh-notify-config.toml and create dictionary.
def get_config() -> dict: def get_config() -> dict:
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
this_config_path: str = "" this_config_path: str = ""
@ -117,8 +129,8 @@ def get_config() -> dict:
with open(this_config_path, 'rb') as ntfier_config: with open(this_config_path, 'rb') as ntfier_config:
config: dict = tomli.load(ntfier_config) config: dict = tomli.load(ntfier_config)
except (FileNotFoundError, PermissionError, OSError): except (FileNotFoundError, PermissionError, OSError):
logger(2, config, me, him, "Error accessing configuration file: " + this_config_path) logger(level=2, config=config, me=me, him=him, message=f"Error accessing configuration: %s" % this_config_path)
logger(2, config, me, him, "Reading TOML configuration file: " + this_config_path) logger(level=2, config=config, me=me, him=him, message=f"Reading TOML configuration file: %s" % this_config_path)
config['targets']: str = config.get('general').get('targets', 'discord, slack, ntfy') config['targets']: str = config.get('general').get('targets', 'discord, slack, ntfy')
config['full_alert']: bool = config.get('general').get('full_alert', False) config['full_alert']: bool = config.get('general').get('full_alert', False)
@ -145,19 +157,17 @@ def view_config() -> None:
with open(this_config_path, 'r') as ntfier_config: with open(this_config_path, 'r') as ntfier_config:
print(ntfier_config.read()) print(ntfier_config.read())
except (FileNotFoundError, PermissionError, OSError): except (FileNotFoundError, PermissionError, OSError):
print(this_config_path + " does not exist or is not accessible") print(f"%s does not exist or is not accessible" % this_config_path)
return return
# Get script arguments during execution. Params found here override config settings. # Get script arguments during execution. Params found here override config settings.
def get_arguments() -> dict: def get_arguments() -> dict:
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
# Retrieve the configuration information
config: dict = get_config() config: dict = get_config()
logger(2, config, me, him, "Configuration retrieved to dictionary") logger(level=2, config=config, me=me, him=him, message="Configuration retrieved to dictionary")
# Short options # Short options
options: str = "u:s:p:m:t:c:hv" options: str = "u:s:p:m:t:c:hv"
@ -198,10 +208,11 @@ def get_arguments() -> dict:
# Fetch the arguments from the command line, skipping the first argument (name of the script). # Fetch the arguments from the command line, skipping the first argument (name of the script).
argument_list: list = sys.argv[1:] argument_list: list = sys.argv[1:]
logger(2, config, me, him, "Found arguments:" + str(argument_list)) logger(level=2, config=config, me=me, him=him, message=f"Found arguments: %s" % argument_list)
if not argument_list: if not argument_list:
logger(1, config, me, him, 'No argument list found (no arguments provided with script execution') logger(level=1, config=config, me=me, him=him,
message='No argument list found (no arguments provided with script execution')
# Store defaults for the non-existing arguments in the arguments dictionary to avoid None errors. # Store defaults for the non-existing arguments in the arguments dictionary to avoid None errors.
arguments: dict = {'url': url, arguments: dict = {'url': url,
@ -216,7 +227,7 @@ def get_arguments() -> dict:
try: try:
# Parsing arguments # Parsing arguments
p_arguments, values = getopt.getopt(argument_list, options, long_options) p_arguments, values = getopt.getopt(argument_list, options, long_options)
logger(2, config, me, him, "Parsing arguments") logger(level=2, config=config, me=me, him=him, message="Parsing arguments")
# Check each argument. Arguments that are present will override the defaults. # Check each argument. Arguments that are present will override the defaults.
for current_argument, current_value in p_arguments: for current_argument, current_value in p_arguments:
@ -243,8 +254,8 @@ def get_arguments() -> dict:
except getopt.error as err: except getopt.error as err:
# Output error, and return error code # Output error, and return error code
logger(0, config, me, him, "Error during argument parsing:" + str(err)) logger(level=0, config=config, me=me, him=him, message=f"Error during argument parsing: %s" % err)
logger(2, config, me, him, "Arguments returned as dictionary") logger(level=2, config=config, me=me, him=him, message="Arguments returned as dictionary")
# Store the arguments in the arguments dictionary. # Store the arguments in the arguments dictionary.
arguments: dict = {'url': url, 'sender': sender, 'targets': targets, 'message': message, arguments: dict = {'url': url, 'sender': sender, 'targets': targets, 'message': message,
@ -265,7 +276,7 @@ def load_message() -> dict:
config: dict = get_config() config: dict = get_config()
# get alert from stdin # get alert from stdin
logger(2, config, me, him, "Loading event message from STDIN") logger(level=2, config=config, me=me, him=him, message="Loading event message from STDIN")
input_str: str = "" input_str: str = ""
for line in sys.stdin: for line in sys.stdin:
@ -275,11 +286,11 @@ def load_message() -> dict:
data: json = json.loads(input_str) data: json = json.loads(input_str)
if data.get("command") == "add": if data.get("command") == "add":
logger(1, config, me, him, "Relevant event data found") logger(level=1, config=config, me=me, him=him, message="Relevant event data found")
return data return data
else: else:
# Event came in, but wasn't processed. # Event came in, but wasn't processed.
logger(0, config, me, him, "Event data not found") logger(level=0, config=config, me=me, him=him, message="Event data not found")
sys.exit(1) sys.exit(1)
@ -289,31 +300,33 @@ def check_test_mode(config) -> dict:
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
if config.get('python').get('test_mode'): if config.get('python').get('test_mode'):
logger(1, config, me, him, "Running in test mode: using test message wazuh-notify-test-event.json") logger(level=1, config=config, me=me, him=him,
message="Running in test mode: using test message wazuh-notify-test-event.json")
# Load the test event data. # Load the test event data.
home_path, _, _ = set_environment() home_path, _, _ = set_environment()
with (open(home_path + '/etc/wazuh-notify-test-event.json') as event_file): with (open(home_path + '/etc/wazuh-notify-test-event.json') as event_file):
data: dict = json.loads(event_file.read()) data: dict = json.loads(event_file.read())
else: else:
# We are running live. Load the data from the Wazuh process. # We are running live. Load the data from the Wazuh process.
logger(2, config, me, him, "Running in live mode: using live message") logger(level=2, config=config, me=me, him=him, message="Running in live mode: using live message")
data: dict = load_message() data: dict = load_message()
return data return data
# Check if there are reasons not to process this event. Check exclusions for rules, agents, days and hours. # Check if there are reasons not to process this event. Check exclusions for rules, agents, days and hours.
def exclusions_check(config, alert) -> None: def exclusions_check(config, alert) -> bool:
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging. # The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
# Set some environment # Set some environment
now_message, now_logging, now_weekday, now_time = set_time_format() now_message, now_logging, now_weekday, now_time = set_time_format()
logger(2, config, me, him, "Setting pretty datetime formats for notifications and logging.") logger(level=2, config=config, me=me, him=him,
message="Setting pretty datetime formats for notifications and logging.")
# Check the exclusion records from the configuration yaml. # Check the exclusion records from the configuration yaml.
logger(1, config, me, him, "Checking if we are outside of the exclusion rules: ") logger(level=1, config=config, me=me, him=him, message="Checking if we are outside of the exclusion rules: ")
ex_hours: tuple = config.get('python').get('excluded_hours') ex_hours: tuple = config.get('python').get('excluded_hours')
# Start hour may not be later than end hours. End hour may not exceed 00:00 midnight to avoid day jump. # Start hour may not be later than end hours. End hour may not exceed 00:00 midnight to avoid day jump.
@ -333,30 +346,23 @@ def exclusions_check(config, alert) -> None:
# Evaluate the conditions for sending a notification. Any False will cause the notification to be discarded. # Evaluate the conditions for sending a notification. Any False will cause the notification to be discarded.
if (now_time > ex_hours[0]) and (now_time < ex_hours[1]): if (now_time > ex_hours[0]) and (now_time < ex_hours[1]):
logger(2, config, me, him, "excluded: event inside exclusion time frame") logger(level=2, config=config, me=me, him=him, message="excluded: event inside exclusion time frame")
ex_hours_eval = False ex_hours_eval = False
elif now_weekday in ex_days: elif now_weekday in ex_days:
logger(2, config, me, him, "excluded: event inside excluded weekdays") logger(level=2, config=config, me=me, him=him, message="excluded: event inside excluded weekdays")
ex_weekday_eval = False ex_weekday_eval = False
elif ev_rule in ex_rules: elif ev_rule in ex_rules:
logger(2, config, me, him, "excluded: event id inside exclusion list") logger(level=2, config=config, me=me, him=him, message="excluded: event id inside exclusion list")
ev_rule_eval = False ev_rule_eval = False
elif ev_agent in ex_agents: elif ev_agent in ex_agents:
logger(2, config, me, him, "excluded: event agent inside exclusion list") logger(level=2, config=config, me=me, him=him, message="excluded: event agent inside exclusion list")
ev_rule_eval = False ev_rule_eval = False
notification_eval = True if (ex_hours_eval and ex_weekday_eval and ev_rule_eval and ev_agent_eval) else False notification_eval = True if (ex_hours_eval and ex_weekday_eval and ev_rule_eval and ev_agent_eval) else False
logger(1, config, me, him, "Exclusion rules evaluated. Process event is " + str(notification_eval)) logger(level=1, config=config, me=me, him=him,
message=f"Exclusion rules evaluated. Process event is %s" % notification_eval)
if not notification_eval: return notification_eval
# The event was excluded by the exclusion rules in the configuration.
logger(1, config, me, him, "Event excluded, no notification sent. Exiting")
exit()
else:
# The event was not excluded by the exclusion rules in the configuration. Keep processing.
logger(2, config, me, him, "Event NOT excluded, notification will be sent")
return
# Map the event threat level to the appropriate 5-level priority scale and color for use in the notification platforms. # Map the event threat level to the appropriate 5-level priority scale and color for use in the notification platforms.
@ -367,34 +373,37 @@ def threat_mapping(config, threat_level, fired_times):
# Map threat level to priority. Enters Homeland Security :-). # Map threat level to priority. Enters Homeland Security :-).
p_map = config.get('priority_map') p_map = config.get('priority_map')
logger(2, config, me, him, "Threat mapping: priority mapped to: " + str(p_map)) logger(level=2, config=config, me=me, him=him, message=f"Threat mapping: priority mapped to: %s" % p_map)
for i in range(len(p_map)): for i in range(len(p_map)):
logger(2, config, me, him, "Threat mapping: list loop counter: " + str(i)) logger(level=2, config=config, me=me, him=him, message=f"Threat mapping: list loop counter: %s" % i)
logger(2, config, me, him, "Threat mapping: threat level found: " + str(threat_level)) logger(level=2, config=config, me=me, him=him, message=f"Threat mapping: threat level found: %s" % threat_level)
if threat_level in p_map[i]["threat_map"]: if threat_level in p_map[i]["threat_map"]:
color_mapping = p_map[i]["color"] color_mapping = p_map[i]["color"]
priority_mapping = 5 - i priority_mapping = 5 - i
logger(2, config, me, him, "Threat mapping: priority: " + str(priority_mapping)) logger(level=2, config=config, me=me, him=him, message=f"Threat mapping: priority: %s" % priority_mapping)
logger(2, config, me, him, "Threat mapping: color: " + str(color_mapping)) logger(level=2, config=config, me=me, him=him, message=f"Threat mapping: color: %s" % color_mapping)
if fired_times >= p_map[i]["notify_threshold"]: if fired_times >= p_map[i]["notify_threshold"]:
logger(2, config, me, him, "The notification_threshold prevents this message from sending") logger(level=2, config=config, me=me, him=him,
message="The notification_threshold prevents this message from sending")
exit(0) exit(0)
if fired_times >= p_map[i]["mention_threshold"]: if fired_times >= p_map[i]["mention_threshold"]:
# When this flag is set, Discord!! recipients get a stronger message (DM). # When this flag is set, Discord!! recipients get a stronger message (DM).
mention_flag = "@here" mention_flag = "@here"
logger(2, config, me, him, "Threat mapping: mention flag: " + str(mention_flag)) logger(level=2, config=config, me=me, him=him,
message=f"Threat mapping: mention flag: %s" % mention_flag)
else: else:
mention_flag = "" mention_flag = ""
logger(2, config, me, him, "Threat level mapped as: " + "priority:" + str(priority_mapping) + logger(level=2, config=config, me=me, him=him,
" color: " + str(color_mapping) + " mention: " + str(mention_flag)) message=f"Threat level mapped as: priority: %s color: %s mention: %s"
% (priority_mapping, color_mapping, mention_flag))
return priority_mapping, color_mapping, mention_flag return priority_mapping, color_mapping, mention_flag
logger(0, config, me, him, "Threat level mapping failed! Returning garbage (99, 99, 99)") logger(level=0, config=config, me=me, him=him, message="Threat level mapping failed! Returning (99, 99, 99)")
return 99, 99, "99" return 99, 99, "99"
@ -409,13 +418,13 @@ def construct_message_body(caller, config, arguments, alert) -> str:
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(2, config, me, him, caller + ": Constructing message body") logger(level=2, config=config, me=me, him=him, message=f"%s: Constructing message body" % caller)
# Include a specific control sequence for markdown bold parameter names. # Include a specific control sequence for markdown bold parameter names.
# todo To be fixed # todo To be fixed
md_map = config.get('markdown_emphasis', '') md_map = config.get('markdown_emphasis', '')
md_e = md_map[caller] md_e = md_map[caller]
logger(2, config, me, him, caller + "Emphasis string used: " + md_e) logger(level=2, config=config, me=me, him=him, message=caller + "Emphasis string used: " + md_e)
# If the --message (-m) argument was fulfilled, use this message to be sent. # If the --message (-m) argument was fulfilled, use this message to be sent.
if arguments['message']: if arguments['message']:
@ -432,7 +441,8 @@ def construct_message_body(caller, config, arguments, alert) -> str:
md_e + "Threat level:" + md_e + " " + str(alert["rule"]["level"]) + "\n" + md_e + "Threat level:" + md_e + " " + str(alert["rule"]["level"]) + "\n" +
md_e + "Times fired:" + md_e + " " + str(alert["rule"]["firedtimes"]) + "\n" md_e + "Times fired:" + md_e + " " + str(alert["rule"]["firedtimes"]) + "\n"
) )
logger(2, config, me, him, caller + " basic message constructed. Returning: " + message_body) logger(level=2, config=config, me=me, him=him,
message=f"%s basic message constructed.\n %s" % (caller, message_body))
return message_body return message_body
@ -442,11 +452,11 @@ def prepare_payload(caller, config, arguments, message_body, alert, priority):
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(2, config, me, him, "Payload being constructed.") logger(level=2, config=config, me=me, him=him, message="Payload being constructed.")
md_map = config.get('markdown_emphasis', '') md_map = config.get('markdown_emphasis', '')
md_e = md_map[caller] md_e = md_map[caller]
logger(2, config, me, him, caller + "Emphasis string used: " + md_e) logger(level=2, config=config, me=me, him=him, message=f"%s: emphasis string used: %s" % (caller, md_e))
priority: str = str(priority) priority: str = str(priority)
tags = (str(alert['rule']['groups']).replace("[", "") tags = (str(alert['rule']['groups']).replace("[", "")
@ -454,7 +464,7 @@ def prepare_payload(caller, config, arguments, message_body, alert, priority):
.replace("'", "") .replace("'", "")
.replace(",", ", ") .replace(",", ", ")
) )
logger(2, config, me, him, caller + " full event formatted.") logger(level=2, config=config, me=me, him=him, message="Full event formatted.")
full_event: str = str(json.dumps(alert, indent=4) full_event: str = str(json.dumps(alert, indent=4)
.replace('"', '') .replace('"', '')
@ -475,7 +485,7 @@ def prepare_payload(caller, config, arguments, message_body, alert, priority):
# Add the full alert data to the notification. # Add the full alert data to the notification.
if caller in config["full_alert"]: if caller in config["full_alert"]:
logger(2, config, me, him, caller + "Full alert data will be attached.") logger(level=2, config=config, me=me, him=him, message=caller + "Full alert data will be attached.")
# Add the full alert data to the notification body # Add the full alert data to the notification body
notification_body: str = ("\n\n" + message_body + "\n" + notification_body: str = ("\n\n" + message_body + "\n" +
md_e + "__Full event__" + md_e + "\n" + "```\n" + full_event + "```") md_e + "__Full event__" + md_e + "\n" + "```\n" + full_event + "```")
@ -485,7 +495,7 @@ def prepare_payload(caller, config, arguments, message_body, alert, priority):
# Add priority & tags to the notification body # Add priority & tags to the notification body
notification_body = (notification_body + "\n\n" + md_e + "Priority:" + md_e + " " + str(priority) + "\n" + notification_body = (notification_body + "\n\n" + md_e + "Priority:" + md_e + " " + str(priority) + "\n" +
md_e + "Tags:" + md_e + " " + tags + "\n\n" + click) md_e + "Tags:" + md_e + " " + tags + "\n\n" + click)
logger(2, config, me, him, caller + " adding priority and tags") logger(level=2, config=config, me=me, him=him, message="Adding priority and tags")
config["targets"] = arguments['targets'] if arguments['targets'] != "" else config["targets"] config["targets"] = arguments['targets'] if arguments['targets'] != "" else config["targets"]
@ -502,14 +512,14 @@ def build_discord_notification(config, notification_body, color, mention, sender
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging. # The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(2, config, me, him, "Discord payload created") logger(level=2, config=config, me=me, him=him, message="Discord payload created")
payload_json = {"username": sender, payload_json = {"username": sender,
"content": mention, "content": mention,
"embeds": [{"description": notification_body, "embeds": [{"description": notification_body,
"color": color, "color": color,
"title": sender}]} "title": sender}]}
logger(2, config, me, him, "Discord notification built") logger(level=2, config=config, me=me, him=him, message="Discord notification built")
return "", "", payload_json return "", "", payload_json
@ -518,14 +528,14 @@ def build_ntfy_notification(config, notification_body, priority, click, sender):
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging. # The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(2, config, me, him, "Ntfy payloads created") logger(level=2, config=config, me=me, him=him, message="Ntfy payloads created")
payload_data = notification_body payload_data = notification_body
payload_headers = {"Markdown": "yes", payload_headers = {"Markdown": "yes",
"Title": sender, "Title": sender,
"Priority": str(priority), "Priority": str(priority),
"Click": click} "Click": click}
logger(2, config, me, him, "Ntfy notification built") logger(level=2, config=config, me=me, him=him, message="Ntfy notification built")
return payload_headers, payload_data, "" return payload_headers, payload_data, ""
@ -534,14 +544,14 @@ def build_slack_notification(config, notification_body, color, mention, sender):
# The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging. # The 'me' variable sets the called function (current function), the 'him' the calling function. Used for logging.
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(2, config, me, him, "Slack payload created") logger(level=2, config=config, me=me, him=him, message="Slack payload created")
payload_json = {"username": sender, payload_json = {"username": sender,
"content": mention, "content": mention,
"embeds": [{"description": notification_body, "embeds": [{"description": notification_body,
"color": color, "color": color,
"title": sender}]} "title": sender}]}
logger(2, config, me, him, "Slack notification built") logger(level=2, config=config, me=me, him=him, message="Slack notification built")
return "", "", payload_json return "", "", payload_json
@ -551,7 +561,7 @@ def handle_discord_notification(config, arguments, alert, color, priority, menti
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(1, config, me, him, "Process discord notification: start") logger(level=1, config=config, me=me, him=him, message="Process discord notification: start")
# Load the url/webhook from the configuration. # Load the url/webhook from the configuration.
discord_url, _, _ = get_env() discord_url, _, _ = get_env()
@ -568,7 +578,7 @@ def handle_discord_notification(config, arguments, alert, color, priority, menti
_, _, payload_json = build_discord_notification(config=config, notification_body=notification_body, color=color, _, _, payload_json = build_discord_notification(config=config, notification_body=notification_body, color=color,
mention=mention, sender=sender) mention=mention, sender=sender)
logger(1, config, me, him, "Process discord notification: done") logger(level=1, config=config, me=me, him=him, message="Process discord notification: done")
return payload_json, discord_url return payload_json, discord_url
@ -578,7 +588,7 @@ def handle_ntfy_notification(config, arguments, alert, priority):
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(1, config, me, him, "Process ntfy notification: start") logger(level=1, config=config, me=me, him=him, message="Process ntfy notification: start")
# Load the url/webhook from the configuration. # Load the url/webhook from the configuration.
_, ntfy_url, _ = get_env() _, ntfy_url, _ = get_env()
@ -598,7 +608,7 @@ def handle_ntfy_notification(config, arguments, alert, priority):
payload_headers, payload_data, _ = build_ntfy_notification(config=config, notification_body=notification_body, payload_headers, payload_data, _ = build_ntfy_notification(config=config, notification_body=notification_body,
priority=priority, click=click, sender=sender) priority=priority, click=click, sender=sender)
logger(1, config, me, him, "Process ntfy notification: done") logger(level=1, config=config, me=me, him=him, message="Process ntfy notification: done")
return payload_data, payload_headers, ntfy_url return payload_data, payload_headers, ntfy_url
@ -608,7 +618,7 @@ def handle_slack_notification(config, arguments, alert, color, priority, mention
me: str = frame(0).f_code.co_name me: str = frame(0).f_code.co_name
him: str = frame(1).f_code.co_name him: str = frame(1).f_code.co_name
logger(1, config, me, him, "Process slack notification: start") logger(level=1, config=config, me=me, him=him, message="Process slack notification: start")
# Load the url/webhook from the configuration. # Load the url/webhook from the configuration.
_, _, slack_url = get_env() _, _, slack_url = get_env()
@ -625,5 +635,5 @@ def handle_slack_notification(config, arguments, alert, color, priority, mention
_, _, payload_json = build_slack_notification(config=config, notification_body=notification_body, color=color, _, _, payload_json = build_slack_notification(config=config, notification_body=notification_body, color=color,
mention=mention, sender=sender) mention=mention, sender=sender)
logger(1, config, me, him, "Process slack notification: done") logger(level=1, config=config, me=me, him=him, message="Process slack notification: done")
return payload_json, slack_url return payload_json, slack_url