diff --git a/.caldera-shim/app/__init__.py b/.caldera-shim/app/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/app/objects/__init__.py b/.caldera-shim/app/objects/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/app/objects/c_ability.py b/.caldera-shim/app/objects/c_ability.py
new file mode 100644
index 0000000..6647dbb
--- /dev/null
+++ b/.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 []
diff --git a/.caldera-shim/app/objects/c_adversary.py b/.caldera-shim/app/objects/c_adversary.py
new file mode 100644
index 0000000..5f18d11
--- /dev/null
+++ b/.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 {}
diff --git a/.caldera-shim/app/objects/c_agent.py b/.caldera-shim/app/objects/c_agent.py
new file mode 100644
index 0000000..d56b8c6
--- /dev/null
+++ b/.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
diff --git a/.caldera-shim/app/objects/c_operation.py b/.caldera-shim/app/objects/c_operation.py
new file mode 100644
index 0000000..9cdba0b
--- /dev/null
+++ b/.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
diff --git a/.caldera-shim/app/objects/secondclass/__init__.py b/.caldera-shim/app/objects/secondclass/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/app/objects/secondclass/c_executor.py b/.caldera-shim/app/objects/secondclass/c_executor.py
new file mode 100644
index 0000000..77ff081
--- /dev/null
+++ b/.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
diff --git a/.caldera-shim/app/objects/secondclass/c_link.py b/.caldera-shim/app/objects/secondclass/c_link.py
new file mode 100644
index 0000000..df073e1
--- /dev/null
+++ b/.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)
diff --git a/.caldera-shim/app/service/__init__.py b/.caldera-shim/app/service/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/app/service/auth_svc.py b/.caldera-shim/app/service/auth_svc.py
new file mode 100644
index 0000000..2c6e4a2
--- /dev/null
+++ b/.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
diff --git a/.caldera-shim/app/utility/__init__.py b/.caldera-shim/app/utility/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/app/utility/base_object.py b/.caldera-shim/app/utility/base_object.py
new file mode 100644
index 0000000..d27b6ec
--- /dev/null
+++ b/.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'
diff --git a/.caldera-shim/app/utility/base_service.py b/.caldera-shim/app/utility/base_service.py
new file mode 100644
index 0000000..ad3cfeb
--- /dev/null
+++ b/.caldera-shim/app/utility/base_service.py
@@ -0,0 +1,5 @@
+"""Stub for Caldera BaseService."""
+
+class BaseService:
+ def get_config(self, **kwargs):
+ return {}
diff --git a/.caldera-shim/app/utility/base_world.py b/.caldera-shim/app/utility/base_world.py
new file mode 100644
index 0000000..6ead54e
--- /dev/null
+++ b/.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 [{}]
diff --git a/.caldera-shim/plugins/__init__.py b/.caldera-shim/plugins/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/.caldera-shim/plugins/debrief b/.caldera-shim/plugins/debrief
new file mode 120000
index 0000000..6581736
--- /dev/null
+++ b/.caldera-shim/plugins/debrief
@@ -0,0 +1 @@
+../../
\ No newline at end of file
diff --git a/app/debrief_svc.py b/app/debrief_svc.py
index daf4c17..c956782 100644
--- a/app/debrief_svc.py
+++ b/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'))
diff --git a/app/objects/c_story.py b/app/objects/c_story.py
index 0109405..b1a6023 100644
--- a/app/objects/c_story.py
+++ b/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):
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..ef41db4
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+asyncio_mode = auto
+pythonpath = .caldera-shim
+markers =
+ asyncio: mark a test as an asyncio coroutine
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..6131ccc
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,313 @@
+"""Shared fixtures for debrief plugin test suite."""
+import pytest
+import asyncio
+import importlib
+import json
+import os
+import sys
+
+from base64 import b64encode
+from datetime import datetime
+from enum import Enum
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from plugins.debrief.attack_mapper import Attack18Map
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+LINK_DECIDE_TIME = '2021-01-01T08:00:00Z'
+LINK_COLLECT_TIME = '2021-01-01T08:01:00Z'
+LINK_FINISH_TIME = '2021-01-01T08:02:00Z'
+
+
+class _OriginType(Enum):
+ IMPORTED = 'IMPORTED'
+ LEARNED = 'LEARNED'
+ SEEDED = 'SEEDED'
+
+
+def _encode(s):
+ return b64encode(s.encode('utf-8')).decode()
+
+
+def _make_ability(ability_id='ab-1', tactic='discovery', technique_id='T1082',
+ technique_name='System Information Discovery', name='Test ability',
+ cleanup=False):
+ ab = SimpleNamespace(
+ ability_id=ability_id,
+ tactic=tactic,
+ technique_id=technique_id,
+ technique_name=technique_name,
+ name=name,
+ )
+ return ab
+
+
+def _make_link(paw='paw1', ability=None, status=0, cleanup=0, unique='link1',
+ command=None, facts=None, created='2021-01-01 08:00:00', finish=LINK_FINISH_TIME,
+ link_id=None):
+ if ability is None:
+ ability = _make_ability()
+ if command is None:
+ command = _encode('whoami')
+ lnk = SimpleNamespace(
+ paw=paw,
+ ability=ability,
+ status=status,
+ cleanup=cleanup,
+ unique=unique,
+ command=command,
+ facts=facts or [],
+ created=created,
+ finish=finish,
+ id=link_id or unique,
+ )
+ lnk.decode_bytes = staticmethod(lambda c: c)
+ return lnk
+
+
+def _make_agent(paw='paw1', platform='windows', host='WORKSTATION', group='red',
+ display_name='Agent1', unique='agent1', username='user1',
+ privilege='User', exe_name='agent.exe',
+ created=None, origin_link_id=None):
+ if created is None:
+ created = datetime(2021, 1, 1, 8, 0, 0)
+ return SimpleNamespace(
+ paw=paw,
+ platform=platform,
+ host=host,
+ group=group,
+ display_name=display_name,
+ unique=unique,
+ username=username,
+ privilege=privilege,
+ exe_name=exe_name,
+ created=created,
+ origin_link_id=origin_link_id,
+ )
+
+
+def _make_source(source_id='src-1', facts=None):
+ return SimpleNamespace(id=source_id, facts=facts or [])
+
+
+def _make_fact(trait='host.user.name', value='admin', score=1, unique='fact1',
+ origin_type=None, source='src-1', links=None, collected_by=None,
+ created=None):
+ if origin_type is None:
+ origin_type = _OriginType.LEARNED
+ if created is None:
+ created = datetime(2021, 1, 1, 8, 1, 0)
+ return SimpleNamespace(
+ trait=trait,
+ value=value,
+ score=score,
+ unique=unique,
+ origin_type=origin_type,
+ source=source,
+ links=links or [],
+ collected_by=collected_by or [],
+ created=created,
+ )
+
+
+def _make_planner(name='atomic'):
+ return SimpleNamespace(name=name)
+
+
+def _make_objective(name='default'):
+ return SimpleNamespace(name=name)
+
+
+def _make_operation(name='TestOp', agents=None, chain=None, source=None,
+ state='finished', finish='2021-01-01 09:00:00',
+ created='2021-01-01 08:00:00', op_id='op-1',
+ planner=None, objective=None, facts=None, relationships=None):
+ if agents is None:
+ agents = [_make_agent()]
+ if chain is None:
+ chain = [_make_link()]
+ if source is None:
+ source = _make_source()
+ if planner is None:
+ planner = _make_planner()
+ if objective is None:
+ objective = _make_objective()
+
+ async def all_facts():
+ return facts or []
+
+ async def all_relationships():
+ return relationships or []
+
+ op = SimpleNamespace(
+ id=op_id,
+ name=name,
+ agents=agents,
+ chain=chain,
+ source=source,
+ state=state,
+ finish=finish,
+ created=created,
+ planner=planner,
+ objective=objective,
+ display={'name': name, 'id': op_id},
+ all_facts=all_facts,
+ all_relationships=all_relationships,
+ )
+ return op
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def make_ability():
+ return _make_ability
+
+
+@pytest.fixture
+def make_link():
+ return _make_link
+
+
+@pytest.fixture
+def make_agent():
+ return _make_agent
+
+
+@pytest.fixture
+def make_operation():
+ return _make_operation
+
+
+@pytest.fixture
+def make_fact():
+ return _make_fact
+
+
+@pytest.fixture
+def make_source():
+ return _make_source
+
+
+@pytest.fixture
+def encode_command():
+ return _encode
+
+
+@pytest.fixture
+def sample_agent():
+ return _make_agent()
+
+
+@pytest.fixture
+def sample_agent_linux():
+ return _make_agent(paw='paw2', platform='linux', host='SERVER', unique='agent2',
+ display_name='Agent2', username='linuser', exe_name='agent')
+
+
+@pytest.fixture
+def sample_ability():
+ return _make_ability()
+
+
+@pytest.fixture
+def sample_link():
+ return _make_link()
+
+
+@pytest.fixture
+def sample_operation():
+ return _make_operation()
+
+
+@pytest.fixture
+def two_op_chain():
+ """Two links in chain with different tactics/techniques."""
+ ab1 = _make_ability(ability_id='ab-1', tactic='discovery', technique_id='T1082',
+ technique_name='System Information Discovery', name='Sys Info')
+ ab2 = _make_ability(ability_id='ab-2', tactic='collection', technique_id='T1005',
+ technique_name='Data from Local System', name='Collect Data')
+ link1 = _make_link(paw='paw1', ability=ab1, unique='lnk1', created='2021-01-01 08:00:00')
+ link2 = _make_link(paw='paw1', ability=ab2, unique='lnk2', created='2021-01-01 08:01:00')
+ return [link1, link2]
+
+
+@pytest.fixture
+def multi_op_operations(sample_agent, two_op_chain):
+ op1 = _make_operation(name='Op1', op_id='op-1', agents=[sample_agent], chain=two_op_chain)
+ op2 = _make_operation(name='Op2', op_id='op-2', agents=[sample_agent],
+ chain=[_make_link(paw='paw1', ability=_make_ability(
+ ability_id='ab-3', tactic='discovery', technique_id='T1082',
+ technique_name='System Information Discovery', name='Sys Info 2'),
+ unique='lnk3')])
+ return [op1, op2]
+
+
+@pytest.fixture
+def mock_services():
+ """Minimal mock services dict for DebriefService / DebriefGui."""
+ data_svc = AsyncMock()
+ data_svc.locate = AsyncMock(return_value=[])
+
+ app_svc = AsyncMock()
+ app_svc.find_op_with_link = AsyncMock(return_value=None)
+ app_svc.application = MagicMock()
+ app_svc.application.router = MagicMock()
+
+ auth_svc = AsyncMock()
+ auth_svc.get_permissions = AsyncMock(return_value=[])
+
+ file_svc = AsyncMock()
+ knowledge_svc = AsyncMock()
+
+ return {
+ 'data_svc': data_svc,
+ 'app_svc': app_svc,
+ 'auth_svc': auth_svc,
+ 'file_svc': file_svc,
+ 'knowledge_svc': knowledge_svc,
+ }
+
+
+@pytest.fixture
+def mock_attack18_map():
+ """Return an Attack18Map with empty indexes for isolated unit testing."""
+ return Attack18Map({
+ 'techniques_by_id': {},
+ 'strategies_by_tid': {},
+ 'analytics_by_tid': {},
+ })
+
+
+@pytest.fixture
+def populated_attack18_map():
+ """Return an Attack18Map with some sample data."""
+ return Attack18Map({
+ 'techniques_by_id': {
+ 'T1082': {'technique_id': 'T1082', 'name': 'System Information Discovery'},
+ 'T1005': {'technique_id': 'T1005', 'name': 'Data from Local System'},
+ 'T1547.001': {'technique_id': 'T1547.001', 'name': 'Registry Run Keys'},
+ },
+ 'strategies_by_tid': {
+ 'T1082': [{'id': 's1', 'name': 'Detect SysInfo', 'det_id': 'DET0001',
+ 'external_references': []}],
+ 'T1547': [{'id': 's2', 'name': 'Detect Persistence', 'det_id': 'DET0002',
+ 'external_references': []}],
+ },
+ 'analytics_by_tid': {
+ 'T1082': [
+ {'id': 'a1', 'an_id': 'AN0001', 'name': 'Analytic 0001',
+ 'platform': 'windows', 'platforms': ['windows'],
+ 'statement': 'Monitor sysinfo', 'tunables': [],
+ 'dc_elements': [{'name': 'WinEventLog:Security', 'channel': '4688',
+ 'data_component': 'Process Creation'}],
+ 'det_id': 'DET0001'},
+ ],
+ },
+ })
diff --git a/tests/test_attack_mapper_extended.py b/tests/test_attack_mapper_extended.py
new file mode 100644
index 0000000..717f71a
--- /dev/null
+++ b/tests/test_attack_mapper_extended.py
@@ -0,0 +1,441 @@
+"""Extended tests for attack_mapper.py — covers Attack18Map, index_bundle, utilities, fetch_and_cache."""
+import pytest
+import json
+import os
+from unittest.mock import AsyncMock, MagicMock, patch, mock_open
+
+from plugins.debrief.attack_mapper import (
+ Attack18Map,
+ NormalizeAnalyticException,
+ index_bundle,
+ fetch_and_cache,
+ load_attack18_cache,
+ get_attack18,
+ _parent_tid,
+ _extract_tid,
+ _normalize_analytic,
+ CACHE_PATH,
+)
+
+
+# ===========================================================================
+# Attack18Map
+# ===========================================================================
+class TestAttack18Map:
+ def test_empty_init(self):
+ m = Attack18Map({})
+ assert m.techniques_by_id == {}
+ assert m.strategies_by_tid == {}
+ assert m.analytics_by_tid == {}
+
+ def test_none_init(self):
+ m = Attack18Map(None)
+ assert m.techniques_by_id == {}
+
+ def test_get_strategies_present(self, populated_attack18_map):
+ result = populated_attack18_map.get_strategies('T1082')
+ assert len(result) == 1
+ assert result[0]['det_id'] == 'DET0001'
+
+ def test_get_strategies_missing(self, populated_attack18_map):
+ result = populated_attack18_map.get_strategies('T9999')
+ assert result == []
+
+ def test_get_parent_strategies(self, populated_attack18_map):
+ result = populated_attack18_map.get_parent_strategies('T1547.001')
+ assert len(result) == 1
+ assert result[0]['det_id'] == 'DET0002'
+
+ def test_get_parent_strategies_no_sub(self, populated_attack18_map):
+ result = populated_attack18_map.get_parent_strategies('T1082')
+ assert len(result) == 1 # parent of T1082 is T1082
+
+ def test_get_analytics(self, populated_attack18_map):
+ result = populated_attack18_map.get_analytics('T1082')
+ assert len(result) == 1
+
+ def test_get_analytics_with_platform_filter(self, populated_attack18_map):
+ result = populated_attack18_map.get_analytics('T1082', platform='windows')
+ assert len(result) == 1
+
+ def test_get_analytics_wrong_platform(self, populated_attack18_map):
+ result = populated_attack18_map.get_analytics('T1082', platform='macos')
+ assert len(result) == 0
+
+ def test_get_analytics_missing_tid(self, populated_attack18_map):
+ result = populated_attack18_map.get_analytics('T9999')
+ assert result == []
+
+ def test_get_parent_analytics(self, populated_attack18_map):
+ result = populated_attack18_map.get_parent_analytics('T1082.001', platform='windows')
+ # Parent is T1082
+ assert len(result) == 1
+
+ def test_get_analytics_empty_platform(self, populated_attack18_map):
+ result = populated_attack18_map.get_analytics('T1082', platform='')
+ assert len(result) == 1 # Empty platform returns all
+
+
+# ===========================================================================
+# _parent_tid
+# ===========================================================================
+class TestParentTid:
+ def test_main_technique(self):
+ assert _parent_tid('T1082') == 'T1082'
+
+ def test_subtechnique(self):
+ assert _parent_tid('T1547.001') == 'T1547'
+
+ def test_empty(self):
+ assert _parent_tid('') == ''
+
+ def test_none(self):
+ assert _parent_tid(None) == ''
+
+ def test_lowercase(self):
+ assert _parent_tid('t1082') == 'T1082'
+
+
+# ===========================================================================
+# _extract_tid
+# ===========================================================================
+class TestExtractTid:
+ def test_valid_technique(self):
+ obj = {'external_references': [
+ {'source_name': 'mitre-attack', 'external_id': 'T1082'}
+ ]}
+ assert _extract_tid(obj) == 'T1082'
+
+ def test_no_mitre_source(self):
+ obj = {'external_references': [
+ {'source_name': 'other', 'external_id': 'T1082'}
+ ]}
+ assert _extract_tid(obj) is None
+
+ def test_no_external_refs(self):
+ assert _extract_tid({}) is None
+ assert _extract_tid({'external_references': None}) is None
+
+ def test_non_technique_id(self):
+ obj = {'external_references': [
+ {'source_name': 'mitre-attack', 'external_id': 'S0001'}
+ ]}
+ assert _extract_tid(obj) is None
+
+ def test_subtechnique(self):
+ obj = {'external_references': [
+ {'source_name': 'mitre-attack', 'external_id': 'T1547.001'}
+ ]}
+ assert _extract_tid(obj) == 'T1547.001'
+
+
+# ===========================================================================
+# _normalize_analytic
+# ===========================================================================
+class TestNormalizeAnalytic:
+ def test_basic_analytic(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0001',
+ 'description': 'Test statement',
+ 'x_mitre_platforms': ['windows'],
+ 'x_mitre_mutable_elements': [],
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_log_source_references': [],
+ }
+ row, dcs = _normalize_analytic(a, {})
+ assert row['an_id'] == 'AN0001'
+ assert row['platform'] == 'windows'
+ assert row['platforms'] == ['windows']
+ assert row['statement'] == 'Test statement'
+ assert dcs == []
+
+ def test_analytic_with_dash_in_id(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN-0001'}],
+ 'x_mitre_log_source_references': [],
+ }
+ row, _ = _normalize_analytic(a, {})
+ assert row['an_id'] == 'AN0001'
+
+ def test_analytic_id_from_name_fallback(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0042',
+ 'external_references': [],
+ 'x_mitre_log_source_references': [],
+ }
+ row, _ = _normalize_analytic(a, {})
+ assert row['an_id'] == 'AN0042'
+
+ def test_raises_if_no_id(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'NoIdHere',
+ 'external_references': [],
+ 'x_mitre_log_source_references': [],
+ }
+ with pytest.raises(NormalizeAnalyticException):
+ _normalize_analytic(a, {})
+
+ def test_with_data_components(self):
+ dc_id = 'dc-1'
+ dc = {'name': 'Process Creation'}
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_log_source_references': [
+ {
+ 'name': 'WinEventLog:Security',
+ 'channel': '4688',
+ 'x_mitre_data_component_ref': dc_id,
+ }
+ ],
+ }
+ row, dcs = _normalize_analytic(a, {dc_id: dc})
+ assert len(dcs) == 1
+ assert dcs[0]['data_component'] == 'Process Creation'
+
+ def test_multi_platform(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0001',
+ 'x_mitre_platforms': ['windows', 'linux'],
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_log_source_references': [],
+ }
+ row, _ = _normalize_analytic(a, {})
+ assert row['platforms'] == ['windows', 'linux']
+ assert row['platform'] == 'windows'
+
+ def test_string_platform(self):
+ a = {
+ 'id': 'analytic--1',
+ 'name': 'Analytic 0001',
+ 'platform': 'Linux',
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_log_source_references': [],
+ }
+ row, _ = _normalize_analytic(a, {})
+ assert row['platform'] == 'linux'
+
+
+# ===========================================================================
+# index_bundle
+# ===========================================================================
+class TestIndexBundle:
+ def _make_bundle(self, objects):
+ return {'type': 'bundle', 'objects': objects}
+
+ def test_empty_bundle_raises(self):
+ with pytest.raises(Exception, match='v18'):
+ index_bundle(self._make_bundle([]))
+
+ def test_no_strategies_or_analytics_raises(self):
+ bundle = self._make_bundle([
+ {'type': 'attack-pattern', 'id': 'ap-1',
+ 'external_references': [{'source_name': 'mitre-attack', 'external_id': 'T1082'}]},
+ ])
+ with pytest.raises(Exception, match='v18'):
+ index_bundle(bundle)
+
+ def test_revoked_objects_skipped(self):
+ bundle = self._make_bundle([
+ {'type': 'x-mitre-detection-strategy', 'id': 'ds-1', 'revoked': True,
+ 'external_references': [{'external_id': 'DET0001'}]},
+ {'type': 'x-mitre-analytic', 'id': 'an-1', 'revoked': False,
+ 'name': 'Analytic 0001', 'external_references': [{'external_id': 'AN0001'}]},
+ ])
+ # Only analytic remains, which is enough to not raise
+ idx = index_bundle(bundle)
+ assert 'analytics_by_tid' in idx
+
+ def test_indexes_techniques(self):
+ bundle = self._make_bundle([
+ {'type': 'attack-pattern', 'id': 'ap-1',
+ 'name': 'SysInfo',
+ 'external_references': [{'source_name': 'mitre-attack', 'external_id': 'T1082'}]},
+ {'type': 'x-mitre-analytic', 'id': 'an-1',
+ 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN0001'}]},
+ ])
+ idx = index_bundle(bundle)
+ assert 'T1082' in idx['techniques_by_id']
+
+ def test_strategy_det_id_normalization(self):
+ bundle = self._make_bundle([
+ {'type': 'x-mitre-detection-strategy', 'id': 'ds-1',
+ 'name': 'Test Strat',
+ 'external_references': [{'external_id': 'DET-0012'}],
+ 'x_mitre_analytic_refs': []},
+ {'type': 'x-mitre-analytic', 'id': 'an-1',
+ 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN0001'}]},
+ ])
+ idx = index_bundle(bundle)
+ # DET-0012 should be normalized to DET0012
+ assert idx is not None
+ all_det_ids = [
+ s.get('det_id', '') for strats in idx.get('strategies_by_tid', {}).values() for s in strats
+ ]
+ assert any('DET0012' in d for d in all_det_ids), (
+ f"Expected normalized DET0012 in strategies, got: {all_det_ids}"
+ )
+
+ def test_full_strategy_to_technique_mapping(self):
+ bundle = self._make_bundle([
+ {'type': 'attack-pattern', 'id': 'ap-1', 'name': 'T',
+ 'external_references': [{'source_name': 'mitre-attack', 'external_id': 'T1082'}]},
+ {'type': 'x-mitre-detection-strategy', 'id': 'ds-1', 'name': 'S',
+ 'external_references': [{'external_id': 'DET0001'}],
+ 'x_mitre_analytic_refs': ['an-1']},
+ {'type': 'x-mitre-analytic', 'id': 'an-1', 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_platforms': ['windows'],
+ 'x_mitre_log_source_references': []},
+ {'type': 'relationship', 'id': 'rel-1',
+ 'source_ref': 'ds-1', 'target_ref': 'ap-1',
+ 'relationship_type': 'detects'},
+ ])
+ idx = index_bundle(bundle)
+ assert 'T1082' in idx['strategies_by_tid']
+ assert 'T1082' in idx['analytics_by_tid']
+ assert len(idx['analytics_by_tid']['T1082']) == 1
+
+ def test_deduplication(self):
+ bundle = self._make_bundle([
+ {'type': 'attack-pattern', 'id': 'ap-1', 'name': 'T',
+ 'external_references': [{'source_name': 'mitre-attack', 'external_id': 'T1082'}]},
+ {'type': 'x-mitre-detection-strategy', 'id': 'ds-1', 'name': 'S',
+ 'external_references': [{'external_id': 'DET0001'}],
+ 'x_mitre_analytic_refs': ['an-1']},
+ {'type': 'x-mitre-analytic', 'id': 'an-1', 'name': 'Analytic 0001',
+ 'external_references': [{'external_id': 'AN0001'}],
+ 'x_mitre_platforms': ['windows'],
+ 'x_mitre_log_source_references': []},
+ # Two relationships to same technique
+ {'type': 'relationship', 'source_ref': 'ds-1', 'target_ref': 'ap-1',
+ 'relationship_type': 'detects'},
+ {'type': 'relationship', 'source_ref': 'ds-1', 'target_ref': 'ap-1',
+ 'relationship_type': 'detects'},
+ ])
+ idx = index_bundle(bundle)
+ # Deduplication should keep only one
+ assert len(idx['analytics_by_tid']['T1082']) == 1
+ assert len(idx['strategies_by_tid']['T1082']) == 1
+
+
+# ===========================================================================
+# load_attack18_cache
+# ===========================================================================
+class TestLoadAttack18Cache:
+ def test_load_dict(self, tmp_path):
+ p = tmp_path / 'cache.json'
+ p.write_text(json.dumps({'type': 'bundle', 'objects': []}))
+ result = load_attack18_cache(str(p))
+ assert result['type'] == 'bundle'
+
+ def test_load_list_becomes_bundle(self, tmp_path):
+ p = tmp_path / 'cache.json'
+ p.write_text(json.dumps([{'type': 'attack-pattern'}]))
+ result = load_attack18_cache(str(p))
+ assert result['type'] == 'bundle'
+ assert len(result['objects']) == 1
+
+ def test_load_unexpected_type_raises(self, tmp_path):
+ p = tmp_path / 'cache.json'
+ p.write_text('"just a string"')
+ with pytest.raises(TypeError, match='Unexpected'):
+ load_attack18_cache(str(p))
+
+
+# ===========================================================================
+# fetch_and_cache
+# ===========================================================================
+class TestFetchAndCache:
+ @pytest.mark.asyncio
+ async def test_fetch_success(self, tmp_path):
+ cache_path = str(tmp_path / 'cache.json')
+ mock_resp = AsyncMock()
+ mock_resp.status = 200
+ mock_resp.text = AsyncMock(return_value='{"type": "bundle"}')
+
+ mock_session_get = MagicMock()
+ mock_cm = AsyncMock()
+ mock_cm.__aenter__ = AsyncMock(return_value=mock_resp)
+ mock_cm.__aexit__ = AsyncMock(return_value=False)
+ mock_session_get.return_value = mock_cm
+
+ with patch('plugins.debrief.attack_mapper.CACHE_PATH', cache_path):
+ await fetch_and_cache(mock_session_get, path=cache_path)
+ assert os.path.exists(cache_path)
+
+ @pytest.mark.asyncio
+ async def test_fetch_already_cached(self, tmp_path):
+ cache_path = str(tmp_path / 'cache.json')
+ with open(cache_path, 'w') as f:
+ f.write('{}')
+ mock_session_get = MagicMock()
+ await fetch_and_cache(mock_session_get, path=cache_path)
+ mock_session_get.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_fetch_non_200(self, tmp_path):
+ cache_path = str(tmp_path / 'no_cache.json')
+ mock_resp = AsyncMock()
+ mock_resp.status = 500
+ mock_resp.text = AsyncMock(return_value='error')
+
+ mock_session_get = MagicMock()
+ mock_cm = AsyncMock()
+ mock_cm.__aenter__ = AsyncMock(return_value=mock_resp)
+ mock_cm.__aexit__ = AsyncMock(return_value=False)
+ mock_session_get.return_value = mock_cm
+
+ with pytest.raises(RuntimeError, match='Non-200'):
+ await fetch_and_cache(mock_session_get, path=cache_path)
+
+ @pytest.mark.asyncio
+ async def test_fetch_empty_response(self, tmp_path):
+ cache_path = str(tmp_path / 'no_cache.json')
+ mock_resp = AsyncMock()
+ mock_resp.status = 200
+ mock_resp.text = AsyncMock(return_value='')
+
+ mock_session_get = MagicMock()
+ mock_cm = AsyncMock()
+ mock_cm.__aenter__ = AsyncMock(return_value=mock_resp)
+ mock_cm.__aexit__ = AsyncMock(return_value=False)
+ mock_session_get.return_value = mock_cm
+
+ with pytest.raises(RuntimeError, match='Empty'):
+ await fetch_and_cache(mock_session_get, path=cache_path)
+
+
+# ===========================================================================
+# get_attack18
+# ===========================================================================
+class TestGetAttack18:
+ def test_cache_missing_raises(self, tmp_path):
+ with patch('plugins.debrief.attack_mapper.CACHE_PATH', str(tmp_path / 'missing.json')), \
+ patch('plugins.debrief.attack_mapper._attack18_global', None):
+ with pytest.raises(FileNotFoundError):
+ get_attack18()
+
+ def test_returns_cached_instance(self):
+ sentinel = Attack18Map({})
+ with patch('plugins.debrief.attack_mapper._attack18_global', sentinel):
+ assert get_attack18() is sentinel
+
+
+# ===========================================================================
+# NormalizeAnalyticException
+# ===========================================================================
+class TestNormalizeAnalyticException:
+ def test_is_exception(self):
+ ex = NormalizeAnalyticException('test')
+ assert isinstance(ex, Exception)
+ assert str(ex) == 'test'
diff --git a/tests/test_base_report_section.py b/tests/test_base_report_section.py
new file mode 100644
index 0000000..b095918
--- /dev/null
+++ b/tests/test_base_report_section.py
@@ -0,0 +1,81 @@
+"""Tests for app/utility/base_report_section.py — BaseReportSection."""
+import pytest
+from unittest.mock import patch, MagicMock
+
+from reportlab.lib.styles import getSampleStyleSheet
+from reportlab.platypus.flowables import KeepTogetherSplitAtTop
+
+from plugins.debrief.app.utility.base_report_section import BaseReportSection
+
+
+@pytest.fixture
+def section():
+ with patch('plugins.debrief.app.utility.base_report_section.get_attack18') as mock_a18:
+ mock_a18.return_value = MagicMock()
+ s = BaseReportSection()
+ return s
+
+
+class TestBaseReportSectionInit:
+ def test_default_fields(self, section):
+ assert section.id == 'base-section-template'
+ assert section.display_name == 'Base Section Template'
+ assert section.description == 'Base class for debrief report section'
+ assert section.section_title == 'BASE SECTION HEADER'
+
+
+class TestStatusName:
+ def test_success(self):
+ assert BaseReportSection.status_name(0) == 'success'
+
+ def test_failure(self):
+ assert BaseReportSection.status_name(1) == 'failure'
+
+ def test_discarded(self):
+ assert BaseReportSection.status_name(-2) == 'discarded'
+
+ def test_timeout(self):
+ assert BaseReportSection.status_name(124) == 'timeout'
+
+ def test_collected(self):
+ assert BaseReportSection.status_name(-3) == 'collected'
+
+ def test_untrusted(self):
+ assert BaseReportSection.status_name(-4) == 'untrusted'
+
+ def test_visibility(self):
+ assert BaseReportSection.status_name(-5) == 'visibility'
+
+ def test_unknown_returns_queued(self):
+ assert BaseReportSection.status_name(999) == 'queued'
+ assert BaseReportSection.status_name(-999) == 'queued'
+
+
+class TestGenerateSectionTitleAndDescription:
+ def test_returns_grouped_flowable(self, section):
+ styles = getSampleStyleSheet()
+ result = section.generate_section_title_and_description(styles)
+ assert isinstance(result, KeepTogetherSplitAtTop)
+
+
+class TestGroupElements:
+ def test_returns_keep_together(self, section):
+ from reportlab.platypus import Paragraph
+ styles = getSampleStyleSheet()
+ elems = [Paragraph('test', styles['Normal'])]
+ result = section.group_elements(elems)
+ assert isinstance(result, KeepTogetherSplitAtTop)
+
+
+class TestGenerateTable:
+ def test_generates_table(self, section):
+ from reportlab.platypus import Table
+ data = [['Col1', 'Col2'], ['val1', 'val2'], ['val3', 'val4']]
+ result = section.generate_table(data, '*')
+ assert isinstance(result, Table)
+
+ def test_alternating_row_colors(self, section):
+ from reportlab.platypus import Table
+ data = [['Col1'], ['row1'], ['row2'], ['row3']]
+ result = section.generate_table(data, '*')
+ assert isinstance(result, Table)
diff --git a/tests/test_debrief_gui.py b/tests/test_debrief_gui.py
new file mode 100644
index 0000000..e306163
--- /dev/null
+++ b/tests/test_debrief_gui.py
@@ -0,0 +1,243 @@
+"""Exhaustive tests for app/debrief_gui.py — DebriefGui and helpers."""
+import pytest
+import os
+import re
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
+
+from plugins.debrief.app.debrief_gui import (
+ DebriefGui,
+ UseTemplateMarker,
+ LockTemplateMarker,
+ TemplateSwitchDoc,
+)
+
+
+# ---------------------------------------------------------------------------
+# Template marker flowables
+# ---------------------------------------------------------------------------
+class TestUseTemplateMarker:
+ def test_wrap_returns_zero(self):
+ m = UseTemplateMarker('Portrait')
+ assert m.wrap(100, 100) == (0, 0)
+
+ def test_name_stored(self):
+ m = UseTemplateMarker('Landscape')
+ assert m.name == 'Landscape'
+
+ def test_draw_does_not_raise(self):
+ m = UseTemplateMarker('Portrait')
+ m.draw()
+
+
+class TestLockTemplateMarker:
+ def test_wrap_returns_zero(self):
+ m = LockTemplateMarker('Landscape')
+ assert m.wrap(200, 200) == (0, 0)
+
+ def test_name_stored(self):
+ m = LockTemplateMarker('Landscape')
+ assert m.name == 'Landscape'
+
+ def test_draw_does_not_raise(self):
+ m = LockTemplateMarker('Landscape')
+ m.draw()
+
+
+# ---------------------------------------------------------------------------
+# TemplateSwitchDoc.afterFlowable
+# ---------------------------------------------------------------------------
+class TestTemplateSwitchDoc:
+ def _make_doc(self):
+ from io import BytesIO
+ from reportlab.lib.pagesizes import letter
+ buf = BytesIO()
+ doc = TemplateSwitchDoc(buf, pagesize=letter)
+ return doc, buf
+
+ def test_after_flowable_use_template(self):
+ doc, buf = self._make_doc()
+ marker = UseTemplateMarker('Portrait')
+ doc.handle_nextPageTemplate = MagicMock()
+ doc.afterFlowable(marker)
+ doc.handle_nextPageTemplate.assert_called()
+ buf.close()
+
+ def test_after_flowable_lock_template(self):
+ doc, buf = self._make_doc()
+ marker = LockTemplateMarker('Landscape')
+ doc.handle_nextPageTemplate = MagicMock()
+ doc.afterFlowable(marker)
+ assert doc._locked_template == 'Landscape'
+ buf.close()
+
+ def test_after_flowable_pagebreak_with_locked(self):
+ from reportlab.platypus import PageBreak
+ doc, buf = self._make_doc()
+ doc._locked_template = 'Landscape'
+ doc.handle_nextPageTemplate = MagicMock()
+ doc.afterFlowable(PageBreak())
+ doc.handle_nextPageTemplate.assert_called_with('Landscape')
+ buf.close()
+
+ def test_after_flowable_error_handling(self):
+ doc, buf = self._make_doc()
+ doc.handle_nextPageTemplate = MagicMock(side_effect=Exception('test'))
+ marker = UseTemplateMarker('Bad')
+ # Should not raise
+ doc.afterFlowable(marker)
+ buf.close()
+
+
+# ---------------------------------------------------------------------------
+# DebriefGui._sanitize_filename
+# ---------------------------------------------------------------------------
+class TestSanitizeFilename:
+ def test_basic_filename(self):
+ assert DebriefGui._sanitize_filename('report.pdf') == 'report.pdf'
+
+ def test_strips_path(self):
+ result = DebriefGui._sanitize_filename('/path/to/evil.pdf')
+ assert '/' not in result
+ assert result == 'evil.pdf'
+
+ def test_removes_special_chars(self):
+ result = DebriefGui._sanitize_filename('my report (1).pdf')
+ # Only word chars, hyphens, dots, underscores allowed
+ assert re.match(r'^[\w._-]+$', result)
+
+ def test_empty_after_sanitize(self):
+ result = DebriefGui._sanitize_filename('###')
+ assert result == ''
+
+ def test_spaces_to_underscores(self):
+ result = DebriefGui._sanitize_filename('my file.pdf')
+ # Spaces become underscores or are removed
+ assert ' ' not in result
+
+
+# ---------------------------------------------------------------------------
+# DebriefGui._save_svgs / _clean_downloads
+# ---------------------------------------------------------------------------
+class TestSvgOperations:
+ def test_save_svgs(self, tmp_path):
+ import base64
+ svg_content = b''
+ encoded = base64.b64encode(svg_content).decode()
+ save_dir = str(tmp_path) + '/'
+
+ svgs = {'test_graph': encoded}
+ with patch('plugins.debrief.app.debrief_gui.open', create=True) as mock_open:
+ mock_fh = MagicMock()
+ mock_open.return_value.__enter__ = MagicMock(return_value=mock_fh)
+ mock_open.return_value.__exit__ = MagicMock(return_value=False)
+ DebriefGui._save_svgs(svgs)
+ mock_open.assert_called_once()
+ mock_fh.write.assert_called_once_with(base64.b64decode(encoded))
+
+ def test_clean_downloads(self, tmp_path):
+ # Create test files
+ (tmp_path / 'test.png').write_text('png')
+ (tmp_path / 'test.svg').write_text('svg')
+ (tmp_path / 'keep.txt').write_text('keep')
+
+ with patch('glob.glob') as mock_glob:
+ mock_glob.side_effect = [
+ [str(tmp_path / 'test.png')],
+ [str(tmp_path / 'test.svg')],
+ ]
+ with patch('os.remove') as mock_remove:
+ DebriefGui._clean_downloads()
+ assert mock_remove.call_count == 2
+
+
+# ---------------------------------------------------------------------------
+# DebriefGui._suppress_logs
+# ---------------------------------------------------------------------------
+class TestSuppressLogs:
+ def test_sets_info_level(self):
+ import logging
+ DebriefGui._suppress_logs('test_lib_debrief')
+ logger = logging.getLogger('test_lib_debrief')
+ assert logger.level == logging.INFO
+
+
+# ---------------------------------------------------------------------------
+# DebriefGui._get_runtime_agents
+# ---------------------------------------------------------------------------
+class TestGetRuntimeAgents:
+ def _make_gui(self, mock_services):
+ with patch('plugins.debrief.app.debrief_gui.BaseWorld'), \
+ patch('plugins.debrief.app.debrief_gui.for_all_public_methods', lambda x: lambda cls: cls), \
+ patch('plugins.debrief.app.debrief_gui.check_authorization'), \
+ patch('plugins.debrief.app.debrief_gui.DebriefService'), \
+ patch('plugins.debrief.app.debrief_gui.rl_settings'), \
+ patch('plugins.debrief.app.debrief_gui.BaseWorld.get_config', return_value=None):
+ gui = DebriefGui.__new__(DebriefGui)
+ gui.services = mock_services
+ gui.debrief_svc = MagicMock()
+ gui.auth_svc = mock_services['auth_svc']
+ gui.data_svc = mock_services['data_svc']
+ gui.file_svc = mock_services['file_svc']
+ gui.knowledge_svc = mock_services.get('knowledge_svc')
+ gui.log = MagicMock()
+ gui.uploads_dir = '/tmp/uploads'
+ gui.report_section_modules = {}
+ gui.report_section_names = []
+ gui.loaded_report_sections = False
+ gui._a18 = None
+ return gui
+
+ def test_runtime_agents_filters_correctly(self, mock_services, make_agent, make_link, make_operation):
+ gui = self._make_gui(mock_services)
+ agent1 = make_agent(paw='paw1')
+ agent2 = make_agent(paw='paw2', unique='agent2')
+ # Only agent1 has links
+ lnk = make_link(paw='paw1')
+ op = make_operation(agents=[agent1, agent2], chain=[lnk])
+ result = gui._get_runtime_agents([op])
+ paws = [a.paw for a in result]
+ assert 'paw1' in paws
+ assert 'paw2' not in paws
+
+ def test_runtime_agents_empty_ops(self, mock_services):
+ gui = self._make_gui(mock_services)
+ result = gui._get_runtime_agents([])
+ assert result == []
+
+ def test_runtime_agents_none_ops(self, mock_services):
+ gui = self._make_gui(mock_services)
+ result = gui._get_runtime_agents(None)
+ assert result == []
+
+ def test_no_duplicate_agents(self, mock_services, make_agent, make_link, make_operation):
+ gui = self._make_gui(mock_services)
+ agent = make_agent(paw='paw1')
+ l1 = make_link(paw='paw1', unique='u1')
+ l2 = make_link(paw='paw1', unique='u2')
+ op1 = make_operation(agents=[agent], chain=[l1], op_id='op1')
+ op2 = make_operation(agents=[agent], chain=[l2], op_id='op2')
+ result = gui._get_runtime_agents([op1, op2])
+ assert len(result) == 1
+
+
+# ---------------------------------------------------------------------------
+# DebriefGui._pretty_name
+# ---------------------------------------------------------------------------
+class TestPrettyName:
+ def _make_gui(self):
+ gui = DebriefGui.__new__(DebriefGui)
+ return gui
+
+ def test_dotted_trait(self):
+ gui = self._make_gui()
+ assert gui._pretty_name('server.malicious.url') == 'Server Malicious Url'
+
+ def test_underscore_trait(self):
+ gui = self._make_gui()
+ result = gui._pretty_name('host_user_name')
+ assert 'Host' in result
+
+ def test_empty_string(self):
+ gui = self._make_gui()
+ assert gui._pretty_name('') == ''
diff --git a/tests/test_debrief_svc.py b/tests/test_debrief_svc.py
new file mode 100644
index 0000000..1a647f1
--- /dev/null
+++ b/tests/test_debrief_svc.py
@@ -0,0 +1,355 @@
+"""Exhaustive tests for app/debrief_svc.py — DebriefService."""
+import pytest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from plugins.debrief.app.debrief_svc import DebriefService
+
+
+# ---------------------------------------------------------------------------
+# Helper builders (lightweight, no Caldera imports)
+# ---------------------------------------------------------------------------
+def _svc(mock_services):
+ with patch.object(DebriefService, 'get_config', return_value={'app.name': 'caldera', 'app.port': '8888'}):
+ svc = DebriefService(mock_services)
+ return svc
+
+
+# ===========================================================================
+# generate_ttps — static method, no async
+# ===========================================================================
+class TestGenerateTtps:
+ def test_empty_operations(self):
+ assert DebriefService.generate_ttps([]) == {}
+
+ def test_single_operation_single_link(self, sample_operation):
+ ttps = DebriefService.generate_ttps([sample_operation])
+ assert 'discovery' in ttps
+ entry = ttps['discovery']
+ assert entry['name'] == 'discovery'
+ assert 'System Information Discovery' in entry['techniques']
+ assert 'TestOp' in entry['steps']
+ assert 'Test ability' in entry['steps']['TestOp']
+
+ def test_cleanup_links_excluded(self, make_ability, make_link, make_operation):
+ ab = make_ability(tactic='execution')
+ link_clean = make_link(ability=ab, cleanup=1, unique='clean1')
+ op = make_operation(chain=[link_clean])
+ ttps = DebriefService.generate_ttps([op])
+ assert ttps == {}
+
+ def test_multiple_operations_merge(self, multi_op_operations):
+ ttps = DebriefService.generate_ttps(multi_op_operations)
+ assert 'discovery' in ttps
+ assert 'collection' in ttps
+ disc = ttps['discovery']
+ assert 'Op1' in disc['steps']
+ assert 'Op2' in disc['steps']
+
+ def test_key_by_tid(self, sample_operation):
+ ttps = DebriefService.generate_ttps([sample_operation], key_by_tid=True)
+ entry = ttps['discovery']
+ assert 'T1082' in entry['techniques']
+
+ def test_sorted_output(self, multi_op_operations):
+ ttps = DebriefService.generate_ttps(multi_op_operations)
+ keys = list(ttps.keys())
+ assert keys == sorted(keys)
+
+ def test_duplicate_ability_not_repeated(self, make_ability, make_link, make_operation):
+ ab = make_ability(tactic='discovery', name='Same ability')
+ l1 = make_link(ability=ab, unique='u1')
+ l2 = make_link(ability=ab, unique='u2')
+ op = make_operation(chain=[l1, l2])
+ ttps = DebriefService.generate_ttps([op])
+ assert len(ttps['discovery']['steps']['TestOp']) == 1
+
+ def test_multiple_techniques_same_tactic(self, make_ability, make_link, make_operation):
+ ab1 = make_ability(tactic='discovery', technique_id='T1082', technique_name='SysInfo', name='A1')
+ ab2 = make_ability(tactic='discovery', technique_id='T1016', technique_name='System Network Config', name='A2')
+ l1 = make_link(ability=ab1, unique='u1')
+ l2 = make_link(ability=ab2, unique='u2')
+ op = make_operation(chain=[l1, l2])
+ ttps = DebriefService.generate_ttps([op])
+ assert len(ttps['discovery']['techniques']) == 2
+
+
+# ===========================================================================
+# _generate_new_tactic_entry / _update_tactic_entry — static helpers
+# ===========================================================================
+class TestTacticEntryHelpers:
+ def test_generate_new_tactic_entry(self, sample_operation, sample_link):
+ entry = DebriefService._generate_new_tactic_entry(
+ sample_operation, 'discovery', sample_link)
+ assert entry['name'] == 'discovery'
+ assert 'System Information Discovery' in entry['techniques']
+ assert entry['techniques']['System Information Discovery'] == 'T1082'
+
+ def test_generate_new_tactic_entry_key_by_tid(self, sample_operation, sample_link):
+ entry = DebriefService._generate_new_tactic_entry(
+ sample_operation, 'discovery', sample_link, key_by_tid=True)
+ assert 'T1082' in entry['techniques']
+
+ def test_update_tactic_entry_new_technique(self, make_ability, make_link):
+ entry = {'name': 'discovery', 'techniques': {'SysInfo': 'T1082'}, 'steps': {'Op': ['A1']}}
+ ab = make_ability(technique_id='T1016', technique_name='NetConfig', name='A2')
+ lnk = make_link(ability=ab)
+ DebriefService._update_tactic_entry(entry, 'Op', lnk)
+ assert 'NetConfig' in entry['techniques']
+ assert 'A2' in entry['steps']['Op']
+
+ def test_update_tactic_entry_new_op(self, sample_link):
+ entry = {'name': 'discovery', 'techniques': {'SysInfo': 'T1082'}, 'steps': {'Op1': ['A1']}}
+ DebriefService._update_tactic_entry(entry, 'Op2', sample_link)
+ assert 'Op2' in entry['steps']
+
+ def test_update_tactic_entry_tid_stripped_and_uppercased(self, make_ability, make_link):
+ ab = make_ability(technique_id=' t1082 ', technique_name='SysInfo')
+ lnk = make_link(ability=ab)
+ entry = DebriefService._generate_new_tactic_entry(
+ SimpleNamespace(name='Op'), 'discovery', lnk, key_by_tid=True)
+ assert 'T1082' in entry['techniques']
+
+
+# ===========================================================================
+# _add_agents_to_d3 — static method
+# ===========================================================================
+class TestAddAgentsToD3:
+ def test_adds_agent_node_and_link(self, sample_agent):
+ id_store = {'c2': 0}
+ graph = {'nodes': [], 'links': []}
+ DebriefService._add_agents_to_d3([sample_agent], id_store, graph)
+ assert len(graph['nodes']) == 1
+ assert graph['nodes'][0]['type'] == 'agent'
+ assert len(graph['links']) == 1
+ assert graph['links'][0]['source'] == 0
+
+ def test_no_duplicate_agents(self, sample_agent):
+ id_store = {'c2': 0}
+ graph = {'nodes': [], 'links': []}
+ DebriefService._add_agents_to_d3([sample_agent, sample_agent], id_store, graph)
+ assert len(graph['nodes']) == 1
+
+ def test_multiple_agents(self, sample_agent, sample_agent_linux):
+ id_store = {'c2': 0}
+ graph = {'nodes': [], 'links': []}
+ DebriefService._add_agents_to_d3([sample_agent, sample_agent_linux], id_store, graph)
+ assert len(graph['nodes']) == 2
+ assert len(graph['links']) == 2
+
+
+# ===========================================================================
+# _format_timestamp
+# ===========================================================================
+class TestFormatTimestamp:
+ def test_replaces_space_with_t(self):
+ assert DebriefService._format_timestamp('2021-01-01 08:00:00') == '2021-01-01T08:00:00'
+
+ def test_already_formatted(self):
+ assert DebriefService._format_timestamp('2021-01-01T08:00:00') == '2021-01-01T08:00:00'
+
+
+# ===========================================================================
+# _get_by_prop_order
+# ===========================================================================
+class TestGetByPropOrder:
+ def test_single_link(self, sample_link):
+ result = DebriefService._get_by_prop_order([sample_link], 'tactic')
+ assert len(result) == 1
+ assert result[0][0] == 'discovery'
+ assert result[0][1] == [sample_link]
+
+ def test_two_different_tactics(self, two_op_chain):
+ result = DebriefService._get_by_prop_order(two_op_chain, 'tactic')
+ assert len(result) == 2
+ assert result[0][0] == 'discovery'
+ assert result[1][0] == 'collection'
+
+ def test_same_tactic_grouped(self, make_ability, make_link):
+ ab1 = make_ability(tactic='discovery', name='A1')
+ ab2 = make_ability(tactic='discovery', name='A2')
+ l1 = make_link(ability=ab1, unique='u1', cleanup=0)
+ l2 = make_link(ability=ab2, unique='u2', cleanup=0)
+ result = DebriefService._get_by_prop_order([l1, l2], 'tactic')
+ assert len(result) == 1
+ assert len(result[0][1]) == 2
+
+ def test_prop_technique_name(self, two_op_chain):
+ result = DebriefService._get_by_prop_order(two_op_chain, 'technique_name')
+ assert len(result) == 2
+
+
+# ===========================================================================
+# _link_nontargeted_facts
+# ===========================================================================
+class TestLinkNontargetedFacts:
+ def test_untargeted_fact_gets_link(self):
+ nodes = [{'id': 10, 'name': 'fact1'}, {'id': 11, 'name': 'fact2'}]
+ links = [{'source': 1, 'target': 10, 'type': 'relationship'}]
+ DebriefService._link_nontargeted_facts(nodes, links, op_id=1)
+ assert len(links) == 2
+ assert links[1]['target'] == 11
+ assert links[1]['source'] == 1
+
+ def test_all_targeted_no_extra_links(self):
+ nodes = [{'id': 10, 'name': 'fact1'}]
+ links = [{'source': 1, 'target': 10, 'type': 'relationship'}]
+ DebriefService._link_nontargeted_facts(nodes, links, op_id=1)
+ assert len(links) == 1
+
+
+# ===========================================================================
+# build_steps_d3 — async
+# ===========================================================================
+class TestBuildStepsD3:
+ @pytest.mark.asyncio
+ async def test_empty_operations(self, mock_services):
+ svc = _svc(mock_services)
+ mock_services['data_svc'].locate.return_value = []
+ result = await svc.build_steps_d3([])
+ assert result['nodes'][0]['type'] == 'c2'
+ assert len(result['links']) == 0
+
+ @pytest.mark.asyncio
+ async def test_single_operation(self, mock_services, sample_operation):
+ svc = _svc(mock_services)
+ mock_services['data_svc'].locate.return_value = [sample_operation]
+ result = await svc.build_steps_d3(['op-1'])
+ node_types = [n['type'] for n in result['nodes']]
+ assert 'c2' in node_types
+ assert 'agent' in node_types
+ assert 'link' in node_types
+
+ @pytest.mark.asyncio
+ async def test_cleanup_link_labeled(self, mock_services, make_ability, make_link, make_operation):
+ svc = _svc(mock_services)
+ ab = make_ability(name='DoClean')
+ lnk = make_link(ability=ab, cleanup=1, unique='cl1')
+ op = make_operation(chain=[lnk])
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_steps_d3(['op-1'])
+ link_nodes = [n for n in result['nodes'] if n['type'] == 'link']
+ assert any('(cleanup)' in n['attrs']['name'] for n in link_nodes)
+
+ @pytest.mark.asyncio
+ async def test_multiple_links_chained(self, mock_services, make_operation, two_op_chain):
+ svc = _svc(mock_services)
+ op = make_operation(chain=two_op_chain)
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_steps_d3(['op-1'])
+ next_links = [l for l in result['links'] if l['type'] == 'next_link']
+ assert len(next_links) >= 2
+
+
+# ===========================================================================
+# build_tactic_d3 / build_technique_d3 — async
+# ===========================================================================
+class TestBuildTacticAndTechniqueD3:
+ @pytest.mark.asyncio
+ async def test_tactic_d3(self, mock_services, sample_operation):
+ svc = _svc(mock_services)
+ mock_services['data_svc'].locate.return_value = [sample_operation]
+ result = await svc.build_tactic_d3(['op-1'])
+ tactic_nodes = [n for n in result['nodes'] if n['type'] == 'tactic']
+ assert len(tactic_nodes) >= 1
+
+ @pytest.mark.asyncio
+ async def test_technique_d3(self, mock_services, sample_operation):
+ svc = _svc(mock_services)
+ mock_services['data_svc'].locate.return_value = [sample_operation]
+ result = await svc.build_technique_d3(['op-1'])
+ technique_nodes = [n for n in result['nodes'] if n['type'] == 'technique_name']
+ assert len(technique_nodes) >= 1
+
+ @pytest.mark.asyncio
+ async def test_empty_chain(self, mock_services, make_operation):
+ svc = _svc(mock_services)
+ op = make_operation(chain=[])
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_tactic_d3(['op-1'])
+ # Only operation node, no tactic nodes
+ assert len([n for n in result['nodes'] if n['type'] == 'tactic']) == 0
+
+
+# ===========================================================================
+# build_fact_d3 — async
+# ===========================================================================
+class TestBuildFactD3:
+ @pytest.mark.asyncio
+ async def test_fact_graph_with_facts(self, mock_services, make_operation, make_fact, make_source):
+ svc = _svc(mock_services)
+ f1 = make_fact(trait='host.user', value='admin', unique='f1')
+ src = make_source(facts=[f1])
+ op = make_operation(facts=[f1], source=src)
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_fact_d3(['op-1'])
+ fact_nodes = [n for n in result['nodes'] if n['type'] == 'fact']
+ assert len(fact_nodes) == 1
+
+ @pytest.mark.asyncio
+ async def test_fact_graph_no_facts(self, mock_services, make_operation):
+ svc = _svc(mock_services)
+ op = make_operation(facts=[], relationships=[])
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_fact_d3(['op-1'])
+ fact_nodes = [n for n in result['nodes'] if n['type'] == 'fact']
+ assert len(fact_nodes) == 0
+
+ @pytest.mark.asyncio
+ async def test_fact_relationship_links(self, mock_services, make_operation, make_fact, make_source):
+ svc = _svc(mock_services)
+ f1 = make_fact(trait='host.user', value='admin', unique='f1')
+ f2 = make_fact(trait='host.password', value='pass', unique='f2')
+ rel = SimpleNamespace(edge='has_password', source=f1, target=f2)
+ src = make_source(facts=[f1])
+ op = make_operation(facts=[f1, f2], relationships=[rel], source=src)
+ mock_services['data_svc'].locate.return_value = [op]
+ result = await svc.build_fact_d3(['op-1'])
+ rel_links = [l for l in result['links'] if l['type'] == 'relationship']
+ assert len(rel_links) >= 1
+
+
+# ===========================================================================
+# build_attackpath_d3 — async
+# ===========================================================================
+class TestBuildAttackpathD3:
+ @pytest.mark.asyncio
+ async def test_empty_operations(self, mock_services):
+ svc = _svc(mock_services)
+ mock_services['data_svc'].locate.return_value = []
+ result = await svc.build_attackpath_d3([])
+ assert result['nodes'][0]['type'] == 'c2'
+
+ @pytest.mark.asyncio
+ async def test_agent_with_origin_link(self, mock_services, make_operation, make_agent, make_link, make_ability):
+ svc = _svc(mock_services)
+ ab = make_ability(tactic='lateral-movement', technique_id='T1021',
+ technique_name='Remote Services')
+ # link paw must match parent_agent.paw, and agent.unique used as id_store key
+ parent_agent = make_agent(paw='paw1', unique='paw1')
+ lnk = make_link(paw='paw1', ability=ab, unique='origin-lnk', link_id='origin-lnk')
+ spawned_agent = make_agent(paw='paw2', unique='paw2', origin_link_id='origin-lnk')
+ op = make_operation(agents=[parent_agent, spawned_agent], chain=[lnk])
+ mock_services['data_svc'].locate.return_value = [op]
+ mock_services['app_svc'].find_op_with_link.return_value = op
+ result = await svc.build_attackpath_d3(['op-1'])
+ link_nodes = [n for n in result['nodes'] if n['type'] == 'link']
+ assert len(link_nodes) >= 1
+
+
+# ===========================================================================
+# _get_pub_attrs — static
+# ===========================================================================
+class TestGetPubAttrs:
+ def test_filters_private_attrs(self, make_fact):
+ f = make_fact()
+ # add a private attr
+ f._private = 'hidden'
+ result = DebriefService._get_pub_attrs(f)
+ assert '_private' not in result
+ assert 'trait' in result
+
+ def test_origin_type_name(self, make_fact):
+ f = make_fact()
+ result = DebriefService._get_pub_attrs(f)
+ assert result['origin_type'] == 'LEARNED'
diff --git a/tests/test_hook.py b/tests/test_hook.py
new file mode 100644
index 0000000..5d27105
--- /dev/null
+++ b/tests/test_hook.py
@@ -0,0 +1,96 @@
+"""Tests for hook.py — plugin registration and ATT&CK cache init."""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import plugins.debrief.hook as hook
+
+
+class TestHookMetadata:
+ def test_name(self):
+ assert hook.name == 'Debrief'
+
+ def test_description(self):
+ assert hook.description == 'Operation debrief functionality'
+
+ def test_address(self):
+ assert hook.address == '/plugin/debrief/gui'
+
+ def test_access(self):
+ # Access should be RED
+ assert hook.access is not None
+
+
+class TestInitAttack18Cache:
+ @pytest.mark.asyncio
+ async def test_successful_cache_init(self):
+ with patch('plugins.debrief.hook.fetch_and_cache', new_callable=AsyncMock) as mock_fetch:
+ mock_fetch.return_value = None
+ # Should not raise
+ await hook._init_attack18_cache()
+ mock_fetch.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_cache_init_failure_logged(self):
+ with patch('plugins.debrief.hook.fetch_and_cache', new_callable=AsyncMock,
+ side_effect=Exception('network error')):
+ with patch.object(hook.log, 'warning') as mock_warn:
+ await hook._init_attack18_cache()
+ mock_warn.assert_called_once()
+
+
+class TestEnable:
+ @pytest.mark.asyncio
+ async def test_enable_registers_routes(self):
+ mock_app = MagicMock()
+ mock_router = MagicMock()
+ mock_app.router = mock_router
+
+ app_svc = MagicMock()
+ app_svc.application = mock_app
+
+ services = {
+ 'app_svc': app_svc,
+ 'auth_svc': MagicMock(),
+ 'data_svc': MagicMock(),
+ 'file_svc': MagicMock(),
+ 'knowledge_svc': MagicMock(),
+ }
+
+ with patch('plugins.debrief.hook.BaseWorld') as mock_bw, \
+ patch('plugins.debrief.hook.DebriefGui') as mock_gui_cls, \
+ patch('plugins.debrief.hook.asyncio.create_task') as mock_task:
+ mock_bw.strip_yml.return_value = [{}]
+ mock_gui_instance = MagicMock()
+ mock_gui_cls.return_value = mock_gui_instance
+
+ await hook.enable(services)
+
+ # Check routes registered
+ assert mock_router.add_route.call_count >= 7
+ assert mock_router.add_static.call_count >= 2
+
+ # Check cache warmup task created
+ mock_task.assert_called_once()
+
+ @pytest.mark.asyncio
+ async def test_enable_applies_config(self):
+ mock_app = MagicMock()
+ mock_app.router = MagicMock()
+ app_svc = MagicMock()
+ app_svc.application = mock_app
+
+ services = {
+ 'app_svc': app_svc,
+ 'auth_svc': MagicMock(),
+ 'data_svc': MagicMock(),
+ 'file_svc': MagicMock(),
+ 'knowledge_svc': MagicMock(),
+ }
+
+ with patch('plugins.debrief.hook.BaseWorld') as mock_bw, \
+ patch('plugins.debrief.hook.DebriefGui'), \
+ patch('plugins.debrief.hook.asyncio.create_task'):
+ mock_bw.strip_yml.return_value = [{}]
+ await hook.enable(services)
+ mock_bw.apply_config.assert_called_once_with(
+ 'debrief', mock_bw.strip_yml.return_value[0])
diff --git a/tests/test_sections.py b/tests/test_sections.py
new file mode 100644
index 0000000..688d5d9
--- /dev/null
+++ b/tests/test_sections.py
@@ -0,0 +1,441 @@
+"""Tests for ALL debrief report section modules in app/debrief-sections/."""
+import pytest
+import importlib
+from unittest.mock import patch, MagicMock, AsyncMock
+from types import SimpleNamespace
+
+from reportlab.lib.styles import getSampleStyleSheet
+from reportlab.platypus.flowables import KeepTogetherSplitAtTop
+
+# Import section modules
+agents_mod = importlib.import_module('plugins.debrief.app.debrief-sections.agents')
+attackpath_mod = importlib.import_module('plugins.debrief.app.debrief-sections.attackpath_graph')
+fact_graph_mod = importlib.import_module('plugins.debrief.app.debrief-sections.fact_graph')
+facts_table_mod = importlib.import_module('plugins.debrief.app.debrief-sections.facts_table')
+main_summary_mod = importlib.import_module('plugins.debrief.app.debrief-sections.main_summary')
+statistics_mod = importlib.import_module('plugins.debrief.app.debrief-sections.statistics')
+steps_graph_mod = importlib.import_module('plugins.debrief.app.debrief-sections.steps_graph')
+steps_table_mod = importlib.import_module('plugins.debrief.app.debrief-sections.steps_table')
+tactic_graph_mod = importlib.import_module('plugins.debrief.app.debrief-sections.tactic_graph')
+tactic_technique_mod = importlib.import_module('plugins.debrief.app.debrief-sections.tactic_technique_table')
+technique_graph_mod = importlib.import_module('plugins.debrief.app.debrief-sections.technique_graph')
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def styles():
+ return getSampleStyleSheet()
+
+
+def _mock_section(module):
+ with patch('plugins.debrief.app.utility.base_report_section.get_attack18') as mock_a18:
+ mock_a18.return_value = MagicMock(
+ get_strategies=MagicMock(return_value=[]),
+ get_analytics=MagicMock(return_value=[]),
+ get_parent_strategies=MagicMock(return_value=[]),
+ get_parent_analytics=MagicMock(return_value=[]),
+ strategies_by_tid={},
+ analytics_by_tid={},
+ techniques_by_id={},
+ )
+ section = module.DebriefReportSection()
+ return section
+
+
+# ===========================================================================
+# Agents Section
+# ===========================================================================
+class TestAgentsSection:
+ def test_fields(self):
+ section = _mock_section(agents_mod)
+ assert section.id == 'agents'
+ assert section.display_name == 'Agents'
+
+ @pytest.mark.asyncio
+ async def test_generate_with_agents(self, styles, make_agent):
+ section = _mock_section(agents_mod)
+ agents = [make_agent(paw='p1'), make_agent(paw='p2', unique='a2')]
+ result = await section.generate_section_elements(styles, agents=agents)
+ assert len(result) == 2 # title+desc, table
+
+ @pytest.mark.asyncio
+ async def test_generate_no_agents(self, styles):
+ section = _mock_section(agents_mod)
+ result = await section.generate_section_elements(styles)
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_empty_agents(self, styles):
+ section = _mock_section(agents_mod)
+ result = await section.generate_section_elements(styles, agents=[])
+ # 'agents' key IS in kwargs even if empty, so section header + table still generated
+ assert len(result) == 2
+
+
+# ===========================================================================
+# Attack Path Graph Section
+# ===========================================================================
+class TestAttackpathGraphSection:
+ def test_fields(self):
+ section = _mock_section(attackpath_mod)
+ assert section.id == 'attackpath-graph'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_graph(self, styles):
+ section = _mock_section(attackpath_mod)
+ result = await section.generate_section_elements(styles, graph_files={})
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_graph(self, styles, tmp_path):
+ section = _mock_section(attackpath_mod)
+ with patch.object(section, 'generate_grouped_graph_section_flowables',
+ return_value=MagicMock()) as mock_gen:
+ result = await section.generate_section_elements(
+ styles, graph_files={'attackpath': '/fake/path.svg'})
+ assert len(result) == 1
+ mock_gen.assert_called_once()
+
+
+# ===========================================================================
+# Fact Graph Section
+# ===========================================================================
+class TestFactGraphSection:
+ def test_fields(self):
+ section = _mock_section(fact_graph_mod)
+ assert section.id == 'fact-graph'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_graph(self, styles):
+ section = _mock_section(fact_graph_mod)
+ result = await section.generate_section_elements(styles, graph_files={})
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_graph(self, styles):
+ section = _mock_section(fact_graph_mod)
+ with patch.object(section, 'generate_grouped_graph_section_flowables',
+ return_value=MagicMock()):
+ result = await section.generate_section_elements(
+ styles, graph_files={'fact': '/fake/fact.svg'})
+ assert len(result) == 1
+
+
+# ===========================================================================
+# Facts Table Section
+# ===========================================================================
+class TestFactsTableSection:
+ def test_fields(self):
+ section = _mock_section(facts_table_mod)
+ assert section.id == 'facts-table'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_operations(self, styles):
+ section = _mock_section(facts_table_mod)
+ result = await section.generate_section_elements(styles)
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_operation(self, styles, make_operation, make_fact):
+ section = _mock_section(facts_table_mod)
+ f1 = make_fact(trait='host.user', value='admin')
+ op = make_operation(facts=[f1])
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op])
+ assert len(result) >= 2 # title + table per operation
+
+ @pytest.mark.asyncio
+ async def test_generate_multiple_operations(self, styles, make_operation, make_fact):
+ section = _mock_section(facts_table_mod)
+ f1 = make_fact(trait='t1', value='v1')
+ op1 = make_operation(name='Op1', op_id='o1', facts=[f1])
+ op2 = make_operation(name='Op2', op_id='o2', facts=[])
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op1, op2])
+ assert len(result) >= 4 # title + table for each
+
+
+class TestFactsTableHelpers:
+ def test_origin_name_enum(self):
+ from enum import Enum
+ class OT(Enum):
+ LEARNED = 'LEARNED'
+ section = _mock_section(facts_table_mod)
+ fact = SimpleNamespace(origin_type=OT.LEARNED)
+ assert section._origin_name(fact) == 'LEARNED'
+
+ def test_origin_name_none(self):
+ section = _mock_section(facts_table_mod)
+ fact = SimpleNamespace(origin_type=None)
+ assert section._origin_name(fact) is None
+
+ def test_generate_fact_table_cmd_no_links(self):
+ section = _mock_section(facts_table_mod)
+ from enum import Enum
+ class OT(Enum):
+ LEARNED = 'LEARNED'
+ fact = SimpleNamespace(origin_type=OT.LEARNED, links=[])
+ result = section._generate_fact_table_cmd(fact, {})
+ assert 'No Command' in result
+
+ def test_generate_fact_table_cmd_with_link(self, make_link):
+ section = _mock_section(facts_table_mod)
+ from enum import Enum
+ class OT(Enum):
+ LEARNED = 'LEARNED'
+ lnk = make_link(command='whoami')
+ lnk.decode_bytes = lambda c: c
+ fact = SimpleNamespace(origin_type=OT.LEARNED, links=['lnk1'])
+ result = section._generate_fact_table_cmd(fact, {'lnk1': lnk})
+ assert 'whoami' in result
+
+ def test_value_truncation(self):
+ section = _mock_section(facts_table_mod)
+ from enum import Enum
+ class OT(Enum):
+ LEARNED = 'LEARNED'
+ long_val = 'x' * 2000
+ fact = SimpleNamespace(
+ trait='t', value=long_val, score=1, origin_type=OT.LEARNED,
+ source='src', links=[], collected_by=[])
+ row = section._generate_fact_table_row(
+ fact, False, {}, 'src', set(), set())
+ assert len(row[1]) < len(long_val), "Long values should be truncated"
+
+
+# ===========================================================================
+# Main Summary Section
+# ===========================================================================
+class TestMainSummarySection:
+ def test_fields(self):
+ section = _mock_section(main_summary_mod)
+ assert section.id == 'main-summary'
+ assert section.display_name == 'Main Summary'
+
+ @pytest.mark.asyncio
+ async def test_generate_returns_flowables(self, styles):
+ section = _mock_section(main_summary_mod)
+ result = await section.generate_section_elements(styles)
+ assert len(result) == 1
+ assert isinstance(result[0], KeepTogetherSplitAtTop)
+
+
+# ===========================================================================
+# Statistics Section
+# ===========================================================================
+class TestStatisticsSection:
+ def test_fields(self):
+ section = _mock_section(statistics_mod)
+ assert section.id == 'statistics'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_operations(self, styles):
+ section = _mock_section(statistics_mod)
+ result = await section.generate_section_elements(styles)
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_operation(self, styles, make_operation):
+ section = _mock_section(statistics_mod)
+ op = make_operation()
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op])
+ assert len(result) >= 2 # title+desc, table
+
+ @pytest.mark.asyncio
+ async def test_unfinished_operation(self, styles, make_operation):
+ section = _mock_section(statistics_mod)
+ op = make_operation(finish=None)
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op])
+ assert len(result) >= 2
+
+
+# ===========================================================================
+# Steps Graph Section
+# ===========================================================================
+class TestStepsGraphSection:
+ def test_fields(self):
+ section = _mock_section(steps_graph_mod)
+ assert section.id == 'steps-graph'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_graph(self, styles):
+ section = _mock_section(steps_graph_mod)
+ result = await section.generate_section_elements(styles, graph_files={})
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_graph(self, styles):
+ section = _mock_section(steps_graph_mod)
+ with patch.object(section, 'generate_grouped_graph_section_flowables',
+ return_value=MagicMock()):
+ result = await section.generate_section_elements(
+ styles, graph_files={'steps': '/fake/steps.svg'})
+ assert len(result) == 1
+
+
+# ===========================================================================
+# Steps Table Section
+# ===========================================================================
+class TestStepsTableSection:
+ def test_fields(self):
+ section = _mock_section(steps_table_mod)
+ assert section.id == 'steps-table'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_operations(self, styles):
+ section = _mock_section(steps_table_mod)
+ result = await section.generate_section_elements(styles)
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_operation(self, styles, make_operation, make_link, make_ability):
+ section = _mock_section(steps_table_mod)
+ ab = make_ability()
+ lnk = make_link(ability=ab, facts=[SimpleNamespace(score=2)])
+ op = make_operation(chain=[lnk])
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op])
+ assert len(result) >= 2
+
+ @pytest.mark.asyncio
+ async def test_step_with_no_facts(self, styles, make_operation, make_link):
+ section = _mock_section(steps_table_mod)
+ lnk = make_link(facts=[])
+ op = make_operation(chain=[lnk])
+ with patch.object(section, 'generate_table', return_value=MagicMock()):
+ result = await section.generate_section_elements(styles, operations=[op])
+ assert len(result) >= 2
+
+
+# ===========================================================================
+# Tactic Graph Section
+# ===========================================================================
+class TestTacticGraphSection:
+ def test_fields(self):
+ section = _mock_section(tactic_graph_mod)
+ assert section.id == 'tactic-graph'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_graph(self, styles):
+ section = _mock_section(tactic_graph_mod)
+ result = await section.generate_section_elements(styles, graph_files={})
+ assert result == []
+
+
+# ===========================================================================
+# Technique Graph Section
+# ===========================================================================
+class TestTechniqueGraphSection:
+ def test_fields(self):
+ section = _mock_section(technique_graph_mod)
+ assert section.id == 'technique-graph'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_graph(self, styles):
+ section = _mock_section(technique_graph_mod)
+ result = await section.generate_section_elements(styles, graph_files={})
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_graph(self, styles):
+ section = _mock_section(technique_graph_mod)
+ with patch.object(section, 'generate_grouped_graph_section_flowables',
+ return_value=MagicMock()):
+ result = await section.generate_section_elements(
+ styles, graph_files={'technique': '/fake/tech.svg'})
+ assert len(result) == 1
+
+
+# ===========================================================================
+# Tactic/Technique Table Section
+# ===========================================================================
+class TestTacticTechniqueTableSection:
+ def test_fields(self):
+ section = _mock_section(tactic_technique_mod)
+ assert section.id == 'tactic-technique-table'
+
+ @pytest.mark.asyncio
+ async def test_generate_no_operations(self, styles):
+ section = _mock_section(tactic_technique_mod)
+ result = await section.generate_section_elements(styles)
+ assert result == []
+
+ @pytest.mark.asyncio
+ async def test_generate_with_operation(self, styles, make_operation):
+ section = _mock_section(tactic_technique_mod)
+ op = make_operation()
+ result = await section.generate_section_elements(
+ styles, operations=[op], selected_sections=[])
+ assert len(result) >= 1
+
+ @pytest.mark.asyncio
+ async def test_generate_with_detections_link(self, styles, make_operation):
+ section = _mock_section(tactic_technique_mod)
+ op = make_operation()
+ result = await section.generate_section_elements(
+ styles, operations=[op], selected_sections=['ttps-detections'])
+ assert len(result) >= 1
+
+ def test_stacked_text(self):
+ section = _mock_section(tactic_technique_mod)
+ result = section._stacked(['line1', 'line2'])
+ assert 'line1' in result
+ assert '
' in result
+
+ def test_stacked_html(self):
+ section = _mock_section(tactic_technique_mod)
+ result = section._stacked(['A'], html=True)
+ assert '' in result
+
+ def test_stacked_empty(self):
+ section = _mock_section(tactic_technique_mod)
+ assert section._stacked([]) == '\u2014'
+ assert section._stacked(None) == '\u2014'
+
+ def test_generate_ttp_detection_info_no_strategies(self, styles):
+ section = _mock_section(tactic_technique_mod)
+ section._a18 = MagicMock()
+ section._a18.get_strategies.return_value = []
+ section.styles = styles
+ section.tt_body = MagicMock()
+ result = section._generate_ttp_detection_info('T9999', False)
+ assert result == ['\u2014']
+
+ def test_generate_ttp_detection_info_with_strategies(self, styles):
+ section = _mock_section(tactic_technique_mod)
+ section._a18 = MagicMock()
+ section._a18.get_strategies.return_value = [
+ {'det_id': 'DET0001', 'name': 'Test Strategy'}
+ ]
+ section.styles = styles
+ section.tt_body = MagicMock()
+ result = section._generate_ttp_detection_info('T1082', include_det_links=False)
+ assert 'DET0001' in result[0]
+
+ def test_generate_ttp_detection_info_with_links(self, styles):
+ section = _mock_section(tactic_technique_mod)
+ section._a18 = MagicMock()
+ section._a18.get_strategies.return_value = [
+ {'det_id': 'DET0001', 'name': 'Test Strategy'}
+ ]
+ section.styles = styles
+ section.tt_body = MagicMock()
+ result = section._generate_ttp_detection_info('T1082', include_det_links=True)
+ assert 'href' in result[0].lower(), "Detection info with links should contain href attributes"
+
+ def test_parent_fallback(self, styles):
+ section = _mock_section(tactic_technique_mod)
+ section._a18 = MagicMock()
+ # No sub-technique strategies, has parent
+ section._a18.get_strategies.side_effect = lambda tid: (
+ [] if '.' in tid else [{'det_id': 'DET0002', 'name': 'Parent'}])
+ section.styles = styles
+ section.tt_body = MagicMock()
+ result = section._generate_ttp_detection_info('T1547.001', False)
+ # Should contain DET0002 with parent TID reference
+ assert 'DET0002' in result[0]
+ assert 'T1547' in result[0]
diff --git a/tests/test_story.py b/tests/test_story.py
new file mode 100644
index 0000000..0ca9a41
--- /dev/null
+++ b/tests/test_story.py
@@ -0,0 +1,92 @@
+"""Tests for app/objects/c_story.py — Story class."""
+import pytest
+from unittest.mock import MagicMock, patch
+
+from reportlab.platypus import Paragraph, Spacer, PageBreak
+from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
+
+from plugins.debrief.app.objects.c_story import Story
+
+
+class TestStoryInit:
+ def test_empty_story(self):
+ s = Story()
+ assert s.story_arr == []
+
+ def test_header_logo_path_default(self):
+ Story.set_header_logo_path(None) # reset to known state
+ assert Story._header_logo_path is None
+
+
+class TestStoryAppend:
+ def test_append_adds_data_and_spacer(self):
+ s = Story()
+ para = Paragraph('hello', getSampleStyleSheet()['Normal'])
+ s.append(para)
+ assert len(s.story_arr) == 2
+ assert isinstance(s.story_arr[1], Spacer)
+
+ def test_append_custom_spacing(self):
+ s = Story()
+ para = Paragraph('test', getSampleStyleSheet()['Normal'])
+ s.append(para, spacing=24)
+ # Second item is a spacer
+ assert isinstance(s.story_arr[1], Spacer)
+
+ def test_append_zero_spacing(self):
+ s = Story()
+ para = Paragraph('test', getSampleStyleSheet()['Normal'])
+ s.append(para, spacing=0)
+ assert len(s.story_arr) == 2
+
+
+class TestStoryAppendText:
+ def test_append_text(self):
+ s = Story()
+ styles = getSampleStyleSheet()
+ s.append_text('Hello World', styles['Normal'], 12)
+ assert len(s.story_arr) == 2
+ assert isinstance(s.story_arr[0], Paragraph)
+ assert isinstance(s.story_arr[1], Spacer)
+
+
+class TestStoryPageBreak:
+ def test_page_break(self):
+ s = Story()
+ s.page_break()
+ assert len(s.story_arr) == 1
+ assert isinstance(s.story_arr[0], PageBreak)
+
+
+class TestSetHeaderLogoPath:
+ def test_set_path(self):
+ Story.set_header_logo_path('/some/path/logo.png')
+ assert Story._header_logo_path == '/some/path/logo.png'
+
+ def test_set_none(self):
+ Story.set_header_logo_path(None)
+ assert Story._header_logo_path is None
+
+
+class TestGetTableObject:
+ def test_string_input(self):
+ result = Story.get_table_object('hello')
+ assert isinstance(result, Paragraph)
+
+ def test_list_input(self):
+ result = Story.get_table_object(['item1', 'item2'])
+ assert isinstance(result, Paragraph)
+
+ def test_dict_input(self):
+ result = Story.get_table_object({'key': ['val1', 'val2']})
+ assert isinstance(result, Paragraph)
+
+
+class TestGetDescription:
+ def test_get_description_delegates_to_descriptions(self):
+ """get_description() calls self._descriptions() which is not defined
+ on the base Story class. Verify the delegation attempt occurs."""
+ s = Story()
+ assert hasattr(s, 'get_description')
+ # _descriptions is not implemented on Story — subclasses would provide it
+ assert not hasattr(s, '_descriptions')
diff --git a/tox.ini b/tox.ini
new file mode 100644
index 0000000..5c16db1
--- /dev/null
+++ b/tox.ini
@@ -0,0 +1,14 @@
+[tox]
+envlist = py3
+skipsdist = true
+
+[testenv]
+deps =
+ pytest>=7.0,<9.0
+ pytest-asyncio>=0.21,<1.0
+ aiohttp>=3.8,<4.0
+ reportlab>=4.0,<5.0
+ svglib>=1.5,<2.0
+ lxml>=4.9
+commands =
+ pytest {posargs:-v}