Merge branch 'main' into marshall-get-gpos

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

nxc/helpers/modules.py

@@ -0,0 +1,43 @@
+from nxc.paths import NXC_PATH
+from os import makedirs
+import logging
+import datetime
+
+
+def create_log_dir(module_name):
+    makedirs(f"{NXC_PATH}/logs/{module_name}", exist_ok=True)
+
+
+def create_loot_dir(module_name):
+    makedirs(f"{NXC_PATH}/loot/{module_name}", exist_ok=True)
+
+
+def generate_module_log_file(module_name):
+    create_log_dir(module_name)
+    return f"{NXC_PATH}/logs/{module_name}/{datetime.now().strftime('%Y-%m-%d')}.log"
+
+
+def create_module_logger(module_name):
+    log_file = generate_module_log_file(module_name)
+    module_logger = logging.getLogger(module_name)
+    module_logger.propagate = False
+    module_logger.setLevel(logging.INFO)
+    module_file_handler = logging.FileHandler(log_file)
+    module_file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
+    module_logger.addHandler(module_file_handler)
+    return module_logger
+
+
+def add_loot_data(module_name, filename, data):
+    create_loot_dir(module_name)
+    loot_file = get_loot_data_filepath(module_name, filename)
+    with open(loot_file, "a") as file:
+        file.write(data)
+
+
+def get_loot_data_filepath(module_name, filename):
+    return f"{NXC_PATH}/loot/{module_name}/{filename}"
+
+
+def get_loot_data_folder(module_name):
+    return f"{NXC_PATH}/loot/{module_name}"

nxc/modules/example_module.py

@@ -1,3 +1,4 @@
+from nxc.helpers.modules import add_loot_data
 from nxc.helpers.misc import CATEGORY
 
 
@@ -8,7 +9,7 @@ class NXCModule:
     Module by @yomama
     """
 
-    name = "example module"
+    name = "example_module"  # Make sure this is unique and one word (no spaces)
     description = "I do something"
     supported_protocols = []  # Example: ['smb', 'mssql']
     category = CATEGORY.ENUMERATION  # Must be one of "Enumeration", "Privilege Escalation" or "Credential Dumping". Use what fits best.
@@ -45,8 +46,11 @@ class NXCModule:
         except Exception as e:
             context.log.exception(f"Exception occurred: {e}")  # This will display an exception traceback screen after an exception was raised and should only be used for critical errors
 
+        # Use this function to add loot data you want to save to $NXC_PATH/loot/$MODULE_NAME/$FILENAME
+        add_loot_data(self.name, "custom_loot_file.txt", "Data can be anything you want, passwords, hashes, or anything")
+
     def on_admin_login(self, context, connection):
         """Concurrent.
         Required if on_login is not present
