Add exhaustive pytest test suite with fixes

.caldera-shim/app/objects/c_ability.py

@@ -0,0 +1,12 @@
+"""Stub for Caldera Ability."""
+
+class Ability:
+    def __init__(self, ability_id='', tactic='', technique_id='', technique_name='',
+                 name='', description='', executors=None, **kwargs):
+        self.ability_id = ability_id
+        self.tactic = tactic
+        self.technique_id = technique_id
+        self.technique_name = technique_name
+        self.name = name
+        self.description = description
+        self.executors = executors or []

.caldera-shim/app/objects/c_adversary.py

@@ -0,0 +1,8 @@
+"""Stub for Caldera Adversary."""
+
+class Adversary:
+    def __init__(self, adversary_id='', name='', description='', atomic_ordering=None, **kwargs):
+        self.adversary_id = adversary_id
+        self.name = name
+        self.description = description
+        self.atomic_ordering = atomic_ordering or {}

.caldera-shim/app/objects/c_agent.py

@@ -0,0 +1,27 @@
+"""Stub for Caldera Agent."""
+from datetime import datetime
+
+class Agent:
+    def __init__(self, sleep_min=30, sleep_max=60, watchdog=0, platform='', host='',
+                 username='', architecture='', group='', location='', pid=0, ppid=0,
+                 executors=None, privilege='', exe_name='', contact='', paw='', **kwargs):
+        self.sleep_min = sleep_min
+        self.sleep_max = sleep_max
+        self.watchdog = watchdog
+        self.platform = platform
+        self.host = host
+        self.username = username
+        self.architecture = architecture
+        self.group = group
+        self.location = location
+        self.pid = pid
+        self.ppid = ppid
+        self.executors = executors or []
+        self.privilege = privilege
+        self.exe_name = exe_name
+        self.contact = contact
+        self.paw = paw
+        self.unique = paw
+        self.display_name = paw
+        self.created = datetime.now()
+        self.origin_link_id = None

.caldera-shim/app/objects/c_operation.py

@@ -0,0 +1,20 @@
+"""Stub for Caldera Operation."""
+from datetime import datetime
+
+class Operation:
+    def __init__(self, name='', agents=None, adversary=None, **kwargs):
+        self.name = name
+        self.agents = agents or []
+        self.adversary = adversary
+        self.chain = []
+        self.id = kwargs.get('id', name)
+        self.state = 'finished'
+        self.finish = None
+        self.created = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        self.source = None
+        self.planner = None
+        self.objective = None
+        self.display = {'name': name}
+
+    def set_start_details(self):
+        pass

.caldera-shim/app/objects/secondclass/c_executor.py

@@ -0,0 +1,7 @@
+"""Stub for Caldera Executor."""
+
+class Executor:
+    def __init__(self, name='', platform='', command='', **kwargs):
+        self.name = name
+        self.platform = platform
+        self.command = command

.caldera-shim/app/objects/secondclass/c_link.py

@@ -0,0 +1,31 @@
+"""Stub for Caldera Link."""
+import base64
+from datetime import datetime
+
+
+class Link:
+    def __init__(self, command='', plaintext_command='', paw='', ability=None,
+                 executor=None, **kwargs):
+        self.command = command
+        self.plaintext_command = plaintext_command
+        self.paw = paw
+        self.ability = ability
+        self.executor = executor
+        self.status = kwargs.get('status', -3)
+        self.host = kwargs.get('host', '')
+        self.pid = 0
+        self.decide = kwargs.get('decide', datetime.now())
+        self.collect = kwargs.get('collect', None)
+        self.finish = kwargs.get('finish', '')
+        self.unique = kwargs.get('unique', str(id(self)))
+        self.id = self.unique
+        self.cleanup = kwargs.get('cleanup', 0)
+        self.facts = kwargs.get('facts', [])
+        self.created = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
+    @staticmethod
+    def decode_bytes(encoded_cmd):
+        try:
+            return base64.b64decode(encoded_cmd).decode('utf-8')
+        except Exception:
+            return str(encoded_cmd)

.caldera-shim/app/service/auth_svc.py

