Initial commit

src/security_demo/database.py

@@ -0,0 +1,92 @@
+"""Database module - streamlined version.
+
+FINDING CLASSIFICATIONS:
+- TRUE POSITIVE (TP): Actual security vulnerability
+- FALSE POSITIVE (FP): Flagged but not a real issue in context
+- UNCERTAIN: Could be either depending on deployment context
+"""
+
+import hashlib
+import hmac
+import secrets
+from typing import Any, List, Optional
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+
+class DatabaseManager:
+    """Database operations with SQL patterns."""
+
+    def __init__(self, db_url: str = "sqlite:///app.db"):
+        self.engine = create_engine(db_url)
+        self.Session = sessionmaker(bind=self.engine)
+
+    # =========================================================================
+    # SQL INJECTION PATTERNS
+    # =========================================================================
+
+    def find_by_username_unsafe(self, username: str) -> Optional[dict]:
+        """TP: SQL injection via string formatting."""
+        session = self.Session()
+        query = f"SELECT * FROM users WHERE username = '{username}'"
+        result = session.execute(text(query))
+        return result.fetchone()
+
+    def search_users_unsafe(self, search_term: str) -> List[dict]:
+        """TP: SQL injection in LIKE clause."""
+        session = self.Session()
+        query = f"SELECT * FROM users WHERE username LIKE '%{search_term}%'"
+        result = session.execute(text(query))
+        return result.fetchall()
+
+    def find_by_id_safe(self, user_id: int) -> Optional[dict]:
+        """FP: Parameterized query is safe."""
+        session = self.Session()
+        result = session.execute(
+            text("SELECT * FROM users WHERE id = :id"), {"id": user_id}
+        )
+        return result.fetchone()
+
+    def dynamic_column_sort(self, column: str, order: str = "ASC") -> List[dict]:
+        """UNCERTAIN: Column name from allowlist but still uses f-string."""
+        allowed_columns = ["username", "email", "created_at"]
+        if column not in allowed_columns:
+            raise ValueError("Invalid column")
+        session = self.Session()
+        query = f"SELECT * FROM users ORDER BY {column} {order}"
+        result = session.execute(text(query))
+        return result.fetchall()
+
+
+class PasswordManager:
+    """Password hashing patterns."""
+
+    def hash_password_md5(self, password: str) -> str:
+        """TP: MD5 is cryptographically broken for passwords."""
+        return hashlib.md5(password.encode()).hexdigest()
+
+    def hash_password_sha1(self, password: str) -> str:
+        """TP: SHA1 is weak for password hashing."""
+        return hashlib.sha1(password.encode()).hexdigest()
+
+    def compute_file_checksum_md5(self, filepath: str) -> str:
+        """FP: MD5 acceptable for file integrity (non-security)."""
+        hasher = hashlib.md5(usedforsecurity=False)
+        with open(filepath, "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
+                hasher.update(chunk)
+        return hasher.hexdigest()
+
+    def verify_signature_sha256(
+        self, message: bytes, signature: str, key: bytes
+    ) -> bool:
+        """FP: HMAC-SHA256 for signatures is secure."""
+        expected = hmac.new(key, message, hashlib.sha256).hexdigest()
+        return hmac.compare_digest(expected, signature)
+
+    def hash_password_pbkdf2(self, password: str) -> tuple:
+        """FP: PBKDF2 is a proper password hash."""
+        salt = secrets.token_bytes(32)
+        key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 600000)
+        return key.hex(), salt.hex()