fix: B3 summary counting, computer SID caching, OpenGraph placement

B3 Summary Counting Logic (skip detection):
- Add TaskType.SKIPPED enum value for dual-homed hosts
- Add TaskRow.skipped() factory method
- Add SKIPPED row when dual-homed detection triggers in online.py
- Fix async_runner.py to check for SKIPPED rows instead of empty results
- Tests updated for new type and factory method

Computer SID Caching:
- Add persistent cache for computer name→SID resolution in sid_resolver.py
- Uses 'computers' category in SQLite cache
- Cache key format: name:{HOSTNAME}:{DOMAIN}
- Eliminates redundant LDAP queries for repeated scans

OpenGraph Message Placement:
- Move JSON auto-generation messages from pre-summary to BloodHound section
- Add Rich-formatted 'BloodHound OpenGraph Integration' section header
- Use Panel and Rule for visual consistency

Author: r0BIT

taskhound/cli.py

@@ -261,25 +261,19 @@ def main():
                         laps_failures.append(laps_result)
 
     # Exports
-    # Auto-generate JSON if OpenGraph is enabled and no explicit JSON output was specified
+    # Track if we need to auto-generate JSON for OpenGraph (defer messages until OpenGraph section)
+    opengraph_json_path = None
+    opengraph_json_overwrites = False
     if args.bh_opengraph and not args.json:
         # Create output directory if it doesn't exist
         import os
 
         os.makedirs(args.bh_output, exist_ok=True)
 
-        # Generate JSON path
-        json_path = f"{args.bh_output}/taskhound_data.json"
-
-        # Warn if file already exists (will be overwritten)
-        if os.path.exists(json_path):
-            warn(f"OpenGraph will overwrite existing file: {json_path}")
-
-        # Inform user about auto-generation and how to customize
-        info(f"Auto-generating JSON for OpenGraph: {json_path}")
-        info("To use a different path, specify --json <path>")
-
-        args.json = json_path
+        # Generate JSON path (messages will be shown in OpenGraph section)
+        opengraph_json_path = f"{args.bh_output}/taskhound_data.json"
+        opengraph_json_overwrites = os.path.exists(opengraph_json_path)
+        args.json = opengraph_json_path
 
     if args.json:
         write_json(args.json, all_rows)
@@ -315,11 +309,24 @@ def main():
 
     # BloodHound OpenGraph Integration
     if args.bh_opengraph:
+        from rich.console import Console
+        from rich.panel import Panel
+
         from .config_model import BloodHoundConfig
 
+        console = Console()
         print()
-        info("BloodHound OpenGraph Integration")
-        print("-" * 50)
+        console.print(Panel.fit(
+            "[bold]BloodHound OpenGraph Integration[/bold]",
+            border_style="blue",
+        ))
+
+        # Show JSON auto-generation messages (deferred from earlier)
+        if opengraph_json_path:
+            if opengraph_json_overwrites:
+                warn(f"OpenGraph will overwrite existing file: {opengraph_json_path}")
+            info(f"Auto-generating JSON for OpenGraph: {opengraph_json_path}")
+            info("To use a different path, specify --json <path>")
 
         # Create consolidated config from args
         bh_config = BloodHoundConfig.from_args_and_config(args)

taskhound/engine/async_runner.py

@@ -194,14 +194,21 @@ def _process_single(
                 for row in target_rows
             )
 
-            result.success = not has_failure
+            # Check if target was skipped (dual-homed duplicate)
+            # process_target adds TaskRow.skipped() rows for dual-homed hosts
+            has_skipped = any(
+                row.type == "SKIPPED"
+                for row in target_rows
+            )
+
+            result.success = not has_failure and not has_skipped
             result.lines = lines
             result.rows = target_rows
             result.laps_result = laps_result
 
-            # Detect skipped targets (dual-homed duplicates)
-            # These return empty lines and no rows, but are not failures
-            if not has_failure and not lines and not target_rows:
+            # Mark as skipped only if explicitly flagged via SKIPPED row
+            # (not just because results are empty after filtering)
+            if has_skipped:
                 result.skipped = True
 
             if has_failure:
@@ -399,13 +406,19 @@ def _run_sequential(
                         for row in target_rows
                     )
 
-                    result.success = not has_failure
+                    # Check if target was skipped (dual-homed duplicate)
+                    has_skipped = any(
+                        row.type == "SKIPPED"
+                        for row in target_rows
+                    )
+
+                    result.success = not has_failure and not has_skipped
                     result.lines = lines
                     result.rows = target_rows
                     result.laps_result = laps_result
 
-                    # Detect skipped targets (dual-homed duplicates)
-                    if not has_failure and not lines and not target_rows:
+                    # Mark as skipped only if explicitly flagged via SKIPPED row
+                    if has_skipped:
                         result.skipped = True
 
                     if has_failure:

taskhound/engine/online.py

