From 6cbd6af33abf3c07668503e1958a9d97f5e3fed5 Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:53 -0300 Subject: [PATCH 1/6] Checkmarx Bot - Auto Pull Request --- app/commands/IgnoreCommand.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/app/commands/IgnoreCommand.py b/app/commands/IgnoreCommand.py index 75afc5f..430a18a 100644 --- a/app/commands/IgnoreCommand.py +++ b/app/commands/IgnoreCommand.py @@ -16,20 +16,20 @@ def __init__(self, cursor, username): @override def execute(self, blabberUsername): - sqlQuery = "DELETE FROM listeners WHERE blabber='%s' AND listener='%s';" + sqlQuery = "DELETE FROM listeners WHERE blabber=%s AND listener=%s;" self.logger.info(sqlQuery) try : - self.cursor.execute(sqlQuery % (blabberUsername, self.username)) + self.cursor.execute(sqlQuery, [blabberUsername, self.username]) - sqlQuery = "SELECT blab_name FROM users WHERE username = '" + blabberUsername + "'" + sqlQuery = "SELECT blab_name FROM users WHERE username = %s" self.logger.info(sqlQuery) - self.cursor.execute(sqlQuery) + self.cursor.execute(sqlQuery, [blabberUsername]) result = self.cursor.fetchone() event = self.username + " is now ignoring " + blabberUsername + " (" + result[0] + ")" - sqlQuery = "INSERT INTO users_history (blabber, event) VALUES (\"" + self.username + "\", \"" + event + "\")" + sqlQuery = "INSERT INTO users_history (blabber, event) VALUES (%s, %s)" self.logger.info(sqlQuery) - self.cursor.execute(sqlQuery) + self.cursor.execute(sqlQuery, [self.username, event]) except: From 136bf5cfc8ff0113272e6503e76d07227f8f093d Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:54 -0300 Subject: [PATCH 2/6] Checkmarx Bot - Auto Pull Request --- app/views/blabController.py | 53 ++++++++++++++++++++++--------------- 1 file changed, 31 insertions(+), 22 deletions(-) diff --git a/app/views/blabController.py b/app/views/blabController.py index f0f4f01..4249407 100644 --- a/app/views/blabController.py +++ b/app/views/blabController.py @@ -24,12 +24,12 @@ sqlBlabsByMe = ("SELECT blabs.content, blabs.timestamp, COUNT(comments.blabber), blabs.blabid " "FROM blabs LEFT JOIN comments ON blabs.blabid = comments.blabid " - "WHERE blabs.blabber = '%s' GROUP BY blabs.blabid ORDER BY blabs.timestamp DESC;") + "WHERE blabs.blabber = %s GROUP BY blabs.blabid ORDER BY blabs.timestamp DESC;") sqlBlabsForMe = ("SELECT users.username, users.blab_name, blabs.content, blabs.timestamp, COUNT(comments.blabber), blabs.blabid " "FROM blabs INNER JOIN users ON blabs.blabber = users.username INNER JOIN listeners ON blabs.blabber = listeners.blabber " - "LEFT JOIN comments ON blabs.blabid = comments.blabid WHERE listeners.listener = '%s' " - "GROUP BY blabs.blabid ORDER BY blabs.timestamp DESC LIMIT {} OFFSET {};") + "LEFT JOIN comments ON blabs.blabid = comments.blabid WHERE listeners.listener = %s " + "GROUP BY blabs.blabid ORDER BY blabs.timestamp DESC LIMIT {limit} OFFSET {offset};") # # handles the getting and updating of information from feed page @@ -47,8 +47,8 @@ def feed(request): with connection.cursor() as cursor: logger.info("Executing query to get all 'Blabs for me'") - blabsForMe = sqlBlabsForMe.format(10, 0) - cursor.execute(blabsForMe % (username,)) + blabsForMe = sqlBlabsForMe.format(limit=10, offset=0) + cursor.execute(blabsForMe, [username]) blabsForMeResults = cursor.fetchall() feedBlabs = [] @@ -72,7 +72,7 @@ def feed(request): # Find the Blabs by this user logger.info("Executing query to get all of user's Blabs") - cursor.execute(sqlBlabsByMe % (username,)) + cursor.execute(sqlBlabsByMe, [username]) blabsByMeResults = cursor.fetchall() myBlabs = [] @@ -114,10 +114,10 @@ def feed(request): with connection.cursor() as cursor: logger.info("Creating query to add new Blab") - addBlabSql = "INSERT INTO blabs (blabber, content, timestamp) values ('%s', '%s', datetime('now'));" + addBlabSql = "INSERT INTO blabs (blabber, content, timestamp) values (%s, %s, datetime('now'));" logger.info("Executing query to add new blab") - cursor.execute(addBlabSql % (username, blab)) + cursor.execute(addBlabSql, [username, blab]) if not cursor.rowcount: request.error = "Failed to add blab" @@ -158,8 +158,8 @@ def morefeed(request): with connection.cursor() as cursor: logger.info("Executing query to see more Blabs") - blabsForMe = sqlBlabsForMe.format(len, cnt) - cursor.execute(blabsForMe % (username,)) + blabsForMe = sqlBlabsForMe.format(limit=len, offset=cnt) + cursor.execute(blabsForMe, [username]) results = cursor.fetchall() ret = "" for blab in results: @@ -187,18 +187,19 @@ def blab(request): logger.info("User is Logged In - continuing... UA=" + request.headers["User-Agent"] + " U=" + username) blabDetailsSql = ("SELECT blabs.content, users.blab_name " - "FROM blabs INNER JOIN users ON blabs.blabber = users.username " + "WHERE blabs.blabid = '%s';") + "FROM blabs INNER JOIN users ON blabs.blabber = users.username " + "WHERE blabs.blabid = %s;") blabCommentsSql = ("SELECT users.username, users.blab_name, comments.content, comments.timestamp " "FROM comments INNER JOIN users ON comments.blabber = users.username " - "WHERE comments.blabid = '%s' ORDER BY comments.timestamp DESC;") + "WHERE comments.blabid = %s ORDER BY comments.timestamp DESC;") try : logger.info("Creating the Database connection") with connection.cursor() as cursor: logger.info("Executing query to see Blab details") - cursor.execute(blabDetailsSql % (blabid,)) + cursor.execute(blabDetailsSql, [blabid]) blabDetailsResults = cursor.fetchone() if (blabDetailsResults): @@ -208,7 +209,7 @@ def blab(request): # Get comments logger.info("Executing query to get all comments") - cursor.execute(blabCommentsSql % (blabid,)) + cursor.execute(blabCommentsSql, [blabid]) blabCommentsResults = cursor.fetchall() comments = [] @@ -247,14 +248,14 @@ def blab(request): logger.info("User is Logged In - continuing... UA=" + request.headers["User-Agent"] + " U=" + username) - addCommentSql = "INSERT INTO comments (blabid, blabber, content, timestamp) values ('%s', '%s', '%s', '%s');" + addCommentSql = "INSERT INTO comments (blabid, blabber, content, timestamp) values (%s, %s, %s, %s);" try : logger.info("Creating the Database connection") with connection.cursor() as cursor: logger.info("Executing addComment") - cursor.execute(addCommentSql % (blabid, username, comment, moment.now().format("YYYY-MM-DD hh:mm:ss"))) + cursor.execute(addCommentSql, [blabid, username, comment, moment.now().format("YYYY-MM-DD hh:mm:ss")]) if not cursor.rowcount: request.error = "Failed to add comment" @@ -273,7 +274,13 @@ def blab(request): def blabbers(request): if request.method == "GET": sort = request.GET.get('sort') - if (sort is None or not sort): + # Use an allowlist to prevent ORDER BY injection; default to safe value + SORT_ALLOWLIST = { + "blab_name ASC", "blab_name DESC", + "created_at ASC", "created_at DESC", + "username ASC", "username DESC", + } + if sort not in SORT_ALLOWLIST: sort = "blab_name ASC" response = redirect('feed') logger.info("Showing Blabbers") @@ -282,14 +289,16 @@ def blabbers(request): if not username: logger.info("User is not Logged In - redirecting...") return redirect("/login?target=blabbers") - + logger.info("User is Logged In - continuing... UA=" + request.headers["User-Agent"] + " U=" + username) - blabbersSql = ("SELECT users.username," + " users.blab_name," + " users.created_at," - " SUM(iif(listeners.listener='%s', 1, 0)) as listeners," + blabbersSql = ("SELECT users.username, users.blab_name, users.created_at," + " SUM(iif(listeners.listener=%s, 1, 0)) as listeners," " SUM(iif(listeners.status='Active',1,0)) as listening" " FROM users LEFT JOIN listeners ON users.username = listeners.blabber" - " WHERE users.username NOT IN (\"admin\",\"admin-totp\",'%s')" + " GROUP BY users.username" + " ORDER BY " + sort + ";") + " WHERE users.username NOT IN ('admin','admin-totp',%s)" + " GROUP BY users.username" + " ORDER BY " + sort + ";") try: logger.info("Creating database connection") @@ -297,7 +306,7 @@ def blabbers(request): logger.info(blabbersSql) logger.info("Executing query to see Blab details") - cursor.execute(blabbersSql % (username, username)) + cursor.execute(blabbersSql, [username, username]) blabbersResults = cursor.fetchall() blabbers = [] From 52839c759865be018f09943771e940b4527a4704 Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:55 -0300 Subject: [PATCH 3/6] Checkmarx Bot - Auto Pull Request --- app/commands/ListenCommand.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/app/commands/ListenCommand.py b/app/commands/ListenCommand.py index 04e8102..fdfe273 100644 --- a/app/commands/ListenCommand.py +++ b/app/commands/ListenCommand.py @@ -16,20 +16,20 @@ def __init__(self, cursor, username): @override def execute(self, blabberUsername): - sqlQuery = "INSERT INTO listeners (blabber, listener, status) values ('%s', '%s', 'Active');" + sqlQuery = "INSERT INTO listeners (blabber, listener, status) values (%s, %s, 'Active');" self.logger.info(sqlQuery) try : - self.cursor.execute(sqlQuery % (blabberUsername, self.username)) + self.cursor.execute(sqlQuery, [blabberUsername, self.username]) - sqlQuery = "SELECT blab_name FROM users WHERE username = '" + blabberUsername + "'" + sqlQuery = "SELECT blab_name FROM users WHERE username = %s" self.logger.info(sqlQuery) - self.cursor.execute(sqlQuery) + self.cursor.execute(sqlQuery, [blabberUsername]) result = self.cursor.fetchone() event = self.username + " started listening to " + blabberUsername + " (" + result[0] + ")" - sqlQuery = "INSERT INTO users_history (blabber, event) VALUES (\"" + self.username + "\", \"" + event + "\")" + sqlQuery = "INSERT INTO users_history (blabber, event) VALUES (%s, %s)" self.logger.info(sqlQuery) - self.cursor.execute(sqlQuery) + self.cursor.execute(sqlQuery, [self.username, event]) except: From 9f0a723e5335ecc4ae4e1eca63df7326671bb922 Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:56 -0300 Subject: [PATCH 4/6] Checkmarx Bot - Auto Pull Request --- app/views/resetController.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/app/views/resetController.py b/app/views/resetController.py index cdd3a0c..e4b2b92 100644 --- a/app/views/resetController.py +++ b/app/views/resetController.py @@ -93,27 +93,26 @@ def processReset(request): with connection.cursor() as cursor: with transaction.atomic(): # Add the users - logger.info("Preparing the Stetement for adding users") - usersStatement = "INSERT INTO users (username, password, password_hint, totp_secret, created_at, last_login, real_name, blab_name) values ('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s');" + logger.info("Preparing the Statement for adding users") + usersStatement = "INSERT INTO users (username, password, password_hint, totp_secret, created_at, last_login, real_name, blab_name) values (%s, %s, %s, %s, %s, %s, %s, %s);" if users[0].password == "21232f297a57a5a743894a0e4a801fc3": logger.info("Encryption successful!") for user in users: logger.info("Adding user " + user.username) - cursor.execute(usersStatement % (user.username, user.password, user.password_hint, user.totp_secret, user.created_at, - user.last_login, user.real_name, user.blab_name)) + cursor.execute(usersStatement, [user.username, user.password, user.password_hint, user.totp_secret, user.created_at, + user.last_login, user.real_name, user.blab_name]) # Add the listeners logger.info("Preparing the Statement for adding listeners") with transaction.atomic(): - listenersStatement = "INSERT INTO listeners (blabber, listener, status) values ('%s', '%s', 'Active');" + listenersStatement = "INSERT INTO listeners (blabber, listener, status) values (%s, %s, 'Active');" for blabber in users[2:]: for listener in users[2:]: if rand.choice([False, True]) and (blabber != listener): - logger.info("Adding " + listener.username + " as a listener of " + blabber.username) - cursor.execute(listenersStatement % (blabber.username, listener.username)) + cursor.execute(listenersStatement, [blabber.username, listener.username]) # Fetch pre-loaded Blabs logger.info("Reading blabs from file") From 0e134dac0eaa3e75b76174de5335cd88f282dcb7 Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:57 -0300 Subject: [PATCH 5/6] Checkmarx Bot - Auto Pull Request --- app/views/userController.py | 111 +++++++++++++++--------------------- 1 file changed, 46 insertions(+), 65 deletions(-) diff --git a/app/views/userController.py b/app/views/userController.py index a7b1292..f9a2eb1 100644 --- a/app/views/userController.py +++ b/app/views/userController.py @@ -7,7 +7,6 @@ import pyotp import pickle, base64 -import sqlparse from email.mime.multipart import MIMEMultipart from passeo import passeo from ecdsa import SigningKey @@ -99,24 +98,10 @@ def login(request): logger.info("Creating the Database connection") with connection.cursor() as cursor: logger.info("Creating database query") - # START VULN CODE - sqlQuery = "select username, password, password_hint, created_at, last_login, \ - real_name, blab_name from users where username='" + username + "' \ - and password='" + hashlib.md5(password.encode('utf-8')).hexdigest() + "';" + sqlQuery = ("SELECT username, password, password_hint, created_at, last_login, " + "real_name, blab_name FROM users WHERE username=%s AND password=%s;") logger.info(sqlQuery) - - parsed = sqlparse.parse(sqlQuery)[0] - logger.info("Attempted login with username and password: " + parsed[8].value) - - cursor.execute(sqlQuery) - # END VULN CODE - # GOOD CODE - # sqlQuery = "select username, password, password_hint, created_at, last_login, \ - # real_name, blab_name from users where username=:username and password=:password;" - - # logger.info(sqlQuery, {"username": username, "password": hashlib.md5(password.encode('utf-8')).hexdigest()}) - # cursor.execute(sqlQuery, {"username": username, "password": hashlib.md5(password.encode('utf-8')).hexdigest()}) - # END GOOD CODE + cursor.execute(sqlQuery, [username, hashlib.md5(password.encode('utf-8')).hexdigest()]) row = cursor.fetchone() if (row): @@ -135,8 +120,8 @@ def login(request): blab_name=row["blab_name"]) response = updateInResponse(currentUser, response) - update = "UPDATE users SET last_login=datetime('now') WHERE username='" + row['username'] + "';" - cursor.execute(update) + update = "UPDATE users SET last_login=datetime('now') WHERE username=%s;" + cursor.execute(update, [row['username']]) # if the username ends with "totp", add the TOTP login step if username[-4:].lower() == "totp": @@ -181,9 +166,9 @@ def showPasswordHint(request): try: logger.info("Creating the Database connection") with connection.cursor() as cursor: - sql = "SELECT password_hint FROM users WHERE username = '" + username + "'" + sql = "SELECT password_hint FROM users WHERE username = %s" logger.info(sql) - cursor.execute(sql) + cursor.execute(sql, [username]) row = cursor.fetchone() if (row): @@ -222,9 +207,9 @@ def showTotp(request): #Create db connection with connection.cursor() as cursor: - sql = "SELECT totp_secret FROM users WHERE username = '" + username + "'" + sql = "SELECT totp_secret FROM users WHERE username = %s" logger.info(sql) - cursor.execute(sql) + cursor.execute(sql, [username]) result = cursor.fetchone() if result: @@ -256,9 +241,9 @@ def processTotp(request): with connection.cursor() as cursor: - sql = "SELECT totp_secret FROM users WHERE username = '" + username + "'" + sql = "SELECT totp_secret FROM users WHERE username = %s" logger.info(sql) - cursor.execute(sql) + cursor.execute(sql, [username]) result = cursor.fetchone() if result: @@ -338,8 +323,8 @@ def processRegister(request): logger.info("Creating the Database connection") try: with connection.cursor() as cursor: - sqlQuery = "SELECT username FROM users WHERE username = '" + username + "'" - cursor.execute(sqlQuery) + sqlQuery = "SELECT username FROM users WHERE username = %s" + cursor.execute(sqlQuery, [username]) row = cursor.fetchone() if (row): request.error = "Username '" + username + "' already exists!" @@ -405,22 +390,20 @@ def processRegisterFinish(request): realName = form.cleaned_data.get('realName') blabName = form.cleaned_data.get('blabName') #mysqlCurrentDateTime = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - #create query - query = '' - query += "insert into users (username, password, password_hint, totp_secret, created_at, real_name, blab_name) values(" - query += ("'" + username + "',") - query += ("'" + hashlib.md5(password.encode('utf-8')).hexdigest() + "',") - query += ("'" + password + "',") - query += ("'" + pyotp.random_base32() + "',") - query += ("datetime('now'),") - query += ("'" + realName + "',") - query += ("'" + blabName + "'") - query += (");") - #execute query - cursor.execute(query) - sqlStatement = cursor.fetchone() #<- variable for response + # Use parameterized query to prevent SQL injection + query = ("INSERT INTO users (username, password, password_hint, totp_secret, created_at, real_name, blab_name) " + "VALUES (%s, %s, %s, %s, datetime('now'), %s, %s);") + query_params = [ + username, + hashlib.md5(password.encode('utf-8')).hexdigest(), + password, + pyotp.random_base32(), + realName, + blabName, + ] logger.info(query) - # END EXAMPLE VULNERABILITY + cursor.execute(query, query_params) + sqlStatement = cursor.fetchone() except IntegrityError as ie: logger.error("Integrity error", ie) return render(request, 'app/register.html') @@ -482,16 +465,16 @@ def showProfile(request): myHecklers = None myInfo = None sqlMyHecklers = '' - sqlMyHecklers += "SELECT users.username, users.blab_name, users.created_at " - sqlMyHecklers += "FROM users LEFT JOIN listeners ON users.username = listeners.listener " - sqlMyHecklers += "WHERE listeners.blabber='%s' AND listeners.status='Active';" + sqlMyHecklers += "SELECT users.username, users.blab_name, users.created_at " + sqlMyHecklers += "FROM users LEFT JOIN listeners ON users.username = listeners.listener " + sqlMyHecklers += "WHERE listeners.blabber=%s AND listeners.status='Active';" try: logger.info("Getting Database connection") with connection.cursor() as cursor: # Find the Blabbers that this user listens to logger.info(sqlMyHecklers) - cursor.execute(sqlMyHecklers % username) + cursor.execute(sqlMyHecklers, [username]) myHecklersResults = cursor.fetchall() hecklers=[] for i in myHecklersResults: @@ -507,20 +490,18 @@ def showProfile(request): # Get the audit trail for this user events = [] - # START EXAMPLE VULNERABILITY - sqlMyEvents = "select event from users_history where blabber=\"" + username + "\" ORDER BY eventid DESC; " + sqlMyEvents = "SELECT event FROM users_history WHERE blabber=%s ORDER BY eventid DESC;" logger.info(sqlMyEvents) - cursor.execute(sqlMyEvents) + cursor.execute(sqlMyEvents, [username]) userHistoryResult = cursor.fetchall() - # END EXAMPLE VULNERABILITY for result in userHistoryResult : events.append(result[0]) # Get the users information - sql = "SELECT username, real_name, blab_name, totp_secret FROM users WHERE username = '" + username + "'" + sql = "SELECT username, real_name, blab_name, totp_secret FROM users WHERE username = %s" logger.info(sql) - cursor.execute(sql) + cursor.execute(sql, [username]) myInfoResults = cursor.fetchone() if not myInfoResults: return JsonResponse({'message':'Error, no Inforesults found'}) @@ -581,9 +562,9 @@ def processProfile(request): # TODO: Error in SQL execution with connection.cursor() as cursor: logger.info("Preparing the update Prepared Statement") - update = "UPDATE users SET real_name='%s', blab_name='%s' WHERE username='%s';" + update = "UPDATE users SET real_name=%s, blab_name=%s WHERE username=%s;" logger.info("Executing the update Prepared Statement") - cursor.execute(update % (realName,blabName,sessionUsername)) + cursor.execute(update, [realName, blabName, sessionUsername]) updateResult = cursor.fetchone() # If there is a record... @@ -729,8 +710,8 @@ def usernameExists(username): logger.info("Getting Database connection") with connection.cursor() as cursor: logger.info("Preparing the duplicate username check Prepared Statement") - sqlStatement = "SELECT username FROM users WHERE username='%s'" - cursor.execute(sqlStatement % (username,)) + sqlStatement = "SELECT username FROM users WHERE username=%s" + cursor.execute(sqlStatement, [username]) result = cursor.fetchone() if not result: # username does not exist @@ -761,17 +742,17 @@ def updateUsername(oldUsername, newUsername): # Update all references to this user sqlStrQueries = [ - "UPDATE users SET username='%s' WHERE username='%s'", - "UPDATE blabs SET blabber='%s' WHERE blabber='%s'", - "UPDATE comments SET blabber='%s' WHERE blabber='%s'", - "UPDATE listeners SET blabber='%s' WHERE blabber='%s'", - "UPDATE listeners SET listener='%s' WHERE listener='%s'", - "UPDATE users_history SET blabber='%s' WHERE blabber='%s'" ] - + "UPDATE users SET username=%s WHERE username=%s", + "UPDATE blabs SET blabber=%s WHERE blabber=%s", + "UPDATE comments SET blabber=%s WHERE blabber=%s", + "UPDATE listeners SET blabber=%s WHERE blabber=%s", + "UPDATE listeners SET listener=%s WHERE listener=%s", + "UPDATE users_history SET blabber=%s WHERE blabber=%s" ] + # Execute updates as part of a batch transaction # This will roll back all changes if one query fails for query in sqlStrQueries: - cursor.execute(query % (newUsername,oldUsername)) + cursor.execute(query, [newUsername, oldUsername]) # Rename the user profile image to match new username From 76f5b16803d11a3fca5356a6965539bf03d361d3 Mon Sep 17 00:00:00 2001 From: Lucas Santos Ferreira <215082630+cx-lucas-ferreira@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:42:58 -0300 Subject: [PATCH 6/6] Checkmarx Bot - Auto Pull Request --- app/tests/test_sql_injection.py | 812 ++++++++++++++++++++++++++++++++ 1 file changed, 812 insertions(+) create mode 100644 app/tests/test_sql_injection.py diff --git a/app/tests/test_sql_injection.py b/app/tests/test_sql_injection.py new file mode 100644 index 0000000..2ba7f9f --- /dev/null +++ b/app/tests/test_sql_injection.py @@ -0,0 +1,812 @@ +""" +Tests for Second Order SQL Injection remediation (CWE-89). + +These tests verify that all SQL queries in the application use Django's +parameterized query API (cursor.execute(sql, params)) rather than string +concatenation or %-formatting, thereby preventing SQL injection attacks. + +Second-order SQL injection is particularly dangerous because: + 1. Attacker stores a malicious payload in the database (e.g., via registration). + 2. The application later retrieves that value and uses it in a new SQL query + via string concatenation — executing the injected SQL. + +The fix uses Django's cursor.execute(sql, [params]) throughout, which passes +user-controlled values as bind parameters, never interpolating them into the +SQL string. +""" + +import hashlib +from unittest.mock import MagicMock, patch, call + +import django +from django.test import TestCase, RequestFactory, override_settings +from django.db import connection + + +# --------------------------------------------------------------------------- +# Helper: build a minimal fake cursor that records execute() calls +# --------------------------------------------------------------------------- + +class FakeCursor: + """Mock database cursor that records all execute() calls.""" + + def __init__(self, rows=None): + self.rows = rows or [] + self.description = [] + self.rowcount = len(self.rows) + self.execute_calls = [] # list of (sql, params) tuples + + def execute(self, sql, params=None): + self.execute_calls.append((sql, params)) + + def fetchone(self): + return self.rows[0] if self.rows else None + + def fetchall(self): + return self.rows + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +class FakeCursorContext: + """Context manager that yields a FakeCursor.""" + + def __init__(self, cursor): + self.cursor = cursor + + def __enter__(self): + return self.cursor + + def __exit__(self, *args): + pass + + +# --------------------------------------------------------------------------- +# Tests for ListenCommand +# --------------------------------------------------------------------------- + +class ListenCommandParameterizedTest(TestCase): + """Verify ListenCommand uses parameterized queries (no string formatting).""" + + def _make_cursor(self): + """Return a FakeCursor that returns a blab_name for the SELECT.""" + cursor = FakeCursor(rows=[("Test Blabber",)]) + cursor.description = [("blab_name",)] + return cursor + + def test_insert_listeners_uses_params(self): + """INSERT INTO listeners must pass username as a parameter, not inline.""" + from app.commands.ListenCommand import ListenCommand + + cursor = self._make_cursor() + cmd = ListenCommand(cursor, "alice") + cmd.execute("bob") + + insert_call = cursor.execute_calls[0] + sql, params = insert_call + + # The SQL must contain %s placeholders, not the literal values + self.assertIn("%s", sql) + self.assertNotIn("alice", sql) + self.assertNotIn("bob", sql) + # Values must be passed as parameters + self.assertIsNotNone(params) + self.assertIn("bob", params) + self.assertIn("alice", params) + + def test_select_blab_name_uses_params(self): + """SELECT blab_name must pass blabberUsername as a parameter.""" + from app.commands.ListenCommand import ListenCommand + + cursor = self._make_cursor() + cmd = ListenCommand(cursor, "alice") + cmd.execute("bob") + + select_call = cursor.execute_calls[1] + sql, params = select_call + + self.assertIn("%s", sql) + self.assertNotIn("bob", sql) + self.assertIsNotNone(params) + self.assertIn("bob", params) + + def test_insert_history_uses_params(self): + """INSERT INTO users_history must pass blabber and event as parameters.""" + from app.commands.ListenCommand import ListenCommand + + cursor = self._make_cursor() + cmd = ListenCommand(cursor, "alice") + cmd.execute("bob") + + history_call = cursor.execute_calls[2] + sql, params = history_call + + # SQL must not contain literal username values + self.assertIn("%s", sql) + self.assertNotIn("alice", sql) + self.assertIsNotNone(params) + # alice is the listener (stored as blabber in history) + self.assertIn("alice", params) + + def test_second_order_injection_payload_in_username_is_neutralized(self): + """ + A username containing SQL metacharacters stored in DB must be passed + safely as a parameter — the event string ends up as a bound parameter, + not interpolated into SQL. + """ + from app.commands.ListenCommand import ListenCommand + + # Simulate a malicious username retrieved from the database + malicious_username = "alice'; DROP TABLE users; --" + cursor = self._make_cursor() + cmd = ListenCommand(cursor, malicious_username) + cmd.execute("bob") + + # Verify no call contains the raw SQL payload in the query string + for sql, params in cursor.execute_calls: + self.assertNotIn("DROP TABLE", sql, + "SQL payload must not appear in the query string") + + # The malicious value appears only in params (safely bound) + history_call = cursor.execute_calls[2] + sql, params = history_call + self.assertIn(malicious_username, params) + + +# --------------------------------------------------------------------------- +# Tests for IgnoreCommand +# --------------------------------------------------------------------------- + +class IgnoreCommandParameterizedTest(TestCase): + """Verify IgnoreCommand uses parameterized queries.""" + + def _make_cursor(self): + cursor = FakeCursor(rows=[("Test Blabber",)]) + cursor.description = [("blab_name",)] + return cursor + + def test_delete_listeners_uses_params(self): + """DELETE FROM listeners must use bind parameters.""" + from app.commands.IgnoreCommand import IgnoreCommand + + cursor = self._make_cursor() + cmd = IgnoreCommand(cursor, "alice") + cmd.execute("bob") + + delete_call = cursor.execute_calls[0] + sql, params = delete_call + + self.assertIn("%s", sql) + self.assertNotIn("alice", sql) + self.assertNotIn("bob", sql) + self.assertIsNotNone(params) + self.assertIn("bob", params) + self.assertIn("alice", params) + + def test_insert_history_uses_params(self): + """INSERT INTO users_history must use bind parameters.""" + from app.commands.IgnoreCommand import IgnoreCommand + + cursor = self._make_cursor() + cmd = IgnoreCommand(cursor, "alice") + cmd.execute("bob") + + history_call = cursor.execute_calls[2] + sql, params = history_call + + self.assertIn("%s", sql) + self.assertNotIn("alice", sql) + self.assertIsNotNone(params) + self.assertIn("alice", params) + + def test_second_order_injection_payload_neutralized(self): + """Malicious blabberUsername retrieved from DB must not inject SQL.""" + from app.commands.IgnoreCommand import IgnoreCommand + + malicious_blabber = "bob\"); INSERT INTO users VALUES ('hacker','pwned','hint',NULL,NULL,NULL,NULL); --" + cursor = self._make_cursor() + cmd = IgnoreCommand(cursor, "alice") + cmd.execute(malicious_blabber) + + for sql, params in cursor.execute_calls: + self.assertNotIn("INSERT INTO users", sql, + "Injection payload must not appear in the SQL string") + + +# --------------------------------------------------------------------------- +# Tests for userController SQL queries (unit-level with mock cursor) +# --------------------------------------------------------------------------- + +class UserControllerParameterizedQueryTest(TestCase): + """Unit tests verifying userController queries use parameterized execution.""" + + def _mock_connection(self, cursor): + """Return a context-manager mock that yields the given cursor.""" + ctx = MagicMock() + ctx.__enter__ = MagicMock(return_value=cursor) + ctx.__exit__ = MagicMock(return_value=False) + conn_mock = MagicMock() + conn_mock.cursor.return_value = ctx + return conn_mock + + # ------------------------------------------------------------------ + # showPasswordHint + # ------------------------------------------------------------------ + + @patch("app.views.userController.connection") + def test_password_hint_uses_parameterized_query(self, mock_conn): + """showPasswordHint must use %s + params, never string concatenation.""" + from app.views.userController import showPasswordHint + + cursor = FakeCursor(rows=[("mysecretpassword",)]) + cursor.description = [("password_hint",)] + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.get("/password-hint", {"username": "alice"}) + + showPasswordHint(request) + + self.assertEqual(len(cursor.execute_calls), 1) + sql, params = cursor.execute_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("alice", sql, + "Username must not be interpolated into the SQL string") + self.assertEqual(params, ["alice"]) + + @patch("app.views.userController.connection") + def test_password_hint_injection_payload_not_in_sql(self, mock_conn): + """SQL metacharacters in username must not reach the SQL string.""" + from app.views.userController import showPasswordHint + + malicious = "' OR '1'='1" + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.get("/password-hint", {"username": malicious}) + showPasswordHint(request) + + sql, params = cursor.execute_calls[0] + self.assertNotIn(malicious, sql) + self.assertEqual(params, [malicious]) + + # ------------------------------------------------------------------ + # processRegister (username duplicate check) + # ------------------------------------------------------------------ + + @patch("app.views.userController.connection") + def test_process_register_username_check_parameterized(self, mock_conn): + """processRegister duplicate username check must use bind params.""" + from app.views.userController import processRegister + + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.post("/register", {"username": "newuser"}) + + processRegister(request) + + self.assertGreaterEqual(len(cursor.execute_calls), 1) + sql, params = cursor.execute_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("newuser", sql) + self.assertEqual(params, ["newuser"]) + + # ------------------------------------------------------------------ + # showProfile — sqlMyEvents (the canonical second-order sink) + # ------------------------------------------------------------------ + + @patch("app.views.userController.getProfileImageNameFromUsername", + return_value=None) + @patch("app.views.userController.connection") + def test_show_profile_events_query_parameterized(self, mock_conn, _img): + """ + showProfile's users_history query must use parameterized execution. + + This is the primary second-order injection sink: username comes from + the session (set from a DB value), and was previously concatenated + directly into the SQL string. + """ + from app.views.userController import showProfile + + # Cursor returns: hecklers=[], events=[], user info row + heckler_rows = [] + event_rows = [] + user_rows = [("alice", "Alice Real", "AliceBlab", "TOTP_SECRET")] + + # We need the cursor to return different results for each execute call. + call_count = [0] + results_sequence = [heckler_rows, event_rows, user_rows] + + class SequencedCursor: + execute_calls = [] + description = [ + ("username",), ("real_name",), ("blab_name",), ("totp_secret",) + ] + rowcount = 0 + + def execute(self, sql, params=None): + self.execute_calls.append((sql, params)) + + def fetchone(self): + idx = len(self.execute_calls) - 1 + rows = results_sequence[min(idx, len(results_sequence) - 1)] + return rows[0] if rows else None + + def fetchall(self): + idx = len(self.execute_calls) - 1 + return results_sequence[min(idx, len(results_sequence) - 1)] + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + seq_cursor = SequencedCursor() + mock_conn.cursor.return_value = seq_cursor + + factory = RequestFactory() + request = factory.get("/profile") + request.session = {"username": "alice"} + + showProfile(request) + + # The second execute call is the users_history / events query + # (first is hecklers, second is events, third is user info) + self.assertGreaterEqual(len(seq_cursor.execute_calls), 2) + + # Find the users_history query + history_calls = [ + (sql, params) + for sql, params in seq_cursor.execute_calls + if "users_history" in sql.lower() + ] + self.assertTrue(history_calls, + "Expected a users_history query to be executed") + + sql, params = history_calls[0] + # Must use %s placeholder + self.assertIn("%s", sql) + # Username must NOT be embedded in the SQL string + self.assertNotIn("alice", sql, + "Username must not be interpolated into the SQL string") + # Must be passed as a parameter + self.assertIsNotNone(params) + self.assertIn("alice", params) + + @patch("app.views.userController.getProfileImageNameFromUsername", + return_value=None) + @patch("app.views.userController.connection") + def test_show_profile_hecklers_query_parameterized(self, mock_conn, _img): + """showProfile's listeners/hecklers query must use parameterized execution.""" + from app.views.userController import showProfile + + class SequencedCursor2: + execute_calls = [] + description = [ + ("username",), ("real_name",), ("blab_name",), ("totp_secret",) + ] + rowcount = 0 + + def execute(self, sql, params=None): + self.execute_calls.append((sql, params)) + + def fetchone(self): + if self.execute_calls and "users" in self.execute_calls[-1][0].lower(): + return ("alice", "Alice", "AliceBlab", "SECRET") + return None + + def fetchall(self): + return [] + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + seq_cursor = SequencedCursor2() + mock_conn.cursor.return_value = seq_cursor + + factory = RequestFactory() + request = factory.get("/profile") + request.session = {"username": "alice"} + + showProfile(request) + + listener_calls = [ + (sql, params) + for sql, params in seq_cursor.execute_calls + if "listeners" in sql.lower() + ] + self.assertTrue(listener_calls) + sql, params = listener_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("alice", sql) + self.assertIsNotNone(params) + self.assertIn("alice", params) + + # ------------------------------------------------------------------ + # updateUsername — batch UPDATE queries + # ------------------------------------------------------------------ + + @patch("app.views.userController.getProfileImageNameFromUsername", + return_value=None) + @patch("app.views.userController.transaction") + @patch("app.views.userController.connection") + def test_update_username_queries_parameterized(self, mock_conn, + mock_txn, _img): + """All UPDATE queries in updateUsername must use bind parameters.""" + from app.views.userController import updateUsername + + cursor = FakeCursor() + # transaction.atomic() context manager + mock_txn.atomic.return_value.__enter__ = MagicMock(return_value=None) + mock_txn.atomic.return_value.__exit__ = MagicMock(return_value=False) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + updateUsername("alice", "alice_new") + + self.assertGreater(len(cursor.execute_calls), 0, + "Expected at least one UPDATE to be executed") + + for sql, params in cursor.execute_calls: + # Every query must use %s placeholders + self.assertIn("%s", sql, + f"Query must use %s placeholders: {sql!r}") + # Old/new usernames must not be embedded in the SQL string + self.assertNotIn("alice_new", sql, + f"New username must not be in SQL: {sql!r}") + self.assertNotIn("alice", sql, + f"Old username must not be in SQL: {sql!r}") + # Values must be passed as parameters + self.assertIsNotNone(params) + self.assertIn("alice_new", params) + self.assertIn("alice", params) + + def test_update_username_second_order_payload_not_in_sql(self): + """ + A username that was stored in DB with SQL injection payload must not + cause SQL injection when updateUsername is called. + + This exercises the second-order scenario: attacker registers with a + username like alice'; UPDATE users SET password='pwned' WHERE '1'='1 + and the app later calls updateUsername() with that stored value. + """ + from unittest.mock import patch, MagicMock + + malicious_old = "alice'; UPDATE users SET password='pwned' WHERE '1'='1" + malicious_new = "alice_new" + + cursor = FakeCursor() + + with patch("app.views.userController.transaction") as mock_txn, \ + patch("app.views.userController.connection") as mock_conn, \ + patch("app.views.userController.getProfileImageNameFromUsername", + return_value=None): + + mock_txn.atomic.return_value.__enter__ = MagicMock(return_value=None) + mock_txn.atomic.return_value.__exit__ = MagicMock(return_value=False) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + updateUsername(malicious_old, malicious_new) + + for sql, params in cursor.execute_calls: + self.assertNotIn("UPDATE users SET password", sql, + "Injection payload must not appear in query string") + + +# --------------------------------------------------------------------------- +# Tests for processProfile UPDATE query +# --------------------------------------------------------------------------- + +class ProcessProfileParameterizedTest(TestCase): + """Verify processProfile UPDATE query is parameterized.""" + + @patch("app.views.userController.connection") + def test_update_profile_parameterized(self, mock_conn): + """UPDATE users SET real_name/blab_name must use bind parameters.""" + from app.views.userController import processProfile + + cursor = FakeCursor(rows=[]) # fetchone returns None → success path + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.post( + "/profile", + { + "username": "alice", + "realName": "Alice Wonderland", + "blabName": "AliceBlab", + }, + HTTP_X_REQUESTED_WITH="XMLHttpRequest", + ) + request.session = {"username": "alice"} + + processProfile(request) + + update_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "UPDATE users" in sql + ] + self.assertTrue(update_calls, + "Expected an UPDATE users query") + sql, params = update_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("Alice Wonderland", sql) + self.assertNotIn("AliceBlab", sql) + self.assertNotIn("alice", sql) + self.assertIsNotNone(params) + self.assertIn("Alice Wonderland", params) + self.assertIn("AliceBlab", params) + self.assertIn("alice", params) + + @patch("app.views.userController.connection") + def test_update_profile_second_order_realname_not_in_sql(self, mock_conn): + """ + Real name stored with SQL payload must not inject SQL on profile update. + + Scenario: attacker previously set real_name to a SQL payload via a + stored value. On profile update, that value is used again — it must be + a bound parameter, not interpolated. + """ + from app.views.userController import processProfile + + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + malicious_real_name = "hacker', 'hacked'); DROP TABLE users; --" + request = factory.post( + "/profile", + { + "username": "alice", + "realName": malicious_real_name, + "blabName": "safe", + }, + HTTP_X_REQUESTED_WITH="XMLHttpRequest", + ) + request.session = {"username": "alice"} + + processProfile(request) + + for sql, params in cursor.execute_calls: + self.assertNotIn("DROP TABLE", sql, + "SQL payload must not appear in the query string") + + +# --------------------------------------------------------------------------- +# Tests for blabController SQL queries +# --------------------------------------------------------------------------- + +class BlabControllerParameterizedTest(TestCase): + """Verify blabController uses parameterized queries throughout.""" + + @patch("app.views.blabController.connection") + def test_feed_post_add_blab_parameterized(self, mock_conn): + """INSERT INTO blabs must pass blab content as a parameter.""" + from app.views.blabController import feed + + cursor = FakeCursor() + cursor.rowcount = 1 + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.post("/feed", {"blab": "Hello World"}) + request.session = {"username": "alice"} + request.headers # ensure headers dict exists + + feed(request) + + insert_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "INSERT INTO blabs" in sql + ] + self.assertTrue(insert_calls, "Expected INSERT INTO blabs") + sql, params = insert_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("Hello World", sql) + self.assertNotIn("alice", sql) + self.assertIsNotNone(params) + self.assertIn("alice", params) + self.assertIn("Hello World", params) + + @patch("app.views.blabController.connection") + def test_blab_get_details_parameterized(self, mock_conn): + """SELECT blabs/comments by blabid must use bind parameters.""" + from app.views.blabController import blab + + cursor = FakeCursor(rows=[("Content here", "SomeBlab")]) + cursor.description = [("content",), ("blab_name",)] + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.get("/blab", {"blabid": "42"}) + request.session = {"username": "alice"} + + blab(request) + + detail_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "blabs" in sql.lower() and "WHERE" in sql + ] + self.assertTrue(detail_calls) + sql, params = detail_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("42", sql) + self.assertIsNotNone(params) + self.assertIn("42", params) + + @patch("app.views.blabController.connection") + def test_blab_post_add_comment_parameterized(self, mock_conn): + """INSERT INTO comments must pass all values as parameters.""" + from app.views.blabController import blab + + cursor = FakeCursor() + cursor.rowcount = 1 + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.post("/blab", { + "blabid": "42", + "comment": "Great post!", + }) + request.session = {"username": "alice"} + + blab(request) + + comment_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "INSERT INTO comments" in sql + ] + self.assertTrue(comment_calls, "Expected INSERT INTO comments") + sql, params = comment_calls[0] + self.assertIn("%s", sql) + self.assertNotIn("Great post!", sql) + self.assertNotIn("alice", sql) + self.assertNotIn("42", sql) + self.assertIsNotNone(params) + self.assertIn("alice", params) + self.assertIn("Great post!", params) + + @patch("app.views.blabController.connection") + def test_blabbers_sort_allowlist_rejects_injection(self, mock_conn): + """ + ORDER BY parameter must be validated against an allowlist. + An attempt to inject SQL via the sort parameter must be rejected. + """ + from app.views.blabController import blabbers + + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + # Attempt ORDER BY injection + malicious_sort = "blab_name; DROP TABLE users; --" + request = factory.get("/blabbers", {"sort": malicious_sort}) + request.session = {"username": "alice"} + + blabbers(request) + + # The malicious sort must not appear in any executed SQL + for sql, params in cursor.execute_calls: + self.assertNotIn("DROP TABLE", sql, + "Sort injection payload must not appear in SQL") + + @patch("app.views.blabController.connection") + def test_blabbers_sort_allowlist_accepts_valid_sort(self, mock_conn): + """A valid sort value from the allowlist must be accepted.""" + from app.views.blabController import blabbers + + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.get("/blabbers", {"sort": "created_at DESC"}) + request.session = {"username": "alice"} + + blabbers(request) + + sort_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "ORDER BY" in sql + ] + self.assertTrue(sort_calls, "Expected an ORDER BY query") + sql, _ = sort_calls[0] + self.assertIn("created_at DESC", sql) + + @patch("app.views.blabController.connection") + def test_blabbers_sort_defaults_to_safe_value(self, mock_conn): + """Missing or unknown sort value must default to the safe default.""" + from app.views.blabController import blabbers + + cursor = FakeCursor(rows=[]) + mock_conn.cursor.return_value = FakeCursorContext(cursor) + + factory = RequestFactory() + request = factory.get("/blabbers") # no sort param + request.session = {"username": "alice"} + + blabbers(request) + + sort_calls = [ + (sql, params) + for sql, params in cursor.execute_calls + if "ORDER BY" in sql + ] + self.assertTrue(sort_calls) + sql, _ = sort_calls[0] + self.assertIn("blab_name ASC", sql) + + +# --------------------------------------------------------------------------- +# Integration-style tests: verify no raw string interpolation remains +# --------------------------------------------------------------------------- + +class NoStringInterpolationInQueriesTest(TestCase): + """ + Regression tests that confirm the fix pattern is in place by inspecting + the source code itself. These tests will fail if string concatenation / + %-formatting is re-introduced into SQL queries. + """ + + def _load_source(self, path): + import os + base = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) + with open(os.path.join(base, path)) as f: + return f.read() + + def _count_parameterized(self, source): + """Count cursor.execute() calls that pass a params argument.""" + import re + # Match: cursor.execute(something, [... or (... or params_var + return len(re.findall( + r'cursor\.execute\s*\([^)]+,\s*[\[\(]', source)) + + def _count_unsafe_execute(self, source): + """ + Count cursor.execute() calls that use % or + to build the SQL. + We look for execute calls where the argument contains % or + + followed by a variable (simplified heuristic). + """ + import re + # Detect cursor.execute(sql % ... or cursor.execute("..." + var + return len(re.findall( + r'cursor\.execute\s*\([^,)]*(?:%\s*[(\w]|\+\s*\w)', source)) + + def test_listen_command_no_format_string_sql(self): + source = self._load_source("app/commands/ListenCommand.py") + unsafe = self._count_unsafe_execute(source) + self.assertEqual(unsafe, 0, + "ListenCommand must not use %-format or + for SQL queries") + + def test_ignore_command_no_format_string_sql(self): + source = self._load_source("app/commands/IgnoreCommand.py") + unsafe = self._count_unsafe_execute(source) + self.assertEqual(unsafe, 0, + "IgnoreCommand must not use %-format or + for SQL queries") + + def test_listen_command_has_parameterized_executes(self): + source = self._load_source("app/commands/ListenCommand.py") + count = self._count_parameterized(source) + self.assertGreaterEqual(count, 3, + "ListenCommand should have at least 3 parameterized executes") + + def test_ignore_command_has_parameterized_executes(self): + source = self._load_source("app/commands/IgnoreCommand.py") + count = self._count_parameterized(source) + self.assertGreaterEqual(count, 3, + "IgnoreCommand should have at least 3 parameterized executes")