hohn/codeql
1
0
Fork 0
mirror of https://github.com/github/codeql.git synced 2026-05-03 04:39:29 +02:00
Code Issues Packages Projects Releases Wiki Activity

Revert "deprecate SqlConstruction"

Browse Source 
This reverts commit c0eca0d09a.
This commit is contained in:
Erik Krogh Kristensen
2022-05-03 22:49:33 +02:00
parent 1062aae21c
commit 571fc3e73b
15 changed files with 85 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
python/ql/test/library-tests/frameworks/aiomysql/test.py
View File

@@ -5,24 +5,24 @@ async def test_cursor():
# Create connection directly
conn = await aiomysql.connect()
cur = await conn.cursor()
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Create connection via pool
async with aiomysql.create_pool() as pool:
# Create Cursor via Connection
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Create Cursor directly
async with pool.cursor() as cur:
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# variants using as few `async with` as possible
pool = await aiomysql.create_pool()
conn = await pool.acquire()
cur = await conn.cursor()
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Test SQLAlchemy integration
from aiomysql.sa import create_engine
@@ -30,4 +30,4 @@ from aiomysql.sa import create_engine
async def test_engine():
engine = await create_engine()
conn = await engine.acquire()
await conn.execute("sql") # $ getSql="sql"
await conn.execute("sql") # $ getSql="sql" constructedSql="sql"

10
python/ql/test/library-tests/frameworks/aiopg/test.py
View File

@@ -5,24 +5,24 @@ async def test_cursor():
# Create connection directly
conn = await aiopg.connect()
cur = await conn.cursor()
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Create connection via pool
async with aiopg.create_pool() as pool:
# Create Cursor via Connection
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Create Cursor directly
async with pool.cursor() as cur:
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# variants using as few `async with` as possible
pool = await aiopg.create_pool()
conn = await pool.acquire()
cur = await conn.cursor()
await cur.execute("sql") # $ getSql="sql"
await cur.execute("sql") # $ getSql="sql" constructedSql="sql"
# Test SQLAlchemy integration
from aiopg.sa import create_engine
@@ -30,4 +30,4 @@ from aiopg.sa import create_engine
async def test_engine():
engine = await create_engine()
conn = await engine.acquire()
await conn.execute("sql") # $ getSql="sql"
await conn.execute("sql") # $ getSql="sql" constructedSql="sql"

6
python/ql/test/library-tests/frameworks/asyncpg/test.py
View File

@@ -43,20 +43,20 @@ async def test_cursor():
try:
async with conn.transaction():
cursor = await conn.cursor("sql") # $ getSql="sql"
cursor = await conn.cursor("sql") # $ getSql="sql" constructedSql="sql"
await cursor.fetch()
pstmt = await conn.prepare("psql") # $ getSql="psql"
pcursor = await pstmt.cursor() # $ getSql="psql"
await pcursor.fetch()
async for record in conn.cursor("sql"): # $ getSql="sql"
async for record in conn.cursor("sql"): # $ getSql="sql" constructedSql="sql"
pass
async for record in pstmt.cursor(): # $ getSql="psql"
pass
cursor_factory = conn.cursor("sql") # $ getSql="sql"
cursor_factory = conn.cursor("sql") # $ constructedSql="sql"
cursor = await cursor_factory # $ getSql="sql"
pcursor_factory = pstmt.cursor()

6
python/ql/test/library-tests/frameworks/flask_sqlalchemy/SqlExecution.py
View File

@@ -12,7 +12,7 @@ db = SQLAlchemy(app)
# - https://github.com/pallets/flask-sqlalchemy/blob/931ec00d1e27f51508e05706eef41cc4419a0b32/src/flask_sqlalchemy/__init__.py#L765
# - https://github.com/pallets/flask-sqlalchemy/blob/931ec00d1e27f51508e05706eef41cc4419a0b32/src/flask_sqlalchemy/__init__.py#L99-L109
assert str(type(db.text("Foo"))) == "<class 'sqlalchemy.sql.elements.TextClause'>" # $ getSql="Foo"
assert str(type(db.text("Foo"))) == "<class 'sqlalchemy.sql.elements.TextClause'>" # $ constructedSql="Foo"
# also has engine/session instantiated
@@ -44,8 +44,8 @@ assert result.fetchall() == [("Foo",)]
# text
t = db.text("foo") # $ getSql="foo"
t = db.text("foo") # $ constructedSql="foo"
assert isinstance(t, sqlalchemy.sql.expression.TextClause)
t = db.text(text="foo") # $ getSql="foo"
t = db.text(text="foo") # $ constructedSql="foo"
assert isinstance(t, sqlalchemy.sql.expression.TextClause)

