Merge branch 'Pennyw0rth:main' into dumpSecurityQuestionsModule

Adamkadaban · web-flow · commit 21b1cfbb8c6f · 2024-06-30T17:49:55.000-07:00
diff --git a/nxc/modules/bitlocker.py b/nxc/modules/bitlocker.py
@@ -0,0 +1,131 @@
+import re
+from impacket.dcerpc.v5.dcom import wmi
+from impacket.dcerpc.v5.dtypes import NULL
+from impacket.dcerpc.v5.dcomrt import DCOMConnection
+from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_PKT_PRIVACY
+
+class NXCModule:
+    name = "bitlocker"
+    description = "Enumerating BitLocker Status on target(s) If it is enabled or disabled."
+    supported_protocols = ["smb", "wmi"]
+    opsec_safe = True
+    multiple_hosts = True
+
+    def __init__(self, context=None, module_options=None):
+        self.context = context
+        self.module_options = module_options
+
+    def options(self, context, module_options):
+        """ 
+        USAGE:
+        
+        NetExec smb <IP> -u <username> -p <password> -M bitlocker
+        NetExec wmi <IP> -u <username> -p <password> -M bitlocker (Better option to use on real life.)
+        """
+
+    def on_admin_login(self, context, connection):
+        if context.protocol == "smb":
+            bitlocker_smb = BitLockerSMB(context, connection)
+            bitlocker_smb.check_bitlocker_status()
+        elif context.protocol == "wmi":
+            bitlocker_wmi = BitLockerWMI(context, connection)
+            bitlocker_wmi.check_bitlocker_status()
+
+
+class BitLockerSMB:
+    def __init__(self, context, connection):
+        self.context = context
+        self.connection = connection
+
+    def check_bitlocker_status(self):
+        # PowerShell command to check BitLocker volumes status.
+        check_bitlocker_command_str = 'powershell.exe "Get-BitLockerVolume | Select-Object MountPoint, EncryptionMethod, ProtectionStatus"'
+
+        try:
+            # Executing the PowerShell command to get BitLocker volumes status.
+            check_bitlocker_command_str_output = self.connection.execute(check_bitlocker_command_str, True)
+            
+            if "'Get-BitLockerVolume' is not recognized" in check_bitlocker_command_str_output:
+                self.context.log.fail("BitLockerVolume not found on target.")
+                return
+
+            # Splitting the output into lines.
+            lines = str(check_bitlocker_command_str_output).splitlines()
+            data_lines = [line for line in lines if re.match(r"\w:", line)]
+            
+            for line in data_lines:
+                # Checking every line for starting with drive
+                if line[1] == ":": 
+                    parts = line.split()
+                    MountPoint, EncryptionMethod, protection_status = parts[0], parts[1], parts[2]
+
+                    # Checking if BitLocker is enabled.
+                    if protection_status == "On":
+                        self.context.log.highlight(f"BitLocker is enabled on drive {MountPoint} (Encryption Method: {EncryptionMethod})")
+                    else:
+                        self.context.log.highlight(f"BitLocker is disabled on drive {MountPoint}")
+        except Exception as e:
+            self.context.log.exception(f"Exception occurred: {e}")
+
+
+class BitLockerWMI:
+    def __init__(self, context, connection):
+        self.context = context
+        self.connection = connection
+
+    def check_bitlocker_status(self):
+        try:
+            # Create a DCOM connection
+            dcom_conn = DCOMConnection(
+                self.connection.host,
+                self.connection.username,
+                self.connection.password,
+                self.connection.domain,
+                self.connection.lmhash,
+                self.connection.nthash,
+                oxidResolver=True,
+                doKerberos=self.connection.kerberos,
+                kdcHost=self.connection.kdcHost)
+                
+            try:
+                # CoCreateInstanceEx for WMI login
+                i_interface = dcom_conn.CoCreateInstanceEx(wmi.CLSID_WbemLevel1Login, wmi.IID_IWbemLevel1Login)
+                iWbemLevel1Login = wmi.IWbemLevel1Login(i_interface)
+
+                # Specify the namespace for BitLocker
+                bitlockerNamespace = "root\\CIMv2\\Security\\MicrosoftVolumeEncryption"
+                
+                # NTLM login for WMI
+                iWbemServices = iWbemLevel1Login.NTLMLogin(bitlockerNamespace, NULL, NULL)
+
+                # Set authentication level
+                iWbemServices.get_dce_rpc().set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
+
+                # Query to get BitLocker status
+                classQuery = "SELECT DriveLetter, ProtectionStatus, EncryptionMethod FROM Win32_EncryptableVolume"
+                iEnumWbemClassObject = iWbemServices.ExecQuery(classQuery)
+                encryptionTypeMapping = {0: "None", 1: "AES_256_WITH_DIFFUSER", 2: "AES_256_WITH_DIFFUSER", 3: "AES_128", 4: "AES_256", 5: "HARDWARE_ENCRYPTION", 6: "XTS_AES_128", 7: "XTS_AES_256"}
+                
+                try:
+                    while True:
+                        iWbemClassObject = iEnumWbemClassObject.Next(0xffffffff, 1)
+                        encryptionMethod = int(iWbemClassObject[0].EncryptionMethod)
+                        if iWbemClassObject[0].ProtectionStatus == 1:
+                            self.context.log.highlight(f"BitLocker is enabled on drive {iWbemClassObject[0].DriveLetter} (Encryption Method: {encryptionTypeMapping.get(encryptionMethod, 'Unknown')})")
+                        else:
+                            if encryptionMethod == 0:  # Should be 0 if disabled
+                                self.context.log.highlight(f"BitLocker is disabled on drive {iWbemClassObject[0].DriveLetter}")
+                except Exception:
+                    pass  # Using pass because if try to log or printing, getting "WMI Session Error: code: 0x1 - WBEM_S_FALSE"
+
+                # Release resources
+                iWbemLevel1Login.RemRelease()
+                iWbemServices.RemRelease()
+                dcom_conn.disconnect()
+            except Exception as e:
+                if "WBEM_E_INVALID_NAMESPACE" in str(e):
+                    self.context.log.fail("BitLockerNamespace not found on target.")
+                    dcom_conn.disconnect()
+        except Exception as e:
+            self.context.log.error(f"Error occurred during BitLocker check: {e}")
+            dcom_conn.disconnect()
diff --git a/nxc/modules/pso.py b/nxc/modules/pso.py
@@ -53,7 +53,7 @@ def on_login(self, context, connection):
         for attrs in resp:
             if not isinstance(attrs, ldapasn1_impacket.SearchResultEntry):
                 continue
-            policyName, description, passwordLength, passwordhistorylength, lockoutThreshold, obersationWindow, lockoutDuration, complexity, minPassAge, maxPassAge, reverseibleEncryption, precedence, policyApplies = ("",) * 13
+            policyName, description, passwordLength, passwordhistorylength, lockoutThreshold, observationWindow, lockoutDuration, complexity, minPassAge, maxPassAge, reverseibleEncryption, precedence, policyApplies = ("",) * 13
             for attr in attrs["attributes"]:
                 if str(attr["type"]) == "name":
                     policyName = attr["vals"][0]
diff --git a/nxc/protocols/ldap/proto_args.py b/nxc/protocols/ldap/proto_args.py
@@ -11,7 +11,7 @@ def proto_args(parser, parents):
     dgroup.add_argument("-d", metavar="DOMAIN", dest="domain", type=str, default=None, help="domain to authenticate to")
     dgroup.add_argument("--local-auth", action="store_true", help="authenticate locally to each target")
 
-    egroup = ldap_parser.add_argument_group("Retrevie hash on the remote DC", "Options to get hashes from Kerberos")
+    egroup = ldap_parser.add_argument_group("Retrieve hash on the remote DC", "Options to get hashes from Kerberos")
     egroup.add_argument("--asreproast", help="Output AS_REP response to crack with hashcat to file")
     egroup.add_argument("--kerberoasting", help="Output TGS ticket to crack with hashcat to file")
 
@@ -26,7 +26,7 @@ def proto_args(parser, parents):
     vgroup.add_argument("--get-sid", action="store_true", help="Get domain sid")
     vgroup.add_argument("--active-users", nargs="*", help="Get Active Domain Users Accounts")
 
-    ggroup = ldap_parser.add_argument_group("Retrevie gmsa on the remote DC", "Options to play with gmsa")
+    ggroup = ldap_parser.add_argument_group("Retrieve gmsa on the remote DC", "Options to play with gmsa")
     ggroup.add_argument("--gmsa", action="store_true", help="Enumerate GMSA passwords")
     ggroup.add_argument("--gmsa-convert-id", help="Get the secret name of specific gmsa or all gmsa if no gmsa provided")
     ggroup.add_argument("--gmsa-decrypt-lsa", help="Decrypt the gmsa encrypted value from LSA")
diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py
@@ -14,7 +14,7 @@
     NTDSHashes,
 )
 from impacket.nmb import NetBIOSError, NetBIOSTimeout
-from impacket.dcerpc.v5 import transport, lsat, lsad, scmr
+from impacket.dcerpc.v5 import transport, lsat, lsad, scmr, rrp
 from impacket.dcerpc.v5.rpcrt import DCERPCException
 from impacket.dcerpc.v5.transport import DCERPCTransportFactory, SMBTransport
 from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
@@ -273,7 +273,7 @@ def enum_host_info(self):
             self.conn.logoff()
         except Exception as e:
             self.logger.debug(f"Error logging off system: {e}")
-        
+
         # DCOM connection with kerberos needed
         self.remoteName = self.host if not self.kerberos else f"{self.hostname}.{self.domain}"
 
@@ -572,6 +572,9 @@ def check_if_admin(self):
                 self.admin_privs = True
             except scmr.DCERPCException:
                 self.admin_privs = False
+            except Exception as e:
+                self.logger.fail(f"Error checking if user is admin on {self.host}: {e}")
+                self.admin_privs = False
 
     def gen_relay_list(self):
         if self.server_os.lower().find("windows") != -1 and self.signing is False:
@@ -707,11 +710,11 @@ def execute(self, payload=None, get_output=False, methods=None):
             except UnicodeDecodeError:
                 self.logger.debug("Decoding error detected, consider running chcp.com at the target, map the result with https://docs.python.org/3/library/codecs.html#standard-encodings")
                 output = output.decode("cp437")
-                
+
             self.logger.debug(f"Raw Output: {output}")
             output = "\n".join([ll.rstrip() for ll in output.splitlines() if ll.strip()])
             self.logger.debug(f"Cleaned Output: {output}")
-            
+
             if "This script contains malicious content" in output:
                 self.logger.fail("Command execution blocked by AMSI")
                 return None
@@ -732,24 +735,24 @@ def ps_execute(self, payload=None, get_output=False, methods=None, force_ps32=Fa
         if not payload:
             self.logger.error("No command to execute specified!")
             return None
-        
+
         response = []
         obfs = obfs if obfs else self.args.obfs
         encode = encode if encode else not self.args.no_encode
         force_ps32 = force_ps32 if force_ps32 else self.args.force_ps32
         get_output = True if not self.args.no_output else get_output
-                
+
         self.logger.debug(f"Starting ps_execute(): {payload=} {get_output=} {methods=} {force_ps32=} {obfs=} {encode=}")
         amsi_bypass = self.args.amsi_bypass[0] if self.args.amsi_bypass else None
         self.logger.debug(f"AMSI Bypass: {amsi_bypass}")
-        
+
         if os.path.isfile(payload):
             self.logger.debug(f"File payload set: {payload}")
             with open(payload) as commands:
                 response = [self.execute(create_ps_command(c.strip(), force_ps32=force_ps32, obfs=obfs, custom_amsi=amsi_bypass, encode=encode), get_output, methods) for c in commands]
         else:
             response = [self.execute(create_ps_command(payload, force_ps32=force_ps32, obfs=obfs, custom_amsi=amsi_bypass, encode=encode), get_output, methods)]
-            
+
         self.logger.debug(f"ps_execute response: {response}")
         return response
 
@@ -834,6 +837,74 @@ def shares(self):
             self.logger.highlight(f"{name:<15} {','.join(perms):<15} {remark}")
         return permissions
 
+    def interfaces(self):
+        """
+        Retrieve the list of network interfaces info (Name, IP Address, Subnet Mask, Default Gateway) from remote Windows registry'
+        Made by: @Sant0rryu, @NeffIsBack
+        """
+        try:
+            remoteOps = RemoteOperations(self.conn, False)
+            remoteOps.enableRegistry()
+
+            if remoteOps._RemoteOperations__rrp:
+                reg_handle = rrp.hOpenLocalMachine(remoteOps._RemoteOperations__rrp)["phKey"]
+                key_handle = rrp.hBaseRegOpenKey(remoteOps._RemoteOperations__rrp, reg_handle, "SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters\\Interfaces")["phkResult"]
+                sub_key_list = rrp.hBaseRegQueryInfoKey(remoteOps._RemoteOperations__rrp, key_handle)["lpcSubKeys"]
+                sub_keys = [rrp.hBaseRegEnumKey(remoteOps._RemoteOperations__rrp, key_handle, i)["lpNameOut"][:-1] for i in range(sub_key_list)]
+
+                self.logger.highlight(f"{'-Name-':<11} | {'-IP Address-':<15} | {'-SubnetMask-':<15} | {'-Gateway-':<15} | -DHCP-")
+                for sub_key in sub_keys:
+                    interface = {}
+                    try:
+                        interface_key = f"SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters\\Interfaces\\{sub_key}"
+                        interface_handle = rrp.hBaseRegOpenKey(remoteOps._RemoteOperations__rrp, reg_handle, interface_key)["phkResult"]
+
+                        # Retrieve Interace Name
+                        interface_name_key = f"SYSTEM\\ControlSet001\\Control\\Network\\{{4D36E972-E325-11CE-BFC1-08002BE10318}}\\{sub_key}\\Connection"
+                        interface_name_handle = rrp.hBaseRegOpenKey(remoteOps._RemoteOperations__rrp, reg_handle, interface_name_key)["phkResult"]
+                        interface_name = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, interface_name_handle, "Name")[1].rstrip("\x00")
+                        interface["Name"] = str(interface_name)
+                        if "Kernel" in interface_name:
+                            continue
+
+                        # Retrieve DHCP
+                        try:
+                            dhcp_enabled = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, interface_handle, "EnableDHCP")[1]
+                        except DCERPCException:
+                            dhcp_enabled = False
+                        interface["DHCP"] = bool(dhcp_enabled)
+
+                        # Retrieve IPAddress
+                        try:
+                            ip_address = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, interface_handle, "DhcpIPAddress" if dhcp_enabled else "IPAddress")[1].rstrip("\x00").replace("\x00", ", ")
+                        except DCERPCException:
+                            ip_address = None
+                        interface["IPAddress"] = ip_address if ip_address else None
+
+                        # Retrieve SubnetMask
+                        try:
+                            subnetmask = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, interface_handle, "SubnetMask")[1].rstrip("\x00").replace("\x00", ", ")
+                        except DCERPCException:
+                            subnetmask = None
+                        interface["SubnetMask"] = subnetmask if subnetmask else None
+
+                        # Retrieve DefaultGateway
+                        try:
+                            default_gateway = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, interface_handle, "DhcpDefaultGateway")[1].rstrip("\x00").replace("\x00", ", ")
+                        except DCERPCException:
+                            default_gateway = None
+                        interface["DefaultGateway"] = default_gateway if default_gateway else None
+
+                        self.logger.highlight(f"{interface['Name']:<11} | {interface['IPAddress']!s:<15} | {interface['SubnetMask']!s:<15} | {interface['DefaultGateway']!s:<15} | {interface['DHCP']}")
+
+                    except DCERPCException as e:
+                        self.logger.info(f"Failed to retrieve the network interface info for {sub_key}: {e!s}")
+
+            with contextlib.suppress(Exception):
+                remoteOps.finish()
+        except DCERPCException as e:
+            self.logger.error(f"Failed to connect to the target: {e!s}")
+
     def get_dc_ips(self):
         dc_ips = [dc[1] for dc in self.db.get_domain_controllers(domain=self.domain)]
         if not dc_ips:
@@ -1302,7 +1373,7 @@ def put_file_single(self, src, dst):
                 self.logger.success(f"Created file {src} on \\\\{self.args.share}\\{dst}")
             except Exception as e:
                 self.logger.fail(f"Error writing file to share {self.args.share}: {e}")
-    
+
     def put_file(self):
         for src, dest in self.args.put_file:
             self.put_file_single(src, dest)
@@ -1325,7 +1396,6 @@ def get_file(self):
         for src, dest in self.args.get_file:
             self.get_file_single(src, dest)
 
-
     def enable_remoteops(self):
         try:
             self.remote_ops = RemoteOperations(self.conn, self.kerberos, self.kdcHost)
@@ -1408,7 +1478,7 @@ def sccm(self):
         except Exception as e:
             self.logger.debug(f"Could not upgrade connection: {e}")
             return
-        
+
         try:
             self.logger.display("Collecting Machine masterkeys, grab a coffee and be patient...")
             masterkeys_triage = MasterkeysTriage(
@@ -1423,7 +1493,7 @@ def sccm(self):
         if len(masterkeys) == 0:
             self.logger.fail("No masterkeys looted")
             return
-        
+
         self.logger.success(f"Got {highlight(len(masterkeys))} decrypted masterkeys. Looting SCCM Credentials through {self.args.sccm}")
         try:
             # Collect Chrome Based Browser stored secrets
@@ -1613,7 +1683,6 @@ def dpapi(self):
                     "Google Refresh Token",
                 )
 
-
         if dump_cookies and cookies:
             self.logger.display("Start Dumping Cookies")
             for cookie in cookies:
@@ -1801,4 +1870,4 @@ def add_ntds_hash(ntds_hash, host_id):
         NTDS.finish()
 
     def mark_guest(self):
-        return highlight(f"{highlight('(Guest)')}" if self.is_guest else "")
+        return highlight(f"{highlight('(Guest)')}" if self.is_guest else "")
diff --git a/nxc/protocols/smb/proto_args.py b/nxc/protocols/smb/proto_args.py
@@ -34,6 +34,7 @@ def proto_args(parser, parents):
 
     mapping_enum_group = smb_parser.add_argument_group("Mapping/Enumeration", "Options for Mapping/Enumerating")
     mapping_enum_group.add_argument("--shares", action="store_true", help="enumerate shares and access")
+    mapping_enum_group.add_argument("--interfaces", action="store_true", help="enumerate network interfaces")
     mapping_enum_group.add_argument("--no-write-check", action="store_true", help="Skip write check on shares (avoid leaving traces when missing delete permissions)")
     mapping_enum_group.add_argument("--filter-shares", nargs="+", help="Filter share by access, option 'read' 'write' or 'read,write'")
     mapping_enum_group.add_argument("--sessions", action="store_true", help="enumerate active sessions")
diff --git a/tests/e2e_commands.txt b/tests/e2e_commands.txt
