Merge pull request Pennyw0rth#1053 from XiaoliChan/list-snapshot

[ShadowCopy] Add `list-snapshots` function for SMB & WMI

Author: NeffIsBack

nxc/modules/enum_dns.py

@@ -29,16 +29,16 @@ def options(self, context, module_options):
     def on_admin_login(self, context, connection):
         if not self.domains:
             domains = []
-            output = connection.wmi("Select Name FROM MicrosoftDNS_Zone", "root\\microsoftdns")
+            output = connection.wmi_query("Select Name FROM MicrosoftDNS_Zone", "root\\microsoftdns")
             domains = [result["Name"]["value"] for result in output] if output else []
             context.log.success(f"Domains retrieved: {domains}")
         else:
             domains = [self.domains]
         data = ""
 
         for domain in domains:
-            output = connection.wmi(
-                f"Select TextRepresentation FROM MicrosoftDNS_ResourceRecord WHERE DomainName = {domain}",
+            output = connection.wmi_query(
+                f"Select TextRepresentation FROM MicrosoftDNS_ResourceRecord WHERE DomainName = '{domain}'",
                 "root\\microsoftdns",
             )
 

nxc/modules/get_netconnections.py

@@ -22,7 +22,7 @@ def options(self, context, module_options):
 
     def on_admin_login(self, context, connection):
         data = []
-        cards = connection.wmi("select DNSDomainSuffixSearchOrder, IPAddress from win32_networkadapterconfiguration")
+        cards = connection.wmi_query("select DNSDomainSuffixSearchOrder, IPAddress from win32_networkadapterconfiguration")
         if cards:
             for c in cards:
                 if c["IPAddress"].get("value"):

nxc/modules/wcc.py

@@ -399,8 +399,8 @@ def check_laps(self):
         return success, reasons
 
     def check_last_successful_update(self):
-        records = self.connection.wmi(wmi_query="Select TimeGenerated FROM Win32_ReliabilityRecords Where EventIdentifier=19", namespace="root\\cimv2")
-        if isinstance(records, bool) or len(records) == 0:
+        records = self.connection.wmi_query(wql="Select TimeGenerated FROM Win32_ReliabilityRecords Where EventIdentifier=19", namespace="root\\cimv2")
+        if not records:
             return False, ["No update found"]
         most_recent_update_date = records[0]["TimeGenerated"]["value"]
         most_recent_update_date = most_recent_update_date.split(".")[0]

nxc/protocols/smb.py

@@ -1718,10 +1718,10 @@ def pass_pol(self):
         return PassPolDump(self).dump()
 
     @requires_admin
-    def wmi(self, wmi_query=None, namespace=None):
+    def wmi_query(self, wql=None, namespace=None, callback_func=None):
         records = []
-        if not wmi_query:
-            wmi_query = self.args.wmi.strip("\n")
+        if not wql:
+            wql = self.args.wmi_query.strip("\n")
 
         if not namespace:
             namespace = self.args.wmi_namespace
@@ -1742,28 +1742,29 @@ def wmi(self, wmi_query=None, namespace=None):
             iWbemLevel1Login = IWbemLevel1Login(iInterface)
             iWbemServices = iWbemLevel1Login.NTLMLogin(namespace, NULL, NULL)
             iWbemLevel1Login.RemRelease()
-            iEnumWbemClassObject = iWbemServices.ExecQuery(wmi_query)
+            iEnumWbemClassObject = iWbemServices.ExecQuery(wql)
         except Exception as e:
             self.logger.fail(f"Execute WQL error: {e}")
             if "iWbemLevel1Login" in locals():
                 dcom.disconnect()
         else:
-            self.logger.info(f"Executing WQL syntax: {wmi_query}")
-            while True:
-                try:
-                    wmi_results = iEnumWbemClassObject.Next(0xFFFFFFFF, 1)[0]
-                    record = wmi_results.getProperties()
-                    records.append(record)
-                    for k, v in record.items():
-                        if k != "TimeGenerated":  # from the wcc module, but this is a small hack to get it to stop spamming - TODO: add in method to disable output for this function
-                            self.logger.highlight(f"{k} => {v['value']}")
-                except Exception as e:
-                    if str(e).find("S_FALSE") < 0:
-                        raise e
-                    else:
-                        break
+            self.logger.info(f"Executing WQL syntax: {wql}")
+            try:
+                if not callback_func:
+                    while True:
+                        wmi_results = iEnumWbemClassObject.Next(0xFFFFFFFF, 1)[0]
+                        record = wmi_results.getProperties()
+                        records.append(record)
+                        for k, v in record.items():
+                            if k != "TimeGenerated":  # from the wcc module, but this is a small hack to get it to stop spamming - TODO: add in method to disable output for this function
+                                self.logger.highlight(f"{k} => {v['value']}")
+                else:
+                    callback_func(iEnumWbemClassObject, records)
+            except Exception as e:
+                if str(e).find("S_FALSE") < 0:
+                    self.logger.debug(e)
             dcom.disconnect()
-        return records if records else False
+        return records
 
     def spider(
         self,
@@ -2235,6 +2236,20 @@ def firefox_callback(secret):
         if self.output_file:
             self.output_file.close()
 
+    @requires_admin
+    def list_snapshots(self):
+        drive = self.args.list_snapshots
+
+        self.logger.info(f"Retrieving volume shadow copies of drive {drive}.")
+        snapshots = self.conn.listSnapshots(self.conn.connectTree(drive), "/")
+        if not snapshots:
+            self.logger.info("No volume shadow copies found.")
+            return
+        self.logger.highlight(f"{'Drive':<8}{'Shadow Copies GMT SMB PATH':<26}")
+        self.logger.highlight(f"{'------':<8}{'--------------------------':<26}")
+        for i in snapshots:
+            self.logger.highlight(f"{drive:<8}{i:<26}")
+
     @requires_admin
     def lsa(self):
         try:

nxc/protocols/smb/proto_args.py

@@ -47,6 +47,7 @@ def proto_args(parser, parents):
     cred_gathering_group.add_argument("--sccm", choices={"wmi", "disk"}, nargs="?", const="disk", help="dump SCCM secrets from target systems")
     cred_gathering_group.add_argument("--mkfile", action="store", help="DPAPI option. File with masterkeys in form of {GUID}:SHA1")
     cred_gathering_group.add_argument("--pvk", action="store", help="DPAPI option. File with domain backupkey")
+    cred_gathering_group.add_argument("--list-snapshots", nargs="?", dest="list_snapshots", const="ADMIN$", help="Lists the VSS snapshots (default: %(const)s)")
 
     mapping_enum_group = smb_parser.add_argument_group("Mapping/Enumeration")
     mapping_enum_group.add_argument("--shares", type=str, nargs="?", const="", help="Enumerate shares and access, filter on specified argument (read ; write ; read,write)")
@@ -72,8 +73,8 @@ def proto_args(parser, parents):
     mapping_enum_group.add_argument("--taskkill", type=str, help="Kills a specific PID or a proces name's PID's")
 
     wmi_group = smb_parser.add_argument_group("WMI Queries")
-    wmi_group.add_argument("--wmi", metavar="QUERY", type=str, help="issues the specified WMI query")
-    wmi_group.add_argument("--wmi-namespace", metavar="NAMESPACE", default="root\\cimv2", help="WMI Namespace")
+    wmi_group.add_argument("--wmi-query", metavar="QUERY", dest="wmi_query", type=str, help="Issues the specified WMI query")
+    wmi_group.add_argument("--wmi-namespace", metavar="NAMESPACE", default="root\\cimv2", help="WMI Namespace (default: %(default)s)")
 
     spidering_group = smb_parser.add_argument_group("Spidering Shares")
     spidering_group.add_argument("--spider", metavar="SHARE", type=str, help="share to spider")

nxc/protocols/wmi.py

@@ -45,6 +45,8 @@ def __init__(self, args, db, host):
             "0000052B": "STATUS_WRONG_PASSWORD",
             "00000721": "RPC_S_SEC_PKG_ERROR"
         }
+        self.iWbemLevel1Login = None
+        self.dcom_conn = None
 
         connection.__init__(self, args, db, host)
 
@@ -58,6 +60,14 @@ def proto_logger(self):
             }
         )
 
