feat: Add AES key authentication support

r0BIT · r0BIT · commit b0618b764b0d · 2025-12-11T23:28:43.000+01:00
- Add --aes-key CLI option for Kerberos AES-128/256 key authentication
- Thread aes_key through SMB, LDAP, and RPC authentication paths
- Fix Kerberos RPC auth by setting RPC_C_AUTHN_GSS_NEGOTIATE before connect
- Fix 'Could not resolve FQDN' warning when target is already FQDN
diff --git a/README.md b/README.md
@@ -234,6 +234,20 @@ taskhound -u homer.simpson --hashes :5d41402abc4b2a76b9719d911017c592 -d thesimp
 taskhound -u Administrator -p L0c@lAdm1n! -d . -t moe.thesimpsons.local --ldap-user bart.simpson --ldap-password B@rtP@ss --ldap-domain thesimpsons.local
 ```
 
+### AES Key Authentication
+
+Authenticate using Kerberos AES keys (from a keytab, secretsdump, etc.):
+
+```bash
+# AES-256 key (64 hex characters)
+taskhound -u svc_backup -d corp.local --aes-key 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef -t dc01.corp.local --dc-ip 10.0.0.1
+
+# AES-128 key (32 hex characters) 
+taskhound -u admin -d corp.local --aes-key 0123456789abcdef0123456789abcdef -t server01.corp.local --dc-ip 10.0.0.1
+```
+
+> **NOTE**: AES key authentication implies `-k` (Kerberos). The `--dc-ip` flag is recommended for reliable KDC resolution.
+
 **Why separate LDAP credentials?**
 - LDAP SID resolution now uses Impacket's LDAP implementation with NTLM hash support
 - You might only have local admin access but need domain LDAP for SID resolution
@@ -331,6 +345,7 @@ Authentication:
   -d, --domain          Domain (required for online mode)
   --hashes HASHES       NTLM hashes (LM:NT or NT-only format)
   -k, --kerberos        Use Kerberos authentication (supports ccache)
+  --aes-key AES_KEY     AES key for Kerberos (128-bit: 32 hex, 256-bit: 64 hex)
 
 Targets:
   -t, --target          Single target hostname/IP
diff --git a/config/taskhound.toml.example b/config/taskhound.toml.example
@@ -11,6 +11,7 @@
 # domain = "CORP.LOCAL"
 # password = "Password123!"
 # hashes = "aad3b435b51404eeaad3b435b51404ee:..." (optional)
+# aes_key = "0123456789abcdef..." (AES-128: 32 hex chars, AES-256: 64 hex chars)
 # kerberos = true
 
 [target]
diff --git a/taskhound/auth/context.py b/taskhound/auth/context.py
@@ -48,6 +48,7 @@ class AuthContext:
     password: Optional[str] = None
     domain: str = ""
     hashes: Optional[str] = None
+    aes_key: Optional[str] = None  # AES key for Kerberos (128-bit or 256-bit)
     kerberos: bool = False
     dc_ip: Optional[str] = None
     timeout: int = 60
@@ -62,7 +63,7 @@ class AuthContext:
     @property
     def has_credentials(self) -> bool:
         """Check if valid credentials are configured."""
-        return bool(self.username and (self.password or self.hashes or self.kerberos))
+        return bool(self.username and (self.password or self.hashes or self.aes_key or self.kerberos))
 
     @property
     def ldap_effective_domain(self) -> str:
@@ -104,5 +105,6 @@ def __repr__(self) -> str:
             f"AuthContext(username={self.username!r}, domain={self.domain!r}, "
             f"kerberos={self.kerberos}, dc_ip={self.dc_ip!r}, "
             f"has_password={self.password is not None}, "
-            f"has_hashes={self.hashes is not None})"
+            f"has_hashes={self.hashes is not None}, "
+            f"has_aes_key={self.aes_key is not None})"
         )
diff --git a/taskhound/cli.py b/taskhound/cli.py
@@ -192,12 +192,15 @@ def main():
         targets = normalize_targets(targets, args.domain)
 
         # Build AuthContext from args
+        # AES key implies Kerberos authentication
+        kerberos_enabled = args.kerberos or getattr(args, "aes_key", None) is not None
         auth = AuthContext(
             username=args.username,
             password=args.password,
             domain=args.domain,
             hashes=args.hashes,
-            kerberos=args.kerberos,
+            aes_key=getattr(args, "aes_key", None),
+            kerberos=kerberos_enabled,
             dc_ip=args.dc_ip,
             timeout=args.timeout,
             dns_tcp=getattr(args, "dns_tcp", False),
diff --git a/taskhound/config.py b/taskhound/config.py
@@ -180,6 +180,8 @@ def load_config() -> Dict[str, Any]:
         defaults["hashes"] = auth["hashes"]
     if "kerberos" in auth:
         defaults["kerberos"] = auth["kerberos"]
+    if "aes_key" in auth:
+        defaults["aes_key"] = auth["aes_key"]
 
     # Target
     target = config_data.get("target", {})
@@ -354,6 +356,11 @@ def build_parser() -> argparse.ArgumentParser:
         "--hashes", help="NTLM hashes in LM:NT format (or NT-only 32-hex) to use instead of password"
     )
     auth.add_argument("-k", "--kerberos", action="store_true", help="Use Kerberos authentication (supports ccache)")
+    auth.add_argument(
+        "--aes-key",
+        dest="aes_key",
+        help="AES key for Kerberos authentication (AES-128: 32 hex chars, AES-256: 64 hex chars). Implies -k."
+    )
 
     # Target selection
     target = ap.add_argument_group("Target options")
@@ -773,12 +780,13 @@ def validate_args(args):
         print("[!] Either --target or --targets-file is required for online mode")
         sys.exit(1)
 
-    # Authentication method validation - require either password, hash, or Kerberos
-    if not args.password and not args.hashes and not args.kerberos:
+    # Authentication method validation - require either password, hash, AES key, or Kerberos
+    if not args.password and not args.hashes and not getattr(args, "aes_key", None) and not args.kerberos:
         print("[!] ERROR: Authentication required for online mode")
         print("[!] You must specify one of:")
         print("[!]   -p PASSWORD     (password authentication)")
         print("[!]   --hashes HASH   (NTLM hash authentication)")
+        print("[!]   --aes-key KEY   (Kerberos with AES key)")
         print("[!]   -k              (Kerberos authentication with ccache)")
         print()
         if "KRB5CCNAME" in os.environ:
diff --git a/taskhound/engine/online.py b/taskhound/engine/online.py
@@ -161,6 +161,7 @@ def process_target(
     username = auth.username
     password = auth.password
     hashes = auth.hashes
+    aes_key = auth.aes_key
     kerberos = auth.kerberos
     dc_ip = auth.dc_ip
     timeout = auth.timeout
@@ -267,7 +268,7 @@ def process_target(
         else:
             # Standard mode: Direct SMB connection
             smb = smb_connect(
-                target, domain, username, hashes or password, kerberos=kerberos, dc_ip=dc_ip, timeout=timeout
+                target, domain, username, hashes or password, kerberos=kerberos, dc_ip=dc_ip, timeout=timeout, aes_key=aes_key
             )
 
         good(f"{target}: Connected via SMB")
@@ -421,48 +422,55 @@ def process_target(
 
     # Credential validation via Task Scheduler RPC (if requested and not in OPSEC mode)
     if validate_creds and not opsec and password_task_paths:
-        info(f"{target}: Querying Task Scheduler RPC for credential validation...")
-        try:
-            # Parse hashes for RPC auth
-            lm_hash = ""
-            nt_hash = ""
-            if hashes:
-                hash_parts = hashes.split(":")
-                if len(hash_parts) == 2:
-                    lm_hash, nt_hash = hash_parts
-                elif len(hash_parts) == 1 and len(hash_parts[0]) == 32:
-                    nt_hash = hash_parts[0]
-
-            rpc_client = TaskSchedulerRPC(
-                target=target,
-                domain=domain,
-                username=username,
-                password=password or "",
-                lm_hash=lm_hash,
-                nt_hash=nt_hash,
-            )
+        # Skip if using ccache-only Kerberos (no credentials to use for RPC)
+        if kerberos and not password and not hashes and not aes_key:
+            warn(f"{target}: Credential validation not supported with ccache-only Kerberos (use password, --hashes, or --aes-key)")
+        else:
+            info(f"{target}: Querying Task Scheduler RPC for credential validation...")
+            try:
+                # Parse hashes for RPC auth
+                lm_hash = ""
+                nt_hash = ""
+                if hashes:
+                    hash_parts = hashes.split(":")
+                    if len(hash_parts) == 2:
+                        lm_hash, nt_hash = hash_parts
+                    elif len(hash_parts) == 1 and len(hash_parts[0]) == 32:
+                        nt_hash = hash_parts[0]
+
+                rpc_client = TaskSchedulerRPC(
+                    target=target,
+                    domain=domain,
+                    username=username,
+                    password=password or "",
+                    lm_hash=lm_hash,
+                    nt_hash=nt_hash,
+                    aes_key=aes_key or "",
+                    kerberos=kerberos,
+                    dc_ip=dc_ip or "",
+                )
 
-            if rpc_client.connect():
-                # Validate only the tasks we know have Password logon type
-                cred_validation_results = rpc_client.validate_specific_tasks(password_task_paths)
-                rpc_client.disconnect()
-
-                if cred_validation_results:
-                    valid_count = sum(1 for r in cred_validation_results.values() if r.password_valid)
-                    invalid_count = sum(1 for r in cred_validation_results.values()
-                                       if r.credential_status == CredentialStatus.INVALID)
-                    unknown_count = sum(1 for r in cred_validation_results.values()
-                                       if r.credential_status == CredentialStatus.UNKNOWN)
-                    good(f"{target}: Validated {len(cred_validation_results)} password tasks "
-                         f"({valid_count} valid, {invalid_count} invalid, {unknown_count} unknown)")
+                if rpc_client.connect():
+                    # Validate only the tasks we know have Password logon type
+                    cred_validation_results = rpc_client.validate_specific_tasks(password_task_paths)
+                    rpc_client.disconnect()
+
+                    if cred_validation_results:
+                        valid_count = sum(1 for r in cred_validation_results.values() if r.password_valid)
+                        invalid_count = sum(1 for r in cred_validation_results.values()
+                                           if r.credential_status == CredentialStatus.INVALID)
+                        unknown_count = sum(1 for r in cred_validation_results.values()
+                                           if r.credential_status == CredentialStatus.UNKNOWN)
+                        good(f"{target}: Validated {len(cred_validation_results)} password tasks "
+                             f"({valid_count} valid, {invalid_count} invalid, {unknown_count} unknown)")
+                    else:
+                        info(f"{target}: No run info available for password tasks")
                 else:
-                    info(f"{target}: No run info available for password tasks")
-            else:
-                warn(f"{target}: Failed to connect to Task Scheduler RPC")
-        except Exception as e:
-            warn(f"{target}: Credential validation failed: {e}")
-            if debug:
-                traceback.print_exc()
+                    warn(f"{target}: Failed to connect to Task Scheduler RPC")
+            except Exception as e:
+                warn(f"{target}: Credential validation failed: {e}")
+                if debug:
+                    traceback.print_exc()
     elif validate_creds and not password_task_paths:
         info(f"{target}: No password-authenticated tasks found - skipping credential validation")
     elif validate_creds and opsec:
@@ -594,6 +602,7 @@ def process_target(
                     password=ldap_auth_pass,
                     hashes=ldap_auth_hashes,
                     kerberos=kerberos,
+                    aes_key=aes_key,
                     attributes=["pwdLastSet", "sAMAccountName"],
                 )
 
@@ -634,6 +643,7 @@ def process_target(
                 auth_password=ldap_auth_pass,
                 hashes=ldap_auth_hashes,
                 kerberos=kerberos,
+                aes_key=aes_key,
             )
 
             if tier0_cache:
diff --git a/taskhound/smb/connection.py b/taskhound/smb/connection.py
@@ -46,24 +46,27 @@ def smb_connect(
     kerberos: bool = False,
     dc_ip: str = None,
     timeout: int = 60,
+    aes_key: str = None,
 ) -> SMBConnection:
     # Create and authenticate an SMBConnection to `target`.
     #
     # This function prefers passing an explicit lm/nthash when provided and
     # falls back to a cleartext password. For Kerberos mode we delegate to
     # Impacket's kerberosLogin (which supports a KDC host if provided).
+    # If an AES key is provided, Kerberos authentication is used automatically.
     smb = SMBConnection(remoteName=target, remoteHost=target, sess_port=445, timeout=timeout)
 
     pwd, lmhash, nthash = _parse_hashes(password)
 
-    if kerberos:
+    # AES key implies Kerberos authentication
+    if kerberos or aes_key:
         smb.kerberosLogin(
             user=username,
             password=pwd,
             domain=domain,
             lmhash=lmhash,
             nthash=nthash,
-            aesKey=None,
+            aesKey=aes_key or "",
             TGT=None,
             TGS=None,
             kdcHost=dc_ip,
@@ -106,6 +109,7 @@ def smb_login(
     password: str = None,
     kerberos: bool = False,
     dc_ip: str = None,
+    aes_key: str = None,
 ) -> None:
     """
     Authenticate an existing SMBConnection.
@@ -120,17 +124,19 @@ def smb_login(
         password: Password or NTLM hash
         kerberos: Use Kerberos authentication
         dc_ip: Domain controller IP (for Kerberos)
+        aes_key: AES key for Kerberos authentication (128 or 256 bit)
     """
     pwd, lmhash, nthash = _parse_hashes(password)
 
-    if kerberos:
+    # AES key implies Kerberos authentication
+    if kerberos or aes_key:
         smb.kerberosLogin(
             user=username,
             password=pwd,
             domain=domain,
             lmhash=lmhash,
             nthash=nthash,
-            aesKey=None,
+            aesKey=aes_key or "",
             TGT=None,
             TGS=None,
             kdcHost=dc_ip,
@@ -357,10 +363,11 @@ def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip:
     Extract the server's FQDN from an established SMB connection.
 
     Attempts multiple resolution methods in order:
-    1. SMB DNS hostname (most reliable)
-    2. Constructed from SMB hostname + DNS domain
-    3. DNS PTR lookup using DC as nameserver (if dc_ip provided)
-    4. System DNS PTR lookup (fallback)
+    1. If target is already an FQDN (has dots, not an IP), use it directly
+    2. SMB DNS hostname (most reliable)
+    3. Constructed from SMB hostname + DNS domain
+    4. DNS PTR lookup using DC as nameserver (if dc_ip provided)
+    5. System DNS PTR lookup (fallback)
 
     Args:
         smb: Established SMBConnection
@@ -371,6 +378,10 @@ def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip:
     Returns:
         FQDN string (e.g., "DC.badsuccessor.lab") or "UNKNOWN_HOST"
     """
+    # Method 0: If target already looks like an FQDN (has dots and isn't an IP), use it
+    if target_ip and "." in target_ip and not _is_ip_address(target_ip):
+        return target_ip
+
     try:
         # Method 1: Try to get the full DNS hostname directly from SMB
         fqdn = smb.getServerDNSHostName()
diff --git a/taskhound/smb/task_rpc.py b/taskhound/smb/task_rpc.py
diff --git a/taskhound/utils/ldap.py b/taskhound/utils/ldap.py
diff --git a/taskhound/utils/sid_resolver.py b/taskhound/utils/sid_resolver.py
