Source code for owtf.managers.plugin


This module manages the plugins and their dependencies

import imp
import json
import os

from sqlalchemy import or_

from owtf.db import models
from owtf.settings import PLUGINS_DIR
from owtf.utils.error import abort_framework
from owtf.utils.file import FileOperations
from owtf.utils.timer import timer

TEST_GROUPS = ['web', 'network', 'auxiliary']

[docs]def get_test_groups_config(file_path): """Reads the test groups from a config file .. note:: This needs to be a list instead of a dictionary to preserve order in python < 2.7 :param file_path: The path to the config file :type file_path: `str` :return: List of test groups :rtype: `list` """ test_groups = [] config_file =, 'r').read().splitlines() for line in config_file: if '#' == line[0]: continue # Skip comments try: code, priority, descrip, hint, url = line.strip().split(' | ') except ValueError: abort_framework("Problem in Test Groups file: '%s' -> Cannot parse line: %s" % (file_path, line)) if len(descrip) < 2: descrip = hint if len(hint) < 2: hint = "" test_groups.append({'code': code, 'priority': priority, 'descrip': descrip, 'hint': hint, 'url': url}) return test_groups
[docs]def load_test_groups(session, file_default, file_fallback, plugin_group): """Load test groups into the DB. :param test_groups_file: The path to the test groups config :type test_groups_file: `str` :param plugin_group: Plugin group to load :type plugin_group: `str` :return: None :rtype: None """ file_path = file_default if not os.path.isfile(file_default): file_path = file_fallback test_groups = get_test_groups_config(file_path) for group in test_groups: session.merge( models.TestGroup( code=group['code'], priority=group['priority'], descrip=group['descrip'], hint=group['hint'], url=group['url'], group=plugin_group)) session.commit()
[docs]def load_plugins(session): """Loads the plugins from the filesystem and updates their info. .. note:: Walks through each sub-directory of `PLUGINS_DIR`. For each file, loads it thanks to the imp module. Updates the database with the information for each plugin: + 'title': the title of the plugin + 'name': the name of the plugin + 'code': the internal code of the plugin + 'group': the group of the plugin (ex: web) + 'type': the type of the plugin (ex: active, passive, ...) + 'descrip': the description of the plugin + 'file': the filename of the plugin + 'internet_res': does the plugin use internet resources? :return: None :rtype: None """ # TODO: When the -t, -e or -o is given to OWTF command line, only load # the specific plugins (and not all of them like below). # Retrieve the list of the plugins (sorted) from the directory given by # 'PLUGIN_DIR'. plugins = [] for root, _, files in os.walk(PLUGINS_DIR): plugins.extend([os.path.join(root, filename) for filename in files if filename.endswith('py')]) plugins = sorted(plugins) # Retrieve the information of the plugin. for plugin_path in plugins: # Only keep the relative path to the plugin plugin = plugin_path.replace(PLUGINS_DIR, '') # TODO: Using os.path.sep might not be portable especially on # Windows platform since it allows '/' and '\' in the path. # Retrieve the group, the type and the file of the plugin. # Ensure all empty strings are removed from the list chunks = list(filter(None, plugin.split(os.path.sep))) # TODO: Ensure that the variables group, type and file exist when # the length of chunks is less than 3. if len(chunks) == 3: group, type, file = chunks # Retrieve the internal name and code of the plugin. name, code = os.path.splitext(file)[0].split('@') # Only load the plugin if in XXX_TEST_GROUPS configuration (e.g. web_testgroups.cfg) if session.query(models.TestGroup).get(code) is None: continue # Load the plugin as a module. filename, pathname, desc = imp.find_module( os.path.splitext(os.path.basename(plugin_path))[0], [os.path.dirname(plugin_path)]) plugin_module = imp.load_module(os.path.splitext(file)[0], filename, pathname, desc) # Try te retrieve the `attr` dictionary from the module and convert # it to json in order to save it into the database. attr = None try: attr = json.dumps(plugin_module.ATTR) except AttributeError: # The plugin didn't define an attr dict. pass # Save the plugin into the database. session.merge( models.Plugin( key='%s@%s' % (type, code), group=group, type=type, title=name.title().replace('_', ' '), name=name, code=code, file=file, descrip=plugin_module.DESCRIPTION, attr=attr)) session.commit()
[docs]def derive_test_group_dict(obj): """Fetch the test group dict from the obj :param obj: The test group object :type obj: :return: Test group dict :rtype: `dict` """ if obj: pdict = dict(obj.__dict__) pdict.pop("_sa_instance_state") return pdict
[docs]def derive_test_group_dicts(obj_list): """Fetch the test group dicts from the obj list :param obj_list: The test group object list :type obj_list: `list` :return: Test group dicts in a list :rtype: `list` """ dict_list = [] for obj in obj_list: dict_list.append(derive_test_group_dict(obj)) return dict_list
[docs]def get_test_group(session, code): """Get the test group based on plugin code :param code: Plugin code :type code: `str` :return: Test group dict :rtype: `dict` """ group = session.query(models.TestGroup).get(code) return derive_test_group_dict(group)
[docs]def get_all_test_groups(session): """Get all test groups from th DB :return: :rtype: """ test_groups = session.query(models.TestGroup).order_by(models.TestGroup.priority.desc()).all() return derive_test_group_dicts(test_groups)
[docs]def get_all_plugin_groups(session): """Get all plugin groups from the DB :return: List of available plugin groups :rtype: `list` """ groups = session.query( groups = [i[0] for i in groups] return groups
[docs]def get_all_plugin_types(session): """Get all plugin types from the DB :return: All available plugin types :rtype: `list` """ plugin_types = session.query(models.Plugin.type).distinct().all() plugin_types = [i[0] for i in plugin_types] # Necessary because of sqlalchemy return plugin_types
[docs]def get_types_for_plugin_group(session, plugin_group): """Get available plugin types for a plugin group :param plugin_group: Plugin group :type plugin_group: `str` :return: List of available plugin types :rtype: `list` """ plugin_types = session.query(models.Plugin.type).filter_by(group=plugin_group).distinct().all() plugin_types = [i[0] for i in plugin_types] return plugin_types
[docs]def derive_plugin_dict(obj): """Fetch the plugin dict from an object :param obj: Plugin object :type obj: :return: Plugin dict :rtype: `dict` """ if obj: pdict = dict(obj.__dict__) pdict.pop("_sa_instance_state") # Remove outputs array if present if "outputs" in list(pdict.keys()): pdict.pop("outputs") pdict["min_time"] = None min_time = obj.min_time if min_time is not None: pdict["min_time"] = timer.get_time_as_str(min_time) return pdict
[docs]def derive_plugin_dicts(obj_list): """Fetch plugin dicts from a obj list :param obj_list: List of plugin objects :type obj_list: `list` :return: List of plugin dicts :rtype: `list` """ plugin_dicts = [] for obj in obj_list: plugin_dicts.append(derive_plugin_dict(obj)) return plugin_dicts
[docs]def plugin_gen_query(session, criteria): """Generate a SQLAlchemy query based on the filter criteria :param criteria: Filter criteria :type criteria: `dict` :return: :rtype: """ query = session.query(models.Plugin).join(models.TestGroup) if criteria.get("type", None): if isinstance(criteria["type"], str): query = query.filter(models.Plugin.type == criteria["type"]) if isinstance(criteria["type"], list): query = query.filter(models.Plugin.type.in_(criteria["type"])) if criteria.get("group", None): if isinstance(criteria["group"], str): query = query.filter( == criteria["group"]) if isinstance(criteria["group"], list): query = query.filter(["group"])) if criteria.get("code", None): if isinstance(criteria["code"], str): query = query.filter(models.Plugin.code == criteria["code"]) if isinstance(criteria["code"], list): query = query.filter(models.Plugin.code.in_(criteria["code"])) if criteria.get("name", None): if isinstance(criteria["name"], str): query = query.filter( == criteria["name"]) if isinstance(criteria["name"], list): query = query.filter(["name"])) return query.order_by(models.TestGroup.priority.desc())
[docs]def plugin_name_to_code(session, codes): """Given list of names, get the corresponding codes :param codes: The codes to fetch :type codes: `list` :return: Corresponding plugin codes as a list :rtype: `list` """ checklist = ["OWTF-", "PTES-"] query = session.query(models.Plugin.code) for count, name in enumerate(codes): if all(check not in name for check in checklist): code = query.filter( == name).first() codes[count] = str(code[0]) return codes
[docs]def get_all_plugin_dicts(session, criteria=None): """Get plugin dicts based on filter criteria :param criteria: Filter criteria :type criteria: `dict` :return: List of plugin dicts :rtype: `list` """ if criteria is None: criteria = {} if "code" in criteria: criteria["code"] = plugin_name_to_code(session, criteria["code"]) query = plugin_gen_query(session, criteria) plugin_obj_list = query.all() return derive_plugin_dicts(plugin_obj_list)
[docs]def get_plugins_by_type(session, plugin_type): """Get plugins based on type argument :param plugin_type: Plugin type :type plugin_type: `str` :return: List of plugin dicts :rtype: `list` """ return get_all_plugin_dicts(session, {"type": plugin_type})
[docs]def get_plugins_by_group(session, plugin_group): """Get plugins by plugin group :param plugin_group: Plugin group :type plugin_group: `str` :return: List of plugin dicts :rtype: `list` """ return get_all_plugin_dicts(session, {"group": plugin_group})
[docs]def get_plugins_by_group_type(session, plugin_group, plugin_type): """Get plugins by group and plugin type :param plugin_group: Plugin group :type plugin_group: `str` :param plugin_type: plugin type :type plugin_type: `str` :return: List of plugin dicts :rtype: `list` """ return get_all_plugin_dicts(session, {"type": plugin_type, "group": plugin_group})
[docs]def get_groups_for_plugins(session, plugins): """Gets available groups for selected plugins :param plugins: Plugins selected :type plugins: `list` :return: List of available plugin groups :rtype: `list` """ groups = session.query( or_(models.Plugin.code.in_(plugins), groups = [i[0] for i in groups] return groups