Skip to content

security: add whitelist validation for SQLite PRAGMA settings #1529

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mitre88 wants to merge 3 commits into from
Closed

security: add whitelist validation for SQLite PRAGMA settings #1529

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4be6e94
security: fix yaml.load deprecation and SQL injection in database cre…
Apr 23, 2026
65e27fb
bugfix: fix file handle leaks in arg_parser.py and messages.py
Apr 24, 2026
f7b4bcc
security: add whitelist validation for SQLite PRAGMA settings
Apr 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions nettacker/core/arg_parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -612,9 +612,8 @@ def parse_arguments(self):
options.targets = list(set(options.targets.split(",")))
if options.targets_list:
try:
options.targets = list(
set(open(options.targets_list, "rb").read().decode().split())
)
with open(options.targets_list, "rb") as f:
options.targets = list(set(f.read().decode().split()))
except Exception:
die_failure(_("error_target_file").format(options.targets_list))
Comment on lines +615 to 618

@coderabbitai coderabbitai Bot Apr 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Tighten the file-read exception scope.

The new with blocks are good, but these catches still swallow decode and I/O failures under a generic Exception, which makes bad input look like the same die_failure(...) path. Narrow them to the file errors you actually expect and apply the same change to all four read-from-file paths.

♻️ Example tightening 
-            except Exception:
+            except (OSError, UnicodeDecodeError):
                 die_failure(_("error_username").format(options.usernames_list))

Apply the same narrowing to the targets_list, passwords_list, and read_from_file blocks.

Also applies to: 717-721, 726-730, 733-737

🧰 Tools
🪛 Ruff (0.15.12)

[warning] 617-617: Do not catch blind exception: Exception

