Merge pull request Pennyw0rth#824 from MaxToffy/reg-sessions

Add --reg-sessions option to SMB protocol

Author: NeffIsBack

nxc/protocols/smb.py

@@ -63,7 +63,7 @@
 from dploot.lib.target import Target
 from dploot.triage.sccm import SCCMTriage, SCCMCred, SCCMSecret, SCCMCollection
 
-from time import time, ctime
+from time import time, ctime, sleep
 from traceback import format_exc
 from termcolor import colored
 import contextlib
@@ -1149,6 +1149,147 @@ def format_row(procInfo):
         except SessionError:
             self.logger.fail("Cannot list remote tasks, RDP is probably disabled.")
 
+    def reg_sessions(self):
+
+        def output(sessions):
+            if sessions:
+                # Calculate max lengths for formatting
+                maxSidLen = max(len(key) + 1 for key in sessions)
+                maxSidLen = max(maxSidLen, len("SID") + 1)
+                maxUsernameLen = max(len(vals["Username"] + vals["Domain"]) + 1 for vals in sessions.values()) + 1
+                maxUsernameLen = max(maxUsernameLen, len("USERNAME") + 1)
+
+                # Create the template for formatting
+                template = (f"{{USERNAME: <{maxUsernameLen}}} {{SID: <{maxSidLen}}}")
+
+                # Create headers
+                header = template.format(USERNAME="USERNAME", SID="SID")
+                header2 = template.replace(" <", "=<").format(USERNAME="", SID="")
+
+                # Store result
+                result = [header, header2]
+
+                for sid, vals in sessions.items():
+                    username = vals["Username"]
+                    domain = vals["Domain"]
+                    user_full = f"{domain}\\{username}" if username else ""
+
+                    row = template.format(USERNAME=user_full, SID=sid)
+                    result.append(row)
+
+                self.logger.success("Remote Registry enumerated sessions")
+                for row in result:
+                    self.logger.highlight(row)
+            else:
+                self.logger.info(f"No active session found for specified user(s) using the Remote Registry service on {self.hostname}.")
+
+        # Bind to the Remote Registry Pipe
+        rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\winreg", smb_connection=self.conn)
+        for binding_attempts in range(2, 0, -1):
+            dce = rpctransport.get_dce_rpc()
+            try:
+                dce.connect()
+                dce.bind(rrp.MSRPC_UUID_RRP)
+                break
+            except SessionError as e:
+                self.logger.debug(f"Could not bind to the Remote Registry on {self.hostname}: {e}")
+                if binding_attempts == 1:   # Last attempt
+                    self.logger.info(f"The Remote Registry service seems to be disabled on {self.hostname}.")
+                    return
+            # STATUS_PIPE_NOT_AVAILABLE : Waiting 1 second for the service to start (if idle and set to 'Automatic' startup type)
+            sleep(1)
+
+        # Open HKU hive
+        try:
+            resp = rrp.hOpenUsers(dce)
+        except DCERPCException as e:
+            if "rpc_s_access_denied" in str(e).lower():
+                self.logger.info(f"Access denied while enumerating session using the Remote Registry on {self.hostname}.")
+                return
+            else:
+                self.logger.fail(f"Exception connecting to RPC on {self.hostname}: {e}")
+        except Exception as e:
+            self.logger.fail(f"Exception connecting to RPC on {self.hostname}: {e}")
+
+        # Enumerate HKU subkeys and recover SIDs
+        sid_filter = "^S-1-.*\\d$"
+        exclude_sid = ["S-1-5-18", "S-1-5-19", "S-1-5-20"]
+
+        key_handle = resp["phKey"]
+        index = 1
+        sessions = {}
+
+        while True:
+            try:
+                resp = rrp.hBaseRegEnumKey(dce, key_handle, index)
+                sid = resp["lpNameOut"].rstrip("\0")
+                if re.match(sid_filter, sid) and sid not in exclude_sid:
+                    self.logger.info(f"User with SID {sid} is logged in on {self.hostname}")
+                    sessions.setdefault(sid, {"Username": "", "Domain": ""})
+                index += 1
+            except rrp.DCERPCSessionError as e:
+                if "ERROR_NO_MORE_ITEMS" in str(e):
+                    self.logger.debug(f"No more items found in HKU on {self.hostname}.")
+                    break
+                else:
+                    self.logger.fail(f"Error enumerating HKU subkeys on {self.hostname}: {e}")
+                    break
+
+        rrp.hBaseRegCloseKey(dce, key_handle)
+        dce.disconnect()
+
+        if not sessions:
+            self.logger.info(f"No sessions found via the Remote Registry service on {self.hostname}.")
+            return
+
+        # Bind to the LSARPC Pipe for SID resolution
+        rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\lsarpc", smb_connection=self.conn)
+        dce = rpctransport.get_dce_rpc()
+        try:
+            dce.connect()
+            dce.bind(lsat.MSRPC_UUID_LSAT)
+        except Exception as e:
+            self.logger.debug(f"Failed to connect to LSARPC for SID resolution on {self.hostname}: {e}")
+            output(sessions)
+            return
+
+        # Resolve SIDs with names
+        policy_handle = lsad.hLsarOpenPolicy2(dce, MAXIMUM_ALLOWED | lsat.POLICY_LOOKUP_NAMES)["PolicyHandle"]
+        try:
+            resp = lsat.hLsarLookupSids(dce, policy_handle, sessions.keys(), lsat.LSAP_LOOKUP_LEVEL.LsapLookupWksta)
+        except DCERPCException as e:
+            if str(e).find("STATUS_SOME_NOT_MAPPED") >= 0:
+                resp = e.get_packet()
+                self.logger.debug(f"Could not resolve some SIDs: {e}")
+            else:
+                resp = None
+                self.logger.debug(f"Could not resolve SID(s): {e}")
+
+        if resp:
+            for sid, item in zip(sessions.keys(), resp["TranslatedNames"]["Names"], strict=False):
+                if item["DomainIndex"] >= 0:
+                    sessions[sid]["Username"] = item["Name"]
+                    sessions[sid]["Domain"] = resp["ReferencedDomains"]["Domains"][item["DomainIndex"]]["Name"]
+
+        # Filter for usernames
+        if self.args.reg_sessions:
+            arg = self.args.reg_sessions
+            if os.path.isfile(arg):
+                with open(arg) as f:
+                    usernames = [line.strip().lower() for line in f if line.strip()]
+            else:
+                usernames = [arg.lower()]
+
+            filtered_sessions = {}
+            for sid, info in sessions.items():
+                if info["Username"].lower() not in usernames:
+                    continue
+                else:
+                    filtered_sessions[sid] = info
+            output(filtered_sessions)
+        else:
+            output(sessions)
+
     def shares(self):
         temp_dir = ntpath.normpath("\\" + gen_random_string())
         temp_file = ntpath.normpath("\\" + gen_random_string() + ".txt")
@@ -1370,7 +1511,7 @@ def get_dc_ips(self):
         return dc_ips
 
     def smb_sessions(self):
-        self.logger.fail("[REMOVED] Use option --qwinsta or --loggedon-users")
+        self.logger.fail("[REMOVED] Use option --reg-sessions --qwinsta or --loggedon-users")
         return
 
     def disks(self):

nxc/protocols/smb/proto_args.py

@@ -42,17 +42,18 @@ def proto_args(parser, parents):
     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("--smb-sessions", action="store_true", help="Enumerate active smb sessions")
     mapping_enum_group.add_argument("--disks", action="store_true", help="Enumerate disks")
-    mapping_enum_group.add_argument("--loggedon-users-filter", action="store", help="only search for specific user, works with regex")
-    mapping_enum_group.add_argument("--loggedon-users", nargs="?", const="", help="Enumerate logged on users, if a user is specified than a regex filter is applied.")
     mapping_enum_group.add_argument("--users", nargs="*", metavar="USER", help="Enumerate domain users, if a user is specified than only its information is queried.")
     mapping_enum_group.add_argument("--users-export", help="Enumerate domain users and export them to the specified file")
     mapping_enum_group.add_argument("--groups", nargs="?", const="", metavar="GROUP", help="Enumerate domain groups, if a group is specified than its members are Enumerated")
-    mapping_enum_group.add_argument("--computers", nargs="?", const="", metavar="COMPUTER", help="Enumerate computer users")
     mapping_enum_group.add_argument("--local-groups", nargs="?", const="", metavar="GROUP", help="Enumerate local groups, if a group is specified then its members are Enumerated")
+    mapping_enum_group.add_argument("--computers", nargs="?", const="", metavar="COMPUTER", help="Enumerate computer users")
     mapping_enum_group.add_argument("--pass-pol", action="store_true", help="dump password policy")
     mapping_enum_group.add_argument("--rid-brute", nargs="?", type=int, const=4000, metavar="MAX_RID", help="Enumerate users by bruteforcing RIDs")
+    mapping_enum_group.add_argument("--smb-sessions", action="store_true", help="Enumerate active smb sessions")
+    mapping_enum_group.add_argument("--reg-sessions", type=str, nargs="?", const="", help="Enumerate users sessions using the Remote Registry. If a username is given, filter for it. If a file is given, filter for listed usernames. If no value is given, list all.")
+    mapping_enum_group.add_argument("--loggedon-users", nargs="?", const="", help="Enumerate logged on users, if a user is specified than a regex filter is applied.")
+    mapping_enum_group.add_argument("--loggedon-users-filter", action="store", help="only search for specific user, works with regex")
     mapping_enum_group.add_argument("--qwinsta", type=str, nargs="?", const="", help="Enumerate user sessions. If a username is given, filter for it; if a file is given, filter for listed usernames. If no value is given, list all.")
     mapping_enum_group.add_argument("--tasklist", type=str, nargs="?", const=True, help="Enumerate running processes and filter for the specified one if specified")
     mapping_enum_group.add_argument("--taskkill", type=str, help="Kills a specific PID or a proces name's PID's")