fix: improve SID resolution and credential matching for DPAPI loot

r0BIT · r0BIT · commit 915a2c88ac98 · 2025-11-27T01:44:30.000+01:00
- Add resolved_runas field to TaskRow model to store resolved username
- Resolve SIDs once early in processing flow, reuse for credential matching
- Pass resolved_runas to format_block() to avoid redundant resolution
- Update credential summary to show 'username (SID)' format when resolved
- Remove wasteful double SID resolution during credential matching

This ensures SID-based tasks (e.g., SIDOnlyTask2) correctly match to
decrypted credentials using the session cache from the same run.
diff --git a/taskhound/cli.py b/taskhound/cli.py
@@ -22,7 +22,7 @@
 from .opengraph import generate_opengraph_files
 from .output.bloodhound import upload_opengraph_to_bloodhound
 from .output.printer import print_results
-from .output.summary import print_summary_table
+from .output.summary import print_decrypted_credentials, print_summary_table
 from .output.writer import write_csv, write_json, write_plain
 from .parsers.highvalue import HighValueLoader
 from .utils.cache_manager import init_cache
@@ -300,6 +300,10 @@ def main():
     if args.opengraph:
         generate_opengraph_files(args.opengraph, all_rows)
 
+    # Print decrypted credentials summary (always shown when credentials found)
+    # This is printed BEFORE the summary table so high-value findings are visible
+    print_decrypted_credentials(all_rows)
+
     # Print summary by default (unless disabled)
     if not args.no_summary:
         backup_dir = args.backup if hasattr(args, "backup") and args.backup else None
diff --git a/taskhound/engine/online.py b/taskhound/engine/online.py
@@ -38,9 +38,70 @@
 from ..smb.tasks import crawl_tasks, smb_listdir
 from ..utils.logging import debug as log_debug
 from ..utils.logging import good, info, status, warn
+from ..utils.sid_resolver import format_runas_with_sid_resolution, is_sid
 from .helpers import sort_tasks_by_priority
 
 
+def _match_decrypted_password(runas: str, decrypted_creds: List, resolved_runas: Optional[str] = None) -> Optional[str]:
+    """
+    Match a task's runas field to decrypted credentials and return the password.
+
+    Handles various runas formats:
+    - Simple username: "highpriv"
+    - Domain\\user: "DOMAIN\\highpriv"
+    - Raw SID: "S-1-5-21-..." (uses resolved_runas if provided)
+
+    Args:
+        runas: The RunAs field from the task (may be raw SID)
+        decrypted_creds: List of ScheduledTaskCredential objects
+        resolved_runas: Pre-resolved username if runas was a SID (from earlier resolution)
+
+    Returns:
+        Decrypted password if found, None otherwise
+    """
+    if not decrypted_creds or not runas:
+        return None
+
+    # Build list of usernames to try matching
+    usernames_to_try = []
+
+    # If we have a pre-resolved username, use it
+    if resolved_runas:
+        usernames_to_try.append(resolved_runas.lower())
+    
+    # Also try the original runas if it's not a raw SID
+    runas_normalized = runas.lower()
+    if not is_sid(runas):
+        if runas_normalized not in usernames_to_try:
+            usernames_to_try.append(runas_normalized)
+
+    # If no valid usernames to try (unresolved SID), can't match
+    if not usernames_to_try:
+        return None
+
+    for cred in decrypted_creds:
+        if not cred.username:
+            continue
+
+        cred_user_normalized = cred.username.lower()
+
+        for try_username in usernames_to_try:
+            # Match full domain\user or partial matches
+            if cred_user_normalized == try_username:
+                # Exact match
+                return cred.password
+            elif "\\" in cred_user_normalized and "\\" not in try_username:
+                # Cred has domain, try_username doesn't - match on username part only
+                if cred_user_normalized.split("\\")[-1] == try_username:
+                    return cred.password
+            elif "\\" in try_username and "\\" not in cred_user_normalized:
+                # try_username has domain, cred doesn't - match on username part only
+                if try_username.split("\\")[-1] == cred_user_normalized:
+                    return cred.password
+
+    return None
+
+
 def process_target(
     target: str,
     all_rows: List[TaskRow],
@@ -538,6 +599,31 @@ def process_target(
                 else:
                     row.cred_detail = f"Unknown status (code: {row.cred_return_code})"
 
+        # Resolve SID early if runas is a SID - store result for credential matching and output
+        # This ensures we only resolve once per task, and the result is available for all uses
+        if is_sid(runas):
+            _, row.resolved_runas = format_runas_with_sid_resolution(
+                runas,
+                hv_loader=hv,
+                bh_connector=bh_connector,
+                smb_connection=smb,
+                no_ldap=no_ldap,
+                domain=domain,
+                dc_ip=dc_ip,
+                username=username,
+                password=password,
+                hashes=hashes,
+                kerberos=kerberos,
+                ldap_domain=ldap_domain,
+                ldap_user=ldap_user,
+                ldap_password=ldap_password,
+                ldap_hashes=ldap_hashes,
+            )
+
+        # Enrich row with decrypted password if available from DPAPI loot
+        if decrypted_creds:
+            row.decrypted_password = _match_decrypted_password(runas, decrypted_creds, row.resolved_runas)
+
         # Add Credential Guard status to each row
         row.credential_guard = credguard_status
         # Determine if the task stores credentials or runs with token/S4U (no saved credentials)
@@ -597,6 +683,7 @@ def process_target(
                     decrypted_creds=decrypted_creds,
                     concise=concise,
                     cred_validation=row.to_dict() if row.cred_status else None,
+                    resolved_runas=row.resolved_runas,
                 )
             )
             priv_count += 1
@@ -630,6 +717,7 @@ def process_target(
                     decrypted_creds=decrypted_creds,
                     concise=concise,
                     cred_validation=row.to_dict() if row.cred_status else None,
+                    resolved_runas=row.resolved_runas,
                 )
             )
 