(BLE001)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@nettacker/core/arg_parser.py` around lines 615 - 618, The current broad
except Exception around the file-read blocks (e.g., the targets_list read that
sets options.targets) swallows decode and unrelated errors; narrow the catch to
the specific file and decode errors you expect (FileNotFoundError,
PermissionError, OSError and UnicodeDecodeError) and handle them the same way
you currently call die_failure(_("error_target_file").format(...)); apply this
tightening to all four file-read sites (targets_list, passwords_list, and the
two read_from_file places referenced) so only genuine file I/O or decode
failures are caught while other exceptions propagate.


Expand Down Expand Up @@ -708,28 +707,32 @@ def parse_arguments(self):
options.excluded_ports = list(tmp_excluded_ports)

if options.user_agent == "random_user_agent":
options.user_agents = open(Config.path.user_agents_file).read().split("\n")
with open(Config.path.user_agents_file) as f:
options.user_agents = f.read().split("\n")

# Check user list
if options.usernames:
options.usernames = list(set(options.usernames.split(",")))
elif options.usernames_list:
try:
options.usernames = list(set(open(options.usernames_list).read().split("\n")))
with open(options.usernames_list) as f:
options.usernames = list(set(f.read().split("\n")))
except Exception:
die_failure(_("error_username").format(options.usernames_list))
# Check password list
if options.passwords:
options.passwords = list(set(options.passwords.split(",")))
elif options.passwords_list:
try:
options.passwords = list(set(open(options.passwords_list).read().split("\n")))
with open(options.passwords_list) as f:
options.passwords = list(set(f.read().split("\n")))
except Exception:
die_failure(_("error_passwords").format(options.passwords_list))
# Check custom wordlist
if options.read_from_file:
try:
open(options.read_from_file).read().split("\n")
with open(options.read_from_file) as f:
f.read().split("\n")
except Exception:
die_failure(_("error_wordlist").format(options.read_from_file))
# Check output file
Expand Down
3 changes: 2 additions & 1 deletion nettacker/core/messages.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@ def application_language():


def load_yaml(filename):
return yaml.load(StringIO(open(filename, "r").read()), Loader=yaml.FullLoader)
with open(filename, "r") as f:
return yaml.safe_load(f)


def get_languages():
Expand Down
12 changes: 10 additions & 2 deletions nettacker/database/db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,16 @@ def create_connection():
cursor = connection.cursor()

# Performance enhancing configurations. Put WAL cause that helps with concurrency.
cursor.execute(f"PRAGMA journal_mode={Config.db.journal_mode}")
cursor.execute(f"PRAGMA synchronous={Config.db.synchronous_mode}")
allowed_journal_modes = {"DELETE", "WAL", "MEMORY", "OFF"}
allowed_sync_modes = {"OFF", "NORMAL", "FULL", "EXTRA"}
journal_mode = Config.db.journal_mode
sync_mode = Config.db.synchronous_mode
if journal_mode not in allowed_journal_modes:
raise ValueError(f"Invalid journal_mode: {journal_mode}")
if sync_mode not in allowed_sync_modes:
raise ValueError(f"Invalid synchronous_mode: {sync_mode}")
cursor.execute(f"PRAGMA journal_mode={journal_mode}")
cursor.execute(f"PRAGMA synchronous={sync_mode}")
Comment on lines +63 to +72

@coderabbitai coderabbitai Bot Apr 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

SQLite PRAGMA journal_mode valid values list

💡 Result:

The valid values for SQLite PRAGMA journal_mode are: DELETE, TRUNCATE, PERSIST, MEMORY, WAL, OFF. These are explicitly listed in the official SQLite documentation as the options that can be queried or set using PRAGMA journal_mode = DELETE | TRUNCATE | PERSIST | MEMORY | WAL | OFF. DELETE is the default mode, using a rollback journal that is deleted after each transaction. TRUNCATE truncates the journal to zero length. PERSIST overwrites the journal header but keeps the file. MEMORY stores the journal in RAM (risking corruption on crash). WAL uses write-ahead logging for better concurrency. OFF disables journaling entirely (highly risky, disables atomic commits). Note that for in-memory databases, only MEMORY or OFF are applicable, and changes to other modes are ignored. WAL mode is persistent across connections.

Citations:

🏁 Script executed:

# First, locate and read the database file
find . -name "db.py" -path "*/database/*" | head -20

Repository: OWASP/Nettacker

Length of output: 86

🏁 Script executed:

# Read the specific lines from the file
sed -n '50,80p' ./nettacker/database/db.py

Repository: OWASP/Nettacker

Length of output: 1464

🌐 Web query:

SQLite PRAGMA synchronous valid values documentation

💡 Result:

The valid values for the SQLite PRAGMA synchronous are: - 0 or OFF: Minimal synchronization, fastest but risks corruption on OS crash or power loss. - 1 or NORMAL: Syncs at critical moments, good balance for WAL mode, small risk in rollback mode. - 2 or FULL: Ensures data is written to disk before continuing, safe against crashes (default for rollback journal). - 3 or EXTRA: Like FULL but also syncs the directory after unlinking rollback journal, maximum durability. Usage: PRAGMA [schema.]synchronous; (query) or PRAGMA [schema.]synchronous = value; (set). TEMP schema always uses OFF. FULL is recommended for durability in rollback mode; NORMAL for WAL mode performance.

Citations:

Whitelist omits two valid SQLite journal modes and uses case-sensitive comparison.

SQLite's PRAGMA journal_mode accepts DELETE | TRUNCATE | PERSIST | MEMORY | WAL | OFF (case-insensitive). The current whitelist omits TRUNCATE and PERSIST, preventing users with those modes configured from initializing connections. Additionally, the case-sensitive comparison means configs like "wal" or "truncate" will be rejected despite SQLite accepting them. The synchronous_mode whitelist is already complete.

🔧 Proposed fix 
-            allowed_journal_modes = {"DELETE", "WAL", "MEMORY", "OFF"}
-            allowed_sync_modes = {"OFF", "NORMAL", "FULL", "EXTRA"}
-            journal_mode = Config.db.journal_mode
-            sync_mode = Config.db.synchronous_mode
-            if journal_mode not in allowed_journal_modes:
-                raise ValueError(f"Invalid journal_mode: {journal_mode}")
-            if sync_mode not in allowed_sync_modes:
-                raise ValueError(f"Invalid synchronous_mode: {sync_mode}")
-            cursor.execute(f"PRAGMA journal_mode={journal_mode}")
-            cursor.execute(f"PRAGMA synchronous={sync_mode}")
+            allowed_journal_modes = {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}
+            allowed_sync_modes = {"OFF", "NORMAL", "FULL", "EXTRA"}
+            journal_mode = str(Config.db.journal_mode).upper()
+            sync_mode = str(Config.db.synchronous_mode).upper()
+            if journal_mode not in allowed_journal_modes:
+                raise ValueError(f"Invalid journal_mode: {Config.db.journal_mode}")
+            if sync_mode not in allowed_sync_modes:
+                raise ValueError(f"Invalid synchronous_mode: {Config.db.synchronous_mode}")
+            cursor.execute(f"PRAGMA journal_mode={journal_mode}")
+            cursor.execute(f"PRAGMA synchronous={sync_mode}")
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
allowed_journal_modes = {"DELETE", "WAL", "MEMORY", "OFF"}
allowed_sync_modes = {"OFF", "NORMAL", "FULL", "EXTRA"}
journal_mode = Config.db.journal_mode
sync_mode = Config.db.synchronous_mode
if journal_mode not in allowed_journal_modes:
raise ValueError(f"Invalid journal_mode: {journal_mode}")
if sync_mode not in allowed_sync_modes:
raise ValueError(f"Invalid synchronous_mode: {sync_mode}")
cursor.execute(f"PRAGMA journal_mode={journal_mode}")
cursor.execute(f"PRAGMA synchronous={sync_mode}")
allowed_journal_modes = {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}
allowed_sync_modes = {"OFF", "NORMAL", "FULL", "EXTRA"}
journal_mode = str(Config.db.journal_mode).upper()
sync_mode = str(Config.db.synchronous_mode).upper()
if journal_mode not in allowed_journal_modes:
raise ValueError(f"Invalid journal_mode: {Config.db.journal_mode}")
if sync_mode not in allowed_sync_modes:
raise ValueError(f"Invalid synchronous_mode: {Config.db.synchronous_mode}")
cursor.execute(f"PRAGMA journal_mode={journal_mode}")
cursor.execute(f"PRAGMA synchronous={sync_mode}")
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@nettacker/database/db.py` around lines 63 - 72, The whitelist for SQLite
journal modes is missing "TRUNCATE" and "PERSIST" and comparisons are
case-sensitive; update allowed_journal_modes to include "TRUNCATE" and
"PERSIST", normalize journal_mode and synchronous_mode (e.g., journal_mode =
Config.db.journal_mode.upper(), sync_mode = Config.db.synchronous_mode.upper())
before checking against allowed_journal_modes and allowed_sync_modes, and use
the normalized values in the cursor.execute PRAGMA calls
(cursor.execute(f"PRAGMA journal_mode={journal_mode}") and
cursor.execute(f"PRAGMA synchronous={sync_mode}")) so lower/upper-cased config
values are accepted.


