fix: address code quality issues identified in GitHub Advanced Security review

- Remove unused imports across multiple files
- Improve try-except-continue patterns with proper logging
- Fix context manager usage for file operations
- Remove unused noqa directives
- Add proper SQL injection warning suppressions with noqa comments
- Maintain all SchemaPin functionality and backward compatibility

Author: jaschadub

examples/schemapin/advanced_usage.py

@@ -164,7 +164,9 @@ async def _attempt_key_recovery(self, domain: str) -> str | None:
                 print(f"Trying alternative endpoint: {endpoint}")
                 # Simulate discovery attempt
                 await asyncio.sleep(0.1)
-            except Exception:
+            except Exception as e:
+                # Log the exception for debugging
+                print(f"Failed to connect to {endpoint}: {e}")
                 continue
 
         return None

src/mockloop_mcp/generator.py

@@ -460,23 +460,23 @@ async def export_data():
         import io
         import zipfile
         from fastapi.responses import StreamingResponse
-        
+
         try:
             # Create in-memory zip file
             zip_buffer = io.BytesIO()
-            
+
             with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
                 # Export request logs
                 conn = sqlite3.connect(str(DB_PATH))
                 conn.row_factory = sqlite3.Row
                 cursor = conn.cursor()
-                
+
                 # Get all request logs
                 cursor.execute('''
                     SELECT * FROM request_logs ORDER BY created_at DESC
                 ''')
                 logs = cursor.fetchall()
-                
+
                 # Convert to JSON
                 logs_data = []
                 for row in logs:
@@ -487,28 +487,28 @@ async def export_data():
                         except:
                             pass
                     logs_data.append(log_entry)
-                
+
                 # Add logs to zip
                 zip_file.writestr("request_logs.json", json.dumps(logs_data, indent=2))
-                
+
                 # Export performance metrics if available
                 try:
                     cursor.execute('SELECT * FROM performance_metrics ORDER BY recorded_at DESC')
                     metrics = [dict(row) for row in cursor.fetchall()]
                     zip_file.writestr("performance_metrics.json", json.dumps(metrics, indent=2))
                 except:
                     pass
-                
+
                 # Export test sessions if available
                 try:
                     cursor.execute('SELECT * FROM test_sessions ORDER BY created_at DESC')
                     sessions = [dict(row) for row in cursor.fetchall()]
                     zip_file.writestr("test_sessions.json", json.dumps(sessions, indent=2))
                 except:
                     pass
-                
+
                 conn.close()
-                
+
                 # Add metadata
                 metadata = {
                     "export_timestamp": time.strftime('%Y-%m-%dT%H:%M:%S%z', time.gmtime()),
@@ -521,24 +521,24 @@ async def export_data():
                     }
                 }
                 zip_file.writestr("metadata.json", json.dumps(metadata, indent=2))
-            
+
             zip_buffer.seek(0)
-            
+
             # Return as streaming response
             def iter_zip():
                 yield zip_buffer.getvalue()
-            
+
             timestamp = time.strftime('%Y%m%d_%H%M%S', time.gmtime())
             filename = f"mockloop_export_{timestamp}.zip"
-            
+
             print(f"DEBUG ADMIN: Exported {len(logs_data)} logs to {filename}")
-            
+
             return StreamingResponse(
                 iter_zip(),
                 media_type="application/zip",
                 headers={"Content-Disposition": f"attachment; filename={filename}"}
             )
-            
+
         except Exception as e:
             print(f"DEBUG ADMIN: Error exporting data: {e}")
             return {"error": str(e)}
@@ -549,11 +549,11 @@ async def get_request_logs(limit: int = 100, offset: int = 0):
             conn = sqlite3.connect(str(DB_PATH))
             conn.row_factory = sqlite3.Row
             cursor = conn.cursor()
-            
+
             # Get total count
             cursor.execute("SELECT COUNT(*) FROM request_logs")
             total_count = cursor.fetchone()[0]
-            
+
             # Get paginated logs with all available columns
             cursor.execute('''
                 SELECT id, timestamp, type, method, path, status_code, process_time_ms,
@@ -564,7 +564,7 @@ async def get_request_logs(limit: int = 100, offset: int = 0):
                 ORDER BY created_at DESC
                 LIMIT ? OFFSET ?
             ''', (limit, offset))
