Defaults Facts and Fact Help! UI Element

app/objects/c_ability.py

@@ -24,6 +24,7 @@ class AbilitySchema(ma.Schema):
     repeatable = ma.fields.Bool(missing=None)
     buckets = ma.fields.List(ma.fields.String(), missing=None)
     additional_info = ma.fields.Dict(keys=ma.fields.String(), values=ma.fields.String())
+    fact_descriptions = ma.fields.Dict(keys=ma.fields.String(), values=ma.fields.Dict())
     access = ma.fields.Nested(AccessSchema, missing=None)
     singleton = ma.fields.Bool(missing=None)
     plugin = ma.fields.String(missing=None)
@@ -59,7 +60,7 @@ def executors(self):
 
     def __init__(self, ability_id='', name=None, description=None, tactic=None, technique_id=None, technique_name=None,
                  executors=(), requirements=None, privilege=None, repeatable=False, buckets=None, access=None,
-                 additional_info=None, tags=None, singleton=False, plugin='', delete_payload=True, **kwargs):
+                 additional_info=None, tags=None, singleton=False, plugin='', delete_payload=True, fact_descriptions=None, **kwargs):
         super().__init__()
         self.ability_id = ability_id if ability_id else str(uuid.uuid4())
         self.tactic = tactic.lower() if tactic else None
