feat: align CLI output labels + add missing config mappings

Output alignment:
- Align all output labels to 18 chars for consistent formatting
- Shorten 'Decrypted Password' → 'Decrypted Pwd'
- Shorten 'Password Analysis' → 'Pwd Analysis'

Config mappings:
- Add ldap.tier0 for LDAP-based Tier-0 detection
- Add dpapi.key for DPAPI credential decryption
- Add target.threads for parallel scanning
- Add target.rate_limit for rate limiting
- Add new [dpapi] section to taskhound.toml.example

Tests:
- Update test assertions to match new aligned format

Author: r0BIT

config/taskhound.toml.example

@@ -6,7 +6,7 @@
 # Priority: CLI Args > Env Vars > Local Config > User Config > Defaults
 
 [authentication]
-# Default credentials (use with caution!)
+# Default credentials
 # username = "svc_taskhound"
 # domain = "CORP.LOCAL"
 # password = "Password123!"
@@ -19,6 +19,8 @@
 # target = "10.0.0.1"
 # targets_file = "targets.txt"
 # timeout = 60
+# threads = 10  # Parallel workers for multi-target scans (default: 1)
+# rate_limit = 5.0  # Max targets per second (default: unlimited)
 
 [scanning]
 # Default scan options
@@ -75,6 +77,11 @@ color = "#8B5CF6"
 # password = "password"
 # hashes = "LM:NT"
 # domain = "domain.local"
+# tier0 = false  # Enable LDAP-based Tier-0 detection via group membership queries
+
+[dpapi]
+# DPAPI credential decryption settings
+# key = "0x51e43225e5b43b25d3768a2ae7f99934cb35d3ea"  # DPAPI_SYSTEM userkey from LSA secrets
 
 [output]
 # plain = "./output/plain"

taskhound/classification.py

@@ -6,7 +6,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 
 from .utils.logging import warn
 from .utils.sid_resolver import looks_like_domain_user
@@ -30,6 +30,9 @@ class ClassificationResult:
 # Type alias for pre-fetched password data: username -> pwdLastSet datetime
 PwdLastSetCache = Dict[str, datetime]
 
+# Type alias for pre-fetched Tier-0 membership data: username -> (is_tier0, group_list)
+Tier0Cache = Dict[str, Tuple[bool, List[str]]]
+
 
 def _get_task_date_for_analysis(meta: Dict) -> Tuple[Optional[str], bool]:
     """
@@ -96,12 +99,12 @@ def _analyze_password_age(
     if pwd_cache and task_date:
         try:
             from .parsers.highvalue import _analyze_password_freshness
-            
+
             # Normalize username for lookup
             norm_user = runas.split("\\")[-1].lower() if "\\" in runas else runas.lower()
-            
+
             pwd_last_set = pwd_cache.get(norm_user)
-            
+
             if pwd_last_set:
                 risk_level, pwd_analysis = _analyze_password_freshness(task_date, pwd_last_set)
                 if risk_level != "UNKNOWN":
@@ -122,6 +125,7 @@ def classify_task(
     show_unsaved_creds: bool,
     include_local: bool,
     pwd_cache: Optional[PwdLastSetCache] = None,
+    tier0_cache: Optional[Tier0Cache] = None,
 ) -> ClassificationResult:
     """
     Classify a task as TIER-0, PRIV, or TASK based on the runas account.