diff --git a/taskhound/models/task.py b/taskhound/models/task.py
@@ -56,6 +56,7 @@ class TaskRow:
         cred_last_run: ISO timestamp of last task run
         cred_return_code: Hex return code from last execution
         cred_detail: Human-readable credential validation detail
+        resolved_runas: Resolved username if runas was a SID (for credential matching)
     """
 
     # Required fields (set during construction)
@@ -73,6 +74,7 @@ class TaskRow:
 
     # Task identity
     runas: Optional[str] = None
+    resolved_runas: Optional[str] = None  # Resolved username if runas was a SID
     command: Optional[str] = None
     arguments: Optional[str] = None
     author: Optional[str] = None
@@ -99,6 +101,9 @@ class TaskRow:
     cred_return_code: Optional[str] = None
     cred_detail: Optional[str] = None
 
+    # DPAPI looted credentials
+    decrypted_password: Optional[str] = None
+
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary for JSON/CSV export."""
         return asdict(self)
diff --git a/taskhound/output/printer.py b/taskhound/output/printer.py
@@ -142,6 +142,7 @@ def format_block(
     decrypted_creds: Optional[List] = None,
     concise: bool = False,
     cred_validation: Optional[Dict[str, Any]] = None,
+    resolved_runas: Optional[str] = None,
 ) -> List[str]:
     # Format a small pretty-print block used by the CLI output.
     #
