vyos-event-handler.py
No OneTemporary
Actions

Size

5 KB

Referenced Files

None

Subscribers

None

vyos-event-handler.py
View Options

	#!/usr/bin/env python3
	#
	# Copyright (C) 2022 VyOS maintainers and contributors
	#
	# This program is free software; you can redistribute it and/or modify
	# it under the terms of the GNU General Public License version 2 or later as
	# published by the Free Software Foundation.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License
	# along with this program. If not, see <http://www.gnu.org/licenses/>.

	import argparse
	import json
	import re
	import select
	from copy import deepcopy
	from os import getpid, environ
	from pathlib import Path
	from signal import signal, SIGTERM, SIGINT
	from sys import exit
	from systemd import journal

	from vyos.util import run, dict_search

	# Identify this script
	my_pid = getpid()
	my_name = Path(__file__).stem


	# handle termination signal
	def handle_signal(signal_type, frame):
	if signal_type == SIGTERM:
	journal.send('Received SIGTERM signal, stopping normally',
	SYSLOG_IDENTIFIER=my_name)
	if signal_type == SIGINT:
	journal.send('Received SIGINT signal, stopping normally',
	SYSLOG_IDENTIFIER=my_name)
	exit(0)


	# Class for analyzing and process messages
	class Analyzer:
	# Initialize settings
	def __init__(self, config: dict) -> None:
	self.config = {}
	# Prepare compiled regex objects
	for event_id, event_config in config.items():
	script = dict_search('script.path', event_config)
	# Check for arguments
	if dict_search('script.arguments', event_config):
	script_arguments = dict_search('script.arguments', event_config)
	script = f'{script} {script_arguments}'
	# Prepare environment
	environment = deepcopy(environ)
	# Check for additional environment options
	if dict_search('script.environment', event_config):
	for env_variable, env_value in dict_search(
	'script.environment', event_config).items():
	environment[env_variable] = env_value.get('value')
	# Create final config dictionary
	pattern_raw = event_config['filter']['pattern']
	pattern_compiled = re.compile(
	rf'{event_config["filter"]["pattern"]}')
	pattern_config = {
	pattern_compiled: {
	'pattern_raw':
	pattern_raw,
	'syslog_id':
	dict_search('filter.syslog-identifier', event_config),
	'pattern_script': {
	'path': script,
	'environment': environment
	}
	}
	}
	self.config.update(pattern_config)

	# Execute script safely
	def script_run(self, pattern: str, script_path: str,
	script_env: dict) -> None:
	try:
	run(script_path, env=script_env)
	journal.send(
	f'Pattern found: "{pattern}", script executed: "{script_path}"',
	SYSLOG_IDENTIFIER=my_name)
	except Exception as err:
	journal.send(
	f'Pattern found: "{pattern}", failed to execute script "{script_path}": {err}',
	SYSLOG_IDENTIFIER=my_name)

	# Analyze a message
	def process_message(self, message: dict) -> None:
	for pattern_compiled, pattern_config in self.config.items():
	# Check if syslog id is presented in config and matches
	syslog_id = pattern_config.get('syslog_id')
	if syslog_id and message['SYSLOG_IDENTIFIER'] != syslog_id:
	continue
	if pattern_compiled.fullmatch(message['MESSAGE']):
	# Add message to environment variables
	pattern_config['pattern_script']['environment'][
	'message'] = message['MESSAGE']
	# Run script
	self.script_run(
	pattern=pattern_config['pattern_raw'],
	script_path=pattern_config['pattern_script']['path'],
	script_env=pattern_config['pattern_script']['environment'])


	if __name__ == '__main__':
	# Parse command arguments and get config
	parser = argparse.ArgumentParser()
	parser.add_argument('-c',
	'--config',
	action='store',
	help='Path to even-handler configuration',
	required=True,
	type=Path)

	args = parser.parse_args()
	try:
	config_path = Path(args.config)
	config = json.loads(config_path.read_text())
	# Create an object for analazyng messages
	analyzer = Analyzer(config)
	except Exception as err:
	print(
	f'Configuration file "{config_path}" does not exist or malformed: {err}'
	)
	exit(1)

	# Prepare for proper exitting
	signal(SIGTERM, handle_signal)
	signal(SIGINT, handle_signal)

	# Set up journal connection
	data = journal.Reader()
	data.seek_tail()
	data.get_previous()
	p = select.poll()
	p.register(data, data.get_events())

	journal.send(f'Started with configuration: {config}',
	SYSLOG_IDENTIFIER=my_name)

	while p.poll():
	if data.process() != journal.APPEND:
	continue
	for entry in data:
	message = entry['MESSAGE']
	pid = entry['_PID']
	# Skip empty messages and messages from this process
	if message and pid != my_pid:
	try:
	analyzer.process_message(entry)
	except Exception as err:
	journal.send(f'Unable to process message: {err}',
	SYSLOG_IDENTIFIER=my_name)

File Metadata

Mime Type: text/x-script.python
Expires: Sat, Oct 17, 12:12 AM (1 d, 22 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4355705
Default Alt Text: vyos-event-handler.py (5 KB)

vyos-event-handler.pyNo OneTemporaryActions

vyos-event-handler.pyView Options

File Metadata

Event Timeline

vyos-event-handler.py
No OneTemporary
Actions

vyos-event-handler.py
View Options