@@ -138,6 +142,7 @@ def classify_task(
         show_unsaved_creds: Whether to include tasks without saved credentials
         include_local: Whether to include local system accounts
         pwd_cache: Pre-fetched dict of username -> pwdLastSet datetime
+        tier0_cache: Pre-fetched dict of username -> (is_tier0, group_list) from LDAP
 
     Returns:
         ClassificationResult with task_type, reason, password_analysis, should_include
@@ -153,8 +158,9 @@ def classify_task(
         )
 
     # Check for Tier 0 first, then high-value
+    # Priority: BloodHound data > LDAP tier0_cache
     if hv and hv.loaded:
-        # Check Tier 0 classification
+        # Check Tier 0 classification via BloodHound
         is_tier0, tier0_reasons = hv.check_tier0(runas)
         if is_tier0:
             reason = "; ".join(tier0_reasons)
@@ -199,6 +205,35 @@ def classify_task(
                 should_include=True,
             )
 
+    # Check LDAP-based Tier-0 detection (when BloodHound not available)
+    elif tier0_cache:
+        # Normalize username for lookup
+        norm_user = runas.split("\\")[-1].lower() if "\\" in runas else runas.lower()
+        tier0_result = tier0_cache.get(norm_user)
+
+        if tier0_result:
+            is_tier0, groups = tier0_result
+            if is_tier0:
+                reason = f"Tier-0 via LDAP: member of {', '.join(groups)}"
+                password_analysis = None
+
+                if has_no_saved_creds:
+                    reason = f"{reason} (no saved credentials — DPAPI dump not applicable; manipulation requires an interactive session)"
+                else:
+                    password_analysis = _analyze_password_age(hv, runas, meta, rel_path, pwd_cache)
+
+                # Update row in place
+                row.type = TaskType.TIER0.value
+                row.reason = reason
+                row.password_analysis = password_analysis
+
+                return ClassificationResult(
+                    task_type="TIER-0",
+                    reason=reason,
+                    password_analysis=password_analysis,
+                    should_include=True,
+                )
+
     # Regular task - still analyze password age if credentials are stored
     password_analysis = None
     if has_stored_creds:

taskhound/cli.py

@@ -209,24 +209,25 @@ def main():
         )
 
         # Common kwargs for process_target
-        process_kwargs = dict(
-            auth=auth,
-            include_ms=args.include_ms,
-            include_local=args.include_local,
-            hv=hv,
-            debug=args.debug,
-            show_unsaved_creds=args.unsaved_creds,
-            backup_dir=args.backup,
-            credguard_detect=args.credguard_detect,
-            no_ldap=args.no_ldap,
-            loot=args.loot,
-            dpapi_key=args.dpapi_key,
-            bh_connector=bh_connector,
-            concise=not args.verbose,
-            opsec=args.opsec,
-            laps_cache=laps_cache,
-            validate_creds=args.validate_creds,
-        )
+        process_kwargs = {
+            "auth": auth,
+            "include_ms": args.include_ms,
+            "include_local": args.include_local,
+            "hv": hv,
+            "debug": args.debug,
+            "show_unsaved_creds": args.unsaved_creds,
+            "backup_dir": args.backup,
+            "credguard_detect": args.credguard_detect,
+            "no_ldap": args.no_ldap,
+            "loot": args.loot,
+            "dpapi_key": args.dpapi_key,
+            "bh_connector": bh_connector,
+            "concise": not args.verbose,
+            "opsec": args.opsec,
+            "laps_cache": laps_cache,
+            "validate_creds": args.validate_creds,
+            "ldap_tier0": args.ldap_tier0,
+        }
 
         # Parallel mode (--threads > 1)
         if args.threads > 1:
@@ -240,7 +241,7 @@ def main():
 
             start_time = time.perf_counter()
             results = async_engine.run(targets, process_target, **process_kwargs)
-            elapsed_ms = (time.perf_counter() - start_time) * 1000
+            _ = (time.perf_counter() - start_time) * 1000  # elapsed_ms for future use
 
             # Aggregate results
             all_rows, laps_failures, laps_successes = aggregate_results(results)
@@ -307,7 +308,9 @@ def main():
     # Print summary by default (unless disabled)
     if not args.no_summary:
         backup_dir = args.backup if hasattr(args, "backup") and args.backup else None
