Restore test coverage to 60.05% (1249 tests)

Added tests for:
- TaskSchedulerRPC context manager (__enter__/__exit__)
- Engine helpers block priority sorting with unknown types
- SMB tasks crawl exception handling
- Date parser zero float strings and overflow errors
- Console format_task_line and collecting_* functions
- Cache manager get_all and invalidate methods

Author: r0BIT

taskhound/classification.py

@@ -5,6 +5,7 @@
 # PRIV (high-value), or TASK (normal) based on the runas account.
 
 from dataclasses import dataclass
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
 
 from .utils.logging import warn
@@ -26,6 +27,10 @@ class ClassificationResult:
     should_include: bool = True  # Whether to include in output
 
 
+# Type alias for pre-fetched password data: username -> pwdLastSet datetime
+PwdLastSetCache = Dict[str, datetime]
+
+
 def _get_task_date_for_analysis(meta: Dict) -> Tuple[Optional[str], bool]:
     """
     Get the best available date for password freshness analysis.
@@ -56,32 +61,54 @@ def _analyze_password_age(
     runas: str,
     meta: Dict,
     rel_path: str,
+    pwd_cache: Optional[PwdLastSetCache] = None,
 ) -> Optional[str]:
     """
     Analyze password age for DPAPI dump viability.
 
+    Uses BloodHound data if available, otherwise uses pre-fetched LDAP data
+    from pwd_cache (if provided).
+
     Args:
-        hv: HighValueLoader instance
+        hv: HighValueLoader instance (can be None)
         runas: The account the task runs as
         meta: Task metadata dict
         rel_path: Task path for warning messages
+        pwd_cache: Pre-fetched dict of username -> pwdLastSet datetime (optional)
 
     Returns:
         Password analysis string or None if not applicable
     """
-    if not hv or not hv.loaded:
-        return None
-
     task_date, is_fallback = _get_task_date_for_analysis(meta)
     if is_fallback and task_date:
         warn(
             f"Task {rel_path} has no explicit creation date - "
             "using trigger StartBoundary for password analysis (may be inaccurate)"
         )
 
-    risk_level, pwd_analysis = hv.analyze_password_age(runas, task_date)
-    if risk_level != "UNKNOWN":
-        return pwd_analysis
+    # Try BloodHound data first
+    if hv and hv.loaded:
+        risk_level, pwd_analysis = hv.analyze_password_age(runas, task_date)
+        if risk_level != "UNKNOWN":
+            return pwd_analysis
+
+    # Fall back to pre-fetched LDAP data if BloodHound not available
+    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":
+                    return pwd_analysis
+        except Exception as e:
+            from .utils.logging import debug
+            debug(f"Password analysis failed for {runas}: {e}")
 
     return None
 
@@ -94,6 +121,7 @@ def classify_task(
     hv: Optional[Any],
     show_unsaved_creds: bool,
     include_local: bool,
+    pwd_cache: Optional[PwdLastSetCache] = None,
 ) -> ClassificationResult:
     """
     Classify a task as TIER-0, PRIV, or TASK based on the runas account.
