from re import findall from re import compile as compileregex from abc import ABC, abstractmethod from tfw.internals.networking import Intent from tfw.internals.inotify import InotifyObserver 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, uplink, histfile): self.histfile = histfile self.history = [] self._last_length = len(self.history) self.uplink = uplink super().__init__(self.histfile) @property @abstractmethod def domain(self): raise NotImplementedError() 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) 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}', 'command': command }, intent=Intent.EVENT) 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 """ @property def domain(self): return 'bash' @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. """ @property def domain(self): return 'gdb' @property def command_pattern(self): return r'(?<=\n)\+(.+)\n'