python.sqlalchemy.security.sqlalchemy-sql-injection.sqlalchemy-sql-injection

Verifed by r2c
Community Favorite
profile photo of returntocorpreturntocorp
Author
99,085
Download Count*

Distinct, Having, Group_by, Order_by, and Filter in SQLAlchemy can cause sql injections if the developer inputs raw SQL into the before-mentioned clauses. This pattern captures relevant cases in which the developer inputs raw SQL into the distinct, having, group_by, order_by or filter clauses and injects user-input into the raw SQL with any function besides "bindparams". Use bindParams to securely bind user-input to SQL statements.

Run Locally

Run in CI

Defintion

rules:
  - id: sqlalchemy-sql-injection
    patterns:
      - pattern-either:
          - pattern: |
              def $FUNC(...,$VAR,...):
                ...
                $SESSION.query(...).$SQLFUNC("...".$FORMATFUNC(...,$VAR,...))
          - pattern: >
              def $FUNC(...,$VAR,...):
                ...
                $SESSION.query.join(...).$SQLFUNC("...".$FORMATFUNC(...,$VAR,...))
          - pattern: |
              def $FUNC(...,$VAR,...):
                ...
                $SESSION.query.$SQLFUNC("...".$FORMATFUNC(...,$VAR,...))
          - pattern: |
              def $FUNC(...,$VAR,...):
                ...
                query.$SQLFUNC("...".$FORMATFUNC(...,$VAR,...))
      - metavariable-regex:
          metavariable: $SQLFUNC
          regex: (group_by|order_by|distinct|having|filter)
      - metavariable-regex:
          metavariable: $FORMATFUNC
          regex: (?!bindparams)
    message: Distinct, Having, Group_by, Order_by, and Filter in SQLAlchemy can
      cause sql injections if the developer inputs raw SQL into the
      before-mentioned clauses. This pattern captures relevant cases in which
      the developer inputs raw SQL into the distinct, having, group_by, order_by
      or filter clauses and injects user-input into the raw SQL with any
      function besides "bindparams". Use bindParams to securely bind user-input
      to SQL statements.
    fix-regex:
      regex: format
      replacement: bindparams
    languages:
      - python
    severity: WARNING
    metadata:
      cwe:
        - "CWE-89: Improper Neutralization of Special Elements used in an SQL
          Command ('SQL Injection')"
      category: security
      technology:
        - sqlalchemy
      owasp:
        - A01:2017 - Injection
        - A03:2021 - Injection
      references:
        - https://owasp.org/Top10/A03_2021-Injection
      cwe2022-top25: true
      cwe2021-top25: true
      subcategory:
        - vuln
      likelihood: LOW
      impact: HIGH
      confidence: MEDIUM
      license: Commons Clause License Condition v1.0[LGPL-2.1-only]

Examples

sqlalchemy-sql-injection.py

# ruleid: sqlalchemy-sql-injection
def bad1(var):
    session.query(MyClass).distinct("foo={}".format(var))

# ruleid: sqlalchemy-sql-injection
def bad2(var):
    query = cls.query.join(DeploymentPermission).having(
        "oops{}".format(var)
    )

# ruleid: sqlalchemy-sql-injection
def bad3(var):
    query = cls.query.group_by(
        "oops{}".format(var)
    )

# ruleid: sqlalchemy-sql-injection
def bad4(var):
    query = query.order_by("oops{}".format(var)).limit(limit)

# ruleid: sqlalchemy-sql-injection
def bad5(var):
    query = query.filter("oops{}".format(var)).limit(limit)

# ok: sqlalchemy-sql-injection
def ok1(cls, deployment: "Deployment", token_name: str) -> str:
    query = cls.query(DeploymentPermission).distinct(
        cls.id == DeploymentPermission.token_id,
    )

# ok: sqlalchemy-sql-injection
def ok2(cls, deployment: "Deployment", token_name: str) -> str:
    query = cls.query.join(DeploymentPermission).having(
        cls == hello
    )

# ok: sqlalchemy-sql-injection
def ok3(var):
    query = cls.query.group_by(
        var=3)

# ok: sqlalchemy-sql-injection
def ok4(var):
    query = query.order_by(desc(Scan.started_at)).limit(limit)

# ok: sqlalchemy-sql-injection
def ok5(var):
    query = query.filter(var==5).limit(limit)

# ok: sqlalchemy-sql-injection
def ok6(var):
    query = query.filter("oops{}".bindparams(var)).limit(limit)