Prevent SQL Injection with SQLAlchemy's Parameterized Queries

from sqlalchemy import create_engine, Column, Integer, String, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 1. Database Setup (using SQLite for example)
DATABASE_URL = "sqlite:///./sql_injection_safe.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 2. Define a simple ORM model
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String)

# Create tables (if they don't exist)
Base.metadata.create_all(bind=engine)

# 3. Safe Interaction with ORM (Recommended)
def get_user_by_username_orm(username: str):
    db = SessionLocal()
    try:
        # SQLAlchemy ORM automatically parameterizes queries
        user = db.query(User).filter(User.username == username).first()
        return user
    finally:
        db.close()

# 4. Safe Interaction with SQLAlchemy Core (Parameterized Text)
def get_user_by_username_core(username: str):
    db = SessionLocal()
    try:
        # Using text() with .bindparams() for explicit parameterization
        query = text("SELECT id, username, email FROM users WHERE username = :user_param")
        result = db.execute(query.bindparams(user_param=username)).fetchone()
        if result:
            # Convert row to a dict or User object if needed
            return User(id=result.id, username=result.username, email=result.email)
        return None
    finally:
        db.close()

# 5. Example Usage
if __name__ == "__main__":
    db = SessionLocal()
    # Add a user for testing
    if not db.query(User).filter(User.username == "alice").first():
        db.add(User(username="alice", email="[email protected]"))
        db.commit()
    db.close()

    # Test ORM method
    print("
--- ORM Method ---")
    found_user_orm = get_user_by_username_orm("alice")
    if found_user_orm:
        print(f"Found user (ORM): {found_user_orm.username}, {found_user_orm.email}")
    else:
        print("User not found (ORM).")

    # Test Core method with parameter
    print("
--- Core Method ---")
    found_user_core = get_user_by_username_core("alice")
    if found_user_core:
        print(f"Found user (Core): {found_user_core.username}, {found_user_core.email}")
    else:
        print("User not found (Core).")

    # Example of a *potential* malicious input (handled safely by parameterization)
    malicious_input = "alice' OR '1'='1"; --"
    print("
--- Malicious Input Test (ORM) ---")
    found_user_malicious_orm = get_user_by_username_orm(malicious_input)
    if found_user_malicious_orm:
        print(f"Found user (ORM, malicious input): {found_user_malicious_orm.username}, {found_user_malicious_orm.email}")
    else:
        print("User not found with malicious input (ORM, safe).")

    print("
--- Malicious Input Test (Core) ---")
    found_user_malicious_core = get_user_by_username_core(malicious_input)
    if found_user_malicious_core:
        print(f"Found user (Core, malicious input): {found_user_malicious_core.username}, {found_user_malicious_core.email}")
    else:
        print("User not found with malicious input (Core, safe).")