-            
+
             logs = []
             for row in cursor.fetchall():
                 log_entry = {
@@ -590,11 +590,11 @@ async def get_request_logs(limit: int = 100, offset: int = 0):
                     "created_at": row["created_at"]
                 }
                 logs.append(log_entry)
-            
+
             conn.close()
             print(f"DEBUG ADMIN: Retrieved {len(logs)} logs from database (total: {total_count})")
             return {"logs": logs, "count": total_count}
-            
+
         except Exception as e:
             print(f"DEBUG ADMIN: Error querying database: {e}")
             return {"logs": [], "count": 0, "error": str(e)}
@@ -605,27 +605,27 @@ async def get_debug_info():
             # Get database info
             conn = sqlite3.connect(str(DB_PATH))
             cursor = conn.cursor()
-            
+
             # Check database tables and counts
             cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
             tables = [row[0] for row in cursor.fetchall()]
-            
+
             table_info = {}
             for table in tables:
                 cursor.execute(f"SELECT COUNT(*) FROM {table}")
                 count = cursor.fetchone()[0]
                 table_info[table] = count
-            
+
             # Get recent logs count
             cursor.execute("SELECT COUNT(*) FROM request_logs WHERE created_at > datetime('now', '-1 hour')")
             recent_logs = cursor.fetchone()[0]
-            
+
             # Get schema version
             cursor.execute("SELECT MAX(version) FROM schema_version")
             schema_version = cursor.fetchone()[0] or 0
-            
+
             conn.close()
-            
+
             debug_info = {
                 "status": "ok",
                 "database": {
@@ -641,13 +641,13 @@ async def get_debug_info():
                 },
                 "timestamp": time.strftime('%Y-%m-%dT%H:%M:%S%z', time.gmtime())
             }
-            
+
             print(f"DEBUG ADMIN: Debug info retrieved successfully")
             return debug_info
-            
+
         except Exception as e:
             print(f"DEBUG ADMIN: Error getting debug info: {e}")
-            return {"status": "error", "error": str(e)}"""
+            return {"status": "error", "error": str(e)}"""  # noqa: S608
         webhook_api_endpoints_str = ""
         if webhooks_enabled_bool and admin_ui_enabled_bool:
             _webhook_api_endpoints_raw = """

src/mockloop_mcp/mcp_audit_logger.py

@@ -245,7 +245,7 @@ def _handle_schema_migrations(self, cursor) -> None:
                                 sql_query = f"""
                                     INSERT INTO mcp_data_lineage ({columns_str})
                                     SELECT {columns_str} FROM mcp_data_lineage_backup
-                                """  # nosec B608  # noqa: S608
+                                """  # noqa: S608
                                 cursor.execute(sql_query)
 
                         # Drop backup table

src/mockloop_mcp/mcp_compliance.py

@@ -462,16 +462,16 @@ def purge_expired_data(self, dry_run: bool = True) -> dict[str, Any]:
                     placeholders = ",".join(["?" for _ in expired_ids])
 
                     # Delete from data lineage table
-                    sql_data_lineage = f"DELETE FROM mcp_data_lineage WHERE entry_id IN ({placeholders})"  # nosec B608  # noqa: S608
+                    sql_data_lineage = f"DELETE FROM mcp_data_lineage WHERE entry_id IN ({placeholders})"  # noqa: S608
                     cursor.execute(sql_data_lineage, expired_ids)
 
                     # Delete from compliance events table
-                    sql_compliance_events = f"DELETE FROM mcp_compliance_events WHERE entry_id IN ({placeholders})"  # nosec B608  # noqa: S608
+                    sql_compliance_events = f"DELETE FROM mcp_compliance_events WHERE entry_id IN ({placeholders})"  # noqa: S608
                     cursor.execute(sql_compliance_events, expired_ids)
 
                     # Delete from audit logs table
                     sql_audit_logs = (
-                        f"DELETE FROM mcp_audit_logs WHERE entry_id IN ({placeholders})"  # nosec B608  # noqa: S608
+                        f"DELETE FROM mcp_audit_logs WHERE entry_id IN ({placeholders})"  # noqa: S608
                     )
                     cursor.execute(sql_audit_logs, expired_ids)
 

src/mockloop_mcp/schemapin/audit.py

@@ -8,7 +8,7 @@
 import logging
 import sqlite3
 import uuid
-from datetime import datetime, timezone, UTC
+from datetime import datetime, UTC
 from typing import Any
 
 from .config import VerificationResult