@@ -80,6 +81,7 @@ def __init__(self, ability_id='', name=None, description=None, tactic=None, tech
             self.access = self.Access(access)
         self.additional_info = additional_info or dict()
         self.additional_info.update(**kwargs)
+        self.fact_descriptions = {}
         self.tags = set(tags) if tags else set()
         self.plugin = plugin
         self.delete_payload = delete_payload

app/objects/secondclass/c_executor.py

@@ -55,7 +55,6 @@ def __init__(self, name, platform, command=None, code=None, language=None, build
         self.code = code
         self.language = language
         self.build_target = build_target
-
         self.payloads = payloads if payloads else []
         self.uploads = uploads if uploads else []
 

app/service/data_svc.py

@@ -10,6 +10,7 @@
 import pathlib
 from importlib import import_module
 
+from app.objects.secondclass.c_fact import Fact
 from app.objects.c_ability import Ability
 from app.objects.c_adversary import Adversary
 from app.objects.c_objective import Objective
@@ -45,6 +46,8 @@
 
 
 class DataService(DataServiceInterface, BaseService):
+    list_of_facts = []
+    fact_descriptions = {}
 
     def __init__(self):
         self.log = self.add_service('data_svc', self)
@@ -116,6 +119,19 @@ async def restore_state(self):
             self.log.debug('Restored data from persistent storage')
         self.log.debug('There are %s jobs in the scheduler' % len(self.ram['schedules']))
 
+    async def create_default_facts(self, data):
+        plugin_facts = []
+        for field in data:
+            default = data[field]["default"]
+            if default:
+                new_fact = Fact(value=default, trait=field, score=1, source="default")
+                plugin_facts.append(new_fact)
+        self.list_of_facts.extend(plugin_facts)
+
+    async def load_default_facts(self):
+        new_source = Source(id="default", name="default", facts=self.list_of_facts, adjustments=[])
+        await self.store(new_source)
+
     async def apply(self, collection):
         if collection not in self.ram:
             self.ram[collection] = []
@@ -169,10 +185,8 @@ async def load_ability_file(self, filename, access):
                 ab.pop('access', None)
                 plugin = self._get_plugin_name(filename)
                 ab.pop('plugin', plugin)
-
                 if tactic and tactic not in filename:
                     self.log.error('Ability=%s has wrong tactic' % ability_id)
-
                 await self._create_ability(ability_id=ability_id, name=name, description=description, tactic=tactic,
                                            technique_id=technique_id, technique_name=technique_name,
                                            executors=executors, requirements=requirements, privilege=privilege,
@@ -209,6 +223,7 @@ async def load_executors_from_platform_dict(self, platforms):
             for executor_names, executor in platform_executors.items():
 
                 command = executor['command'].strip() if executor.get('command') else None
+
                 cleanup = executor['cleanup'].strip() if executor.get('cleanup') else None
 
                 code = executor['code'].strip() if executor.get('code') else None
@@ -288,12 +303,14 @@ async def _load(self, plugins=()):
                 plugins.append(Plugin(data_dir='data'))
             for plug in plugins:
                 await self._load_payloads(plug)
+                await self._load_fact_description_files(plug)
                 await self._load_abilities(plug, async_tasks)
                 await self._load_objectives(plug)
                 await self._load_adversaries(plug)
                 await self._load_planners(plug)
                 await self._load_sources(plug)
                 await self._load_packers(plug)
+            await self.load_default_facts()
             for task in async_tasks:
                 await task
             await self._load_extensions()
@@ -417,8 +434,25 @@ async def _create_ability(self, ability_id, name=None, description=None, tactic=
                           technique_id=technique_id, technique_name=technique_name, executors=executors,
                           requirements=requirements, privilege=privilege, repeatable=repeatable, buckets=buckets,
                           access=access, singleton=singleton, plugin=plugin, **kwargs)
+        if plugin in self.fact_descriptions:
+            ability.fact_descriptions = self.fact_descriptions[plugin]
+        else:
+            ability.fact_descriptions = {}
         return await self.store(ability)
 
+    async def _load_fact_description_files(self, plugin):
+        filename = f"plugins/{plugin.name}/fact_description.yml"
+        if os.path.exists(filename):
+            await self.load_fact_description_file(filename, plugin.name)
+
+    async def load_fact_description_file(self, filename, plugin_name):
+        try:
+            for entries in self.strip_yml(filename):
+                await self.create_default_facts(entries)
+                self.fact_descriptions[plugin_name] = dict(entries)
+        except Exception as e1:
+            print(f"ERROR: Unable to read fact_description.yml for plugin {plugin_name}::", e1)
+
     async def _prune_non_critical_data(self):
         self.ram.pop('plugins')
         self.ram.pop('obfuscators')

initialize_plugin_facts.py

@@ -0,0 +1,71 @@
+import yaml
+import glob
+import os
+from pathlib import Path
+
+
+def get_fact_descriptions():
+    ability_files = glob.glob("./plugins/**/data/abilities/**/*.yml")
+    facts_per_plugin = {}
+    for ability_path in ability_files:
+        with open(ability_path, "r") as file:
+            plugin_name = Path(ability_path).parents[3].name
+            if plugin_name not in facts_per_plugin:
+                facts_per_plugin[plugin_name] = set()
+            r_ability = yaml.safe_load(file)[0]
+            commands = []
+            if "executors" in r_ability:
+                commands.append(r_ability["executors"][0]["command"])
+            else:
+                for platform in r_ability["platforms"]:
+                    try:
+                        for thingy in r_ability["platforms"][platform]:
+                            commands.append(r_ability["platforms"][platform][thingy]["command"])
+                            commands.append(r_ability["platforms"][platform][thingy]["alt_command"])
+
+                    except KeyError:
+                        pass
+                if not commands:
+                    continue
+        plugin_facts = get_facts_from_commands(commands, plugin_name)
+        facts_per_plugin[plugin_name] = facts_per_plugin[plugin_name] | plugin_facts[plugin_name]
+    return facts_per_plugin
+
+
+def get_facts_from_commands(commands, plugin_name):
+    plugin_facts = {plugin_name: set()}
+    for command in commands:
+        split_command = command.split()
+        for cmd in split_command:
+            if "#{" in cmd:
+                fact_start_idx = cmd.index("#{")
+                fact_end_idx = cmd.index("}")
+                unprocessed_fact_name = cmd[fact_start_idx + 2: fact_end_idx]
+                plugin_facts[plugin_name].add(unprocessed_fact_name)
+    return (plugin_facts)
+
+
+def write_descriptions(facts_per_plugin):
+    for plug in facts_per_plugin:
+        descriptions = {x: {"default": None, "description": ""} for x in facts_per_plugin[plug]}
+        descriptions_path = f"./plugins/{plug}/fact_description.yml"
+        to_write = {}
+        if os.path.exists(descriptions_path):
+            with open(descriptions_path, "r") as f:
+                existing_data = yaml.safe_load(f)
+                if existing_data:
+                    to_write = existing_data
+                    for k in descriptions:
+                        if k not in existing_data:
+                            to_write.update({k: descriptions[k]})
+        else:
+            to_write = descriptions
+
+        if to_write:
+            with open(descriptions_path, "w") as fw:
+                yaml.dump(to_write, fw)
+
+
+if __name__ == "__main__":
+    descriptions = get_fact_descriptions()
+    write_descriptions(descriptions)