return connection, cursor
except Exception as e:
Expand Down
6 changes: 5 additions & 1 deletion nettacker/database/mysql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,11 @@ def mysql_create_database():
existing_databases = [d[0] for d in existing_databases]

if Config.db.name not in existing_databases:
conn.execute(text("CREATE DATABASE {0} ".format(Config.db.name)))
db_name = Config.db.name
if db_name.isalnum() and db_name[0].isalpha():

@chatgpt-codex-connector chatgpt-codex-connector Bot Apr 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Permit underscores in MySQL database-name validation

The new check db_name.isalnum() and db_name[0].isalpha() rejects names like test_db, even though underscore-containing names are valid and already expected in this codebase (for example tests/database/test_mysql.py uses Config.db.name = "test_db"). With that input, mysql_create_database() now raises ValueError, so database initialization can fail for existing user configs that previously worked. The validation should still prevent injection but accept valid identifier characters such as _.

Useful? React with 👍 / 👎.

conn.execute(text(f"CREATE DATABASE `{db_name}` "))
else:
raise ValueError(f"Invalid database name: {db_name}")
except Exception as e:
print(e)
Comment on lines 22 to 29

@coderabbitai coderabbitai Bot Apr 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

The raised ValueError is silently swallowed by the surrounding except Exception: print(e).

The new validation raises ValueError (line 27), but the outer handler at lines 28–29 just prints it and returns normally. Execution continues to mysql_create_tables() in the caller, which will then fail with a generic connection error — the security signal is lost and the failure mode becomes confusing. Either re-raise from the handler or narrow the except so the validation error propagates.

🔧 Proposed fix 
-    except Exception as e:
-        print(e)
+    except ValueError:
+        raise
+    except Exception as e:
+        print(e)

