mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-11-12 19:47:17 +00:00
101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
|
# All Rights Reserved. See LICENSE file for details.
|
|
|
|
from os.path import dirname
|
|
from re import findall
|
|
from re import compile as compileregex
|
|
from abc import ABC, abstractmethod
|
|
|
|
from tfw.components.inotify import InotifyObserver
|
|
from tfw.event_handlers import TFWServerUplinkConnector
|
|
|
|
|
|
class HistoryMonitor(ABC, InotifyObserver):
|
|
"""
|
|
Abstract class capable of monitoring and parsing a history file such as
|
|
bash HISTFILEs. Monitoring means detecting when the file was changed and
|
|
notifying subscribers about new content in the file.
|
|
|
|
This is useful for monitoring CLI sessions.
|
|
|
|
To specify a custom HistoryMonitor inherit from this class and override the
|
|
command pattern property and optionally the sanitize_command method.
|
|
See examples below.
|
|
"""
|
|
def __init__(self, domain, histfile):
|
|
self.domain = domain
|
|
self.histfile = histfile
|
|
self._history = []
|
|
self._last_length = len(self._history)
|
|
self.uplink = TFWServerUplinkConnector()
|
|
super().__init__(dirname(self.histfile), [self.histfile])
|
|
|
|
def on_modified(self, event):
|
|
self._fetch_history()
|
|
if self._last_length < len(self._history):
|
|
for command in self._history[self._last_length:]:
|
|
self.send_message(command)
|
|
|
|
@property
|
|
def history(self):
|
|
return self._history
|
|
|
|
def _fetch_history(self):
|
|
self._last_length = len(self._history)
|
|
with open(self.histfile, 'r') as ifile:
|
|
pattern = compileregex(self.command_pattern)
|
|
data = ifile.read()
|
|
self._history = [
|
|
self.sanitize_command(command)
|
|
for command in findall(pattern, data)
|
|
]
|
|
|
|
@property
|
|
@abstractmethod
|
|
def command_pattern(self):
|
|
raise NotImplementedError
|
|
|
|
def sanitize_command(self, command):
|
|
# pylint: disable=no-self-use
|
|
return command
|
|
|
|
def send_message(self, command):
|
|
self.uplink.send_message({
|
|
'key': f'history.{self.domain}',
|
|
'value': command
|
|
})
|
|
|
|
|
|
class BashMonitor(HistoryMonitor):
|
|
"""
|
|
HistoryMonitor for monitoring bash CLI sessions.
|
|
This requires the following to be set in bash
|
|
(note that this is done automatically by TFW):
|
|
PROMPT_COMMAND="history -a"
|
|
shopt -s cmdhist
|
|
shopt -s histappend
|
|
unset HISTCONTROL
|
|
"""
|
|
def __init__(self, histfile):
|
|
super().__init__('bash', histfile)
|
|
|
|
@property
|
|
def command_pattern(self):
|
|
return r'.+'
|
|
|
|
def sanitize_command(self, command):
|
|
return command.strip()
|
|
|
|
|
|
class GDBMonitor(HistoryMonitor):
|
|
"""
|
|
HistoryMonitor to monitor GDB sessions.
|
|
For this to work "set trace-commands on" must be set in GDB.
|
|
"""
|
|
def __init__(self, histfile):
|
|
super().__init__('gdb', histfile)
|
|
|
|
@property
|
|
def command_pattern(self):
|
|
return r'(?<=\n)\+(.+)\n'
|