-        print_summary_table(all_rows, backup_dir, hv_loaded)
+        # Tier-0 detection is available if we have BloodHound data OR --ldap-tier0
+        has_tier0_detection = hv_loaded or args.ldap_tier0
+        print_summary_table(all_rows, backup_dir, has_tier0_detection)
 
         # Print LAPS summary if LAPS was used
         if laps_cache is not None:

taskhound/config.py

@@ -91,6 +91,10 @@ def load_config() -> Dict[str, Any]:
         defaults["target"] = target["target"]
     if "targets_file" in target:
         defaults["targets_file"] = target["targets_file"]
+    if "threads" in target:
+        defaults["threads"] = target["threads"]
+    if "rate_limit" in target:
+        defaults["rate_limit"] = target["rate_limit"]
 
     # Scanning
     scan = config_data.get("scanning", {})
@@ -134,6 +138,11 @@ def load_config() -> Dict[str, Any]:
     if "enabled" in cred_validation:
         defaults["validate_creds"] = cred_validation["enabled"]
 
+    # DPAPI
+    dpapi = config_data.get("dpapi", {})
+    if "key" in dpapi:
+        defaults["dpapi_key"] = dpapi["key"]
+
     # BloodHound
     bh = config_data.get("bloodhound", {})
     if "live" in bh:
@@ -190,6 +199,8 @@ def load_config() -> Dict[str, Any]:
         defaults["ldap_hashes"] = ldap["hashes"]
     if "domain" in ldap:
         defaults["ldap_domain"] = ldap["domain"]
+    if "tier0" in ldap:
+        defaults["ldap_tier0"] = ldap["tier0"]
 
     # Output
     output = config_data.get("output", {})
@@ -421,6 +432,11 @@ def build_parser() -> argparse.ArgumentParser:
     ldap.add_argument(
         "--ldap-domain", help="Alternative domain for SID lookup (can be different from main auth domain)"
     )
