fix(mysql): align command methods with call_cmd_args dispatch (#861)

Rename methods to match argparse dest names so the base class
call_cmd_args() dispatches them correctly. Remove inline command
handling from plaintext_login() which caused dump_database to
execute twice. Fix staticmethod signatures and unused variables
in db_navigator. Add e2e test commands.

Refs #861

nxc/connection.py

@@ -174,7 +174,7 @@ class connection:
         if self.kerberos:
             self.host = self.hostname
 
-        self.logger.info(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
+        self.logger.info(f"Connection __init__ info: host={self.host}, hostname={self.hostname}, port={self.port}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
 
         try:
             self.proto_flow()

nxc/protocols/mysql.py

@@ -0,0 +1,275 @@
+import socket
+import contextlib
+from nxc.config import process_secret
+from nxc.connection import connection
+from nxc.logger import NXCAdapter
+from pymysql.connections import Connection
+from pymysql.err import OperationalError
+
+
+class mysql(connection):
+    def __init__(self, args, db, host):
+        self.protocol = "MySQL"
+        self.remote_version = None
+        self._pinged = False
+        super().__init__(args, db, host)
+
+    def proto_logger(self):
+        self.logger = NXCAdapter(
+            extra={
+                "protocol": "MySQL",
+                "host": self.host,
+                "port": self.port,
+                "hostname": self.hostname,
+            }
+        )
+
+    def create_conn_obj(self):
+        """Create the connection object, but defer the actual connection
+        Because of how proto_flow works, we do an initial custom socket to ensure there is something on the port,
+        then we can get server information with that same socket if there is something there
+        """
+        self.logger.debug(f"Creating connection object for {self.host}:{self.port}")
+
+        # during the initial connection the username and password are actually blank
+        self.conn = Connection(
+            host=self.host,
+            port=self.port,
+            user=self.username,
+            password=self.password,
+            defer_connect=True,
+        )
+        self.logger.debug(f"Connection object created for {self.host}:{self.port}")
+
+        # we don't want to do additional useless TCP handshake requests for connect(), etc, so we only "ping" it once
+        if not self._pinged:
+            self.logger.debug(f"Making initial socket connection to {self.host}:{self.port}")
+            try:
+                self.conn._sock = socket.create_connection((self.host, self.port))
+                self._pinged = True
+            except Exception as e:
+                self.logger.debug(f"Error creating socket connection: {e}")
+                return False
+        return True
+
+    def enum_host_info(self):
+        """Use existing socket but only get the server information
+        The way we do this actually sends less packets than nmap -sV, since they open a socket for ping then don't reuse it
+        We also don't want to send a QUIT message, so we use _force_close()
+        """
+        self.logger.debug("Getting server information")
+
+        try:
+            # setup the socket for the server information request
+            self.conn._rfile = self.conn._sock.makefile("rb")
+            self.conn._next_seq_id = 0
+            self.conn._get_server_information()
+            if self.conn.server_version:
+                self.remote_version = self.conn.server_version
+                self.logger.debug(f"Server information: {self.remote_version}")
+                try:
+                    self.db.add_host(self.host, self.port, self.remote_version)
+                except IndexError as e:
+                    self.logger.debug(f"Error adding host to DB: {e}, trying one more time...")
+                    self.db.add_host(self.host, self.port, self.remote_version)
+
+            with contextlib.suppress(Exception):
+                self.conn._force_close()  # close the socket without sending a QUIT message
+        except OperationalError as e:
+            if "is not allowed to connect to" in str(e):
+                self.logger.fail("Host is not allowed to connect to server")
+                exit(1)
+            else:
+                self.logger.fail(f"Error getting server information: {e}")
+
+    def print_host_info(self):
+        if self.remote_version:
+            self.logger.display(f"MySQL Version: {self.remote_version}")
+
+    def plaintext_login(self, username, password):
+        """Authenticate with MySQL server using username/password via pymysql connect."""
+        self.password = password
+        self.username = username
+        self.create_conn_obj()
+
+        try:
+            self.logger.debug(f"Attempting login with {username}:{process_secret(password)}")
+            self.conn.connect()
+            self.logger.debug(f"Authenticated to {self.host}:{self.port}")
+            self.logger.success(f"{self.username}:{process_secret(self.password)} {self.mark_pwned()}")
+
+            cred_id = self.db.add_credential(username, password)
+            host_id = self.db.get_hosts(self.host)[0].id
+            self.db.add_loggedin_relation(cred_id, host_id)
+            return True
+        except Exception as e:
+            self.logger.fail(f"{username}:{process_secret(password)} (Response:{e})")
+            with contextlib.suppress(Exception):
+                if self.conn:
+                    self.conn.close()
+            return False
+
+    def query(self):
+        """Execute a custom SQL query."""
+        sql = self.args.query
+        try:
+            cursor = self.conn.cursor()
+            self.logger.display(f"Executing query: {sql}")
+            cursor.execute(sql)
+
+            if cursor.description:
+                results = cursor.fetchall()
+                if results:
+                    self.logger.display(f"Query Results ({len(results)} rows):")
+                    for i, row in enumerate(results):
+                        self.logger.highlight(f"Row {i + 1}: {row}")
+                else:
+                    self.logger.display("Query executed successfully (no results)")
+            else:
+                affected_rows = cursor.rowcount
+                self.logger.display(f"Query executed successfully. {affected_rows} rows affected.")
+
+            cursor.close()
+        except Exception as e:
+            self.logger.fail(f"Error executing query: {e}")
+
+    def databases(self):
+        """List all databases."""
+        try:
+            cursor = self.conn.cursor()
+            try:
+                cursor.execute("SHOW DATABASES")
+            except Exception:
+                cursor.execute("SHOW DATABASES LIKE '%'")
+            databases = cursor.fetchall()
+
+            if databases:
+                self.logger.display(f"Available Databases ({len(databases)}):")
+                for db in databases:
+                    self.logger.highlight(db[0])
+            else:
+                self.logger.display("No databases found or access denied.")
+
+            cursor.close()
+        except Exception as e:
+            self.logger.fail(f"Error listing databases: {e}")
+
+    def tables(self):
+        """List tables in specified database."""
+        database = self.args.tables
+        try:
+            cursor = self.conn.cursor()
+            cursor.execute(f"USE `{database}`")
+            try:
+                cursor.execute("SHOW TABLES")
+            except Exception:
+                cursor.execute(f"SHOW TABLES FROM `{database}`")
+            results = cursor.fetchall()
+
+            if results:
+                self.logger.display(f"Tables in {database} ({len(results)}):")
+                for table in results:
+                    self.logger.highlight(table[0])
+            else:
+                self.logger.display(f"No tables found in database '{database}' or access denied.")
+
+            cursor.close()
+        except Exception as e:
+            self.logger.fail(f"Error listing tables: {e}")
+
+    def dump_database(self):
+        """Dump database structure and sample data."""
+        database = self.args.dump_database
+        try:
+            cursor = self.conn.cursor()
+            cursor.execute(f"USE `{database}`")
+            try:
+                cursor.execute("SHOW TABLES")
+            except Exception:
+                cursor.execute(f"SHOW TABLES FROM `{database}`")
+            tables = cursor.fetchall()
+
+            if not tables:
+                self.logger.display(f"No tables found in database '{database}'")
+                cursor.close()
+                return
+
+            self.logger.display(f"Database dump for {database} ({len(tables)} tables):")
+
+            for table in tables:
+                table_name = table[0]
+                self.logger.highlight(f"Table: {table_name}")
+
+                try:
+                    cursor.execute(f"DESCRIBE `{table_name}`")
+                    structure = cursor.fetchall()
+
+                    if structure:
+                        self.logger.display("  Schema:")
+                        self.logger.display(f"    {'Field':<15} | {'Type':<15} | {'Null':<4} | {'Key':<15} | {'Default':<15} | {'Extra':<15}")
+                        self.logger.display(f"    {'-' * 15} | {'-' * 15} | {'-' * 4} | {'-' * 15} | {'-' * 15} | {'-' * 15}")
+
+                        for col in structure:
+                            formatted_values = []
+                            for idx, value in enumerate(col):
+                                # third column is whether the column is nullable, so either YES or NO
+                                if idx != 2:
+                                    formatted_values.append(f"{value!s:<15}")
+                                else:
+                                    formatted_values.append(f"{value!s:<4}")
+                            self.logger.display(f"    {' | '.join(formatted_values)}")
+
+                    try:
+                        cursor.execute(f"SELECT * FROM `{table_name}` LIMIT 5")
+                        data = cursor.fetchall()
+
+                        if data:
+                            self.logger.display(f"  Data ({len(data)} rows):")
+                            for i, row in enumerate(data):
+                                self.logger.display(f"    Row {i + 1}: {row}")
+                        else:
+                            self.logger.display("  Data: No data in table")
+                    except Exception as e:
+                        self.logger.display(f"  Data: Error reading data - {e}")
+                except Exception as e:
+                    self.logger.display(f"  Error processing table: {e}")
+
+                self.logger.display("")
+
+            cursor.close()
+        except Exception as e:
+            self.logger.fail(f"Error dumping database: {e}")
+
+    def server_capabilities(self):
+        """Display MySQL server capabilities."""
+        try:
+            cursor = self.conn.cursor()
+            capabilities = {}
+            try:
+                cursor.execute("SHOW VARIABLES LIKE 'have_ssl'")
+                result = cursor.fetchone()
+                capabilities["ssl"] = result[1] if result else "Unknown"
+            except Exception:
+                capabilities["ssl"] = "Not supported"
+            try:
+                cursor.execute("SHOW PLUGINS")
+                plugins = cursor.fetchall()
+                auth_plugins = [row[1] for row in plugins if "auth" in row[1].lower()]
+                capabilities["auth_plugins"] = auth_plugins
+            except Exception:
+                capabilities["auth_plugins"] = ["Unknown"]
+            try:
+                cursor.execute("SELECT @@sql_mode")
+                result = cursor.fetchone()
+                capabilities["sql_mode"] = result[0] if result else "Unknown"
+            except Exception:
+                capabilities["sql_mode"] = "Unknown"
+            cursor.close()
+
+            self.logger.display("Server capabilities:")
+            self.logger.display(f"    SSL: {capabilities['ssl']}")
+            self.logger.display(f"    Auth plugins: {capabilities['auth_plugins']}")
+            self.logger.display(f"    SQL mode: {capabilities['sql_mode']}")
+        except Exception as e:
+            self.logger.debug(f"Error getting server capabilities: {e}")
+            return {}

nxc/protocols/mysql/__init__.py

@@ -0,0 +1 @@
+

nxc/protocols/mysql/database.py

@@ -0,0 +1,278 @@
+import sys
+
+from sqlalchemy import Table, select, delete, func
+from sqlalchemy.dialects.sqlite import Insert
+from sqlalchemy.exc import (
+    NoInspectionAvailable,
+    NoSuchTableError,
+)
+
+from nxc.database import BaseDB, format_host_query
+from nxc.logger import nxc_logger
+
+
+class database(BaseDB):
+    def __init__(self, db_engine):
+        self.CredentialsTable = None
+        self.HostsTable = None
+        self.LoggedinRelationsTable = None
+
+        super().__init__(db_engine)
+
+    @staticmethod
+    def db_schema(db_conn):
+        db_conn.execute(
+            """CREATE TABLE "credentials" (
+            "id" integer PRIMARY KEY,
+            "username" text,
+            "password" text
+            )"""
+        )
+
+        db_conn.execute(
+            """CREATE TABLE "hosts" (
+            "id" integer PRIMARY KEY,
+            "host" text,
+            "port" integer,
+            "banner" text
+            )"""
+        )
+        db_conn.execute(
+            """CREATE TABLE "loggedin_relations" (
+            "id" integer PRIMARY KEY,
+            "credid" integer,
+            "hostid" integer,
+            FOREIGN KEY(credid) REFERENCES credentials(id),
+            FOREIGN KEY(hostid) REFERENCES hosts(id)
+        )"""
+        )
+
+    def reflect_tables(self):
+        with self.db_engine.connect():
+            try:
+                self.CredentialsTable = Table("credentials", self.metadata, autoload_with=self.db_engine)
+                self.HostsTable = Table("hosts", self.metadata, autoload_with=self.db_engine)
+                self.LoggedinRelationsTable = Table("loggedin_relations", self.metadata, autoload_with=self.db_engine)
+            except (NoInspectionAvailable, NoSuchTableError):
+                print(
+                    f"""
+                    [-] Error reflecting tables for the {self.protocol} protocol - this means there is a DB schema mismatch
+                    [-] This is probably because a newer version of nxc is being run on an old DB schema
+                    [-] Optionally save the old DB data (`cp {self.db_path} ~/nxc_{self.protocol.lower()}.bak`)
+                    [-] Then remove the {self.protocol} DB (`rm -f {self.db_path}`) and run nxc to initialize the new DB"""
+                )
+                sys.exit()
+
+    def add_host(self, host, port, banner):
+        """Check if this host is already in the DB, if not add it"""
+        hosts = []
+        updated_ids = []
+
+        q = select(self.HostsTable).filter(self.HostsTable.c.host == host)
+        results = self.db_execute(q).all()
+
+        # create new host
+        if not results:
+            new_host = {
+                "host": host,
+                "port": port,
+                "banner": banner,
+            }
+            hosts = [new_host]
+        # update existing hosts data
+        else:
+            for host_result in results:
+                host_data = host_result._asdict()
+                nxc_logger.debug(f"host: {host_result}")
+                nxc_logger.debug(f"host_data: {host_data}")
+                # only update column if it is being passed in
+                if host is not None:
+                    host_data["host"] = host
+                if port is not None:
+                    host_data["port"] = port
+                if banner is not None:
+                    host_data["banner"] = banner
+                # only add host to be updated if it has changed
+                if host_data not in hosts:
+                    hosts.append(host_data)
+                    updated_ids.append(host_data["id"])
+        nxc_logger.debug(f"Hosts: {hosts}")
+
+        # TODO: find a way to abstract this away to a single Upsert call
+        q = Insert(self.HostsTable)
+        update_columns = {col.name: col for col in q.excluded if col.name not in "id"}
+        q = q.on_conflict_do_update(index_elements=self.HostsTable.primary_key, set_=update_columns)
+
+        self.db_execute(q, hosts)
+        # we only return updated IDs for now - when RETURNING clause is allowed we can return inserted
+        if updated_ids:
+            nxc_logger.debug(f"add_host() - Host IDs Updated: {updated_ids}")
+            return updated_ids
+
+    def add_credential(self, username, password):
+        """Check if this credential has already been added to the database, if not add it in."""
+        credentials = []
+
+        q = select(self.CredentialsTable).filter(func.lower(self.CredentialsTable.c.username) == func.lower(username),
+                                                 func.lower(self.CredentialsTable.c.password) == func.lower(password))
+        results = self.db_execute(q).all()
+
+        # add new credential
+        if not results:
+            new_cred = {
+                "username": username,
+                "password": password,
+            }
+            credentials = [new_cred]
+        # update existing cred data
+        else:
+            for creds in results:
+                # this will include the id, so we don't touch it
+                cred_data = creds._asdict()
+                # only update column if it is being passed in
+                if username is not None:
+                    cred_data["username"] = username
+                if password is not None:
+                    cred_data["password"] = password
+                # only add cred to be updated if it has changed
+                if cred_data not in credentials:
+                    credentials.append(cred_data)
+
+        # TODO: find a way to abstract this away to a single Upsert call
+        q_users = Insert(self.CredentialsTable)
+        update_columns_users = {col.name: col for col in q_users.excluded if col.name not in "id"}
+        q_users = q_users.on_conflict_do_update(index_elements=self.CredentialsTable.primary_key,
+                                                set_=update_columns_users)
+        nxc_logger.debug(f"Adding credentials: {credentials}")
+
+        self.db_execute(q_users, credentials)
+
+        # hacky way to get cred_id since we can't use returning() yet
+        if len(credentials) == 1:
+            return self.get_credential(username, password)
+        else:
+            return credentials
+
+    def remove_credentials(self, creds_id):
+        """Removes a credential ID from the database"""
+        del_hosts = []
+        for cred_id in creds_id:
+            q = delete(self.CredentialsTable).filter(self.CredentialsTable.c.id == cred_id)
+            del_hosts.append(q)
+        self.db_execute(q)
+
+    def is_credential_valid(self, credential_id):
+        """Check if this credential ID is valid."""
+        q = select(self.CredentialsTable).filter(
+            self.CredentialsTable.c.id == credential_id,
+            self.CredentialsTable.c.password is not None,
+        )
+        results = self.db_execute(q).all()
+        return len(results) > 0
+
+    def get_credential(self, username, password):
+        q = select(self.CredentialsTable).filter(
+            self.CredentialsTable.c.username == username,
+            self.CredentialsTable.c.password == password,
+        )
+        results = self.db_execute(q).first()
+        if results is not None:
+            return results.id
+
+    def get_credentials(self, filter_term=None):
+        """Return credentials from the database."""
+        # if we're returning a single credential by ID
+        if self.is_credential_valid(filter_term):
+            q = select(self.CredentialsTable).filter(self.CredentialsTable.c.id == filter_term)
+        # if we're filtering by username
+        elif filter_term and filter_term != "":
+            like_term = func.lower(f"%{filter_term}%")
+            q = select(self.CredentialsTable).filter(func.lower(self.CredentialsTable.c.username).like(like_term))
+        # otherwise return all credentials
+        else:
+            q = select(self.CredentialsTable)
+
+        return self.db_execute(q).all()
+
+    def is_host_valid(self, host_id):
+        """Check if this host ID is valid."""
+        q = select(self.HostsTable).filter(self.HostsTable.c.id == host_id)
+        results = self.db_execute(q).all()
+        return len(results) > 0
+
+    def get_hosts(self, filter_term=None):
+        """Return hosts from the database."""
+        q = select(self.HostsTable)
+
+        # if we're returning a single host by ID
+        if self.is_host_valid(filter_term):
+            q = q.filter(self.HostsTable.c.id == filter_term)
+            results = self.db_execute(q).first()
+            # all() returns a list, so we keep the return format the same so consumers don't have to guess
+            return [results]
+        # if we're filtering by host
+        elif filter_term and filter_term != "":
+            q = format_host_query(q, filter_term, self.HostsTable)
+
+        results = self.db_execute(q).all()
+        nxc_logger.debug(f"MySQL get_hosts() - results: {results}")
+        return results
+
+    def is_user_valid(self, cred_id):
+        """Check if this User ID is valid."""
+        q = select(self.CredentialsTable).filter(self.CredentialsTable.c.id == cred_id)
+        results = self.db_execute(q).all()
+        return len(results) > 0
+
+    def get_user(self, username):
+        q = select(self.CredentialsTable).filter(func.lower(self.CredentialsTable.c.username) == func.lower(username))
+        return self.db_execute(q).all()
+
+    def get_users(self, filter_term=None):
+        q = select(self.CredentialsTable)
+
+        if self.is_user_valid(filter_term):
+            q = q.filter(self.CredentialsTable.c.id == filter_term)
+        # if we're filtering by username
+        elif filter_term and filter_term != "":
+            like_term = func.lower(f"%{filter_term}%")
+            q = q.filter(func.lower(self.CredentialsTable.c.username).like(like_term))
+        return self.db_execute(q).all()
+
+    def add_loggedin_relation(self, cred_id, host_id):
+        relation_query = select(self.LoggedinRelationsTable).filter(
+            self.LoggedinRelationsTable.c.credid == cred_id,
+            self.LoggedinRelationsTable.c.hostid == host_id,
+        )
+        results = self.db_execute(relation_query).all()
+
+        # only add one if one doesn't already exist
+        if not results:
+            relation = {"credid": cred_id, "hostid": host_id}
+            try:
+                nxc_logger.debug(f"Inserting loggedin_relations: {relation}")
+                # TODO: find a way to abstract this away to a single Upsert call
+                q = Insert(self.LoggedinRelationsTable)
+
+                self.db_execute(q, [relation])
+                inserted_id_results = self.get_loggedin_relations(cred_id, host_id)
+                nxc_logger.debug(f"Checking if relation was added: {inserted_id_results}")
+                return inserted_id_results[0].id
+            except Exception as e:
+                nxc_logger.debug(f"Error inserting LoggedinRelation: {e}")
+
+    def get_loggedin_relations(self, cred_id=None, host_id=None):
+        q = select(self.LoggedinRelationsTable)
+        if cred_id:
+            q = q.filter(self.LoggedinRelationsTable.c.credid == cred_id)
+        if host_id:
+            q = q.filter(self.LoggedinRelationsTable.c.hostid == host_id)
+        return self.db_execute(q).all()
+
+    def remove_loggedin_relations(self, cred_id=None, host_id=None):
+        q = delete(self.LoggedinRelationsTable)
+        if cred_id:
+            q = q.filter(self.LoggedinRelationsTable.c.credid == cred_id)
+        elif host_id:
+            q = q.filter(self.LoggedinRelationsTable.c.hostid == host_id)
+        self.db_execute(q)

nxc/protocols/mysql/db_navigator.py

@@ -0,0 +1,181 @@
+from nxc.nxcdb import DatabaseNavigator, print_table, print_help
+
+
+class navigator(DatabaseNavigator):
+    def display_creds(self, creds):
+        data = [
+            [
+                "CredID",
+                "Total Logins",
+                "Username",
+                "Password",
+            ]
+        ]
+
+        for cred in creds:
+            total_users = self.db.get_loggedin_relations(cred_id=cred[0])
+            data.append(
+                [
+                    cred[0],
+                    str(len(total_users)) + " Host(s)",
+                    cred[1],
+                    cred[2],
+                ]
+            )
+        print_table(data, title="Credentials")
+
+    def display_hosts(self, hosts):
+        data = [
+            [
+                "HostID",
+                "Total Users",
+                "Host",
+                "Port",
+                "Banner",
+            ]
+        ]
+
+        for h in hosts:
+            total_users = self.db.get_loggedin_relations(host_id=h[0])
+            data.append(
+                [
+                    h[0],
+                    str(len(total_users)) + " User(s)",
+                    h[1],
+                    h[2],
+                    h[3],
+                ]
+            )
+        print_table(data, title="Hosts")
+
+    def do_hosts(self, line):
+        filter_term = line.strip()
+
+        if filter_term == "":
+            hosts = self.db.get_hosts()
+            self.display_hosts(hosts)
+        else:
+            hosts = self.db.get_hosts(filter_term=filter_term)
+
+            if len(hosts) > 1:
+                self.display_hosts(hosts)
+            elif len(hosts) == 1:
+                data = [["HostID", "Host", "Port", "Banner"]]
+                host_id_list = [h[0] for h in hosts]
+
+                data += [[h[0], h[1], h[2], h[3]] for h in hosts]
+
+                print_table(data, title="Host")
+
+                login_data = [["CredID", "UserName", "Password"]]
+                for host_id in host_id_list:
+                    login_links = self.db.get_loggedin_relations(host_id=host_id)
+
+                    for link in login_links:
+                        _, cred_id, host_id = link
+                        creds = self.db.get_credentials(filter_term=cred_id)
+                        for cred in creds:
+                            cred_data = [cred[0], cred[1], cred[2]]
+                            if cred_data not in login_data:
+                                login_data.append(cred_data)
+
+                if len(login_data) > 1:
+                    print_table(
+                        login_data,
+                        title="Credential(s) with Logins",
+                    )
+
+    @staticmethod
+    def help_hosts():
+        help_string = """
+        hosts [filter_term]
+        By default prints all hosts
+        Table format:
+        | 'HostID', 'Host', 'Port', 'Banner' |
+        """
+        print_help(help_string)
+
+    def do_creds(self, line):
+        filter_term = line.strip()
+
+        if filter_term == "":
+            creds = self.db.get_credentials()
+            self.display_creds(creds)
+        elif filter_term.split()[0].lower() == "add":
+            # add format: "username password"
+            args = filter_term.split()[1:]
+
+            if len(args) == 2:
+                username, password = args
+                self.db.add_credential(username, password)
+            else:
+                print("[!] Format is 'add username password")
+                return
+        elif filter_term.split()[0].lower() == "remove":
+            args = filter_term.split()[1:]
+            if len(args) != 1:
+                print("[!] Format is 'remove <credID>'")
+                return
+            else:
+                self.db.remove_credentials(args)
+                self.db.remove_loggedin_relations(user_ids=args)
+        else:
+            creds = self.db.get_credentials(filter_term=filter_term)
+            if len(creds) != 1:
+                self.display_creds(creds)
+            elif len(creds) == 1:
+                cred_data = [["CredID", "UserName", "Password"]]
+                cred_id_list = []
+
+                for cred in creds:
+                    cred_id = cred[0]
+                    cred_id_list.append(cred_id)
+                    username = cred[1]
+                    password = cred[2]
+
+                    cred_data.append([cred_id, username, password])
+                print_table(cred_data, title="Credential(s)")
+
+                access_data = [["HostID", "Host", "Port", "Banner"]]
+
+                for cred_id in cred_id_list:
+                    logins = self.db.get_loggedin_relations(cred_id=cred_id)
+
+                    for link in logins:
+                        _, cred_id, host_id = link
+                        hosts = self.db.get_hosts(host_id)
+                        access_data += [[h[0], h[1], h[2], h[3]] for h in hosts]
+
+                # we look if it's greater than one because the header row always exists
+                if len(access_data) > 1:
+                    print_table(access_data, title="Access to Host(s)")
+
+    def help_creds(self):
+        help_string = """
+        creds [add|remove|filter_term]
+        By default prints all creds
+        Table format:
+        | 'CredID', 'Login To', 'UserName', 'Password' |
+        Subcommands:
+            add - format: "add username password"
+            remove - format: "remove <credID>"
+            filter_term - filters creds with filter_term
+                If a single credential is returned (e.g. `creds 15`, it prints the following tables:
+                    Credential(s) | 'CredID', 'UserName', 'Password' |
+                    Access to Host(s) | 'HostID', 'Host', 'Port', 'Banner'
+                Otherwise, it prints the default credential table from a `like` query on the `username` column
+        """
+        print_help(help_string)
+
+    def do_clear_database(self, line):
+        if input("This will destroy all data in the current database, are you SURE you want to run this? (y/n): ") == "y":
+            self.db.clear_database()
+
+    @staticmethod
+    def help_clear_database():
+        help_string = """
+        clear_database
+        THIS COMPLETELY DESTROYS ALL DATA IN THE CURRENTLY CONNECTED DATABASE
+        YOU CANNOT UNDO THIS COMMAND
+        """
+        print_help(help_string)

nxc/protocols/mysql/proto_args.py

@@ -0,0 +1,14 @@
+from nxc.helpers.args import DisplayDefaultsNotNone
+
+
+def proto_args(parser, parents):
+    mysql_parser = parser.add_parser("mysql", help="own stuff using MySQL", parents=parents, formatter_class=DisplayDefaultsNotNone)
+    mysql_parser.add_argument("--port", type=int, default=3306, help="MySQL port")
+
+    cgroup = mysql_parser.add_argument_group("Database Operations", "Options for interacting with MySQL databases")
+    cgroup.add_argument("-q", "--query", metavar="SQL_QUERY", help="Execute a custom SQL query")
+    cgroup.add_argument("--databases", action="store_true", help="List all databases")
+    cgroup.add_argument("--tables", metavar="DATABASE", help="List tables in specified database")
+    cgroup.add_argument("--dump-database", "--dump", dest="dump_database", metavar="DATABASE", help="Dump database structure and data")
+    cgroup.add_argument("--server-capabilities", "--capabilities", dest="server_capabilities", action="store_true", help="Get server capabilities")
+    return parser

poetry.lock

@@ -1742,6 +1742,22 @@ files = [
     {file = "pylnk3-0.4.3.tar.gz", hash = "sha256:fbc5f512b581382c2a4c11e6df3796f9975bcfd99e3fca2afe531ea61b3c4ac2"},
 ]
 
+[[package]]
+name = "pymysql"
+version = "1.1.1"
+description = "Pure Python MySQL Driver"
+optional = false
+python-versions = ">=3.7"
+groups = ["main"]
+files = [
+    {file = "PyMySQL-1.1.1-py3-none-any.whl", hash = "sha256:4de15da4c61dc132f4fb9ab763063e693d521a80fd0e87943b9a453dd4c19d6c"},
+    {file = "pymysql-1.1.1.tar.gz", hash = "sha256:e127611aaf2b417403c60bf4dc570124aeb4a57f5f37b8e95ae399a42f904cd0"},
+]
+
+[package.extras]
+ed25519 = ["PyNaCl (>=1.4.0)"]
+rsa = ["cryptography"]
+
 [[package]]
 name = "pynacl"
 version = "1.5.0"
@@ -2463,4 +2479,4 @@ files = [
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<4.0"
-content-hash = "e02f61c9bb3bccd22fc51f90a9e09552afe50cb309155959dbb3ff06fea4d736"
+content-hash = "7b50455f855e5affd89ac1746a4e2acfe89ce8bc94670e003dfd11fa67dac75a"

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
     "impacket @ git+https://github.com/Pennyw0rth/impacket.git",
     "oscrypto @ git+https://github.com/wbond/oscrypto",
     "pynfsclient @ git+https://github.com/Pennyw0rth/NfsClient",
+    "pymysql (>=1.1.1,<2.0.0)",
 ]
 
 [project.urls]

tests/e2e_commands.txt

@@ -283,6 +283,16 @@ netexec ftp TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD --get test_file.txt
 netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce
 netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce --continue-on-success
 netexec ftp TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE
+##### MySQL
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD --databases
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD --tables mysql
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD --dump-database mysql
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD --server-capabilities
+netexec mysql TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD -q "SELECT VERSION()"
+netexec mysql TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce
+netexec mysql TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE --no-bruteforce --continue-on-success
+netexec mysql TARGET_HOST -u TEST_USER_FILE -p TEST_PASSWORD_FILE
 ##### NFS
 netexec nfs TARGET_HOST -u "" -p "" --shares
 netexec nfs TARGET_HOST -u "" -p "" --enum-shares