Also, db_name.isalnum() rejects underscores, which are common in valid MySQL database names — see the note on nettacker/database/postgresql.py (lines 29–33) for the suggested relaxed check; the same should apply here at line 24.

🧰 Tools
🪛 Ruff (0.15.11)

[warning] 28-28: Do not catch blind exception: Exception

(BLE001)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@nettacker/database/mysql.py` around lines 22 - 29, The try/except around the
database-creation block is swallowing the ValueError raised for invalid names
(Config.db.name) and only prints the exception; change this so validation errors
propagate: either remove the broad except or catch only operational exceptions
and re-raise ValueError so callers like mysql_create_tables() see the failure;
also relax the db name check (replace db_name.isalnum() and db_name[0].isalpha()
logic) to allow underscores (e.g. require the name match a pattern like starting
with a letter then letters, digits or underscores) and validate against
existing_databases before executing the CREATE DATABASE command.


Expand Down
6 changes: 5 additions & 1 deletion nettacker/database/postgresql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,11 @@ def postgres_create_database():
)
conn = engine.connect()
conn = conn.execution_options(isolation_level="AUTOCOMMIT")
conn.execute(text(f"CREATE DATABASE {Config.db.name}"))
db_name = Config.db.name
if db_name.isalnum() and db_name[0].isalpha():

@chatgpt-codex-connector chatgpt-codex-connector Bot Apr 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Allow valid PostgreSQL names with underscores

The same validation pattern in PostgreSQL setup now rejects common valid names like nettacker_db (which this repo itself uses in tests/database/test_postgresql.py). In the fallback path that creates a missing database, this causes a ValueError instead of creating the DB, so first-run setup breaks for underscore-based names that were previously accepted. Keep injection hardening, but broaden validation to include valid PostgreSQL identifier characters.

Useful? React with 👍 / 👎.

conn.execute(text(f'CREATE DATABASE "{db_name}"'))
else:
raise ValueError(f"Invalid database name: {db_name}")
Comment on lines +29 to +33

@coderabbitai coderabbitai Bot Apr 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Validation is too strict — rejects valid PostgreSQL identifiers containing underscores.

db_name.isalnum() returns False for any name containing _ (e.g., nettacker_db, owasp_nettacker). Underscores are standard and extremely common in PostgreSQL database names, so this change can break existing deployments the first time the OperationalError recovery path runs. Consider matching the PostgreSQL unquoted-identifier grammar instead, and chain the exception to satisfy B904.

🔧 Proposed fix 
+import re
 ...
         db_name = Config.db.name
-        if db_name.isalnum() and db_name[0].isalpha():
+        if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", db_name):
             conn.execute(text(f'CREATE DATABASE "{db_name}"'))
         else:
-            raise ValueError(f"Invalid database name: {db_name}")
+            raise ValueError(f"Invalid database name: {db_name}") from None

The same tightening applies symmetrically in nettacker/database/mysql.py (line 24).

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
db_name = Config.db.name
if db_name.isalnum() and db_name[0].isalpha():
conn.execute(text(f'CREATE DATABASE "{db_name}"'))
else:
raise ValueError(f"Invalid database name: {db_name}")
import re
db_name = Config.db.name
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", db_name):
conn.execute(text(f'CREATE DATABASE "{db_name}"'))
else:
raise ValueError(f"Invalid database name: {db_name}") from None
🧰 Tools
🪛 Ruff (0.15.11)

[warning] 33-33: Within an except clause, raise exceptions with raise ... from err or raise ... from None to distinguish them from errors in exception handling

(B904)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@nettacker/database/postgresql.py` around lines 29 - 33, The current
validation rejects valid PostgreSQL names with underscores; replace the
isalnum()/isalpha() checks on db_name with a regex that implements PostgreSQL
unquoted-identifier grammar (e.g. r'^[A-Za-z_][A-Za-z0-9_]*$') in
nettacker/database/postgresql.py (and mirror the same change in
nettacker/database/mysql.py for the symmetric check), and when raising the
ValueError ensure you chain the original exception in the surrounding
error-handling (use "raise ValueError(... ) from err" where err is the caught
exception) so exception chaining requirements are satisfied.

conn.close()
engine = create_engine(
"postgresql+psycopg2://{username}:{password}@{host}:{port}/{name}".format(
Expand Down
Loading