Merge pull request Pennyw0rth#783 from Dfte/qwinsta_grep

Allows --qwinsta to filter for a specific user

Author: NeffIsBack

nxc/protocols/smb.py

@@ -688,7 +688,7 @@ def is_host_dc(self):
         from impacket.dcerpc.v5 import nrpc, epm
 
         self.logger.debug("Performing authentication attempts...")
-        
+
         # First check if port 135 is open
         if self._is_port_open(135):
             self.logger.debug("Port 135 is open, attempting MSRPC connection...")
@@ -964,7 +964,7 @@ def enumerate_sessions_info(self, sessions):
                     sessions[SessionId]["DisconnectTime"] = sessdata["LSMSessionInfoExPtr"]["LSM_SessionInfo_Level1"]["DisconnectTime"]
                     sessions[SessionId]["LogonTime"] = sessdata["LSMSessionInfoExPtr"]["LSM_SessionInfo_Level1"]["LogonTime"]
                     sessions[SessionId]["LastInputTime"] = sessdata["LSMSessionInfoExPtr"]["LSM_SessionInfo_Level1"]["LastInputTime"]
-            
+
             try:
                 with TSTS.RCMPublic(self.conn, self.host, self.kerberos) as rcm:
                     for SessionId in sessions:
@@ -1011,33 +1011,32 @@ def qwinsta(self):
             "WTS_SESSIONSTATE_LOCK": "Locked",
             "WTS_SESSIONSTATE_UNLOCK": "Unlocked",
         }
+
         sessions = self.get_session_list()
-        if not len(sessions):
+        if not sessions:
             return
+
         self.enumerate_sessions_info(sessions)
 
+        # Calculate max lengths for formatting
         maxSessionNameLen = max(len(sessions[i]["SessionName"]) + 1 for i in sessions)
-        maxSessionNameLen = maxSessionNameLen if len("SESSIONNAME") < maxSessionNameLen else len("SESSIONNAME") + 1
+        maxSessionNameLen = max(maxSessionNameLen, len("SESSIONNAME") + 1)
         maxUsernameLen = max(len(sessions[i]["Username"] + sessions[i]["Domain"]) + 1 for i in sessions) + 1
-        maxUsernameLen = maxUsernameLen if len("Username") < maxUsernameLen else len("Username") + 1
+        maxUsernameLen = max(maxUsernameLen, len("USERNAME") + 1)
         maxIdLen = max(len(str(i)) for i in sessions)
-        maxIdLen = maxIdLen if len("ID") < maxIdLen else len("ID") + 1
+        maxIdLen = max(maxIdLen, len("ID") + 1)
         maxStateLen = max(len(sessions[i]["state"]) + 1 for i in sessions)
-        maxStateLen = maxStateLen if len("STATE") < maxStateLen else len("STATE") + 1
-        maxRemoteIp = max(len(sessions[i]["RemoteIp"]) + 1 for i in sessions)
-        maxRemoteIp = maxRemoteIp if len("RemoteAddress") < maxRemoteIp else len("RemoteAddress") + 1
-        maxClientName = max(len(sessions[i]["ClientName"]) + 1 for i in sessions)
-        maxClientName = maxClientName if len("ClientName") < maxClientName else len("ClientName") + 1
-        template = ("{SESSIONNAME: <%d} "  # noqa: UP031
-                    "{USERNAME: <%d} "
-                    "{ID: <%d} "
+        maxStateLen = max(maxStateLen, len("STATE") + 1)
+
+        # Create the template for formatting
+        template = (f"{{SESSIONNAME: <{maxSessionNameLen}}} "
+                    f"{{USERNAME: <{maxUsernameLen}}} "
+                    f"{{ID: <{maxIdLen}}} "
                     "{IPv4: <16} "
-                    "{STATE: <%d} "
+                    f"{{STATE: <{maxStateLen}}} "
                     "{DSTATE: <9} "
                     "{CONNTIME: <20} "
-                    "{DISCTIME: <20} ") % (maxSessionNameLen, maxUsernameLen, maxIdLen, maxStateLen)
-
-        result = []
+                    "{DISCTIME: <20} ")
         header = template.format(
             SESSIONNAME="SESSIONNAME",
             USERNAME="USERNAME",
@@ -1048,7 +1047,6 @@ def qwinsta(self):
             CONNTIME="ConnectTime",
             DISCTIME="DisconnectTime",
         )
-
         header2 = template.replace(" <", "=<").format(
             SESSIONNAME="",
             USERNAME="",
@@ -1059,30 +1057,49 @@ def qwinsta(self):
             CONNTIME="",
             DISCTIME="",
         )
-        result.extend((header, header2))
+        result = [header, header2]
+
+        # Check if we need to filter for usernames
+        usernames = None
+        if self.args.qwinsta:
+            arg = self.args.qwinsta
+            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()]
 
         for i in sessions:
+            username = sessions[i]["Username"]
+            domain = sessions[i]["Domain"]
+            user_full = f"{domain}\\{username}" if username else ""
+
+            # If usernames are provided, filter them
+            if usernames and username.lower() not in usernames:
+                continue
+
             connectTime = sessions[i]["ConnectTime"]
             connectTime = connectTime.strftime(r"%Y/%m/%d %H:%M:%S") if connectTime.year > 1601 else "None"
 
             disconnectTime = sessions[i]["DisconnectTime"]
             disconnectTime = disconnectTime.strftime(r"%Y/%m/%d %H:%M:%S") if disconnectTime.year > 1601 else "None"
-            userName = sessions[i]["Domain"] + "\\" + sessions[i]["Username"] if len(sessions[i]["Username"]) else ""
 
-            result.append(template.format(
+            row = template.format(
                 SESSIONNAME=sessions[i]["SessionName"],
-                USERNAME=userName,
+                USERNAME=user_full,
                 ID=i,
                 IPv4=sessions[i]["RemoteIp"],
                 STATE=sessions[i]["state"],
                 DSTATE=desktop_states[sessions[i]["flags"]],
                 CONNTIME=connectTime,
                 DISCTIME=disconnectTime,
-            ))
+            )
+            result.append(row)
 
-        self.logger.success("Enumerated qwinsta sessions")
-        for row in result:
-            self.logger.highlight(row)
+        if len(result) > 2:
+            self.logger.success("Enumerated qwinsta sessions")
+            for row in result:
+                self.logger.highlight(row)
 
     @requires_admin
     def tasklist(self):

nxc/protocols/smb/proto_args.py

@@ -53,7 +53,7 @@ def proto_args(parser, parents):
     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("--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("--qwinsta", action="store_true", help="Enumerate RDP connections")
+    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")