-        This gets called on each authenticated connection with  Administrative privileges
+        This gets called on each authenticated connection with Administrative privileges
         """

nxc/modules/gpos.py

@@ -0,0 +1,272 @@
+import os
+import ntpath
+import shutil
+from impacket.dcerpc.v5.lsat import DCERPCSessionError
+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.config import process_secret
+from nxc.protocols.smb.samrfunc import LSAQuery
+from nxc.protocols.ldap.constants import ACCESS_MASK_TO_TEXT_LOOKUP
+
+
+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"]
+
+    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"
+
+    def on_login(self, context, connection):
+        context.log.display("Searching for GPOs via SMB in SYSVOL share")
+
+        # list the contents of SYSVOL to find the domain folder
+        policies = connection.conn.listPath("SYSVOL", f"{connection.domain}\\Policies\\*")
+        context.log.debug(f"Found policies path: {connection.domain}\\Policies")
+        context.log.debug(f"Found folders in policies path: {policies}")
+
+        gpos_found = []
+
+        for item in policies:
+            item_name = item.get_longname()
+            context.log.debug(f"Item: {item}, Item name: {item_name}")
+            # make sure it's a directory and not ./.. and check if it's a GUID folder (typical format: {XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX})
+            if item_name not in [".", ".."] and item.is_directory() and item_name.startswith("{") and item_name.endswith("}"):
+                gpo_guid = item_name
+                gpo_path = os.path.join(f"{connection.domain}", "Policies", f"{gpo_guid}")
+
+                gpos_found.append((gpo_guid, gpo_path))
+
+                if self.gpo_name:
+                    # try to find the GPO's display name by checking for gpt.ini
+                    display_name = self.get_gpo_display_name(context, connection, gpo_path)
+
+                    # filter by display name if specified
+                    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
+
+                    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}")
+                else:
+                    # if no GPO name filter, just show all
+                    display_name = self.get_gpo_display_name(context, connection, gpo_path)
+                    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.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")
+                    self.download_gpo(context, connection, gpo_path, self.download_dest, gpo_guid)
+
+        context.log.success(f"GPOs Found: {len(gpos_found)}")
+
+    def get_gpo_display_name(self, context, connection, gpo_path):
+        """Try to get the display name of a GPO by reading gpt.ini file directly into memory"""
+        try:
+            ini_path = f"{gpo_path}\\GPT.ini"
+            context.log.debug(f"GPT.ini path: {ini_path}")
+            ini_content = bytearray()
+
+            # read the file directly into memory through a callback
+            def callback(data):
+                ini_content.extend(data)
+
+            connection.conn.getFile("SYSVOL", ini_path, callback)
+            display_name = "Unknown"
+            ini_text = ini_content.decode("utf-8", errors="replace")
+
+            for line in ini_text.splitlines():
+                if line.lower().startswith("displayname="):
+                    display_name = line.strip().split("=", 1)[1]
+                    break
+
+            return display_name
+        except Exception as e:
+            context.log.debug(f"Could not read GPO display name: {e}")
+            # extract GUID from path as fallback name
+            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}")
+
+        # save the original share to reset it later, since we need to download from SYSVOL
+        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}")
+            # clean up the directory if it was created and download failed
+            if os.path.exists(dest_folder):
+                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:
+            # reset the share to the original share
+            smb_conn.args.share = original_share
+
+    def parse_aces(self, context, sd):
+        """Parse ACEs from security descriptor and display all permissions"""
+        if not sd["Dacl"]:
+            context.log.highlight("  No DACL found - no access restrictions")
+            return
+
+        ace_count = sd["Dacl"]["AceCount"]
+        context.log.highlight(f"  DACL: {ace_count} ACEs (Access Control Entries)")
+
+        context.log.debug(f"DACL AceCount: {ace_count}")
+        context.log.debug(f"DACL Data: {sd['Dacl']}")
+
+        for i in range(ace_count):
+            try:
+                ace = sd["Dacl"]["Data"][i]
+                ace_type = ace["TypeName"]
+
+                # ACCESS_MASK is accessed from the ACE's Ace field
+                # based on ldaptypes structure: ace['Ace']['Mask']['Mask']
+                access_mask = int(ace["Ace"]["Mask"]["Mask"])
+
+                sid = ace["Ace"]["Sid"].formatCanonical()
+                context.log.debug(f"ACE {i} mask: 0x{access_mask:08x}")
+
+                # build permissions list using the lookup dictionary
+                permissions = [mask_name for mask_value, mask_name in ACCESS_MASK_TO_TEXT_LOOKUP.items() if access_mask & mask_value]
+
+                # if no specific permissions found, show the raw mask
+                perm_str = ", ".join(permissions) if permissions else f"0x{access_mask:08x}"
+
+                context.log.highlight(f"    ACE {i + 1}: {ace_type}")
+                context.log.highlight(f"         SID: {sid} - {ace['Ace']['SidName']}")
+                context.log.highlight(f"         Permissions: {perm_str}")
+            except Exception as ace_error:
+                context.log.debug(f"Error parsing ACE {i}: {ace_error}")
+                context.log.debug(f"ACE object structure: {ace}")
+                context.log.debug(f"ACE keys: {list(ace.keys()) if hasattr(ace, 'keys') else 'No keys'}")
+                continue
+
+    def get_folder_security_info(self, context, connection, gpo_guid, gpo_path):
+        """Get security information for a GPO folder and analyze permissions"""
+        rpc_transport = transport.SMBTransport(
+            connection.host,
+            connection.host,
+            filename=r"\srvsvc",  # named pipe for SRVS service
+            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)
+        context.log.debug("Connected to SRVS")
+
+        sec_info = (OWNER_SECURITY_INFORMATION | GROUP_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION)
+
+        # format paths correctly for Windows RPC (note null terminator!)
+        share_name = "SYSVOL\x00"
+        gpo_path_fixed = gpo_path.replace("/", "\\")
+        rpc_path = f"\\{gpo_path_fixed}\x00"
+
+        context.log.debug(f"Getting security info for GPO path: {rpc_path.rstrip(chr(0))}")
+
+        sec_info = srvs.hNetrpGetFileSecurity(dce, share_name, rpc_path, sec_info)
+        context.log.success(f"Retrieved security information ({len(sec_info)} bytes)")
+
+        try:
+            lsa_query = LSAQuery(
+                connection.username,
+                connection.password,
+                connection.domain,
+                connection.port,
+                connection.host,
+                connection.host,
+                connection.kdcHost,
+                connection.aesKey,
+                connection.kerberos,
+                logger=context.log,
+            )
+        except Exception as lsa_error:
+            context.log.debug(f"Error creating LSAQuery: {lsa_error}")
+            return
+
+        sec_desc = ldaptypes.SR_SECURITY_DESCRIPTOR(sec_info)
+        context.log.highlight(f"Security Descriptor for GPO {gpo_guid}:")
+        context.log.debug(f"Security information: {sec_desc}")
+
+        owner_sid = sec_desc["OwnerSid"].formatCanonical()
+        group_sid = sec_desc["GroupSid"].formatCanonical()
+        context.log.debug(f"Owner SID: {owner_sid}")
+        context.log.debug(f"Group SID: {group_sid}")
+
+        sec_desc["OwnerName"] = lsa_query.lookup_sids([owner_sid])[0]
+        sec_desc["GroupName"] = lsa_query.lookup_sids([group_sid])[0]
+
+        for i, ace in enumerate(sec_desc["Dacl"]["Data"]):
+            try:
+                sec_desc["Dacl"]["Data"][i]["Ace"]["SidName"] = lsa_query.lookup_sids([ace["Ace"]["Sid"].formatCanonical()])[0]
+                context.log.debug(f"{sec_desc['Dacl']['Data'][i]['Ace']['SidName']}")
+            except DCERPCSessionError as e:
+                sec_desc["Dacl"]["Data"][i]["Ace"]["SidName"] = "Unknown (potentially deleted user)"
+                context.log.debug(f"Error looking up SID: {e}")
+                continue
+
+        if sec_desc["OwnerSid"]:
+            context.log.highlight(f"  Owner SID: {owner_sid} - {sec_desc['OwnerName']}")
+        if sec_desc["GroupSid"]:
+            context.log.highlight(f"  Group SID: {group_sid} - {sec_desc['GroupName']}")
+        if sec_desc["Dacl"]:
+            self.parse_aces(context, sec_desc)
+
+        dce.disconnect()

nxc/protocols/ldap/constants.py

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

nxc/protocols/smb.py

@@ -8,6 +8,7 @@ from Cryptodome.Hash import MD4
 
 from impacket.smbconnection import SMBConnection, SessionError
 from impacket.smb import SMB_DIALECT
+from impacket.smb3structs import FILE_READ_DATA, FILE_WRITE_DATA
 from impacket.examples.secretsdump import (
     RemoteOperations,
     SAMHashes,
@@ -1876,24 +1877,101 @@ class smb(connection):
         for src, dest in self.args.put_file:
             self.put_file_single(src, dest)
 
-    def get_file_single(self, remote_path, download_path):
+    def download_file(self, share_name, remote_path, dest_file, access_mode=FILE_READ_DATA):
+        try:
+            self.logger.debug(f"Getting file from {share_name}:{remote_path} with access mode {access_mode}")
+            self.conn.getFile(share_name, remote_path, dest_file, shareAccessMode=access_mode)
+            return True
+        except SessionError as e:
+            if "STATUS_SHARING_VIOLATION" in str(e):
+                self.logger.debug(f"Sharing violation on {remote_path}: {e}")
+            else:
+                self.logger.debug(f"SessionError when attempting to download file {remote_path}: {e}")
+            return False
+        except Exception as e:
+            self.logger.debug(f"Other error when attempting to download file {remote_path}: {e}")
+            return False
+
+    def get_file_single(self, remote_path, download_path, silent=False):
         share_name = self.args.share
-        self.logger.display(f'Copying "{remote_path}" to "{download_path}"')
+        if not silent:
+            self.logger.display(f"Copying '{remote_path}' to '{download_path}'")
         if self.args.append_host:
             download_path = f"{self.hostname}-{remote_path}"
         with open(download_path, "wb+") as file:
-            try:
-                self.conn.getFile(share_name, remote_path, file.write)
-                self.logger.success(f'File "{remote_path}" was downloaded to "{download_path}"')
-            except Exception as e:
-                self.logger.fail(f'Error writing file "{remote_path}" from share "{share_name}": {e}')
-                if os.path.getsize(download_path) == 0:
-                    os.remove(download_path)
+            if self.download_file(share_name, remote_path, file.write):
+                if not silent:
+                    self.logger.success(f"File '{remote_path}' was downloaded to '{download_path}'")
+            else:
+                self.logger.debug("Opening with READ alone failed, trying to open file with READ/WRITE access")
+                if self.download_file(share_name, remote_path, file.write, FILE_READ_DATA | FILE_WRITE_DATA):
+                    if not silent:
+                        self.logger.success(f"File '{remote_path}' was downloaded to '{download_path}'")
+                else:
+                    if not silent:
+                        self.logger.fail(f"Error downloading file '{remote_path}' from share '{share_name}'")
 
     def get_file(self):
         for src, dest in self.args.get_file:
             self.get_file_single(src, dest)
 
+    def download_folder(self, folder, dest, recursive=False, silent=False, base_dir=None, ignore_empty=False):
+        self.logger.debug(f"Downloading folder with args: {folder}, {dest}, Recursive: {recursive}, Silent: {silent}, Base dir: {base_dir}, Ignore empty: {ignore_empty}")
+        normalized_folder = ntpath.normpath(folder)
+        base_folder = os.path.basename(normalized_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 = item.get_longname()
+            dir_path = ntpath.normpath(ntpath.join(normalized_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)
+                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)")
+        return
+
+    def get_folder(self):
+        recursive = self.args.recursive
+        ignore_empty = getattr(self.args, "ignore_empty_folders", False)
+        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, False, 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

@@ -74,9 +74,12 @@ def proto_args(parser, parents):
     segroup.add_argument("--pattern", nargs="+", help="pattern(s) to search for in folders, filenames and file content")
     segroup.add_argument("--regex", nargs="+", help="regex(s) to search for in folders, filenames and file content")
 
-    files_group = smb_parser.add_argument_group("Files", "Options for remote file interaction")
+    files_group = smb_parser.add_argument_group("Files", "Options for put and get remote files")
     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=False, action="store_true", 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", "Options for executing commands")

nxc/protocols/smb/samrfunc.py

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

tests/e2e_commands.txt

@@ -164,6 +164,13 @@ netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M winscp
 netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M zerologon
 #netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M change-password -o NEWPASS=Password123
 #netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M change-password -o NEWNTHASH=58A478135A93AC3BF058A5EA0E8FDB71
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o NAME="Default Domain Policy" -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o NAME="Default Domain Policy" -o ALL_PROPS=True -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o NAME="Default" -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o NAME="Default" -o FUZZY=True -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o NAME="Default" -o FUZZY=True -o ALL_PROPS=True -o DOWNLOAD=False
+netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M get_gpos -o ALL_PROPS=True -o 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 runasppl -M uac -M webdav -M wifi -M coerce_plus
@@ -219,6 +226,7 @@ netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M maq
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M subnets
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M user-desc
 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 pso
 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M dump-computers
 ##### WINRM
 netexec winrm TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS # need an extra space after this command due to regex