+    # Redefine disconnect function.
+    def disconnect(self):
+        if self.conn:
+            self.conn.disconnect()
+        if self.dcom_conn:
+            self.dcom_conn.disconnect()
+        return
+
     def create_conn_obj(self):
         connection_target = fr"ncacn_ip_tcp:{self.remoteName}[{self.port!s}]"
         self.logger.debug(f"Creating WMI connection object to {connection_target}")
@@ -145,18 +155,15 @@ def print_host_info(self):
 
     def check_if_admin(self):
         try:
-            dcom = DCOMConnection(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, oxidResolver=True, doKerberos=self.doKerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, remoteHost=self.host)
-            iInterface = dcom.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
+            self.dcom_conn = DCOMConnection(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, oxidResolver=True, doKerberos=self.doKerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, remoteHost=self.host)
+            iInterface = self.dcom_conn.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
             flag, self.stringBinding = dcom_FirewallChecker(iInterface, self.host, self.args.rpc_timeout)
         except Exception as e:
             self.logger.debug(f"Received error while checking admin: {e}")
-            if "dcom" in locals():
-                dcom.disconnect()
             if "access_denied" not in str(e).lower():
                 self.logger.fail(str(e))
         else:
             if not flag or not self.stringBinding:
-                dcom.disconnect()
                 error_msg = f'Check admin error: dcom initialization failed with stringbinding: "{self.stringBinding}", please try "--rpc-timeout" option. (probably is admin)'
 
                 if not self.stringBinding:
@@ -165,16 +172,14 @@ def check_if_admin(self):
                 self.logger.fail(error_msg) if not flag else self.logger.debug(error_msg)
             else:
                 try:
-                    iWbemLevel1Login = IWbemLevel1Login(iInterface)
-                    iWbemServices = iWbemLevel1Login.NTLMLogin("//./root/cimv2", NULL, NULL)
+                    self.iWbemLevel1Login = IWbemLevel1Login(iInterface)
+                    _ = self.iWbemLevel1Login.NTLMLogin("//./root/cimv2", NULL, NULL)
+                    self.iWbemLevel1Login.RemRelease()
                 except Exception as e:
-                    dcom.disconnect()
-
                     if "access_denied" not in str(e).lower():
                         self.logger.fail(str(e))
                         return False
                 else:
-                    dcom.disconnect()
                     self.logger.extra["protocol"] = "WMI"
                     self.admin_privs = True
 
@@ -360,48 +365,60 @@ def hash_login(self, domain, username, ntlm_hash):
                 self.logger.success(out)
                 return True
 
-    # It's very complex to use wmi from rpctansport "convert" to dcom, so let we use dcom directly.
     @requires_admin
-    def wmi(self, wql=None, namespace=None):
-        """Execute WQL syntax via WMI
-
-        This is done via the --wmi flag
-        """
+    def wmi_query(self, wql=None, namespace=None, callback_func=None):
         records = []
         if not wql:
-            wql = self.args.wmi.strip("\n")
+            wql = self.args.wmi_query.strip("\n")
 
         if not namespace:
             namespace = self.args.wmi_namespace
 
         try:
-            dcom = DCOMConnection(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, oxidResolver=True, doKerberos=self.doKerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, remoteHost=self.host)
-            iInterface = dcom.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
-            iWbemLevel1Login = IWbemLevel1Login(iInterface)
-            iWbemServices = iWbemLevel1Login.NTLMLogin(namespace, NULL, NULL)
-            iWbemLevel1Login.RemRelease()
+            iWbemServices = self.iWbemLevel1Login.NTLMLogin(namespace, NULL, NULL)
+            self.iWbemLevel1Login.RemRelease()
             iEnumWbemClassObject = iWbemServices.ExecQuery(wql)
         except Exception as e:
-            dcom.disconnect()
             self.logger.debug(str(e))
             self.logger.fail(f"Execute WQL error: {e}")
             return False
         else:
             self.logger.info(f"Executing WQL syntax: {wql}")
             try:
-                while True:
-                    wmi_results = iEnumWbemClassObject.Next(0xFFFFFFFF, 1)[0]
-                    record = wmi_results.getProperties()
-                    records.append(record)
-                    for k, v in record.items():
-                        self.logger.highlight(f"{k} => {v['value']}")
+                if not callback_func:
+                    while True:
+                        wmi_results = iEnumWbemClassObject.Next(0xFFFFFFFF, 1)[0]
+                        record = wmi_results.getProperties()
+                        records.append(record)
+                        for k, v in record.items():
+                            self.logger.highlight(f"{k} => {v['value']}")
+                else:
+                    callback_func(iEnumWbemClassObject, records)
             except Exception as e:
                 if str(e).find("S_FALSE") < 0:
                     self.logger.debug(e)
+            return records
 
-            dcom.disconnect()
+    def list_snapshots(self):
+        drive = self.args.list_snapshots
+        self.logger.info(f"Retrieving volume shadow copies of drive {drive}.")
+        wql = "select ID, DeviceObject, ClientAccessible, InstallDate from win32_shadowcopy"
 
-            return records
+        def callback_func(iEnumWbemClassObject, records):
+            while True:
+                wmi_results = iEnumWbemClassObject.Next(0xFFFFFFFF, 1)[0]
+                record = dict(wmi_results.getProperties())
+                records.append(record)
+
+        snapshots = self.wmi_query(wql=wql, namespace="root\\cimv2", callback_func=callback_func)
+        if not snapshots:
+            self.logger.info("No volume shadow copies found.")
+            return
+
+        self.logger.highlight(f"{'Drive':<8}{'Shadow Copy ID':<40}{'ClientAccessible':<18}{'InstallDate':<27}{'Device Object':<50}")
+        self.logger.highlight(f"{'------':<8}{'--------------':<40}{'----------------':<18}{'-----------':<27}{'-------------':<50}")
+        for record in snapshots:
+            self.logger.highlight(f"{drive:<8}{record['ID']['value']:<40}{record['ClientAccessible']['value']:<18}{record['InstallDate']['value']:<27}{record['DeviceObject']['value']:<50}")
 
     @requires_admin
     def execute(self, command=None, get_output=False, use_powershell=False):
@@ -422,14 +439,23 @@ def execute(self, command=None, get_output=False, use_powershell=False):
             return ""
 
         if self.args.exec_method == "wmiexec":
-            exec_method = wmiexec.WMIEXEC(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, self.doKerberos, self.kdcHost, self.host, self.aesKey, self.logger, self.args.exec_timeout, self.args.codec)
-            output = exec_method.execute(command, get_output, use_powershell=use_powershell)
-
+            exec_method = wmiexec.WMIEXEC(
+                self.remoteName,
+                self.iWbemLevel1Login,
+                self.logger,
+                self.args.exec_timeout,
+                self.args.codec
+            )
         elif self.args.exec_method == "wmiexec-event":
-            exec_method = wmiexec_event.WMIEXEC_EVENT(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, self.doKerberos, self.kdcHost, self.host, self.aesKey, self.logger, self.args.exec_timeout, self.args.codec)
-            output = exec_method.execute(command, get_output, use_powershell=use_powershell)
+            exec_method = wmiexec_event.WMIEXEC_EVENT(
+                self.remoteName,
+                self.iWbemLevel1Login,
+                self.logger,
+                self.args.exec_timeout,
+                self.args.codec
+            )
+        output = exec_method.execute(command, get_output, use_powershell=use_powershell)
 
-        self.conn.disconnect()
         if self.args.execute and get_output:
             self.logger.success(f'Executed command: "{command}" via {self.args.exec_method}')
             buf = StringIO(output).readlines()

nxc/protocols/wmi/proto_args.py

@@ -9,9 +9,12 @@ def proto_args(parser, parents):
     dgroup.add_argument("-d", metavar="DOMAIN", dest="domain", default=None, type=str, help="Domain to authenticate to")
     dgroup.add_argument("--local-auth", action="store_true", help="Authenticate locally to each target")
 
+    cred_gathering_group = wmi_parser.add_argument_group("Credential Gathering")
+    cred_gathering_group.add_argument("--list-snapshots", nargs="?", dest="list_snapshots", const="ADMIN$", help="Lists the VSS snapshots (default: %(const)s)")
+
     egroup = wmi_parser.add_argument_group("Mapping/Enumeration")
-    egroup.add_argument("--wmi", metavar="QUERY", dest="wmi", type=str, help="Issues the specified WMI query")
-    egroup.add_argument("--wmi-namespace", metavar="NAMESPACE", type=str, default="root\\cimv2", help="WMI Namespace (default: root\\cimv2)")
+    egroup.add_argument("--wmi-query", metavar="QUERY", dest="wmi_query", type=str, help="Issues the specified WMI query")
+    egroup.add_argument("--wmi-namespace", metavar="NAMESPACE", type=str, default="root\\cimv2", help="WMI Namespace (default: %(default)s)")
 
     cgroup = wmi_parser.add_argument_group("Command Execution")
     cgroup.add_argument("--no-output", action="store_true", help="do not retrieve command output")

nxc/protocols/wmi/wmiexec.py

@@ -18,36 +18,22 @@
 import base64
 from nxc.helpers.misc import gen_random_string
 from impacket.dcerpc.v5.dtypes import NULL
-from impacket.dcerpc.v5.dcomrt import DCOMConnection
-from impacket.dcerpc.v5.dcom.wmi import CLSID_WbemLevel1Login, IID_IWbemLevel1Login, IWbemLevel1Login
 
 
 class WMIEXEC:
-    def __init__(self, target, username, password, domain, lmhash, nthash, doKerberos, kdcHost, remoteHost, aesKey, logger, exec_timeout, codec):
+    def __init__(self, target, iWbemLevel1Login, logger, exec_timeout, codec):
         self.__target = target
-        self.__username = username
-        self.__password = password
-        self.__domain = domain
-        self.__lmhash = lmhash
-        self.__nthash = nthash
-        self.__doKerberos = doKerberos
-        self.__kdcHost = kdcHost
-        self.__remoteHost = remoteHost
-        self.__aesKey = aesKey
+        self.__iWbemLevel1Login = iWbemLevel1Login
         self.logger = logger
         self.__exec_timeout = exec_timeout
         self.__registry_Path = ""
         self.__outputBuffer = ""
-
         self.__shell = "cmd.exe /Q /c "
         self.__pwd = "C:\\"
         self.__codec = codec
 
-        self.__dcom = DCOMConnection(self.__target, self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash, oxidResolver=True, doKerberos=self.__doKerberos, kdcHost=self.__kdcHost, aesKey=self.__aesKey, remoteHost=self.__remoteHost)
-        iInterface = self.__dcom.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
-        iWbemLevel1Login = IWbemLevel1Login(iInterface)
-        self.__iWbemServices = iWbemLevel1Login.NTLMLogin("//./root/cimv2", NULL, NULL)
-        iWbemLevel1Login.RemRelease()
+        self.__iWbemServices = self.__iWbemLevel1Login.NTLMLogin("//./root/cimv2", NULL, NULL)
+        self.__iWbemLevel1Login.RemRelease()
         self.__win32Process, _ = self.__iWbemServices.GetObject("Win32_Process")
 
     def execute(self, command, output=False, use_powershell=False):
@@ -65,8 +51,6 @@ def execute(self, command, output=False, use_powershell=False):
             command = self.__shell + command
             self.execute_remote(command)
 
-        self.__dcom.disconnect()
-
         return self.__outputBuffer
 
     def execute_remote(self, command):