sqlalchemy.md

SQLAlchemy Security Guide (TorusGuard v0.4.0)

Scope: Data-layer security guidance for Python applications using SQLAlchemy (ORM & Core). Covers parameterized query construction, text() bindings, tenant query scoping, bulk update mass assignment, session lifecycle, and transaction isolation.

---

🔍 1. Parameterized Queries & text() (TG-INPUT-003)

SQLAlchemy ORM query methods (filter(), filter_by()) parameterize queries automatically. However, using raw SQL clauses with text() requires explicit parameter binding.

❌ Unsafe Pattern: String Interpolation with text()

# VULNERABLE: Direct f-string interpolation into raw SQL
from sqlalchemy import text

def find_user_by_email(session, email: str):
    query = text(f"SELECT * FROM users WHERE email = '{email}'")  # ❌ SQL Injection
    return session.execute(query).fetchall()

✅ Safe Pattern: Parameterized Binding

# SAFE: Named parameter binding
from sqlalchemy import text

def find_user_by_email(session, email: str):
    query = text("SELECT * FROM users WHERE email = :email")  # ✅ Parameterized
    return session.execute(query, {"email": email}).fetchall()

💡 Safe LIKE Search Queries Pattern

Never concatenate wildcards directly into the SQL string:

# ❌ UNSAFE: query = text(f"SELECT * FROM docs WHERE title LIKE '%{user_term}%'")

# ✅ SAFE: Bind wildcard parameter in the dictionary mapping
def search_documents(session, user_term: str):
    query = text("SELECT * FROM docs WHERE title LIKE :pattern")
    return session.execute(query, {"pattern": f"%{user_term}%"}).fetchall()

---

👤 2. Query Scoping & Multi-Tenant Authorization (TG-AUTH-007)

An ORM query is not secure if it omits tenant or user boundaries.

❌ Unsafe Pattern

# VULNERABLE: Any caller can access any order by ID
def get_order(session, order_id: int):
    return session.query(Order).filter(Order.id == order_id).first()

✅ Safe Pattern

# SAFE: Scope query strictly to the current tenant / user
def get_order(session, order_id: int, current_user_id: int):
    return session.query(Order).filter(
        Order.id == order_id, 
        Order.user_id == current_user_id
    ).first()

---

📝 3. Bulk Updates & Mass Assignment (TG-AUTH-006)

Avoid passing raw dictionary payloads into .update().

❌ Unsafe Pattern

# VULNERABLE: Accepts arbitrary dict fields directly into update
def update_profile(session, user_id: int, client_data: dict):
    session.query(User).filter(User.id == user_id).update(client_data)

✅ Safe Pattern

# SAFE: Explicit column whitelist
ALLOWED_FIELDS = {'bio', 'display_name', 'phone_number'}

def update_profile(session, user_id: int, client_data: dict):
    sanitized_updates = {k: v for k, v in client_data.items() if k in ALLOWED_FIELDS}
    session.query(User).filter(User.id == user_id).update(sanitized_updates)

---

🔄 4. Transaction Boundaries & Session Lifecycle

• Always manage session lifecycles using context managers (with Session() as session: or FastAPI dependencies) to ensure sessions and connections are closed.
• Ensure failed operations roll back cleanly to avoid leaving database connections in aborted transaction states.

---

📋 Manual Review Checklist for SQLAlchemy

• All text() constructs use :param bindings rather than f-strings or .format().
• LIKE searches bind wildcards in the parameter values rather than concatenating into SQL.
• Multi-tenant queries include tenant/user ID filter conditions.
• Bulk .update() queries validate and whitelist editable fields.
• Sessions are scoped to request lifecycles and close cleanly upon completion.
• Connection pool sizes and timeouts are configured to prevent connection pool exhaustion.