2
python/ql/test/library-tests/frameworks/sqlalchemy/SqlExecution.py
View File

@@ -46,7 +46,7 @@ with engine.begin() as connection:
connection.execute("some sql") # $ getSql="some sql"
# Injection requiring the text() taint-step
t = text("some sql") # $ getSql="some sql"
t = text("some sql") # $ constructedSql="some sql"
session.query(User).filter(t)
session.query(User).group_by(User.id).having(t)
session.query(User).group_by(t).first()

6
python/ql/test/library-tests/frameworks/sqlalchemy/new_tests.py
View File

@@ -6,7 +6,7 @@ import sqlalchemy.orm
# either v1.4 or v2.0, such that we cover both.
raw_sql = "select 'FOO'"
text_sql = sqlalchemy.text(raw_sql) # $ getSql=raw_sql
text_sql = sqlalchemy.text(raw_sql) # $ constructedSql=raw_sql
Base = sqlalchemy.orm.declarative_base()
@@ -176,7 +176,7 @@ assert session.query(For14).all()[0].id == 14
# and now we can do the actual querying
text_foo = sqlalchemy.text("'FOO'") # $ getSql="'FOO'"
text_foo = sqlalchemy.text("'FOO'") # $ constructedSql="'FOO'"
# filter_by is only vulnerable to injection if sqlalchemy.text is used, which is evident
# from the logs produced if this file is run
@@ -305,7 +305,7 @@ with engine.connect() as conn:
assert scalar_result == "FOO"
# This is a contrived example
select = sqlalchemy.select(sqlalchemy.text("'BAR'")) # $ getSql="'BAR'"
select = sqlalchemy.select(sqlalchemy.text("'BAR'")) # $ constructedSql="'BAR'"
result = conn.execute(select) # $ getSql=select
assert result.fetchall() == [("BAR",)]

16
python/ql/test/library-tests/frameworks/sqlalchemy/taint_test.py
View File

@@ -8,14 +8,14 @@ def test_taint():
ensure_tainted(ts) # $ tainted
t1 = sqlalchemy.text(ts) # $ getSql=ts
t2 = sqlalchemy.text(text=ts) # $ getSql=ts
t3 = sqlalchemy.sql.text(ts) # $ getSql=ts
t4 = sqlalchemy.sql.text(text=ts) # $ getSql=ts
t5 = sqlalchemy.sql.expression.text(ts) # $ getSql=ts
t6 = sqlalchemy.sql.expression.text(text=ts) # $ getSql=ts
t7 = sqlalchemy.sql.expression.TextClause(ts) # $ getSql=ts
t8 = sqlalchemy.sql.expression.TextClause(text=ts) # $ getSql=ts
t1 = sqlalchemy.text(ts) # $ constructedSql=ts
t2 = sqlalchemy.text(text=ts) # $ constructedSql=ts
t3 = sqlalchemy.sql.text(ts) # $ constructedSql=ts
t4 = sqlalchemy.sql.text(text=ts) # $ constructedSql=ts
t5 = sqlalchemy.sql.expression.text(ts) # $ constructedSql=ts
t6 = sqlalchemy.sql.expression.text(text=ts) # $ constructedSql=ts
t7 = sqlalchemy.sql.expression.TextClause(ts) # $ constructedSql=ts
t8 = sqlalchemy.sql.expression.TextClause(text=ts) # $ constructedSql=ts
# Since we flag user-input to a TextClause with its' own query, we don't want to
# have a taint-step for it as that would lead to us also giving an alert for normal