98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
from re import findall
|
|
from re import compile as compileregex
|
|
from abc import ABC, abstractmethod
|
|
|
|
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}',
|
|
'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
|
|
"""
|
|
@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'
|