feat: add --dns-tcp flag for SOCKS proxy compatibility

- Add --dns-tcp CLI flag to force DNS queries over TCP
- Add dns_tcp field to AuthContext dataclass
- Update DNS resolution functions with use_tcp parameter:
  - _dns_ptr_lookup() in smb/connection.py
  - resolve_dc_hostname() in utils/ldap.py
  - get_ldap_connection() in utils/ldap.py
- Thread dns_tcp through engine/online.py call chain
- Add TOML config support for dns_tcp setting
- Update taskhound.toml.example with dns_tcp option

fix: pass hashes parameter correctly in verify_ldap_connection()

- Fix bug where test_hashes was hardcoded to None instead of
  being passed to resolve_sid_via_ldap(), causing --hashes
  authentication to fail during LDAP credential validation

test: add unit tests for TCP DNS functionality

Author: r0BIT

config/taskhound.toml.example

@@ -21,6 +21,7 @@
 # timeout = 60
 # threads = 10  # Parallel workers for multi-target scans (default: 1)
 # rate_limit = 5.0  # Max targets per second (default: unlimited)
+# dns_tcp = false  # Force DNS over TCP (required for SOCKS proxies/proxychains)
 
 [scanning]
 # Default scan options

taskhound/auth/context.py

@@ -51,6 +51,7 @@ class AuthContext:
     kerberos: bool = False
     dc_ip: Optional[str] = None
     timeout: int = 60
+    dns_tcp: bool = False  # Force DNS queries over TCP (for SOCKS proxies)
 
     # LDAP-specific credentials (optional override)
     ldap_domain: Optional[str] = None

taskhound/cli.py

@@ -200,6 +200,7 @@ def main():
             kerberos=args.kerberos,
             dc_ip=args.dc_ip,
             timeout=args.timeout,
+            dns_tcp=getattr(args, "dns_tcp", False),
             ldap_domain=args.ldap_domain,
             ldap_user=args.ldap_user,
             ldap_password=args.ldap_password,

taskhound/config.py

@@ -195,6 +195,8 @@ def load_config() -> Dict[str, Any]:
         defaults["threads"] = target["threads"]
     if "rate_limit" in target:
         defaults["rate_limit"] = target["rate_limit"]
+    if "dns_tcp" in target:
+        defaults["dns_tcp"] = target["dns_tcp"]
 
     # Scanning
     scan = config_data.get("scanning", {})
@@ -372,6 +374,12 @@ def build_parser() -> argparse.ArgumentParser:
         help="Maximum targets per second (default: unlimited). Use to avoid triggering security alerts "
         "or hitting Windows SMB connection limits. Example: --rate-limit 5 limits to 5 targets/second.",
     )
+    target.add_argument(
+        "--dns-tcp",
+        action="store_true",
+        help="Force DNS queries over TCP instead of UDP. Required when using SOCKS proxies or proxychains "
+        "(UDP doesn't traverse SOCKS). Combine with --dc-ip for reliable DNS resolution through tunnels.",
+    )
 
     # High value / scanning options
     scan = ap.add_argument_group("Scanning options")

taskhound/engine/online.py

