Implement basic sanitization, injection tests, improve git-bash lookup on windows

Author: bjthres1

stat_log_db/src/stat_log_db/db.py

@@ -1,4 +1,5 @@
 # from abc import ABC, abstractmethod
+import re
 import sqlite3
 import uuid
 from typing import Any
@@ -317,42 +318,132 @@ def fetchone(self):
     def fetchall(self):
         return self.cursor.fetchall()
 
+    def _validate_sql_identifier(self, identifier: str, identifier_type: str = "identifier") -> str:
+        """
+        Validate and sanitize SQL identifiers (table names, column names) to prevent SQL injection.
+        
+        Args:
+            identifier: The identifier to validate
+            identifier_type: Type of identifier for error messages (e.g., "table name", "column name")
+            
+        Returns:
+            The validated identifier
+            
+        Raises:
+            ValueError: If the identifier is invalid or potentially dangerous
+        """
+        if not isinstance(identifier, str):
+            raise TypeError(f"SQL {identifier_type} must be a string, got {type(identifier).__name__}")
+        
+        if len(identifier) == 0:
+            raise ValueError(f"SQL {identifier_type} cannot be empty")
+        
+        # Check for valid identifier pattern: starts with letter/underscore, contains only alphanumeric/underscore
+        if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', identifier):
+            raise ValueError(f"Invalid SQL {identifier_type}: '{identifier}'. Must start with letter or underscore and contain only letters, numbers, and underscores.")
+        
+        # Check against SQLite reserved words (common ones that could cause issues)
+        reserved_words = {
+            'abort', 'action', 'add', 'after', 'all', 'alter', 'analyze', 'and', 'as', 'asc',
+            'attach', 'autoincrement', 'before', 'begin', 'between', 'by', 'cascade', 'case',
+            'cast', 'check', 'collate', 'column', 'commit', 'conflict', 'constraint', 'create',
+            'cross', 'current', 'current_date', 'current_time', 'current_timestamp', 'database',
+            'default', 'deferrable', 'deferred', 'delete', 'desc', 'detach', 'distinct', 'do',
+            'drop', 'each', 'else', 'end', 'escape', 'except', 'exclusive', 'exists', 'explain',
+            'fail', 'filter', 'following', 'for', 'foreign', 'from', 'full', 'glob', 'group',
+            'having', 'if', 'ignore', 'immediate', 'in', 'index', 'indexed', 'initially', 'inner',
+            'insert', 'instead', 'intersect', 'into', 'is', 'isnull', 'join', 'key', 'left',
+            'like', 'limit', 'match', 'natural', 'no', 'not', 'notnull', 'null', 'of', 'offset',
+            'on', 'or', 'order', 'outer', 'over', 'partition', 'plan', 'pragma', 'preceding',
+            'primary', 'query', 'raise', 'range', 'recursive', 'references', 'regexp', 'reindex',
+            'release', 'rename', 'replace', 'restrict', 'right', 'rollback', 'row', 'rows',
+            'savepoint', 'select', 'set', 'table', 'temp', 'temporary', 'then', 'to', 'transaction',
+            'trigger', 'unbounded', 'union', 'unique', 'update', 'using', 'vacuum', 'values',
+            'view', 'virtual', 'when', 'where', 'window', 'with', 'without'
+        }
+        
+        if identifier.lower() in reserved_words:
+            raise ValueError(f"SQL {identifier_type} '{identifier}' is a reserved word and cannot be used")
+        
+        return identifier
+
+    def _escape_sql_identifier(self, identifier: str) -> str:
+        """
+        Escape SQL identifier by wrapping in double quotes and escaping any internal quotes.
+        This should only be used after validation.
+        """
+        # Escape any double quotes in the identifier by doubling them
+        escaped = identifier.replace('"', '""')
+        return f'"{escaped}"'
+
     def create_table(self, table_name: str, columns: list[tuple[str, str]], temp_table: bool = True, raise_if_exists: bool = True):
         # Validate table_name argument
         if not isinstance(table_name, str):
             raise_auto_arg_type_error("table_name")
         if len(table_name) == 0:
             raise ValueError(f"'table_name' argument of create_table cannot be an empty string!")