+    ldap.add_argument(
+        "--ldap-tier0",
+        action="store_true",
+        help="Enable LDAP-based Tier-0 detection via group membership queries. Checks if runas accounts are members of privileged groups (Domain Admins, Enterprise Admins, etc.) without requiring BloodHound data.",
+    )
 
     # LAPS (Local Administrator Password Solution) options
     laps_group = ap.add_argument_group(

taskhound/engine/online.py

@@ -13,7 +13,7 @@
 from impacket.smbconnection import SessionError
 
 from ..auth import AuthContext
-from ..classification import classify_task, PwdLastSetCache
+from ..classification import PwdLastSetCache, classify_task
 from ..laps import (
     LAPS_ERRORS,
     LAPSCache,
@@ -38,7 +38,10 @@
 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 ..utils.sid_resolver import (
+    format_runas_with_sid_resolution,
+    is_sid,
+)
 from .helpers import sort_tasks_by_priority
 
 
@@ -68,12 +71,11 @@ def _match_decrypted_password(runas: str, decrypted_creds: List, resolved_runas:
     # 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 not is_sid(runas) and 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:
@@ -122,6 +124,7 @@ def process_target(
     opsec: bool = False,
     laps_cache: Optional[LAPSCache] = None,
     validate_creds: bool = False,
+    ldap_tier0: bool = False,
 ) -> Tuple[List[str], Optional[Union[bool, LAPSFailure]]]:
     """
     Connect to `target`, enumerate scheduled tasks, and return printable lines.
@@ -145,6 +148,7 @@ def process_target(
         opsec: Enable OPSEC mode
         laps_cache: LAPS credential cache (if LAPS mode enabled)
         validate_creds: Query Task Scheduler RPC to validate stored credentials
+        ldap_tier0: Check Tier-0 group membership via LDAP (when no BloodHound data)
 
     Returns:
         Tuple of (lines, laps_result) where:
@@ -552,7 +556,7 @@ def process_target(
     if not no_ldap and not opsec and (not hv or not hv.loaded):
         # Collect unique runas users from all tasks with stored credentials
         unique_users = set()
-        for rel_path, xml_bytes in items:
+        for _rel_path, xml_bytes in items:
             meta = parse_task_xml(xml_bytes)
             runas = meta.get("runas")
             if not runas:
@@ -563,17 +567,17 @@ def process_target(
                 # Skip SIDs - we can't look them up by SID in LDAP easily
                 if not is_sid(runas):
                     unique_users.add(runas)
-        
+
         if unique_users:
             info(f"{target}: Querying LDAP for password age data ({len(unique_users)} users)...")
             try:
                 from ..utils.sid_resolver import batch_get_user_attributes
-                
+
                 ldap_auth_domain = ldap_domain or domain
                 ldap_auth_user = ldap_user or username
                 ldap_auth_pass = ldap_password or password
                 ldap_auth_hashes = ldap_hashes or hashes
-                
+
                 results = batch_get_user_attributes(
                     usernames=list(unique_users),
                     domain=ldap_auth_domain,
@@ -584,23 +588,56 @@ def process_target(
                     kerberos=kerberos,
                     attributes=["pwdLastSet", "sAMAccountName"],
                 )
-                
+
                 # Build cache: normalized_username -> pwdLastSet datetime
                 for norm_user, attrs in results.items():
                     pwd_last_set = attrs.get("pwdLastSet")
                     if pwd_last_set:
                         pwd_cache[norm_user] = pwd_last_set
-                
+
                 if pwd_cache:
                     good(f"{target}: Retrieved password age data for {len(pwd_cache)} users")
                 else:
                     info(f"{target}: No password age data available from LDAP")
-                    
+
             except Exception as e:
                 warn(f"{target}: LDAP batch query failed: {e}")
                 if debug:
                     traceback.print_exc()
 
+    # Pre-fetch Tier-0 group members via LDAP (pre-flight approach)
+    # This queries each Tier-0 group once and builds a lookup cache
+    # Only enabled with --ldap-tier0 flag (OPSEC: group membership queries may be logged)
+    tier0_cache: Dict[str, Tuple[bool, list]] = {}  # username -> (is_tier0, group_list)
+    if ldap_tier0 and not no_ldap and (not hv or not hv.loaded):
+        info(f"{target}: Fetching Tier-0 group members via LDAP (pre-flight)...")
+        try:
+            from ..utils.sid_resolver import fetch_tier0_members
+
+            ldap_auth_domain = ldap_domain or domain
+            ldap_auth_user = ldap_user or username
+            ldap_auth_pass = ldap_password or password
+            ldap_auth_hashes = ldap_hashes or hashes
+
+            tier0_cache = fetch_tier0_members(
+                domain=ldap_auth_domain,
+                dc_ip=dc_ip,
+                auth_username=ldap_auth_user,
+                auth_password=ldap_auth_pass,
+                hashes=ldap_auth_hashes,
+                kerberos=kerberos,
+            )
+
+            if tier0_cache:
+                good(f"{target}: Loaded {len(tier0_cache)} Tier-0 users from LDAP")
+            else:
+                info(f"{target}: No Tier-0 users found in domain")
+
+        except Exception as e:
+            warn(f"{target}: LDAP Tier-0 pre-flight failed: {e}")
+            if debug:
+                traceback.print_exc()
+
     for rel_path, xml_bytes in items:
         meta = parse_task_xml(xml_bytes)
         # Save raw XML to backup directory if requested
@@ -697,6 +734,7 @@ def process_target(
 
         # Use shared classification logic
         # pwd_cache was pre-fetched before the loop for password freshness analysis
+        # tier0_cache was pre-fetched for LDAP-based Tier-0 detection (when --ldap-tier0)
         result = classify_task(
             row=row,
             meta=meta,
@@ -706,6 +744,7 @@ def process_target(
             show_unsaved_creds=show_unsaved_creds,
             include_local=include_local,
             pwd_cache=pwd_cache,
+            tier0_cache=tier0_cache,
         )
 
         if not result.should_include: