Add MSSQL and SCCM privilege-escalation analysis

.github/workflows/test.yml

@@ -83,4 +83,4 @@ jobs:
           NEO4J_URI: bolt://localhost:7687
           NEO4J_USER: neo4j
           NEO4J_PASSWORD: testpassword
-        run: pytest tests/checks -v --maxfail=5 -m "neo4j and not integration" --no-cov
+        run: pytest tests/ -v --maxfail=5 -m "neo4j and not integration" --no-cov

.gitignore

@@ -37,3 +37,8 @@ config.ini
 .idea/
 .vscode/
 .DS_Store
+
+# Local notes / drafts
+LAB_REFERENCE.md
+BLOG_POST_V*.md
+PRODUCTION_READINESS_PLAN.md

README.md

@@ -5,8 +5,8 @@
 ADPathfinder is an attack mapping tool for pentesters and red teamers. It analyses SharpHound data and unifies it with OpenGraph plugins to surface attack paths to high-value targets such as Domain Admins and Domain Controllers, starting from low-privileged users and computers. MSSQLHound and ConfigManBearPig are supported natively, extending coverage across AD, ADCS, SCCM, and MSSQL.
 
 <p align="center">
-  <a href="docs/images/demo.mp4">
-    <img src="docs/images/demo.webp" alt="ADPathfinder demo video" width="600">
+  <a href="docs/images/demo-slideshow.mp4">
+    <img src="docs/images/demo-slideshow.webp" alt="ADPathfinder demo video" width="600">
   </a>
 </p>
 
@@ -59,16 +59,16 @@ This stores the BloodHound CE API settings used for import and delete operations
 </details>
 
 <details>
-<summary><strong>Import BloodHound data</strong></summary>
+<summary><strong>Import SharpHound data</strong></summary>
 
 ```bash
-adpathfinder --import BloodHound.zip
+adpathfinder --import SharpHound.zip
 ```
 
-Multiple BloodHound and OpenGraph plugin zip files can be imported in one command:
+Multiple SharpHound and OpenGraph plugin zip files can be imported in one command:
 
 ```bash
-adpathfinder --import BloodHound.zip MSSQLHound.zip ConfigManBearPig.zip
+adpathfinder --import SharpHound.zip MSSQLHound.zip ConfigManBearPig.zip
 ```
 
 </details>
@@ -77,7 +77,7 @@ adpathfinder --import BloodHound.zip MSSQLHound.zip ConfigManBearPig.zip
 <summary><strong>Import data and run a full audit</strong></summary>
 
 ```bash
-adpathfinder -i BloodHound.zip MSSQLHound.zip ConfigManBearPig.zip --ad --pwd Contoso,ContosoIT --ntds ntds.txt -p hashcat.potfile
+adpathfinder -i SharpHound.zip MSSQLHound.zip ConfigManBearPig.zip --ad --pwd Contoso,ContosoIT --ntds ntds.txt -p hashcat.potfile
 ```
 
 Use only the zip files you have. The `--pwd` value is not the AD domain; it is a comma separated list of company, brand, or organisation terms to flag in cracked passwords.
@@ -117,7 +117,7 @@ adpathfinder --ad --pwd Contoso,ContosoIT --ntds ntds.txt -p hashcat.potfile --d
 
 | Stage | What happens |
 | --- | --- |
-| **Ingest** | Merges BloodHound CE and supported OpenGraph zips into one graph. MSSQLHound and ConfigManBearPig data are supported directly |
+| **Ingest** | Merges SharpHound and supported OpenGraph zips into one graph. MSSQLHound and ConfigManBearPig data are supported directly |
 | **Map** | Finds the shortest paths from users and computers to high-value targets. You can [configure the edge set](https://github.com/NetSPI/AD-PathFinder/wiki/Excluded-Relationships) to remove noisy edges like `HasSession` |
 | **Group** | Groups accounts that share a path, so repeated paths are reported as one finding instead of many near-duplicates |
 | **Pair** | Compares NTDS hashes and hashcat potfiles with graph data to show which paths are usable during the assessment |
@@ -131,16 +131,16 @@ adpathfinder --ad --pwd Contoso,ContosoIT --ntds ntds.txt -p hashcat.potfile --d
 ## AD HTML Report
 
 <p align="center">
-  <a href="docs/images/client-report.mp4">
-    <img src="docs/images/client-report.webp" alt="AD HTML report demo video" width="600">
+  <a href="docs/images/ad-report-slideshow.mp4">
+    <img src="docs/images/ad-report-slideshow.webp" alt="AD HTML report demo video" width="600">
   </a>
 </p>
 
 ## Password Audit HTML Report
 
 <p align="center">
-  <a href="docs/images/password-audit.mp4">
-    <img src="docs/images/password-audit.webp" alt="Password audit HTML report demo video" width="600">
+  <a href="docs/images/password-audit-slideshow.mp4">
+    <img src="docs/images/password-audit-slideshow.webp" alt="Password audit HTML report demo video" width="600">
   </a>
 </p>
 

checks/core/base.py

@@ -5,6 +5,22 @@
 
 from .dependencies import CheckDependencies
 from .constants import DataTypes, EntityTypes, DisplayTypes
+from .display.escalation import (
+    EscalationPathDisplayHandler,
+    GroupedEscalationDisplayHandler,
+)
+from .display.group_analysis import GroupAnalysisDisplayHandler
+from .display.shared_graph_paths import SharedGraphPathDisplayHandler
+from .display.simple import SimpleDisplayHandler
+
+_DISPLAY_HANDLERS = {
+    DisplayTypes.GROUPED_ESCALATION_PATHS: GroupedEscalationDisplayHandler,
+    DisplayTypes.ESCALATION_PATHS: EscalationPathDisplayHandler,
+    DisplayTypes.GROUP_ANALYSIS: GroupAnalysisDisplayHandler,
+    DisplayTypes.SHARED_GRAPH_PATHS: SharedGraphPathDisplayHandler,
+    DisplayTypes.SIMPLE: SimpleDisplayHandler,
+}
+
 
 class VulnerabilityCheck:
     RISK_LEVEL = "High"
@@ -187,15 +203,13 @@ def format_relay_finding(self, coerce_source: str | None, target: str, *,
 
     def get_display_method(self) -> Callable[..., Any]:
         suppress = getattr(self, 'suppress_terminal_output', False)
-        if self.DISPLAY_TYPE == DisplayTypes.GROUPED_ESCALATION_PATHS:
-            from checks.core.display.escalation import GroupedEscalationDisplayHandler
-            return GroupedEscalationDisplayHandler(suppress, self, self.sid_mapper).display
-        elif DataTypes.ESCALATION_PATHS in self.REQUIRED_DATA and self.DISPLAY_TYPE == DisplayTypes.ESCALATION_PATHS:
-            from checks.core.display.escalation import EscalationPathDisplayHandler
-            return EscalationPathDisplayHandler(suppress, self, self.sid_mapper).display
-        elif self.DISPLAY_TYPE == DisplayTypes.GROUP_ANALYSIS:
-            from checks.core.display.group_analysis import GroupAnalysisDisplayHandler
-            return GroupAnalysisDisplayHandler(suppress, self, self.sid_mapper).display
-        else:
-            from checks.core.display.simple import SimpleDisplayHandler
-            return SimpleDisplayHandler(suppress, self, self.sid_mapper).display
+        display_type = DisplayTypes(self.DISPLAY_TYPE)
+
+        if (
+            display_type == DisplayTypes.ESCALATION_PATHS
+            and DataTypes.ESCALATION_PATHS not in self.REQUIRED_DATA
+        ):
+            display_type = DisplayTypes.SIMPLE
+
+        handler_cls = _DISPLAY_HANDLERS[display_type]
+        return handler_cls(suppress, self, self.sid_mapper).display

checks/core/constants.py

@@ -28,6 +28,7 @@ class DisplayTypes(StrEnum):
     ESCALATION_PATHS = 'escalation_paths'
     GROUPED_ESCALATION_PATHS = 'grouped_escalation_paths'
     GROUP_ANALYSIS = 'group_analysis'
+    SHARED_GRAPH_PATHS = 'shared_graph_paths'
 
 
 class DisplaySymbols:

checks/core/display/base.py

@@ -0,0 +1,27 @@
+from colorama import Style
+
+
+class BaseDisplayHandler:
+    def __init__(self, suppress_terminal_output=False, check_instance=None, sid_mapper=None):
+        self.suppress_terminal_output = suppress_terminal_output
+        self.check_instance = check_instance
+        self.sid_mapper = sid_mapper
+        self.reset_color = ""
+
+    def _emit_heading(self, category_color, category, content, count, reset_color):
+        if not self.suppress_terminal_output:
+            print(f"\n  {Style.BRIGHT}{category_color}{category}: {count}{reset_color}")
+        content.append(f"\n  {category}: {count}")
+
+    def _emit(self, line, content):
+        if not self.suppress_terminal_output:
+            print(line)
+        content.append(line)
+
+    def _output_line(self, line, content):
+        if not self.suppress_terminal_output:
+            print(f"{self.reset_color}{line}")
+        content.append(line)
+
+    def display(self, results, category_color, category, content, count, reset_color):
+        raise NotImplementedError

checks/core/display/escalation.py

@@ -1,53 +1,51 @@
 from collections import defaultdict
-from colorama import Style
-from ..constants import DataTypes, DisplaySymbols, is_sid
+from .base import BaseDisplayHandler
+from ..constants import DataTypes, DisplaySymbols, EntityTypes, is_sid
 from modules.node_type_cache import extract_ad_report_type_from_labels
 
 
-class EscalationPathDisplayHandler:
-    def __init__(self, suppress_terminal_output=False, check_instance=None, sid_mapper=None):
-        self.suppress_terminal_output = suppress_terminal_output
-        self.check_instance = check_instance
-        self.sid_mapper = sid_mapper
+def _extract_display_info(description):
+    if not description:
+        return ""
+    if " (" in description and description.endswith(")"):
+        return description.split(" (")[-1][:-1]
+    return description
 
+
+class EscalationPathDisplayHandler(BaseDisplayHandler):
     def display(self, results, category_color, category, content, count, reset_color):
-        if not self.suppress_terminal_output:
-            print(f"\n  {Style.BRIGHT}{category_color}{category}: {count}{reset_color}")
-        content.append(f"\n  {category}: {count}")
+        self._emit_heading(category_color, category, content, count, reset_color)
+
+        total_items = len(results)
+        for index, (entity_identifier, item_data) in enumerate(results.items()):
+            if total_items > 1 and index > 0:
+                self._emit("", content)
 
-        for entity_identifier, item_data in results.items():
             if isinstance(item_data, dict):
                 description = item_data.get('description', '')
             else:
                 description = item_data
 
-            display_info = description.split(" (")[-1][:-1] if " (" in description and description.endswith(")") else description if description else ""
+            display_info = _extract_display_info(description)
             display_name = self.sid_mapper.get_display_name(entity_identifier)
 
             if display_info:
                 line = f"{DisplaySymbols.MAIN_ITEM}{display_name} ({display_info})"
             else:
                 line = f"{DisplaySymbols.MAIN_ITEM}{display_name}"
-            if not self.suppress_terminal_output:
-                print(line)
-            content.append(line)
+            self._emit(line, content)
 
             escalation_paths = self._get_escalation_path_details(entity_identifier)
             if escalation_paths:
-                # path is [Node, Rel, Node, Rel, Node] — count relationships not elements
                 steps = len(escalation_paths) // 2
                 final_target = escalation_paths[-1] if escalation_paths else "Unknown"
 
                 leads_line = f"         {DisplaySymbols.PATH_CONNECTOR} LEADS TO{DisplaySymbols.PATH_ARROW}{final_target} (via {steps} steps)"
-                if not self.suppress_terminal_output:
-                    print(leads_line)
-                content.append(leads_line)
+                self._emit(leads_line, content)
 
                 path_str = DisplaySymbols.PATH_ARROW.join(escalation_paths)
                 path_line = f"            {DisplaySymbols.PATH_CONNECTOR} {path_str}"
-                if not self.suppress_terminal_output:
-                    print(path_line)
-                content.append(path_line)
+                self._emit(path_line, content)
 
         return content
 
@@ -86,7 +84,6 @@ def _find_escalation_path_data(self, entity_sid):
 
     def _format_escalation_path(self, full_path):
         path_elements = []
-        # fullPath is interleaved: node, rel, node, rel, ..., final_node
         for item in full_path:
             if isinstance(item, dict):
                 formatted_node = self._format_path_node(item)
@@ -115,22 +112,15 @@ def _format_path_node(self, node):
         return None
 
 
-class GroupedEscalationDisplayHandler:
-    def __init__(self, suppress_terminal_output=False, check_instance=None, sid_mapper=None):
-        self.suppress_terminal_output = suppress_terminal_output
-        self.check_instance = check_instance
-        self.sid_mapper = sid_mapper
-
+class GroupedEscalationDisplayHandler(BaseDisplayHandler):
     def display(self, results, category_color, category, content, count, reset_color):
         if not results:
             return content
 
-        if not self.suppress_terminal_output:
-            print(f"\n  {Style.BRIGHT}{category_color}{category}: {count}{reset_color}")
-        content.append(f"\n  {category}: {count}")
+        self._emit_heading(category_color, category, content, count, reset_color)
 
         entity_type = getattr(self.check_instance, 'ENTITY_TYPE', 'user')
-        entity_label = "Computers" if entity_type == 'computer' else "Users"
+        entity_label = "Computers" if entity_type == EntityTypes.COMPUTER else "Users"
 
         path_to_entities = defaultdict(set)
 
@@ -154,20 +144,20 @@ def display(self, results, category_color, category, content, count, reset_color
             reverse=True
         )
 
-        for path_str, entities_set in sorted_paths:
+        for index, (path_str, entities_set) in enumerate(sorted_paths):
+            if index > 0:
+                self._emit("", content)
+
             sorted_entities = sorted(entities_set)
             entities_str = ", ".join(sorted_entities)
             entities_count = len(entities_set)
 
             line1 = f"    {entity_label} with Shared Path ({entities_count}): {entities_str}"
-            if not self.suppress_terminal_output:
-                print(line1)
-            content.append(line1)
-
-            line2 = f"        Common Escalation Path: {path_str}\n"
-            if not self.suppress_terminal_output:
-                print(line2)
-            content.append(line2)
+            self._emit(line1, content)
+
+            self._emit("", content)
+            line2 = f"        Common Escalation Path: {path_str}"
+            self._emit(line2, content)
 
         return content
 

checks/core/display/group_analysis.py

@@ -1,12 +1,9 @@
-from colorama import Style
+from .base import BaseDisplayHandler
 
 
-class GroupAnalysisDisplayHandler:
+class GroupAnalysisDisplayHandler(BaseDisplayHandler):
     def __init__(self, suppress_terminal_output=False, check_instance=None, sid_mapper=None):
-        self.suppress_terminal_output = suppress_terminal_output
-        self.check_instance = check_instance
-        self.sid_mapper = sid_mapper
-        self.reset_color = ""
+        super().__init__(suppress_terminal_output, check_instance, sid_mapper)
         self._account_analysis = getattr(check_instance, 'account_analysis', None)
 
     def display(self, results, category_color, category, content, count, reset_color):
@@ -15,9 +12,7 @@ def display(self, results, category_color, category, content, count, reset_color
         display_groups = {k: v for k, v in results.items() if not k.startswith("___")}
         group_count = len(display_groups)
 
-        if not self.suppress_terminal_output:
-            print(f"\n  {Style.BRIGHT}{category_color}{category}: {group_count}{reset_color}")
-        content.append(f"\n  {category}: {group_count}")
+        self._emit_heading(category_color, category, content, group_count, reset_color)
 
         enhanced_data_ci = {k.upper(): k for k in enhanced_data.keys()}
 
@@ -42,6 +37,9 @@ def display(self, results, category_color, category, content, count, reset_color
                 if not isinstance(path_obj, dict):
                     continue
 
+                if path_count > 1 and idx > 0:
+                    self._output_line("", content)
+
                 is_last = idx == path_count - 1
                 path_prefix = "      └─" if is_last else "      ├─"
                 cont_prefix = "         " if is_last else "      │  "
@@ -109,8 +107,3 @@ def _render_escalation(self, esc_path, has_esc, cont_prefix, content):
                 content
             )
         self._output_line(f"{cont_prefix}    └─ {esc_str}", content)
-
-    def _output_line(self, line, content):
-        if not self.suppress_terminal_output:
-            print(f"{self.reset_color}{line}")
-        content.append(line)