mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-11-08 20:17:17 +00:00
105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
|
# All Rights Reserved. See LICENSE file for details.
|
|
|
|
from re import findall
|
|
from re import compile as compileregex
|
|
from abc import ABC, abstractmethod
|
|
|
|
from tfw.components.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._domain = ''
|
|
self.histfile = histfile
|
|
self.history = []
|
|
self._last_length = len(self.history)
|
|
self.uplink = uplink
|
|
super().__init__(self.histfile)
|
|
|
|
@property
|
|
def domain(self):
|
|
return self._domain
|
|
|
|
@domain.setter
|
|
def domain(self, domain):
|
|
self._domain = domain
|
|
|
|
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
|
|
"""
|
|
def __init__(self, uplink, histfile):
|
|
super().__init__(uplink, histfile)
|
|
self.domain = '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.
|
|
"""
|
|
def __init__(self, histfile):
|
|
super().__init__(histfile)
|
|
self.domain = 'gdb'
|
|
|
|
@property
|
|
def command_pattern(self):
|
|
return r'(?<=\n)\+(.+)\n'
|