@@ -109,6 +137,7 @@ def classify_task(
         hv: HighValueLoader instance (can be None)
         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
 
     Returns:
         ClassificationResult with task_type, reason, password_analysis, should_include
@@ -134,7 +163,7 @@ def classify_task(
             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)
+                password_analysis = _analyze_password_age(hv, runas, meta, rel_path, pwd_cache)
 
             # Update row in place
             row.type = TaskType.TIER0.value
@@ -156,7 +185,7 @@ def classify_task(
             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)
+                password_analysis = _analyze_password_age(hv, runas, meta, rel_path, pwd_cache)
 
             # Update row in place
             row.type = TaskType.PRIV.value
@@ -172,8 +201,9 @@ def classify_task(
 
     # Regular task - still analyze password age if credentials are stored
     password_analysis = None
-    if hv and hv.loaded and has_stored_creds:
-        password_analysis = _analyze_password_age(hv, runas, meta, rel_path)
+    if has_stored_creds:
+        # Try BloodHound first, then pre-fetched LDAP data
+        password_analysis = _analyze_password_age(hv, runas, meta, rel_path, pwd_cache)
 
     # Determine if we should include this regular task
     should_include = (

taskhound/config.py

@@ -481,10 +481,16 @@ def validate_args(args):
             print("[!] OPSEC mode enabled: Disabling Credential Guard detection")
             args.credguard_detect = False
 
-        # Note: SID lookups are handled dynamically in engine.py, but we can warn here
-        if not args.no_ldap:
-            # We don't force no_ldap=True because engine handles it, but good to know
-            pass
+        # --validate-creds requires RPC queries that are noisy
+        if getattr(args, 'validate_creds', False):
+            print("[!] ERROR: --validate-creds is incompatible with --opsec mode")
+            print("[!] Credential validation requires Task Scheduler RPC queries (\\pipe\\atsvc)")
+            print("[!] These queries may trigger security monitoring and are not OPSEC-safe.")
+            print("[!]")
+            print("[!] Options:")
+            print("[!]   1. Remove --opsec flag to validate credentials")
+            print("[!]   2. Remove --validate-creds flag to maintain OPSEC")
+            sys.exit(1)
 
     # Handle LAPS + OPSEC compatibility
     if getattr(args, "laps", False):

taskhound/engine/offline.py

@@ -281,6 +281,7 @@ def _process_offline_host(
             row.credentials_hint = "no_saved_credentials"
 
         # Use shared classification logic
+        # Note: pwd_cache is None for offline mode since we can't query LDAP
         result = classify_task(
             row=row,
             meta=meta,
@@ -289,6 +290,7 @@ def _process_offline_host(
             hv=hv,
             show_unsaved_creds=show_unsaved_creds,
             include_local=include_local,
+            pwd_cache=None,  # No LDAP in offline mode
         )
 
         if not result.should_include:

taskhound/engine/online.py

@@ -13,7 +13,7 @@
 from impacket.smbconnection import SessionError
 
 from ..auth import AuthContext
-from ..classification import classify_task
+from ..classification import classify_task, PwdLastSetCache
 from ..laps import (
     LAPS_ERRORS,
     LAPSCache,
@@ -545,6 +545,62 @@ def process_target(
     priv_lines: List[str] = []
     task_lines: List[str] = []
 
+    # Pre-fetch pwdLastSet for all unique users via single LDAP batch query
+    # This provides password freshness analysis when BloodHound is not available
+    # Skipped in OPSEC mode to avoid LDAP queries that may be audited
+    pwd_cache: PwdLastSetCache = {}
+    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:
+            meta = parse_task_xml(xml_bytes)
+            runas = meta.get("runas")
+            if not runas:
+                continue
+            logon_type = (meta.get("logon_type") or "").strip().lower()
+            # Only query users from tasks with stored credentials
+            if logon_type == "password":
+                # 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,
+                    dc_ip=dc_ip,
+                    username=ldap_auth_user,
+                    password=ldap_auth_pass,
+                    hashes=ldap_auth_hashes,
+                    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()
+
     for rel_path, xml_bytes in items:
         meta = parse_task_xml(xml_bytes)
         # Save raw XML to backup directory if requested
@@ -601,7 +657,8 @@ def process_target(
 
         # 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):
+        # Skipped in OPSEC mode to avoid SMB/LSARPC and LDAP queries
+        if is_sid(runas) and not opsec:
             _, row.resolved_runas = format_runas_with_sid_resolution(
                 runas,
                 hv_loader=hv,
@@ -639,6 +696,7 @@ def process_target(
             row.credentials_hint = "stored_credentials"
 
         # Use shared classification logic
+        # pwd_cache was pre-fetched before the loop for password freshness analysis
         result = classify_task(
             row=row,
             meta=meta,
@@ -647,6 +705,7 @@ def process_target(
             hv=hv,
             show_unsaved_creds=show_unsaved_creds,
             include_local=include_local,
+            pwd_cache=pwd_cache,
         )
 
         if not result.should_include: