From cfc91231425ad7f87a9d8f83ed85af3cbff66f72 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aur=C3=A9lien=20CHALOT?=
 <aurelien.chalot@orangecyberdefense.com>
Date: Tue, 2 Jun 2026 11:47:37 +0200
Subject: [PATCH] Refactor the --database argument to comply with security
 checks

---
 nxc/protocols/mssql.py | 72 +++++++++++++++++++++---------------------
 1 file changed, 36 insertions(+), 36 deletions(-)

diff --git a/nxc/protocols/mssql.py b/nxc/protocols/mssql.py
index b7edc248ec..a08650c675 100755
--- a/nxc/protocols/mssql.py
+++ b/nxc/protocols/mssql.py
@@ -483,27 +483,27 @@ def _qname(self, ident: str) -> str:
         return "[" + str(ident).replace("]", "]]") + "]"
 
     def list_databases(self):
-        try:
-            q = (
-                "SELECT d.name AS DatabaseName, "
-                "       suser_sname(d.owner_sid) AS Owner "
-                "FROM sys.databases d "
-                "ORDER BY d.name;"
-            )
-            rows = self.conn.sql_query(q) or []
-            if not rows:
-                self.logger.display("No databases returned")
-                return
+        query = """
+        SELECT d.name AS DatabaseName, 
+                suser_sname(d.owner_sid) AS Owner
+        FROM sys.databases d
+        ORDER BY d.name;
+        """
+        rows = self.conn.sql_query(query)
+        if self.conn.lastError:
+            self.logger.fail(f"Error running the SQL query: {self.conn.lastError}")
+        if not rows:
+            self.logger.display("No databases returned")
+            return
 
-            self.logger.display("Enumerated databases")
-            self.logger.highlight(f"{'Database Name':<30} {'Owner':<25}")
-            self.logger.highlight(f"{'-' * 30} {'-' * 25}")
-            for r in rows:
-                self.logger.highlight(f"{r.get('DatabaseName', ''):<30} {r.get('Owner', ''):<25}")
-            self.logger.highlight(f"Total: {len(rows)} database(s)")
-        except Exception as e:
-            self.logger.fail(f"Failed to enumerate databases: {e}")
-            self.logger.debug("list_databases error", exc_info=True)
+        self.logger.display("Enumerated databases")
+        self.logger.highlight(f"{'Database Name':<30} {'Owner':<25}")
+        self.logger.highlight(f"{'-' * 30} {'-' * 25}")
+        for row in rows:
+            database_name = row.get("DatabaseName", "")
+            owner = row.get("Owner", "")
+            self.logger.highlight(f"{database_name:<30} {owner:<25}")
+        self.logger.highlight(f"Total: {len(rows)} database(s)")
 
     def database(self):
         db_arg = self.args.database
@@ -515,24 +515,24 @@ def database(self):
 
         # nxc --database <name> -> tables
         if isinstance(db_arg, str):
-            try:
-                safe = db_arg.replace("'", "''")
-                exists = self.conn.sql_query(f"SELECT 1 FROM sys.databases WHERE name = N'{safe}';")
-                if not exists:
-                    self.logger.fail(f"Database [{db_arg}] does not exist on the server.")
-                    return
-
-                tq = (
-                    f"SELECT t.name AS TableName, t.modify_date "
-                    f"FROM {self._qname(db_arg)}.sys.tables t "
-                    f"ORDER BY t.name;"
-                )
-                rows = self.conn.sql_query(tq) or []
-            except Exception as e:
-                self.logger.fail(f"Insufficient permissions or query error in [{db_arg}]: {e}")
-                self.logger.debug("database() error", exc_info=True)
+            safe = db_arg.replace("'", "''")
+            exists = self.conn.sql_query(f"SELECT 1 FROM sys.databases WHERE name = N'{safe}';")
+            if self.conn.lastError:
+                self.logger.fail(f"Error running the SQL query: {self.conn.lastError}")
+            if not exists:
+                self.logger.fail(f"Database [{db_arg}] does not exist on the server.")
                 return
 
+            query = f"""
+                SELECT t.name AS TableName, t.modify_date
+                FROM {self._qname(db_arg)}.sys.tables t
+                ORDER BY t.name;
+            """
+            rows = self.conn.sql_query(query)
+
+            if self.conn.lastError:
+                self.logger.fail(f"Error running the SQL query: {self.conn.lastError}")
+
             if not rows:
                 self.logger.display(f"Database [{db_arg}] has no user tables.")
                 return