src/mockloop_mcp/schemapin/key_management.py

@@ -8,7 +8,7 @@
 import logging
 import sqlite3
 import time
-from datetime import datetime, timezone, UTC
+from datetime import datetime, UTC
 from pathlib import Path
 from typing import Any
 

tests/integration/test_schemapin_integration.py

@@ -35,25 +35,25 @@ class TestSchemaPinEndToEndWorkflow(unittest.TestCase):
 
     def setUp(self):
         """Set up test environment."""
-        self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db')
-        self.temp_db.close()
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.db') as temp_db:
+            self.temp_db_name = temp_db.name
 
         self.config = SchemaPinConfig(
-            key_pin_storage_path=self.temp_db.name,
+            key_pin_storage_path=self.temp_db_name,
             policy_mode="warn",
             auto_pin_keys=False,
             trusted_domains=["trusted.example.com"],
             interactive_mode=False
         )
 
         self.interceptor = SchemaVerificationInterceptor(self.config)
-        self.key_manager = KeyPinningManager(self.temp_db.name)
+        self.key_manager = KeyPinningManager(self.temp_db_name)
         self.policy_handler = PolicyHandler(self.config)
-        self.audit_logger = SchemaPinAuditLogger(self.temp_db.name)
+        self.audit_logger = SchemaPinAuditLogger(self.temp_db_name)
 
     def tearDown(self):
         """Clean up test environment."""
-        Path(self.temp_db.name).unlink(missing_ok=True)
+        Path(self.temp_db_name).unlink(missing_ok=True)
 
     async def test_first_time_tool_verification_trusted_domain(self):
         """Test TOFU scenario with trusted domain."""
@@ -298,12 +298,12 @@ async def test_database_persistence_across_sessions(self):
         public_key = "-----BEGIN PUBLIC KEY-----\npersistent_key\n-----END PUBLIC KEY-----"
 
         # Pin key with first manager instance
-        manager1 = KeyPinningManager(self.temp_db.name)
+        manager1 = KeyPinningManager(self.temp_db_name)
         success = manager1.pin_key(tool_id, domain, public_key, {"session": "1"})
         assert success is True
 
         # Create new manager instance (simulating new session)
-        manager2 = KeyPinningManager(self.temp_db.name)
+        manager2 = KeyPinningManager(self.temp_db_name)
         retrieved_key = manager2.get_pinned_key(tool_id)
         assert retrieved_key == public_key
 
@@ -365,17 +365,17 @@ class TestSchemaPinMockLoopIntegration(unittest.TestCase):
 
     def setUp(self):
         """Set up test environment."""
-        self.temp_db = tempfile.NamedTemporaryFile(delete=False, suffix='.db')
-        self.temp_db.close()
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.db') as temp_db:
+            self.temp_db_name = temp_db.name
 
         self.config = SchemaPinConfig(
-            key_pin_storage_path=self.temp_db.name,
+            key_pin_storage_path=self.temp_db_name,
             policy_mode="warn"
         )
 
     def tearDown(self):
         """Clean up test environment."""
-        Path(self.temp_db.name).unlink(missing_ok=True)
+        Path(self.temp_db_name).unlink(missing_ok=True)
 
     def test_mcp_tool_schema_extraction(self):
         """Test extracting schemas from MCP tool functions."""
@@ -401,7 +401,7 @@ def sample_mcp_tool(query: str, database: str = "default") -> dict:
 
     async def test_integration_with_mcp_audit_system(self):
         """Test integration with MockLoop's audit system."""
-        audit_logger = SchemaPinAuditLogger(self.temp_db.name)
+        audit_logger = SchemaPinAuditLogger(self.temp_db_name)
 
         # Log various events
         await audit_logger.log_verification_attempt(
@@ -421,10 +421,10 @@ async def test_integration_with_mcp_audit_system(self):
 
     def test_database_schema_compatibility(self):
         """Test that SchemaPin tables integrate with MockLoop database."""
-        SchemaPinAuditLogger(self.temp_db.name)  # Creates tables automatically
+        SchemaPinAuditLogger(self.temp_db_name)  # Creates tables automatically
 
         # Verify tables were created
-        with sqlite3.connect(self.temp_db.name) as conn:
+        with sqlite3.connect(self.temp_db_name) as conn:
             cursor = conn.cursor()
             cursor.execute("""
                 SELECT name FROM sqlite_master