docs and sql injection

Author: millerjp

SQL_INJECTION_FIXES.md

@@ -0,0 +1,125 @@
+# SQL Injection Vulnerability Fixes
+
+## Overview
+
+This document details the SQL injection vulnerabilities found and fixed in the async-python-cassandra-client codebase.
+
+## Critical Vulnerabilities Fixed
+
+### 1. Dynamic UPDATE Query Construction
+**Severity**: CRITICAL
+**Location**: `examples/fastapi_app/main.py:578`
+
+**Vulnerable Code**:
+```python
+query = f"UPDATE users SET {', '.join(update_fields)} WHERE id = ?"
+```
+
+**Issue**: User-controlled field names directly interpolated into SQL query.
+
+**Fix**: Replaced with static prepared statements for all field combinations:
+```python
+# Prepare all possible UPDATE combinations
+update_name_query = await session.prepare("UPDATE users SET name = ? WHERE id = ?")
+update_email_query = await session.prepare("UPDATE users SET email = ? WHERE id = ?")
+update_age_query = await session.prepare("UPDATE users SET age = ? WHERE id = ?")
+# ... etc
+```
+
+### 2. LIMIT Clause Injection
+**Severity**: HIGH
+**Locations**:
+- `examples/fastapi_app/main_enhanced.py:258`
+- `examples/fastapi_app/main_enhanced.py:311`
+
+**Vulnerable Code**:
+```python
+result = await session.execute(f"SELECT * FROM users LIMIT {limit}", timeout=timeout)
+```
+
+**Fix**: Use prepared statements:
+```python
+list_users_query = await session.prepare("SELECT * FROM users LIMIT ?")
+result = await session.execute(list_users_query, [limit])
+```
+
+### 3. Table Name Injection
+**Severity**: HIGH
+**Location**: `examples/export_large_table.py` (multiple)
+
+**Vulnerable Code**:
+```python
+result = await session.execute(f"SELECT COUNT(*) FROM {table_name}")
+```
+
+**Fix**: Validate table names against system schema:
+```python
+# Validate table exists in system schema
+validation_query = await session.prepare(
+    "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?"
+)
+result = await session.execute(validation_query, [keyspace, table_name])
+if not result.one():
+    raise ValueError(f"Table {table_name} does not exist in keyspace {keyspace}")
+```
+
+### 4. Keyspace Interpolation
+**Severity**: MEDIUM
+**Location**: `examples/fastapi_app/main.py` (multiple)
+
+**Vulnerable Code**:
+```python
+query = f"SELECT * FROM {keyspace}.users WHERE age = ? ALLOW FILTERING"
+```
+
+**Fix**: Hardcode keyspace names:
+```python
+query = "SELECT * FROM user_management.users WHERE age = ? ALLOW FILTERING"
+```
+
+## Prevention Measures
+
+### 1. New Test Suite
+Created `tests/unit/test_sql_injection_protection.py` to verify:
+- All queries use prepared statements
+- No string interpolation in SQL
+- Proper validation of identifiers
+- Secure handling of dynamic queries
+
+### 2. Code Patterns to Avoid
+```python
+# ❌ NEVER DO THIS
+query = f"SELECT * FROM {table} WHERE {column} = {value}"
+query = "SELECT * FROM users WHERE id = " + user_id
+query = "SELECT * FROM users LIMIT %s" % limit
+
+# ✅ ALWAYS DO THIS
+query = await session.prepare("SELECT * FROM users WHERE id = ?")
+result = await session.execute(query, [user_id])
+```
+
+### 3. Best Practices
+1. **Always use prepared statements** for any user input
+2. **Validate identifiers** against system schema
+3. **Never use string interpolation** in queries
+4. **Parameterize everything**, including LIMIT clauses
+5. **Review all dynamic SQL** construction carefully
+
+## Impact
+
+- All example applications now secure against SQL injection
+- Test coverage ensures future code maintains security
+- Documentation updated to emphasize secure patterns
+- No functionality lost - all features work with secure implementation
+
+## Verification
+
+Run the SQL injection protection tests:
+```bash
+pytest tests/unit/test_sql_injection_protection.py -v
+```
+
+## References
+
+- [OWASP SQL Injection Prevention](https://cheatsheetseries.owasp.org/cheatsheets/SQL_Injection_Prevention_Cheat_Sheet.html)
+- [CQL Prepared Statements](https://docs.datastax.com/en/developer/python-driver/3.25/getting_started/#prepared-statements)

examples/export_large_table.py

@@ -32,20 +32,30 @@
     logger.warning("aiofiles not installed - using synchronous file I/O")
 
 
-async def count_table_rows(session, table_name: str) -> int:
+async def count_table_rows(session, keyspace: str, table_name: str) -> int:
     """Count total rows in a table (approximate for large tables)."""
     # Note: COUNT(*) can be slow on large tables
     # Consider using token ranges for very large tables
-    result = await session.execute(f"SELECT COUNT(*) FROM {table_name}")
+    # Using system schema to validate table exists and avoid SQL injection
+    validation_query = await session.execute(
+        "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+        [keyspace, table_name],
+    )
+    if not validation_query.one():
+        raise ValueError(f"Table {keyspace}.{table_name} does not exist")
+
+    # Safe to use table name after validation - but still use qualified name
+    # In production, consider using prepared statements even for COUNT queries
+    result = await session.execute(f"SELECT COUNT(*) FROM {keyspace}.{table_name}")
     return result.one()[0]
 
 
-async def export_table_async(session, table_name: str, output_file: str):
+async def export_table_async(session, keyspace: str, table_name: str, output_file: str):
     """Export table using async file I/O (requires aiofiles)."""
-    logger.info(f"Starting async export of {table_name} to {output_file}")
+    logger.info(f"Starting async export of {keyspace}.{table_name} to {output_file}")
 
     # Get approximate row count for progress tracking
-    total_rows = await count_table_rows(session, table_name)
+    total_rows = await count_table_rows(session, keyspace, table_name)
     logger.info(f"Table has approximately {total_rows:,} rows")
 
     # Configure streaming with progress callback
@@ -67,8 +77,16 @@ def progress_callback(page_num: int, rows_so_far: int):
     start_time = datetime.now()
 
     # CRITICAL: Use context manager for streaming to prevent memory leaks
+    # Validate table exists before streaming
+    validation_query = await session.execute(
+        "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+        [keyspace, table_name],
+    )
+    if not validation_query.one():
+        raise ValueError(f"Table {keyspace}.{table_name} does not exist")
+
     async with await session.execute_stream(
-        f"SELECT * FROM {table_name}", stream_config=config
+        f"SELECT * FROM {keyspace}.{table_name}", stream_config=config
     ) as result:
         # Export to CSV
         async with aiofiles.open(output_file, "w", newline="") as f:
@@ -111,13 +129,13 @@ def progress_callback(page_num: int, rows_so_far: int):
     logger.info(f"- File size: {os.path.getsize(output_file):,} bytes")
 
 
-def export_table_sync(session, table_name: str, output_file: str):
+def export_table_sync(session, keyspace: str, table_name: str, output_file: str):
     """Export table using synchronous file I/O."""
-    logger.info(f"Starting sync export of {table_name} to {output_file}")
+    logger.info(f"Starting sync export of {keyspace}.{table_name} to {output_file}")
 
     async def _export():
         # Get approximate row count
-        total_rows = await count_table_rows(session, table_name)
+        total_rows = await count_table_rows(session, keyspace, table_name)
         logger.info(f"Table has approximately {total_rows:,} rows")
 
         # Configure streaming
@@ -133,8 +151,16 @@ async def _export():
         start_time = datetime.now()
 
         # Use context manager for proper streaming cleanup
+        # Validate table exists before streaming
+        validation_query = await session.execute(
+            "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+            [keyspace, table_name],
+        )
+        if not validation_query.one():
+            raise ValueError(f"Table {keyspace}.{table_name} does not exist")
+
         async with await session.execute_stream(
-            f"SELECT * FROM {table_name}", stream_config=config
+            f"SELECT * FROM {keyspace}.{table_name}", stream_config=config
         ) as result:
             # Export to CSV synchronously
             with open(output_file, "w", newline="") as f:
@@ -272,9 +298,13 @@ async def main():
 
         # Export using async I/O if available
         if ASYNC_FILE_IO:
-            await export_table_async(session, "products", str(output_dir / "products_async.csv"))
+            await export_table_async(
+                session, "export_example", "products", str(output_dir / "products_async.csv")
+            )
         else:
-            await export_table_sync(session, "products", str(output_dir / "products_sync.csv"))
+            await export_table_sync(
+                session, "export_example", "products", str(output_dir / "products_sync.csv")
+            )
 
         # Cleanup (optional)
         logger.info("\nCleaning up...")