@@ -295,6 +295,12 @@ def process_target(
             if not was_first:
                 warn(f"{target}: Skipping - already processed as {previous_target} (dual-homed host: {server_fqdn})")
                 status(f"[Collecting] {target} [SKIP] (duplicate of {previous_target})")
+                # Add SKIPPED row so async_runner can detect this was a dual-homed skip
+                all_rows.append(TaskRow.skipped(
+                    host=server_fqdn,
+                    reason=f"duplicate of {previous_target}",
+                    target_ip=target,
+                ))
                 # Close SMB connection before returning
                 if smb:
                     with contextlib.suppress(Exception):

taskhound/models/task.py

@@ -15,6 +15,7 @@ class TaskType(str, Enum):
     PRIV = "PRIV"
     TASK = "TASK"
     FAILURE = "FAILURE"
+    SKIPPED = "SKIPPED"  # Dual-homed duplicate, not processed
 
 
 @dataclass
@@ -186,3 +187,29 @@ def failure(
             type=TaskType.FAILURE.value,
             reason=reason,
         )
+
+    @classmethod
+    def skipped(
+        cls,
+        host: str,
+        reason: str,
+        target_ip: Optional[str] = None,
+    ) -> "TaskRow":
+        """
+        Create a SKIPPED row for dual-homed hosts already processed.
+
+        Args:
+            host: Hostname or IP of the skipped target
+            reason: Skip reason message (e.g., "duplicate of 192.168.1.1")
+            target_ip: Original target IP/hostname
+
+        Returns:
+            TaskRow with type=SKIPPED
+        """
+        return cls(
+            host=host,
+            path="",
+            target_ip=target_ip,
+            type=TaskType.SKIPPED.value,
+            reason=reason,
+        )

taskhound/utils/sid_resolver.py

@@ -383,7 +383,7 @@ def resolve_name_to_sid_via_ldap(
 ) -> Optional[str]:
     """
     Resolve a computer name or username to its SID using LDAP.
-    This is the reverse operation of resolve_sid_via_ldap.
+    Results are cached persistently to avoid redundant LDAP queries.
     NOW SUPPORTS NTLM HASH AUTHENTICATION via Impacket LDAP!
 
     Args:
@@ -399,6 +399,26 @@ def resolve_name_to_sid_via_ldap(
     Returns:
         SID string (e.g., "S-1-5-21-..."), None if resolution fails
     """
+    # Check cache first (before any processing)
+    from ..utils.cache_manager import get_cache
+    cache = get_cache()
+    
+    if cache and is_computer:
+        # Normalize for cache key: strip $ and domain suffix
+        cache_name = name.upper()
+        if cache_name.endswith("$"):
+            cache_name = cache_name[:-1]
+        if "." in cache_name:
+            cache_name = cache_name.split(".")[0]
+        cache_key = f"name:{cache_name}:{domain.upper()}"
+        
+        cached_sid = cache.get("computers", cache_key)
+        if cached_sid:
+            debug(f"Cache hit for computer {name}: {cached_sid}")
+            return cached_sid
+    else:
+        cache_key = None  # Only cache computers for now
+    
     try:
         # Extract just the name part if it's in USER@DOMAIN format
         search_name = name
@@ -489,6 +509,9 @@ def resolve_name_to_sid_via_ldap(
                             if sid_string:
                                 account_name = attributes.get("sAMAccountName") or attributes.get("cn") or name
                                 info(f"Resolved {account_name} to SID {sid_string} via LDAP")
+                                # Cache for future lookups (computers only)
+                                if cache and cache_key:
+                                    cache.set("computers", cache_key, sid_string)
                                 return sid_string
                             else:
                                 debug(f"Failed to convert binary SID to string for {name}")

tests/test_models_task.py

@@ -24,9 +24,13 @@ def test_failure_value(self):
         """FAILURE has correct value."""
         assert TaskType.FAILURE.value == "FAILURE"
 
+    def test_skipped_value(self):
+        """SKIPPED has correct value."""
+        assert TaskType.SKIPPED.value == "SKIPPED"
+
     def test_all_types_exist(self):
         """All expected types exist."""
-        expected = {"TIER0", "PRIV", "TASK", "FAILURE"}
+        expected = {"TIER0", "PRIV", "TASK", "FAILURE", "SKIPPED"}
         actual = {t.name for t in TaskType}
         assert expected == actual
 
@@ -146,3 +150,38 @@ def test_repr_contains_path(self):
         row = TaskRow(host="SERVER", path="\\Windows\\System32\\Tasks\\MyTask")
         repr_str = repr(row)
         assert "MyTask" in repr_str or "path" in repr_str
+
+
+class TestTaskRowFactoryMethods:
+    """Tests for TaskRow factory methods."""
+
+    def test_skipped_factory_basic(self):
+        """skipped() creates correct TaskRow."""
+        row = TaskRow.skipped("SERVER", "Duplicate of OTHER")
+        assert row.host == "SERVER"
+        assert row.type == TaskType.SKIPPED
+        assert row.reason == "Duplicate of OTHER"
+        assert row.path == ""
+
+    def test_skipped_factory_with_target_ip(self):
+        """skipped() can include target_ip."""
+        row = TaskRow.skipped("SERVER", "Dual-homed duplicate", target_ip="192.168.1.100")
+        assert row.host == "SERVER"
+        assert row.type == TaskType.SKIPPED
+        assert row.target_ip == "192.168.1.100"
+        assert row.reason == "Dual-homed duplicate"
+
+    def test_failure_factory_basic(self):
+        """failure() creates correct TaskRow."""
+        row = TaskRow.failure("SERVER", "Connection refused")
+        assert row.host == "SERVER"
+        assert row.type == TaskType.FAILURE
+        assert row.reason == "Connection refused"
+        assert row.path == ""
+
+    def test_failure_factory_with_target_ip(self):
+        """failure() can include target_ip."""
+        row = TaskRow.failure("SERVER", "Access denied", target_ip="10.0.0.1")
+        assert row.host == "SERVER"
+        assert row.type == TaskType.FAILURE
+        assert row.target_ip == "10.0.0.1"