@@ -0,0 +1,9 @@
+"""Stub for Caldera auth_svc."""
+
+def for_all_public_methods(decorator):
+    def wrapper(cls):
+        return cls
+    return wrapper
+
+def check_authorization(func):
+    return func

.caldera-shim/app/utility/base_object.py

@@ -0,0 +1,4 @@
+"""Stub for Caldera BaseObject."""
+
+class BaseObject:
+    TIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ'

.caldera-shim/app/utility/base_service.py

@@ -0,0 +1,5 @@
+"""Stub for Caldera BaseService."""
+
+class BaseService:
+    def get_config(self, **kwargs):
+        return {}

.caldera-shim/app/utility/base_world.py

@@ -0,0 +1,21 @@
+"""Stub for Caldera BaseWorld."""
+from enum import Enum
+
+
+class BaseWorld:
+    class Access(Enum):
+        RED = 'red'
+        BLUE = 'blue'
+        APP = 'app'
+
+    @staticmethod
+    def get_config(prop=None, name=None):
+        return None
+
+    @staticmethod
+    def apply_config(name, config):
+        pass
+
+    @staticmethod
+    def strip_yml(path):
+        return [{}]

.caldera-shim/plugins/debrief

@@ -0,0 +1 @@
+../../

app/debrief_svc.py

@@ -25,14 +25,14 @@ async def build_steps_d3(self, operation_ids):
 
         for operation in operations:
             # Add operation node
-            graph_output['nodes'].append(dict(name=operation.name, type='operation', id=op_id, img='operation',
+            graph_output['nodes'].append(dict(name=operation.name, type='operation', id=operation.id, img='operation',
                                               timestamp=self._format_timestamp(operation.created)))
 
             # Add agents for this operation
             agents = [x for x in operation.agents if x]
             self._add_agents_to_d3(agents, id_store, graph_output)
             for agent in agents:
-                graph_output['links'].append(dict(source=op_id,
+                graph_output['links'].append(dict(source=operation.id,
                                                   target=id_store['agent' + agent.unique],
                                                   type='has_agent'))
 
@@ -42,12 +42,12 @@ async def build_steps_d3(self, operation_ids):
                 link_graph_id = id_store['link' + link.unique] = max(id_store.values()) + 1
                 display_name = link.ability.name + (' (cleanup)' if link.cleanup else '')
                 graph_output['nodes'].append(dict(type='link', name='link:'+link.unique, id=link_graph_id,
-                                                  status=link.status, operation=op_id, img=link.ability.tactic,
+                                                  status=link.status, operation=operation.id, img=link.ability.tactic,
                                                   attrs=dict(status=link.status, name=display_name),
                                                   timestamp=self._format_timestamp(link.created)))
 
                 if not previous_link_graph_id:
-                    graph_output['links'].append(dict(source=op_id, target=link_graph_id, type='next_link'))
+                    graph_output['links'].append(dict(source=operation.id, target=link_graph_id, type='next_link'))
                 else:
                     graph_output['links'].append(dict(source=previous_link_graph_id, target=link_graph_id,
                                                       type='next_link'))

app/objects/c_story.py

@@ -99,12 +99,16 @@ def adjust_icon_svgs(path):
         for icon_svg in svg.getroot().iter("{http://www.w3.org/2000/svg}svg"):
             if icon_svg.get('id') == 'copy-svg':
                 continue
-            viewbox = [int(float(val)) for val in icon_svg.get('viewBox').split()]
+            viewbox_attr = icon_svg.get('viewBox')
+            if not viewbox_attr:
+                continue
+            viewbox = [int(float(val)) for val in viewbox_attr.split()]
             aspect = viewbox[2] / viewbox[3]
             icon_svg.set('width', str(round(float(icon_svg.get('height')) * aspect)))
             if not icon_svg.get('id') or 'legend' not in icon_svg.get('id'):
                 icon_svg.set('x', '-' + str(int(icon_svg.get('width')) / 2))
-        svg.write(open(path, 'wb'))
+        with open(path, 'wb') as f:
+            svg.write(f)
 
     @staticmethod
     def get_table_object(val):

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+asyncio_mode = auto
+pythonpath = .caldera-shim
+markers =
+    asyncio: mark a test as an asyncio coroutine