"""
Module providing the toolset to build graphs of communicating entities.
"""
import gdb
from ..toolbox import my_gdb
from ..toolbox.target import my_system
[docs]class StopRequest:
def __init__(self, message=None, permanent=False, cb=None):
self.message = message
self.permanent = permanent
self.cb = cb
[docs]class Filterable:
"""
Parent class for filterable objects.
Subclasses should override methods `get_parameters` and `add_return_values`.
:var name: Name of the filter.
:var Filterable.filters:
:var Filterable.filterable:
"""
filters = {}
filterables = []
@staticmethod
[docs] def register_filter(name, before=None, after=None):
"""
Registers a new filter.
:param name: Name of the filter.
:param before: Filter function executed before the function call.
:type before: Callable(\*\*params) --> Boolean
:param after: Filter function executed after the function call.
:type after: Callable(\*\*params) --> Boolean
"""
if name not in Filterable.filters.keys():
my_gdb.log.error("Operation '%s' not registered", name)
if (before, after) not in Filterable.filters[name]:
Filterable.filters[name].append((before, after))
def __init__(self, name):
if ":" in name: name = name.partition(":")[2]
self.__class__.filters[name] = []
self.__class__.filterables.append(self)
self.name = name
[docs] def stop_before_filters(self):
"""
Tests if the execution should be stopped before the execution of the function.
Gets the function parameters from the subclass and apply before(\*\*param).
:returns: #1 True if one of the registered `before` filter return True, False otherwise.
:returns: #2 The parameters used for the filtering.
"""
params = self.get_parameters()
if params is None:
params = {}
stop = False
for bfilter, afilter in self.__class__.filters[self.name]:
if bfilter is None: continue
try:
stop = stop or bfilter(**params)
except Exception as e:
log.error("before filter '%s' for primitive '%s' crashed ...",
(afilter, self.name))
return stop, params
[docs] def has_after_filters(self):
"""
:returns: True if some `after` filters were registered for this object.
"""
for (bfilter, afilter) in self.__class__.filters[self.name]:
if afilter is not None:
return True
return False
[docs] def stop_after_filters(self, params):
"""
Tests if the execution should be stopped after the execution of the function.
Adds the return values from the subclass to the `params` object and apply before(\*\*param).
:param params: parameters of the function call.
:returns: True if one of the `after` filter returned True, False otherwise.
"""
self.add_return_values(params)
stop = False
for bfilter, afilter in self.__class__.filters[self.name]:
if afilter is None:
continue
try:
stop = stop or afilter(**params)
except Exception as e:
my_gdb.log.error("after filter '%s' for primitive '%s' crashed ...",
(afilter, self.name))
return stop
@my_gdb.virtual
[docs] def get_parameters(self):
"""
Should return the parameter of the function call.
:returns: parameters and values of the filterable function.
:rtype: dict
"""
return None
@my_gdb.virtual
[docs] def add_return_values(self, data):
"""
Should add information from the return of the function call to `data`.
:param data: object returned by `get_parameters`
:type data: dict
"""
pass