+        
+        # Validate and sanitize table name
+        validated_table_name = self._validate_sql_identifier(table_name, "table name")
+        escaped_table_name = self._escape_sql_identifier(validated_table_name)
+        
         if not isinstance(raise_if_exists, bool):
             raise_auto_arg_type_error("raise_if_exists")
-        # Check if table already exists
+        
+        # Check if table already exists using parameterized query
         if raise_if_exists:
-            self.cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", (table_name,))
+            self.cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", (validated_table_name,))
             if self.cursor.fetchone() is not None:
-                raise ValueError(f"Table '{table_name}' already exists.")
+                raise ValueError(f"Table '{validated_table_name}' already exists.")
+        
         # Validate temp_table argument
         if not isinstance(temp_table, bool):
             raise_auto_arg_type_error("temp_table")
+        
         # Validate columns argument
         if (not isinstance(columns, list)) or (not all(
             isinstance(col, tuple) and len(col) == 2
             and isinstance(col[0], str)
             and isinstance(col[1], str)
         for col in columns)):
             raise_auto_arg_type_error("columns")
-        # Construct columns portion of query
-        # TODO: construct parameters for columns rather than f-string to prevent SQL injection
-        columns_qstr = ""
-        for col in columns:
-            columns_qstr += f"{col[0]} {col[1]},\n"
-        columns_qstr = columns_qstr.rstrip(",\n") # Remove trailing comma and newline
-        # Assemble full query
-        query = f"""--sql
-            CREATE{" TEMPORARY" if temp_table else ""} TABLE IF NOT EXISTS '{table_name}' (
+        
+        # Validate and construct columns portion of query
+        validated_columns = []
+        for col_name, col_type in columns:
+            # Validate column name
+            validated_col_name = self._validate_sql_identifier(col_name, "column name")
+            escaped_col_name = self._escape_sql_identifier(validated_col_name)
+            
+            # Validate column type - allow only safe, known SQLite types
+            allowed_types = {
+                'TEXT', 'INTEGER', 'REAL', 'BLOB', 'NUMERIC',
+                'VARCHAR', 'CHAR', 'NVARCHAR', 'NCHAR',
+                'CLOB', 'DATE', 'DATETIME', 'TIMESTAMP',
+                'BOOLEAN', 'DECIMAL', 'DOUBLE', 'FLOAT',
+                'INT', 'BIGINT', 'SMALLINT', 'TINYINT'
+            }
+            
+            # Allow type specifications with length/precision (e.g., VARCHAR(50), DECIMAL(10,2))
+            base_type = re.match(r'^([A-Z]+)', col_type.upper())
+            if not base_type or base_type.group(1) not in allowed_types:
+                raise ValueError(f"Unsupported column type: '{col_type}'. Must be one of: {', '.join(sorted(allowed_types))}")
+            
+            # Basic validation for type specification format
+            if not re.match(r'^[A-Z]+(\([0-9,\s]+\))?$', col_type.upper()):
+                raise ValueError(f"Invalid column type format: '{col_type}'")
+            
+            validated_columns.append(f"{escaped_col_name} {col_type.upper()}")
+        
+        columns_qstr = ",\n                ".join(validated_columns)
+        
+        # Assemble full query with escaped identifiers
+        temp_keyword = " TEMPORARY" if temp_table else ""
+        query = f"""CREATE{temp_keyword} TABLE IF NOT EXISTS {escaped_table_name} (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 {columns_qstr}
-            );
-        """
+            );"""
+        
         self.execute(query)
 
     def drop_table(self, table_name: str, raise_if_not_exists: bool = False):
@@ -361,13 +452,22 @@ def drop_table(self, table_name: str, raise_if_not_exists: bool = False):
             raise_auto_arg_type_error("table_name")
         if len(table_name) == 0:
             raise ValueError(f"'table_name' argument of drop_table cannot be an empty string!")
+        
+        # Validate and sanitize table name
+        validated_table_name = self._validate_sql_identifier(table_name, "table name")
+        escaped_table_name = self._escape_sql_identifier(validated_table_name)
+        
         if not isinstance(raise_if_not_exists, bool):
             raise_auto_arg_type_error("raise_if_not_exists")
+        
+        # Check if table exists using parameterized query
         if raise_if_not_exists:
-            self.cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';")
+            self.cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", (validated_table_name,))
             if self.cursor.fetchone() is None:
-                raise ValueError(f"Table '{table_name}' does not exist.")
-        self.cursor.execute(f"DROP TABLE IF EXISTS '{table_name}';")
+                raise ValueError(f"Table '{validated_table_name}' does not exist.")
+        
+        # Execute DROP statement with escaped identifier
+        self.cursor.execute(f"DROP TABLE IF EXISTS {escaped_table_name};")
 
     # def read(self):
 

stat_log_db/tests/test_sql_injection.py

@@ -0,0 +1,183 @@
+"""
+Test SQL injection protection in create_table and drop_table methods.
+"""
+
+import pytest
+import sys
+from pathlib import Path
+
+# Add the src directory to the path to import the module
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "stat_log_db" / "src"))
+
+from stat_log_db.db import MemDB
+
+
+@pytest.fixture
+def mem_db():
+    """Create a test in-memory database and clean up after tests."""
+    sl_db = MemDB({
+        "is_mem": True,
+        "fkey_constraint": True
+    })
+    con = sl_db.init_db(True)
+    yield con
+    # Cleanup
+    sl_db.close_db()
+
+
+class TestSQLInjectionProtection:
+    """Test class for SQL injection protection in database operations."""
+
+    def test_malicious_table_name_create(self, mem_db):
+        """Test that malicious SQL injection in table names is rejected."""
+        with pytest.raises(ValueError, match="Invalid SQL table name"):
+            mem_db.create_table("test'; DROP TABLE users; --", [('notes', 'TEXT')], False, True)
+
+    def test_reserved_word_table_name(self, mem_db):
+        """Test that SQL reserved words are rejected as table names."""
+        with pytest.raises(ValueError, match="is a reserved word"):
+            mem_db.create_table("select", [('notes', 'TEXT')], False, True)
+
+    def test_invalid_characters_table_name(self, mem_db):
+        """Test that invalid characters in table names are rejected."""
+        with pytest.raises(ValueError, match="Invalid SQL table name"):
+            mem_db.create_table("test-table", [('notes', 'TEXT')], False, True)
+
+    def test_malicious_column_name(self, mem_db):
+        """Test that malicious SQL injection in column names is rejected."""
+        with pytest.raises(ValueError, match="Invalid SQL column name"):
+            mem_db.create_table("test_table", [('notes\'; DROP TABLE users; --', 'TEXT')], False, True)
+
+    def test_invalid_column_type(self, mem_db):
+        """Test that invalid/malicious column types are rejected."""
+        with pytest.raises(ValueError, match="Unsupported column type"):
+            mem_db.create_table("test_table", [('notes', 'MALICIOUS_TYPE; DROP TABLE users; --')], False, True)
+
+    def test_valid_table_creation(self, mem_db):
+        """Test that valid table creation works correctly."""
+        # This should not raise any exception
+        mem_db.create_table("test_table", [('notes', 'TEXT'), ('count', 'INTEGER')], False, True)
+        
+        # Verify table was created by attempting to insert data
+        mem_db.execute("INSERT INTO test_table (notes, count) VALUES (?, ?);", ("test note", 42))
+        mem_db.commit()
+        
+        # Verify data was inserted
+        mem_db.execute("SELECT * FROM test_table;")
+        result = mem_db.fetchall()
+        assert len(result) == 1
+        assert result[0][1] == "test note"  # Column 0 is auto-increment id
+        assert result[0][2] == 42
+
+    def test_malicious_drop_table_name(self, mem_db):
+        """Test that malicious SQL injection in drop table is rejected."""
+        # First create a valid table
+        mem_db.create_table("test_table", [('notes', 'TEXT')], False, True)
+        
+        # Then try to drop with malicious name
+        with pytest.raises(ValueError, match="Invalid SQL table name"):
+            mem_db.drop_table("test_table'; DROP TABLE sqlite_master; --", False)
+
+    def test_valid_drop_table(self, mem_db):
+        """Test that valid table dropping works correctly."""
+        # Create a table first
+        mem_db.create_table("test_table", [('notes', 'TEXT')], False, True)
+        
+        # Verify it exists by checking sqlite_master
+        mem_db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", ("test_table",))
+        assert mem_db.fetchone() is not None
+        
+        # Drop the table
+        mem_db.drop_table("test_table", False)
+        
+        # Verify it's gone
+        mem_db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", ("test_table",))
+        assert mem_db.fetchone() is None
+
+    def test_empty_table_name_create(self, mem_db):
+        """Test that empty table names are rejected."""
+        with pytest.raises(ValueError, match="cannot be an empty string"):
+            mem_db.create_table("", [('notes', 'TEXT')], False, True)
+
+    def test_empty_table_name_drop(self, mem_db):
+        """Test that empty table names are rejected in drop operations."""
+        with pytest.raises(ValueError, match="cannot be an empty string"):
+            mem_db.drop_table("", False)
+
+    def test_empty_column_name(self, mem_db):
+        """Test that empty column names are rejected."""
+        with pytest.raises(ValueError, match="cannot be empty"):
+            mem_db.create_table("test_table", [('', 'TEXT')], False, True)
+
+    def test_column_name_with_numbers(self, mem_db):
+        """Test that column names with numbers are allowed."""
+        mem_db.create_table("test_table", [('column1', 'TEXT'), ('column_2', 'INTEGER')], False, True)
+
+    def test_table_name_with_underscore(self, mem_db):
+        """Test that table names starting with underscore are allowed."""
+        mem_db.create_table("_test_table", [('notes', 'TEXT')], False, True)
+
+    def test_valid_column_types(self, mem_db):
+        """Test that all supported column types work correctly."""
+        valid_types = [
+            ('text_col', 'TEXT'),
+            ('int_col', 'INTEGER'),
+            ('real_col', 'REAL'),
+            ('blob_col', 'BLOB'),
+            ('numeric_col', 'NUMERIC'),
+            ('varchar_col', 'VARCHAR(255)'),
+            ('decimal_col', 'DECIMAL(10,2)')
+        ]
+        
+        mem_db.create_table("type_test_table", valid_types, False, True)
+
+    def test_case_insensitive_reserved_words(self, mem_db):
+        """Test that reserved words are caught regardless of case."""
+        with pytest.raises(ValueError, match="is a reserved word"):
+            mem_db.create_table("SELECT", [('notes', 'TEXT')], False, True)
+        
+        with pytest.raises(ValueError, match="is a reserved word"):
+            mem_db.create_table("Select", [('notes', 'TEXT')], False, True)
+
+    def test_raise_if_exists_functionality(self, mem_db):
+        """Test the raise_if_exists parameter works correctly."""
+        # Create a table
+        mem_db.create_table("test_table", [('notes', 'TEXT')], False, True)
+        
+        # Try to create the same table with raise_if_exists=True (should fail)
+        with pytest.raises(ValueError, match="already exists"):
+            mem_db.create_table("test_table", [('notes', 'TEXT')], False, True)
+        
+        # Try to create the same table with raise_if_exists=False (should succeed)
+        mem_db.create_table("test_table", [('notes', 'TEXT')], False, False)
+
+    def test_raise_if_not_exists_functionality(self, mem_db):
+        """Test the raise_if_not_exists parameter works correctly."""
+        # Try to drop non-existent table with raise_if_not_exists=True (should fail)
+        with pytest.raises(ValueError, match="does not exist"):
+            mem_db.drop_table("nonexistent_table", True)
+        
+        # Try to drop non-existent table with raise_if_not_exists=False (should succeed)
+        mem_db.drop_table("nonexistent_table", False)
+
+    def test_special_characters_rejection(self, mem_db):
+        """Test that various special characters are properly rejected."""
+        special_chars = [
+            "table;name",
+            "table'name",
+            'table"name',
+            "table name",   # space
+            "table-name",   # hyphen
+            "table.name",   # dot
+            "table(name)",  # parentheses
+            "table[name]",  # brackets
+            "table{name}",  # braces
+            "table@name",   # at symbol
+            "table#name",   # hash
+            "table$name",   # dollar (should be rejected by our implementation)
+        ]
+        
+        for table_name in special_chars:
+            with pytest.raises(ValueError, match="Invalid SQL"):
+                mem_db.create_table(table_name, [('notes', 'TEXT')], False, True)