@@ -164,6 +164,7 @@ def process_target(
     kerberos = auth.kerberos
     dc_ip = auth.dc_ip
     timeout = auth.timeout
+    dns_tcp = auth.dns_tcp
     ldap_domain = auth.ldap_domain
     ldap_user = auth.ldap_user
     ldap_password = auth.ldap_password
@@ -197,9 +198,9 @@ def process_target(
                 if _is_ip_address(target):
                     # Try DC first, then system DNS
                     if dc_ip:
-                        discovered_hostname = _dns_ptr_lookup(target, nameserver=dc_ip)
+                        discovered_hostname = _dns_ptr_lookup(target, nameserver=dc_ip, use_tcp=dns_tcp)
                     if not discovered_hostname:
-                        discovered_hostname = _dns_ptr_lookup(target, nameserver=None)
+                        discovered_hostname = _dns_ptr_lookup(target, nameserver=None, use_tcp=dns_tcp)
 
                     if discovered_hostname:
                         # Extract just the hostname part (before first dot) for LAPS lookup
@@ -275,7 +276,7 @@ def process_target(
         # This is critical when target is an IP address - BloodHound needs FQDNs
         # Tries: 1) SMB hostname, 2) DNS via DC, 3) System DNS
 
-        server_fqdn = get_server_fqdn(smb, target_ip=target, dc_ip=dc_ip)
+        server_fqdn = get_server_fqdn(smb, target_ip=target, dc_ip=dc_ip, dns_tcp=dns_tcp)
         if server_fqdn and server_fqdn != "UNKNOWN_HOST":
             if server_fqdn.upper() != target.upper():
                 info(f"{target}: Resolved FQDN: {server_fqdn}")

taskhound/smb/connection.py

@@ -352,7 +352,7 @@ def get_server_sid(
         return None
 
 
-def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip: Optional[str] = None) -> str:
+def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip: Optional[str] = None, dns_tcp: bool = False) -> str:
     """
     Extract the server's FQDN from an established SMB connection.
 
@@ -366,6 +366,7 @@ def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip:
         smb: Established SMBConnection
         target_ip: Original target IP address (used for DNS fallback)
         dc_ip: Domain Controller IP to use as DNS server
+        dns_tcp: Force DNS queries over TCP (required for SOCKS proxies)
 
     Returns:
         FQDN string (e.g., "DC.badsuccessor.lab") or "UNKNOWN_HOST"
@@ -395,12 +396,12 @@ def get_server_fqdn(smb: SMBConnection, target_ip: Optional[str] = None, dc_ip:
     if target_ip and _is_ip_address(target_ip):
         # Method 3: Try using DC as DNS server (if provided)
         if dc_ip:
-            fqdn = _dns_ptr_lookup(target_ip, nameserver=dc_ip)
+            fqdn = _dns_ptr_lookup(target_ip, nameserver=dc_ip, use_tcp=dns_tcp)
             if fqdn:
                 return fqdn
 
         # Method 4: Try system DNS
-        fqdn = _dns_ptr_lookup(target_ip, nameserver=None)
+        fqdn = _dns_ptr_lookup(target_ip, nameserver=None, use_tcp=dns_tcp)
         if fqdn:
             return fqdn
 
@@ -422,33 +423,38 @@ def _is_ip_address(hostname: str) -> bool:
     return False
 
 
-def _dns_ptr_lookup(ip: str, nameserver: Optional[str] = None) -> Optional[str]:
+def _dns_ptr_lookup(ip: str, nameserver: Optional[str] = None, use_tcp: bool = False) -> Optional[str]:
     """
     Perform DNS PTR lookup to resolve IP to hostname.
 
     Args:
         ip: IP address to resolve
         nameserver: Optional DNS server to query (e.g., DC IP)
+        use_tcp: Force DNS queries over TCP (required for SOCKS proxies)
 
     Returns:
         FQDN if successful, None otherwise
     """
     try:
-        # Try using dnspython if available (supports custom nameserver)
-        if nameserver:
+        # Try using dnspython if available (supports custom nameserver and TCP)
+        if nameserver or use_tcp:
             try:
                 import dns.resolver
                 import dns.reversename
 
                 # Create resolver with custom nameserver
                 resolver = dns.resolver.Resolver(configure=False)
-                resolver.nameservers = [nameserver]
+                if nameserver:
+                    resolver.nameservers = [nameserver]
                 resolver.timeout = 3
                 resolver.lifetime = 3
 
+                # Force TCP if requested (required for SOCKS/proxychains)
+                tcp_flag = use_tcp
+
                 # Perform PTR lookup
                 rev_name = dns.reversename.from_address(ip)
-                answers = resolver.resolve(rev_name, "PTR")
+                answers = resolver.resolve(rev_name, "PTR", tcp=tcp_flag)
 
                 if answers:
                     # Return first answer, strip trailing dot

taskhound/utils/ldap.py

@@ -31,7 +31,7 @@ def parse_ntlm_hashes(hashes: Optional[str]) -> Tuple[str, str]:
         return "", hashes
 
 
-def resolve_dc_hostname(dc_ip: str, domain: str) -> Optional[str]:
+def resolve_dc_hostname(dc_ip: str, domain: str, use_tcp: bool = False) -> Optional[str]:
     """
     Resolve DC IP to hostname for Kerberos SPN construction.
 
@@ -43,6 +43,7 @@ def resolve_dc_hostname(dc_ip: str, domain: str) -> Optional[str]:
     Args:
         dc_ip: Domain controller IP address
         domain: Domain name (for constructing FQDN)
+        use_tcp: Force DNS queries over TCP (required for SOCKS proxies)
 
     Returns:
         DC hostname (FQDN) or None if resolution fails
@@ -59,7 +60,8 @@ def resolve_dc_hostname(dc_ip: str, domain: str) -> Optional[str]:
         resolver.lifetime = 3
 
         rev_name = dns.reversename.from_address(dc_ip)
-        answers = resolver.resolve(rev_name, "PTR")
+        # Force TCP if requested (required for SOCKS/proxychains)
+        answers = resolver.resolve(rev_name, "PTR", tcp=use_tcp)
         if answers:
             hostname = str(answers[0]).rstrip(".")
             # If we got a short name, append the domain
@@ -101,6 +103,7 @@ def get_ldap_connection(
     hashes: Optional[str] = None,
     kerberos: bool = False,
     dc_host: Optional[str] = None,
+    use_tcp: bool = False,
 ) -> ldap_impacket.LDAPConnection:
     """
     Establish LDAP connection to domain controller.
@@ -117,6 +120,7 @@ def get_ldap_connection(
         hashes: NTLM hashes in LM:NT or NT format
         kerberos: Use Kerberos authentication
         dc_host: DC hostname for Kerberos SPN (optional, will try to resolve)
+        use_tcp: Force DNS queries over TCP (required for SOCKS proxies)
 
     Returns:
         LDAPConnection object
@@ -134,7 +138,7 @@ def get_ldap_connection(
     # If not provided, try to resolve
     kerberos_target = dc_host
     if kerberos and not kerberos_target:
-        kerberos_target = resolve_dc_hostname(dc_ip, domain)
+        kerberos_target = resolve_dc_hostname(dc_ip, domain, use_tcp=use_tcp)
         if kerberos_target:
             debug(f"LDAP: Resolved DC hostname for Kerberos SPN: {kerberos_target}")
         else:

taskhound/utils/network.py

@@ -66,7 +66,7 @@ def verify_ldap_connection(
             info("Using domain SID derived from BloodHound data for realistic testing")
 
         info(f"Testing SID resolution with: {test_sid}")
-        result = resolve_sid_via_ldap(test_sid, test_domain, dc_ip, test_username, test_password, None, kerberos)
+        result = resolve_sid_via_ldap(test_sid, test_domain, dc_ip, test_username, test_password, test_hashes, kerberos)
 
         if result:
             good(f"LDAP test successful: {test_sid} -> {result}")

tests/test_ldap_utils.py

@@ -132,6 +132,34 @@ def test_getfqdn_returns_ip_no_hostname(self, mock_gethostbyaddr, mock_getfqdn):
 
         assert result is None
 
+    def test_use_tcp_parameter_accepted(self):
+        """Should accept use_tcp parameter without error"""
+        # Just verify the function accepts the parameter (actual DNS lookup will fail/be mocked)
+        with patch('taskhound.utils.ldap.socket.gethostbyaddr') as mock_gethostbyaddr:
+            mock_gethostbyaddr.return_value = ("dc01.example.com", [], [])
+            result = resolve_dc_hostname("192.168.1.1", "example.com", use_tcp=True)
+            assert result == "dc01.example.com"
+
+    @patch('dns.reversename.from_address')
+    @patch('dns.resolver.Resolver')
+    def test_use_tcp_passes_to_resolver(self, mock_resolver_class, mock_from_address):
+        """Should pass tcp=True to resolver.resolve() when use_tcp=True"""
+        mock_resolver = MagicMock()
+        mock_resolver_class.return_value = mock_resolver
+        mock_from_address.return_value = "1.1.168.192.in-addr.arpa"
+        
+        mock_answer = MagicMock()
+        mock_answer.__str__ = MagicMock(return_value="dc01.example.com.")
+        mock_resolver.resolve.return_value = [mock_answer]
+        
+        result = resolve_dc_hostname("192.168.1.1", "example.com", use_tcp=True)
+        
+        # Verify resolve was called with tcp=True
+        mock_resolver.resolve.assert_called_once()
+        call_args = mock_resolver.resolve.call_args
+        assert call_args[1].get('tcp') is True
+        assert result == "dc01.example.com"
+
 
 # ============================================================================
 # Test: get_ldap_connection