@@ -153,31 +154,66 @@ def format_block(
     else:
         header = "[TASK]"
 
-    # Resolve SID in RunAs field for better display (uses 4-tier fallback: offline BH → API → SMB → LDAP)
-    display_runas, resolved_username = format_runas_with_sid_resolution(
-        runas,
-        hv,
-        bh_connector,
-        smb_connection,
-        no_ldap,
-        domain,
-        dc_ip,
-        username,
-        password,
-        hashes,
-        kerberos,
-        ldap_domain,
-        ldap_user,
-        ldap_password,
-        ldap_hashes,
-    )
+    # Use pre-resolved username if available, otherwise resolve now
+    if resolved_runas:
+        # Already resolved - format display string
+        from ..utils.sid_resolver import is_sid
+        if is_sid(runas):
+            display_runas = f"{resolved_runas} ({runas})"
+        else:
+            display_runas = runas
+        resolved_username = resolved_runas
+    else:
+        # Resolve SID in RunAs field for better display (uses 4-tier fallback: offline BH → API → SMB → LDAP)
+        display_runas, resolved_username = format_runas_with_sid_resolution(
+            runas,
+            hv,
+            bh_connector,
+            smb_connection,
+            no_ldap,
+            domain,
+            dc_ip,
+            username,
+            password,
+            hashes,
+            kerberos,
+            ldap_domain,
+            ldap_user,
+            ldap_password,
+            ldap_hashes,
+        )
 
     if concise:
         # Concise output: One line per task
-        # Format: [KIND] RunAs | Path | What
+        # Format: [KIND] RunAs | Path | What | (optional reason) | (optional password)
         line = f"{header} {display_runas} | {rel_path} | {what}"
         if extra_reason:
             line += f" | {extra_reason}"
+
+        # In concise mode, show decrypted password inline if available
+        if decrypted_creds and kind in ["TIER-0", "PRIV"]:
+            # Normalize the runas for comparison
+            # Handle resolved SID format: "username (S-1-5-21-...)" -> extract just username
+            runas_normalized = runas.lower()
+            if " (s-1-5-" in runas_normalized:
+                runas_normalized = runas_normalized.split(" (s-1-5-")[0].strip()
+
+            for cred in decrypted_creds:
+                if cred.username:
+                    cred_user_normalized = cred.username.lower()
+                    matched = False
+                    if cred_user_normalized == runas_normalized:
+                        matched = True
+                    elif "\\" in cred_user_normalized and "\\" not in runas_normalized:
+                        if cred_user_normalized.split("\\")[-1] == runas_normalized:
+                            matched = True
+                    elif "\\" in runas_normalized and "\\" not in cred_user_normalized:
+                        if runas_normalized.split("\\")[-1] == cred_user_normalized:
+                            matched = True
+                    if matched:
+                        line += f" | PWD: {cred.password}"
+                        break
+
         return [line]
 
     base = [f"\n{header} {rel_path}"]
@@ -252,23 +288,58 @@ def format_block(
         # Check if we have a decrypted password for this user
         decrypted_password = None
         if decrypted_creds:
-            # Normalize the runas for comparison
+            # Use resolved_username if available (handles SID-only runas fields)
+            # Also try the display_runas which may have "username (SID)" format
+            usernames_to_try = []
+            
+            # Add resolved username from SID resolution
+            if resolved_username:
+                usernames_to_try.append(resolved_username.lower())
+            
+            # Also try extracting from display_runas format "username (S-1-5-21-...)"
+            display_runas_lower = display_runas.lower()
+            if " (s-1-5-" in display_runas_lower:
+                username_part = display_runas_lower.split(" (s-1-5-")[0].strip()
+                if username_part and username_part not in usernames_to_try:
+                    usernames_to_try.append(username_part)
+            elif not display_runas_lower.startswith("s-1-5-"):
+                # Not a raw SID, add as-is
+                if display_runas_lower not in usernames_to_try:
+                    usernames_to_try.append(display_runas_lower)
+            
+            # Try the original runas if it's not a raw SID
             runas_normalized = runas.lower()
+            if not runas_normalized.startswith("s-1-5-"):
+                if " (s-1-5-" in runas_normalized:
+                    username_part = runas_normalized.split(" (s-1-5-")[0].strip()
+                    if username_part and username_part not in usernames_to_try:
+                        usernames_to_try.append(username_part)
+                elif runas_normalized not in usernames_to_try:
+                    usernames_to_try.append(runas_normalized)
+            
             for cred in decrypted_creds:
                 if cred.username:
                     cred_user_normalized = cred.username.lower()
-                    # Match full domain\user or partial matches
-                    if cred_user_normalized == runas_normalized:
-                        # Exact match
-                        decrypted_password = cred.password
-                        break
-                    elif "\\" in cred_user_normalized and "\\" not in runas_normalized:
-                        # Cred has domain, runas doesn't - match on username part only
-                        if cred_user_normalized.split("\\")[-1] == runas_normalized:
+                    
+                    for try_username in usernames_to_try:
+                        # Match full domain\user or partial matches
+                        if cred_user_normalized == try_username:
+                            # Exact match
                             decrypted_password = cred.password
                             break
-                    # Note: We DON'T match when runas has domain but cred doesn't
-                    # A credential without domain is likely a local account, not domain account
+                        elif "\\" in cred_user_normalized and "\\" not in try_username:
+                            # Cred has domain, try_username doesn't - match on username part only
+                            if cred_user_normalized.split("\\")[-1] == try_username:
+                                decrypted_password = cred.password
+                                break
+                        elif "\\" in try_username and "\\" not in cred_user_normalized:
+                            # try_username has domain, cred doesn't - match on username part only
+                            if try_username.split("\\")[-1] == cred_user_normalized:
+                                decrypted_password = cred.password
+                                break
+                    
+                    if decrypted_password:
+                        break
 
         # Show decrypted password if available, otherwise show next step
         if decrypted_password:
diff --git a/taskhound/output/summary.py b/taskhound/output/summary.py
diff --git a/tests/test_task_model.py b/tests/test_task_model.py
