hohn/codeql
1
0
Fork 0
mirror of https://github.com/github/codeql.git synced 2026-04-30 11:15:13 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'jorgectf/python/deserialization' of https://github.com/jorgectf/codeql into jorgectf/python/deserialization

Browse Source
This commit is contained in:
jorgectf
2022-01-31 17:48:47 +01:00
parent 9ab6d21757 a1f8acc9bb
commit 080775c873
3887 changed files with 317569 additions and 114448 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
python/ql/test/2/query-tests/Imports/syntax_error/SyntaxError.expected
View File

@@ -1 +1 @@
| nonsense.py:1:14:1:14 | Syntax Error | Syntax Error (in Python 2). |
| nonsense.py:0:1:0:1 | Syntax Error | Syntax Error (in Python 2). |

16
python/ql/test/3/library-tests/with/test.expected Normal file
View File

@@ -0,0 +1,16 @@
| test.py:0:0:0:0 | Module test |
| test.py:1:1:5:2 | With |
| test.py:2:5:2:15 | CtxManager1 |
| test.py:2:5:2:17 | CtxManager1() |
| test.py:2:22:2:29 | example1 |
| test.py:3:5:3:15 | CtxManager2 |
| test.py:3:5:3:17 | CtxManager2() |
| test.py:3:5:3:29 | With |
| test.py:3:22:3:29 | example2 |
| test.py:4:5:4:15 | CtxManager3 |
| test.py:4:5:4:17 | CtxManager3() |
| test.py:4:5:4:29 | With |
| test.py:4:22:4:29 | example3 |
| test.py:4:31:4:30 | |
| test.py:4:31:4:30 | With |
| test.py:6:5:6:8 | Pass |

6
python/ql/test/3/library-tests/with/test.py Normal file
View File

@@ -0,0 +1,6 @@
with (
CtxManager1() as example1,
CtxManager2() as example2,
CtxManager3() as example3,
):
pass

3
python/ql/test/3/library-tests/with/test.ql Normal file
View File

@@ -0,0 +1,3 @@
import python
select any(AstNode n)

2
python/ql/test/3/query-tests/Imports/syntax_error/SyntaxError.expected
View File

@@ -1 +1 @@
| nonsense.py:1:2:1:2 | Syntax Error | Syntax Error (in Python 3). |
| nonsense.py:0:1:0:1 | Syntax Error | Syntax Error (in Python 3). |

22
python/ql/test/experimental/dataflow/ApiGraphs/async_test.py
View File

@@ -3,13 +3,25 @@ import pkg # $ use=moduleImport("pkg")
async def foo():
coro = pkg.async_func() # $ use=moduleImport("pkg").getMember("async_func").getReturn()
coro # $ use=moduleImport("pkg").getMember("async_func").getReturn()
result = await coro # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited()
result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited()
return result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited()
result = await coro # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
return result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
async def bar():
result = await pkg.async_func() # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited()
return result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited()
result = await pkg.async_func() # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
return result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
async def test_async_with():
async with pkg.async_func() as result: # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
return result # $ use=moduleImport("pkg").getMember("async_func").getReturn().getAwaited() awaited=moduleImport("pkg").getMember("async_func").getReturn()
async def test_async_for():
async for _ in pkg.async_func(): # $ use=moduleImport("pkg").getMember("async_func").getReturn() awaited=moduleImport("pkg").getMember("async_func").getReturn()
pass
coro = pkg.async_func() # $ use=moduleImport("pkg").getMember("async_func").getReturn()
async for _ in coro: # $ use=moduleImport("pkg").getMember("async_func").getReturn() MISSING: awaited=moduleImport("pkg").getMember("async_func").getReturn()
pass
def check_annotations():
# Just to make sure how annotations should look like :)

0
python/ql/test/experimental/dataflow/ApiGraphs/awaited.expected Normal file
View File

26
python/ql/test/experimental/dataflow/ApiGraphs/awaited.ql Normal file
View File

@@ -0,0 +1,26 @@
import python
import semmle.python.dataflow.new.DataFlow
import TestUtilities.InlineExpectationsTest
import semmle.python.ApiGraphs
class AwaitedTest extends InlineExpectationsTest {
AwaitedTest() { this = "AwaitedTest" }
override string getARelevantTag() { result = "awaited" }
override predicate hasActualResult(Location location, string element, string tag, string value) {
exists(API::Node awaited, DataFlow::Node use, API::Node pred |
awaited = pred.getAwaited() and
use = awaited.getAUse() and
location = use.getLocation() and
// Module variable nodes have no suitable location, so it's best to simply exclude them entirely
// from the inline tests.
not use instanceof DataFlow::ModuleVariableNode and
exists(location.getFile().getRelativePath())
|
tag = "awaited" and
value = pred.getPath() and
element = use.toString()
)
}
}

2
python/ql/test/experimental/dataflow/TestUtil/FlowTest.qll
View File

@@ -18,7 +18,7 @@ abstract class FlowTest extends InlineExpectationsTest {
location = toNode.getLocation() and
tag = this.flowTag() and
value =
"\"" + prettyNode(fromNode).replaceAll("\"", "'") + lineStr(fromNode, toNode) + " -> " +
"\"" + prettyNode(fromNode).replaceAll("\"", "'") + this.lineStr(fromNode, toNode) + " -> " +
prettyNode(toNode).replaceAll("\"", "'") + "\"" and
element = toNode.toString()
)

8
python/ql/test/experimental/dataflow/TestUtil/RoutingTest.qll
View File

@@ -25,11 +25,13 @@ abstract class RoutingTest extends InlineExpectationsTest {
element = fromNode.toString() and
(
tag = this.flowTag() and
if "\"" + tag + "\"" = fromValue(fromNode) then value = "" else value = fromValue(fromNode)
if "\"" + tag + "\"" = this.fromValue(fromNode)
then value = ""
else value = this.fromValue(fromNode)
or
tag = "func" and
value = toFunc(toNode) and
not value = fromFunc(fromNode)
value = this.toFunc(toNode) and
not value = this.fromFunc(fromNode)
)
)
}

2
python/ql/test/experimental/dataflow/global-flow/known.py
View File

@@ -1 +1 @@
known_attr = [1000]
known_attr = [1000] #$ writes=known_attr

2
python/ql/test/experimental/dataflow/import-star/deux.py Normal file
View File

@@ -0,0 +1,2 @@
from trois import *
print(foo)

15
python/ql/test/experimental/dataflow/import-star/global.expected Normal file
View File

@@ -0,0 +1,15 @@
| test3.py:1:17:1:19 | ControlFlowNode for ImportMember | test3.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:1:1:3 | ControlFlowNode for foo | test1.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:1:1:3 | ControlFlowNode for foo | test3.py:1:17:1:19 | ControlFlowNode for ImportMember |
| three.py:1:1:1:3 | ControlFlowNode for foo | test3.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:1:1:3 | ControlFlowNode for foo | two.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | test1.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | test3.py:1:17:1:19 | ControlFlowNode for ImportMember |
| three.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | test3.py:2:7:2:9 | ControlFlowNode for foo |
| three.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | two.py:2:7:2:9 | ControlFlowNode for foo |
| trois.py:1:1:1:3 | ControlFlowNode for foo | deux.py:2:7:2:9 | ControlFlowNode for foo |
| trois.py:1:1:1:3 | ControlFlowNode for foo | test2.py:2:7:2:9 | ControlFlowNode for foo |
| trois.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | deux.py:2:7:2:9 | ControlFlowNode for foo |
| trois.py:1:7:1:7 | ControlFlowNode for IntegerLiteral | test2.py:2:7:2:9 | ControlFlowNode for foo |
| two.py:2:7:2:9 | ControlFlowNode for foo | test3.py:1:17:1:19 | ControlFlowNode for ImportMember |
| two.py:2:7:2:9 | ControlFlowNode for foo | test3.py:2:7:2:9 | ControlFlowNode for foo |

19
python/ql/test/experimental/dataflow/import-star/global.ql Normal file
View File

@@ -0,0 +1,19 @@
import semmle.python.dataflow.new.DataFlow
/**
* A configuration to find all flows.
* To be used on tiny programs.
*/
class AllFlowsConfig extends DataFlow::Configuration {
AllFlowsConfig() { this = "AllFlowsConfig" }
override predicate isSource(DataFlow::Node node) { any() }
override predicate isSink(DataFlow::Node node) { any() }
}
from DataFlow::CfgNode source, DataFlow::CfgNode sink
where
source != sink and
exists(AllFlowsConfig cfg | cfg.hasFlow(source, sink))
select source, sink

1
python/ql/test/experimental/dataflow/import-star/one.py Normal file
View File

@@ -0,0 +1 @@
from two import *

2
python/ql/test/experimental/dataflow/import-star/test1.py Normal file
View File

@@ -0,0 +1,2 @@
from one import *
print(foo)

2
python/ql/test/experimental/dataflow/import-star/test2.py Normal file
View File

@@ -0,0 +1,2 @@
from un import *
print(foo)

2
python/ql/test/experimental/dataflow/import-star/test3.py Normal file
View File

@@ -0,0 +1,2 @@
from one import foo
print(foo)

1
python/ql/test/experimental/dataflow/import-star/three.py Normal file
View File

@@ -0,0 +1 @@
foo = 5

1
python/ql/test/experimental/dataflow/import-star/trois.py Normal file
View File

@@ -0,0 +1 @@
foo = 6

2
python/ql/test/experimental/dataflow/import-star/two.py Normal file
View File

@@ -0,0 +1,2 @@
from three import *
print(foo)

1
python/ql/test/experimental/dataflow/import-star/un.py Normal file
View File

@@ -0,0 +1 @@
from deux import *

0
python/ql/test/experimental/dataflow/match/dataflowTest.expected Normal file
View File

13
python/ql/test/experimental/dataflow/match/dataflowTest.ql Normal file
View File

@@ -0,0 +1,13 @@
import python
import experimental.dataflow.TestUtil.FlowTest
import experimental.dataflow.testConfig
class DataFlowTest extends FlowTest {
DataFlowTest() { this = "DataFlowTest" }
override string flowTag() { result = "flow" }
override predicate relevantFlow(DataFlow::Node source, DataFlow::Node sink) {
exists(TestConfiguration cfg | cfg.hasFlow(source, sink))
}
}

156
python/ql/test/experimental/dataflow/match/test.py Normal file
View File

@@ -0,0 +1,156 @@
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname((__file__))))
from testlib import *
# These are defined so that we can evaluate the test code.
NONSOURCE = "not a source"
SOURCE = "source"
def is_source(x):
return x == "source" or x == b"source" or x == 42 or x == 42.0 or x == 42j
def SINK(x):
if is_source(x):
print("OK")
else:
print("Unexpected flow", x)
def SINK_F(x):
if is_source(x):
print("Unexpected flow", x)
else:
print("OK")
def test_guard():
match SOURCE:
case x if SINK(x): #$ flow="SOURCE, l:-1 -> x"
pass
@expects(2)
def test_as_pattern():
match SOURCE:
case x as y:
SINK(x) #$ flow="SOURCE, l:-2 -> x"
SINK(y) #$ flow="SOURCE, l:-3 -> y"
def test_or_pattern():
match SOURCE:
# We cannot use NONSOURCE in place of "" below, since it would be seen as a variable.
case ("" as x) | x:
SINK(x) #$ flow="SOURCE, l:-3 -> x"
# No flow for literal pattern
def test_literal_pattern():
match SOURCE:
case 42 as x:
SINK(x) #$ flow="SOURCE, l:-2 -> x" flow="42, l:-1 -> x"
def test_capture_pattern():
match SOURCE:
case x:
SINK(x) #$ flow="SOURCE, l:-2 -> x"
# No flow for wildcard pattern
class Unsafe:
VALUE = SOURCE
def test_value_pattern():
match SOURCE:
case Unsafe.VALUE as x:
SINK(x) #$ flow="SOURCE, l:-2 -> x" MISSING: flow="SOURCE, l:-5 -> x"
@expects(2)
def test_sequence_pattern_tuple():
match (NONSOURCE, SOURCE):
case (x, y):
SINK_F(x)
SINK(y) #$ flow="SOURCE, l:-3 -> y"
@expects(2)
def test_sequence_pattern_list():
match [NONSOURCE, SOURCE]:
case [x, y]:
SINK_F(x) #$ SPURIOUS: flow="SOURCE, l:-2 -> x"
SINK(y) #$ flow="SOURCE, l:-3 -> y"
# Sets are excluded from sequence patterns,
# see https://www.python.org/dev/peps/pep-0635/#sequence-patterns
@expects(2)
def test_star_pattern_tuple():
match (NONSOURCE, SOURCE):
case (x, *y):
SINK_F(x)
SINK(y[0]) #$ flow="SOURCE, l:-3 -> y[0]"
@expects(2)
def test_star_pattern_tuple_exclusion():
match (SOURCE, NONSOURCE):
case (x, *y):
SINK(x) #$ flow="SOURCE, l:-2 -> x"
SINK_F(y[0])
@expects(2)
def test_star_pattern_list():
match [NONSOURCE, SOURCE]:
case [x, *y]:
SINK_F(x) #$ SPURIOUS: flow="SOURCE, l:-2 -> x"
SINK(y[0]) #$ flow="SOURCE, l:-3 -> y[0]"
@expects(2)
def test_star_pattern_list_exclusion():
match [SOURCE, NONSOURCE]:
case [x, *y]:
SINK(x) #$ flow="SOURCE, l:-2 -> x"
SINK_F(y[0]) #$ SPURIOUS: flow="SOURCE, l:-3 -> y[0]"
@expects(2)
def test_mapping_pattern():
match {"a": NONSOURCE, "b": SOURCE}:
case {"a": x, "b": y}:
SINK_F(x)
SINK(y) #$ flow="SOURCE, l:-3 -> y"
# also tests the key value pattern
@expects(2)
def test_double_star_pattern():
match {"a": NONSOURCE, "b": SOURCE}:
case {"a": x, **y}:
SINK_F(x)
SINK(y["b"]) #$ flow="SOURCE, l:-3 -> y['b']"
@expects(2)
def test_double_star_pattern_exclusion():
match {"a": SOURCE, "b": NONSOURCE}:
case {"a": x, **y}:
SINK(x) #$ flow="SOURCE, l:-2 -> x"
SINK_F(y["b"])
try:
SINK_F(y["a"])
except KeyError:
pass
class Cell:
def __init__(self, value):
self.value = value
# also tests the keyword pattern
@expects(2)
def test_class_pattern():
bad_cell = Cell(SOURCE)
good_cell = Cell(NONSOURCE)
match bad_cell:
case Cell(value = x):
SINK(x) #$ flow="SOURCE, l:-5 -> x"
match good_cell:
case Cell(value = x):
SINK_F(x)

13
python/ql/test/experimental/dataflow/sensitive-data/test.py
View File

@@ -112,3 +112,16 @@ print(foo) # $ SensitiveUse=password
harmless = lambda: "bar"
bar = call_wrapper(harmless)
print(bar) # $ SPURIOUS: SensitiveUse=password
# ------------------------------------------------------------------------------
# cross-talk in dictionary.
# ------------------------------------------------------------------------------
from unknown_settings import password # $ SensitiveDataSource=password
print(password) # $ SensitiveUse=password
_config = {"sleep_timer": 5, "mysql_password": password}
# since we have taint-step from store of `password`, we will consider any item in the
# dictionary to be a password :(
print(_config["sleep_timer"]) # $ SPURIOUS: SensitiveUse=password

7
python/ql/test/experimental/dataflow/tainttracking/TestTaintLib.qll
View File

@@ -7,9 +7,16 @@ class TestTaintTrackingConfiguration extends TaintTracking::Configuration {
TestTaintTrackingConfiguration() { this = "TestTaintTrackingConfiguration" }
override predicate isSource(DataFlow::Node source) {
// Standard sources
source.(DataFlow::CfgNode).getNode().(NameNode).getId() in [
"TAINTED_STRING", "TAINTED_BYTES", "TAINTED_LIST", "TAINTED_DICT"
]
or
// User defined sources
exists(CallNode call |
call.getFunction().(NameNode).getId() = "taint" and
source.(DataFlow::CfgNode).getNode() = call.getAnArg()
)
}
override predicate isSink(DataFlow::Node sink) {

55
python/ql/test/experimental/dataflow/tainttracking/commonSanitizer/test_string_const_compare.py
View File

@@ -77,6 +77,57 @@ def test_in_set():
ensure_tainted(ts) # $ tainted
def test_in_local_variable():
ts = TAINTED_STRING
safe = ["safe", "also_safe"]
if ts in safe:
ensure_not_tainted(ts) # $ SPURIOUS: tainted
else:
ensure_tainted(ts) # $ tainted
SAFE = ["safe", "also_safe"]
def test_in_global_variable():
ts = TAINTED_STRING
if ts in SAFE:
ensure_not_tainted(ts) # $ SPURIOUS: tainted
else:
ensure_tainted(ts) # $ tainted
# these global variables can be modified, so should not be considered safe
SAFE_mod_1 = ["safe", "also_safe"]
SAFE_mod_2 = ["safe", "also_safe"]
SAFE_mod_3 = ["safe", "also_safe"]
def make_modification(x):
global SAFE_mod_2, SAFE_mod_3
SAFE_mod_1.append(x)
SAFE_mod_2 += [x]
SAFE_mod_3 = SAFE_mod_3 + [x]
def test_in_modified_global_variable():
ts = TAINTED_STRING
if ts in SAFE_mod_1:
ensure_tainted(ts) # $ tainted
else:
ensure_tainted(ts) # $ tainted
if ts in SAFE_mod_2:
ensure_tainted(ts) # $ tainted
else:
ensure_tainted(ts) # $ tainted
if ts in SAFE_mod_3:
ensure_tainted(ts) # $ tainted
else:
ensure_tainted(ts) # $ tainted
def test_in_unsafe1(xs):
ts = TAINTED_STRING
if ts in xs:
@@ -131,6 +182,10 @@ test_non_eq2()
test_in_list()
test_in_tuple()
test_in_set()
test_in_local_variable()
test_in_global_variable()
make_modification("unsafe")
test_in_modified_global_variable()
test_in_unsafe1(["unsafe", "foo"])
test_in_unsafe2("unsafe")
test_not_in1()

57
python/ql/test/experimental/dataflow/tainttracking/defaultAdditionalTaintStep/test_async.py Normal file
View File

@@ -0,0 +1,57 @@
# Add taintlib to PATH so it can be imported during runtime without any hassle
import sys; import os; sys.path.append(os.path.dirname(os.path.dirname((__file__))))
from taintlib import *
# This has no runtime impact, but allows autocomplete to work
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..taintlib import *
# Actual tests
async def tainted_coro():
return TAINTED_STRING
async def test_await():
coro = tainted_coro()
taint(coro)
s = await coro
ensure_tainted(coro, s) # $ tainted
class AsyncContext:
async def __aenter__(self):
return TAINTED_STRING
async def __aexit__(self, exc_type, exc, tb):
pass
async def test_async_with():
ctx = AsyncContext()
taint(ctx)
async with ctx as tainted:
ensure_tainted(tainted) # $ tainted
class AsyncIter:
def __aiter__(self):
return self
async def __anext__(self):
raise StopAsyncIteration
async def test_async_for():
iter = AsyncIter()
taint(iter)
async for tainted in iter:
ensure_tainted(tainted) # $ tainted
# Make tests runable
import asyncio
asyncio.run(test_await())
asyncio.run(test_async_with())
asyncio.run(test_async_for())

30
python/ql/test/experimental/dataflow/tainttracking/defaultAdditionalTaintStep/test_for.py Normal file
View File

@@ -0,0 +1,30 @@
# Add taintlib to PATH so it can be imported during runtime without any hassle
import sys; import os; sys.path.append(os.path.dirname(os.path.dirname((__file__))))
from taintlib import *
# This has no runtime impact, but allows autocomplete to work
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..taintlib import *
# Actual tests
class Iter:
def __iter__(self):
return self
def __next__(self):
raise StopIteration
def test_for():
iter = Iter()
taint(iter)
for tainted in iter:
ensure_tainted(tainted) # $ tainted
# Make tests runable
test_for()

60
python/ql/test/experimental/dataflow/tainttracking/defaultAdditionalTaintStep/test_with.py Normal file
View File

@@ -0,0 +1,60 @@
# Add taintlib to PATH so it can be imported during runtime without any hassle
import sys; import os; sys.path.append(os.path.dirname(os.path.dirname((__file__))))
from taintlib import *
# This has no runtime impact, but allows autocomplete to work
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..taintlib import *
# Actual tests
class Context:
def __enter__(self):
return ""
def __exit__(self, exc_type, exc, tb):
pass
def test_with():
ctx = Context()
taint(ctx)
with ctx as tainted:
ensure_tainted(tainted) # $ tainted
class Context_taint:
def __enter__(self):
return TAINTED_STRING
def __exit__(self, exc_type, exc, tb):
pass
def test_with_taint():
ctx = Context_taint()
with ctx as tainted:
ensure_tainted(tainted) # $ MISSING: tainted
class Context_arg:
def __init__(self, arg):
self.arg = arg
def __enter__(self):
return self.arg
def __exit__(self, exc_type, exc, tb):
pass
def test_with_arg():
ctx = Context_arg(TAINTED_STRING)
with ctx as tainted:
ensure_tainted(tainted) # $ tainted
# Make tests runable
test_with()
test_with_taint()
test_with_arg()

5
python/ql/test/experimental/dataflow/tainttracking/taintlib.py
View File

@@ -5,6 +5,11 @@ TAINTED_DICT = {"name": TAINTED_STRING, "some key": "foo"}
NOT_TAINTED = "NOT_TAINTED"
# Use this to force expressions to be tainted
def taint(*args):
pass
def ensure_tainted(*args):
print("- ensure_tainted")
for i, arg in enumerate(args):

4
python/ql/test/experimental/dataflow/validTest.py
View File

@@ -57,6 +57,10 @@ if __name__ == "__main__":
check_tests_valid("variable-capture.nonlocal")
check_tests_valid("variable-capture.dict")
check_tests_valid("module-initialization.multiphase")
# The below will fail unless we use Python 3.10 or newer.
# check_tests_valid("match.test")
# The below fails when trying to import modules
# check_tests_valid("module-initialization.test")
# check_tests_valid("module-initialization.testOnce")

46
python/ql/test/experimental/meta/ConceptsTest.qll
View File

@@ -128,6 +128,24 @@ class CodeExecutionTest extends InlineExpectationsTest {
}
}
class SqlConstructionTest extends InlineExpectationsTest {
SqlConstructionTest() { this = "SqlConstructionTest" }
override string getARelevantTag() { result = "constructedSql" }
override predicate hasActualResult(Location location, string element, string tag, string value) {
exists(location.getFile().getRelativePath()) and
exists(SqlConstruction e, DataFlow::Node sql |
exists(location.getFile().getRelativePath()) and
sql = e.getSql() and
location = e.getLocation() and
element = sql.toString() and
value = prettyNodeForInlineTest(sql) and
tag = "constructedSql"
)
}
}
class SqlExecutionTest extends InlineExpectationsTest {
SqlExecutionTest() { this = "SqlExecutionTest" }
@@ -457,3 +475,31 @@ class CryptographicOperationTest extends InlineExpectationsTest {
)
}
}
class HttpClientRequestTest extends InlineExpectationsTest {
HttpClientRequestTest() { this = "HttpClientRequestTest" }
override string getARelevantTag() {
result in ["clientRequestUrlPart", "clientRequestCertValidationDisabled"]
}
override predicate hasActualResult(Location location, string element, string tag, string value) {
exists(location.getFile().getRelativePath()) and
exists(HTTP::Client::Request req, DataFlow::Node url |
url = req.getAUrlPart() and
location = url.getLocation() and
element = url.toString() and
value = prettyNodeForInlineTest(url) and
tag = "clientRequestUrlPart"
)
or
exists(location.getFile().getRelativePath()) and
exists(HTTP::Client::Request req |
req.disablesCertificateValidation(_, _) and
location = req.getLocation() and
element = req.toString() and
value = "" and
tag = "clientRequestCertValidationDisabled"
)
}
}

38
python/ql/test/experimental/meta/InlineTaintTest.qll
View File

@@ -30,24 +30,36 @@ DataFlow::Node shouldNotBeTainted() {
)
}
class TestTaintTrackingConfiguration extends TaintTracking::Configuration {
TestTaintTrackingConfiguration() { this = "TestTaintTrackingConfiguration" }
// this module allows the configuration to be imported in other `.ql` files without the
// top level query predicates of this file coming into scope.
module Conf {
class TestTaintTrackingConfiguration extends TaintTracking::Configuration {
TestTaintTrackingConfiguration() { this = "TestTaintTrackingConfiguration" }
override predicate isSource(DataFlow::Node source) {
source.asCfgNode().(NameNode).getId() in [
"TAINTED_STRING", "TAINTED_BYTES", "TAINTED_LIST", "TAINTED_DICT"
]
or
source instanceof RemoteFlowSource
}
override predicate isSource(DataFlow::Node source) {
source.asCfgNode().(NameNode).getId() in [
"TAINTED_STRING", "TAINTED_BYTES", "TAINTED_LIST", "TAINTED_DICT"
]
or
// User defined sources
exists(CallNode call |
call.getFunction().(NameNode).getId() = "taint" and
source.(DataFlow::CfgNode).getNode() = call.getAnArg()
)
or
source instanceof RemoteFlowSource
}
override predicate isSink(DataFlow::Node sink) {
sink = shouldBeTainted()
or
sink = shouldNotBeTainted()
override predicate isSink(DataFlow::Node sink) {
sink = shouldBeTainted()
or
sink = shouldNotBeTainted()
}
}
}
import Conf
class InlineTaintTest extends InlineExpectationsTest {
InlineTaintTest() { this = "InlineTaintTest" }

4
python/ql/test/experimental/meta/debug/InlineTaintTestPaths.expected Normal file
View File

@@ -0,0 +1,4 @@
edges
nodes
subpaths
#select

25
python/ql/test/experimental/meta/debug/InlineTaintTestPaths.ql Normal file
View File

@@ -0,0 +1,25 @@
/**
* @kind path-problem
*/
// This query is for debugging InlineTaintTestFailures.
// The intended usage is
// 1. load the database of the failing test
// 2. run this query to see actual paths
// 3. if necessary, look at partial paths by (un)commenting appropriate lines
import python
import semmle.python.dataflow.new.DataFlow
import experimental.meta.InlineTaintTest::Conf
// import DataFlow::PartialPathGraph
import DataFlow::PathGraph
class Conf extends TestTaintTrackingConfiguration {
override int explorationLimit() { result = 5 }
}
// from Conf config, DataFlow::PartialPathNode source, DataFlow::PartialPathNode sink
// where config.hasPartialFlow(source, sink, _)
from Conf config, DataFlow::PathNode source, DataFlow::PathNode sink
where config.hasFlowPath(source, sink)
select sink.getNode(), source, sink, "This node receives taint from $@.", source.getNode(),
"this source"

4
python/ql/test/experimental/meta/debug/dataflowTestPaths.expected Normal file
View File

@@ -0,0 +1,4 @@
edges
nodes
subpaths
#select

25
python/ql/test/experimental/meta/debug/dataflowTestPaths.ql Normal file
View File

@@ -0,0 +1,25 @@
/**
* @kind path-problem
*/
// This query is for debugging InlineTaintTestFailures.
// The intended usage is
// 1. load the database of the failing test
// 2. run this query to see actual paths
// 3. if necessary, look at partial paths by (un)commenting appropriate lines
import python
import semmle.python.dataflow.new.DataFlow
import experimental.dataflow.testConfig
// import DataFlow::PartialPathGraph
import DataFlow::PathGraph
class Conf extends TestConfiguration {
override int explorationLimit() { result = 5 }
}
// from Conf config, DataFlow::PartialPathNode source, DataFlow::PartialPathNode sink
// where config.hasPartialFlow(source, sink, _)
from Conf config, DataFlow::PathNode source, DataFlow::PathNode sink
where config.hasFlowPath(source, sink)
select sink.getNode(), source, sink, "This node receives taint from $@.", source.getNode(),
"this source"

8
python/ql/test/experimental/query-tests/Security/CWE-347/JWTEmptyKeyOrAlgorithm.expected Normal file
View File

@@ -0,0 +1,8 @@
| authlib.py:11:1:11:39 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| authlib.py:12:1:12:50 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| pyjwt.py:10:1:10:29 | ControlFlowNode for Attribute() | This JWT encoding has an empty algorithm. |
| pyjwt.py:10:1:10:29 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| pyjwt.py:13:1:13:40 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| pyjwt.py:14:1:14:44 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| python_jose.py:10:1:10:40 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |
| python_jose.py:11:1:11:44 | ControlFlowNode for Attribute() | This JWT encoding has an empty key. |

1
python/ql/test/experimental/query-tests/Security/CWE-347/JWTEmptyKeyOrAlgorithm.qlref Normal file
View File

@@ -0,0 +1 @@
experimental/Security/CWE-347/JWTEmptyKeyOrAlgorithm.ql

3
python/ql/test/experimental/query-tests/Security/CWE-347/JWTMissingSecretOrPublicKeyVerification.expected Normal file
View File

@@ -0,0 +1,3 @@
| pyjwt.py:22:12:22:16 | ControlFlowNode for token | is not verified with a cryptographic secret or public key. |
| pyjwt.py:23:12:23:16 | ControlFlowNode for token | is not verified with a cryptographic secret or public key. |
| python_jose.py:19:12:19:16 | ControlFlowNode for token | is not verified with a cryptographic secret or public key. |

1
python/ql/test/experimental/query-tests/Security/CWE-347/JWTMissingSecretOrPublicKeyVerification.qlref Normal file
View File

@@ -0,0 +1 @@
experimental/Security/CWE-347/JWTMissingSecretOrPublicKeyVerification.ql

18
python/ql/test/experimental/query-tests/Security/CWE-347/authlib.py Normal file
View File

@@ -0,0 +1,18 @@
from authlib.jose import jwt # It is already a JsonWebToken object
from authlib.jose import JsonWebToken
# Encoding
# good - key and algorithm supplied
jwt.encode({"alg": "HS256"}, token, "key")
JsonWebToken().encode({"alg": "HS256"}, token, "key")
# bad - empty key
jwt.encode({"alg": "HS256"}, token, "")
JsonWebToken().encode({"alg": "HS256"}, token, "")
# Decoding
# good - "it will raise BadSignatureError when signature doesn’t match"
jwt.decode(token, key)
JsonWebToken().decode(token, key)

31
python/ql/test/experimental/query-tests/Security/CWE-347/pyjwt.py Normal file
View File

@@ -0,0 +1,31 @@
import jwt
# Encoding
# good - key and algorithm supplied
jwt.encode(token, "key", "HS256")
jwt.encode(token, key="key", algorithm="HS256")
# bad - both key and algorithm set to None
jwt.encode(token, None, None)
# bad - empty key
jwt.encode(token, "", algorithm="HS256")
jwt.encode(token, key="", algorithm="HS256")
# Decoding
# good
jwt.decode(token, "key", "HS256")
# bad - unverified decoding
jwt.decode(token, verify=False)
jwt.decode(token, key, options={"verify_signature": False})
# good - verified decoding
jwt.decode(token, verify=True)
jwt.decode(token, key, options={"verify_signature": True})
def indeterminate(verify):
jwt.decode(token, key, verify)

22
python/ql/test/experimental/query-tests/Security/CWE-347/python_jose.py Normal file
View File

@@ -0,0 +1,22 @@
from jose import jwt
# Encoding
# good - key and algorithm supplied
jwt.encode(token, "key", "HS256")
jwt.encode(token, key="key", algorithm="HS256")
# bad - empty key
jwt.encode(token, "", algorithm="HS256")
jwt.encode(token, key="", algorithm="HS256")
# Decoding
# good
jwt.decode(token, "key", "HS256")
# bad - unverified decoding
jwt.decode(token, key, options={"verify_signature": False})
# good - verified decoding
jwt.decode(token, key, options={"verify_signature": True})

2
python/ql/test/library-tests/PointsTo/api/Constants.ql
View File

@@ -7,7 +7,7 @@ where
or
txt = "b'" + s + "'" and val = Value::forBytes(s)
|
s = "a" or s = "b" or s = "c" or s = "d"
s in ["a", "b", "c", "d"]
)
or
exists(int i | txt = i.toString() and val = Value::forInt(i) |

1
python/ql/test/library-tests/PointsTo/new/ImpliesDataflow.expected
View File

@@ -15,4 +15,3 @@
| code/r_regressions.py:46:1:46:14 | ControlFlowNode for FunctionExpr | code/r_regressions.py:52:9:52:12 | ControlFlowNode for fail |
| code/t_type.py:3:1:3:16 | ControlFlowNode for ClassExpr | code/t_type.py:6:1:6:9 | ControlFlowNode for type() |
| code/t_type.py:3:1:3:16 | ControlFlowNode for ClassExpr | code/t_type.py:13:5:13:13 | ControlFlowNode for type() |
| code/test_package/module2.py:5:5:5:6 | ControlFlowNode for Dict | code/j_convoluted_imports.py:25:1:25:1 | ControlFlowNode for r |

0
python/ql/test/library-tests/frameworks/aiomysql/ConceptsTest.expected Normal file
View File

2
python/ql/test/library-tests/frameworks/aiomysql/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,2 @@
import python
import experimental.meta.ConceptsTest

33
python/ql/test/library-tests/frameworks/aiomysql/test.py Normal file
View File

@@ -0,0 +1,33 @@
import aiomysql
# Only a cursor can execute sql.
async def test_cursor():
# Create connection directly
conn = await aiomysql.connect()
cur = await conn.cursor()
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" constructedSql="sql"
# Create Cursor directly
async with pool.cursor() as cur:
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" constructedSql="sql"
# Test SQLAlchemy integration
from aiomysql.sa import create_engine
async def test_engine():
engine = await create_engine()
conn = await engine.acquire()
await conn.execute("sql") # $ getSql="sql" constructedSql="sql"

0
python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.expected Normal file
View File

2
python/ql/test/library-tests/frameworks/aiopg/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,2 @@
import python
import experimental.meta.ConceptsTest

33
python/ql/test/library-tests/frameworks/aiopg/test.py Normal file
View File

@@ -0,0 +1,33 @@
import aiopg
# Only a cursor can execute sql.
async def test_cursor():
# Create connection directly
conn = await aiopg.connect()
cur = await conn.cursor()
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" constructedSql="sql"
# Create Cursor directly
async with pool.cursor() as cur:
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" constructedSql="sql"
# Test SQLAlchemy integration
from aiopg.sa import create_engine
async def test_engine():
engine = await create_engine()
conn = await engine.acquire()
await conn.execute("sql") # $ getSql="sql" constructedSql="sql"

0
python/ql/test/library-tests/frameworks/asyncpg/ConceptsTest.expected Normal file
View File

2
python/ql/test/library-tests/frameworks/asyncpg/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,2 @@
import python
import experimental.meta.ConceptsTest

105
python/ql/test/library-tests/frameworks/asyncpg/test.py Normal file
View File

@@ -0,0 +1,105 @@
import asyncio
import asyncpg
async def test_connection():
conn = await asyncpg.connect()
try:
# The file-like object is passed in as a keyword-only argument.
# See https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.connection.Connection.copy_from_query
await conn.copy_from_query("sql", output="filepath") # $ getSql="sql" getAPathArgument="filepath"
await conn.copy_from_query("sql", "arg1", "arg2", output="filepath") # $ getSql="sql" getAPathArgument="filepath"
await conn.copy_from_table("table", output="filepath") # $ getAPathArgument="filepath"
await conn.copy_to_table("table", source="filepath") # $ getAPathArgument="filepath"
await conn.execute("sql") # $ getSql="sql"
await conn.executemany("sql") # $ getSql="sql"
await conn.fetch("sql") # $ getSql="sql"
await conn.fetchrow("sql") # $ getSql="sql"
await conn.fetchval("sql") # $ getSql="sql"
finally:
await conn.close()
async def test_prepared_statement():
conn = await asyncpg.connect()
try:
pstmt = await conn.prepare("psql") # $ constructedSql="psql"
pstmt.executemany() # $ getSql="psql"
pstmt.fetch() # $ getSql="psql"
pstmt.fetchrow() # $ getSql="psql"
pstmt.fetchval() # $ getSql="psql"
finally:
await conn.close()
# The sql statement is executed when the `CursorFactory` (obtained by e.g. `conn.cursor()`) is awaited.
# See https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.cursor.CursorFactory
async def test_cursor():
conn = await asyncpg.connect()
try:
async with conn.transaction():
cursor = await conn.cursor("sql") # $ getSql="sql" constructedSql="sql"
await cursor.fetch()
pstmt = await conn.prepare("psql") # $ constructedSql="psql"
pcursor = await pstmt.cursor() # $ getSql="psql"
await pcursor.fetch()
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") # $ constructedSql="sql"
cursor = await cursor_factory # $ getSql="sql"
pcursor_factory = pstmt.cursor()
pcursor = await pcursor_factory # $ getSql="psql"
finally:
await conn.close()
async def test_connection_pool():
pool = await asyncpg.create_pool()
try:
await pool.copy_from_query("sql", output="filepath") # $ getSql="sql" getAPathArgument="filepath"
await pool.copy_from_query("sql", "arg1", "arg2", output="filepath") # $ getSql="sql" getAPathArgument="filepath"
await pool.copy_from_table("table", output="filepath") # $ getAPathArgument="filepath"
await pool.copy_to_table("table", source="filepath") # $ getAPathArgument="filepath"
await pool.execute("sql") # $ getSql="sql"
await pool.executemany("sql") # $ getSql="sql"
await pool.fetch("sql") # $ getSql="sql"
await pool.fetchrow("sql") # $ getSql="sql"
await pool.fetchval("sql") # $ getSql="sql"
async with pool.acquire() as conn:
await conn.execute("sql") # $ getSql="sql"
conn = await pool.acquire()
try:
await conn.fetch("sql") # $ getSql="sql"
finally:
await pool.release(conn)
finally:
await pool.close()
async with asyncpg.create_pool() as pool:
await pool.execute("sql") # $ getSql="sql"
async with pool.acquire() as conn:
await conn.execute("sql") # $ getSql="sql"
conn = await pool.acquire()
try:
await conn.fetch("sql") # $ getSql="sql"
finally:
await pool.release(conn)

16
python/ql/test/library-tests/frameworks/django-v2-v3/response_test.py
View File

@@ -1,6 +1,7 @@
from django.http.response import HttpResponse, HttpResponseRedirect, HttpResponsePermanentRedirect, JsonResponse, HttpResponseNotFound
from django.views.generic import RedirectView
import django.shortcuts
import json
# Not an XSS sink, since the Content-Type is not "text/html"
# FP reported in https://github.com/github/codeql-python-team/issues/38
@@ -13,6 +14,21 @@ def safe__manual_json_response(request):
json_data = '{"json": "{}"}'.format(request.GET.get("foo"))
return HttpResponse(json_data, content_type="application/json") # $HttpResponse mimetype=application/json responseBody=json_data
# reproduction of FP seen here:
# Usage: https://github.com/edx/edx-platform/blob/d70ebe6343a1573c694d6cf68f92c1ad40b73d7d/lms/djangoapps/commerce/api/v0/views.py#L106
# DetailResponse def: https://github.com/edx/edx-platform/blob/d70ebe6343a1573c694d6cf68f92c1ad40b73d7d/lms/djangoapps/commerce/http.py#L9
# JsonResponse def: https://github.com/edx/edx-platform/blob/d70ebe6343a1573c694d6cf68f92c1ad40b73d7d/common/djangoapps/util/json_request.py#L60
class MyJsonResponse(HttpResponse):
def __init__(self, data):
serialized = json.dumps(data).encode("utf-8") # $ encodeFormat=JSON encodeInput=data encodeOutput=json.dumps(..)
super().__init__(serialized, content_type="application/json")
# Not an XSS sink, since the Content-Type is not "text/html"
def safe__custom_json_response(request):
json_data = '{"json": "{}"}'.format(request.GET.get("foo"))
return MyJsonResponse(json_data) # $HttpResponse responseBody=json_data SPURIOUS: mimetype=text/html MISSING: mimetype=application/json
# Not an XSS sink, since the Content-Type is not "text/html"
def safe__manual_content_type(request):
return HttpResponse('<img src="0" onerror="alert(1)">', content_type="text/plain") # $HttpResponse mimetype=text/plain responseBody='<img src="0" onerror="alert(1)">'

0
python/ql/test/library-tests/frameworks/fastapi/ConceptsTest.expected Normal file
View File

12
python/ql/test/library-tests/frameworks/fastapi/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,12 @@
import python
import experimental.meta.ConceptsTest
class DedicatedResponseTest extends HttpServerHttpResponseTest {
DedicatedResponseTest() { file.getShortName() = "response_test.py" }
}
class OtherResponseTest extends HttpServerHttpResponseTest {
OtherResponseTest() { not this instanceof DedicatedResponseTest }
override string getARelevantTag() { result = "HttpResponse" }
}

3
python/ql/test/library-tests/frameworks/fastapi/InlineTaintTest.expected Normal file
View File

@@ -0,0 +1,3 @@
argumentToEnsureNotTaintedNotMarkedAsSpurious
untaintedArgumentToEnsureTaintedNotMarkedAsMissing
failures

1
python/ql/test/library-tests/frameworks/fastapi/InlineTaintTest.ql Normal file
View File

@@ -0,0 +1 @@
import experimental.meta.InlineTaintTest

66
python/ql/test/library-tests/frameworks/fastapi/basic.py Normal file
View File

@@ -0,0 +1,66 @@
# Taking inspiration from https://realpython.com/fastapi-python-web-apis/
# run with
# uvicorn basic:app --reload
# Then visit http://127.0.0.1:8000/docs and http://127.0.0.1:8000/redoc
from fastapi import FastAPI
app = FastAPI()
@app.get("/") # $ routeSetup="/"
async def root(): # $ requestHandler
return {"message": "Hello World"} # $ HttpResponse
@app.get("/non-async") # $ routeSetup="/non-async"
def non_async(): # $ requestHandler
return {"message": "non-async"} # $ HttpResponse
@app.get(path="/kw-arg") # $ routeSetup="/kw-arg"
def kw_arg(): # $ requestHandler
return {"message": "kw arg"} # $ HttpResponse
@app.get("/foo/{foo_id}") # $ routeSetup="/foo/{foo_id}"
async def get_foo(foo_id: int): # $ requestHandler routedParameter=foo_id
# FastAPI does data validation (with `pydantic` PyPI package) under the hood based
# on the type annotation we did for `foo_id`, so it will auto-reject anything that's
# not an int.
return {"foo_id": foo_id} # $ HttpResponse
# this will work as query param, so `/bar?bar_id=123`
@app.get("/bar") # $ routeSetup="/bar"
async def get_bar(bar_id: int = 42): # $ requestHandler routedParameter=bar_id
return {"bar_id": bar_id} # $ HttpResponse
# The big deal is that FastAPI works so well together with pydantic, so you can do stuff like this
from typing import Optional
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
is_offer: Optional[bool] = None
@app.post("/items/") # $ routeSetup="/items/"
async def create_item(item: Item): # $ requestHandler routedParameter=item
# Note: calling `item` a routed parameter is slightly untrue, since it doesn't come
# from the URL itself, but from the body of the POST request
return item # $ HttpResponse
# this also works fine
@app.post("/2items") # $ routeSetup="/2items"
async def create_item2(item1: Item, item2: Item): # $ requestHandler routedParameter=item1 routedParameter=item2
return (item1, item2) # $ HttpResponse
@app.api_route("/baz/{baz_id}", methods=["GET"]) # $ routeSetup="/baz/{baz_id}"
async def get_baz(baz_id: int): # $ requestHandler routedParameter=baz_id
return {"baz_id2": baz_id} # $ HttpResponse
# Docs:
# see https://fastapi.tiangolo.com/tutorial/path-params/
# Things we should look at supporting:
# - https://fastapi.tiangolo.com/tutorial/dependencies/
# - https://fastapi.tiangolo.com/tutorial/background-tasks/
# - https://fastapi.tiangolo.com/tutorial/middleware/
# - https://fastapi.tiangolo.com/tutorial/encoder/

145
python/ql/test/library-tests/frameworks/fastapi/response_test.py Normal file
View File

@@ -0,0 +1,145 @@
# see https://fastapi.tiangolo.com/advanced/response-cookies/
from fastapi import FastAPI, Response
import fastapi.responses
import asyncio
app = FastAPI()
@app.get("/response_parameter") # $ routeSetup="/response_parameter"
async def response_parameter(response: Response): # $ requestHandler
response.set_cookie("key", "value") # $ CookieWrite CookieName="key" CookieValue="value"
response.set_cookie(key="key", value="value") # $ CookieWrite CookieName="key" CookieValue="value"
response.headers.append("Set-Cookie", "key2=value2") # $ CookieWrite CookieRawHeader="key2=value2"
response.headers.append(key="Set-Cookie", value="key2=value2") # $ CookieWrite CookieRawHeader="key2=value2"
response.headers["X-MyHeader"] = "header-value"
response.status_code = 418
return {"message": "response as parameter"} # $ HttpResponse mimetype=application/json responseBody=Dict
@app.get("/resp_parameter") # $ routeSetup="/resp_parameter"
async def resp_parameter(resp: Response): # $ requestHandler
resp.status_code = 418
return {"message": "resp as parameter"} # $ HttpResponse mimetype=application/json responseBody=Dict
@app.get("/response_parameter_no_type") # $ routeSetup="/response_parameter_no_type"
async def response_parameter_no_type(response): # $ requestHandler routedParameter=response
# NOTE: This does in fact not work, since FastAPI relies on the type annotations,
# and not on the name of the parameter
response.status_code = 418
return {"message": "response as parameter"} # $ HttpResponse mimetype=application/json responseBody=Dict
class MyXmlResponse(fastapi.responses.Response):
media_type = "application/xml"
@app.get("/response_parameter_custom_type", response_class=MyXmlResponse) # $ routeSetup="/response_parameter_custom_type"
async def response_parameter_custom_type(response: MyXmlResponse): # $ requestHandler
# NOTE: This is a contrived example of using a wrong annotation for the response
# parameter. It will be passed a `fastapi.responses.Response` value when handling an
# incoming request, so NOT a `MyXmlResponse` value. Cookies/Headers are still
# propagated to the final response though.
print(type(response))
assert type(response) == fastapi.responses.Response
response.set_cookie("key", "value") # $ CookieWrite CookieName="key" CookieValue="value"
response.headers["Custom-Response-Type"] = "yes, but only after function has run"
xml_data = "<foo>FOO</foo>"
return xml_data # $ HttpResponse responseBody=xml_data mimetype=application/xml
# Direct response construction
# see https://fastapi.tiangolo.com/advanced/response-directly/
# see https://fastapi.tiangolo.com/advanced/custom-response/
@app.get("/direct_response") # $ routeSetup="/direct_response"
async def direct_response(): # $ requestHandler
xml_data = "<foo>FOO</foo>"
resp = fastapi.responses.Response(xml_data, 200, None, "application/xml") # $ HttpResponse mimetype=application/xml responseBody=xml_data
resp = fastapi.responses.Response(content=xml_data, media_type="application/xml") # $ HttpResponse mimetype=application/xml responseBody=xml_data
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
@app.get("/direct_response2", response_class=fastapi.responses.Response) # $ routeSetup="/direct_response2"
async def direct_response2(): # $ requestHandler
xml_data = "<foo>FOO</foo>"
return xml_data # $ HttpResponse responseBody=xml_data
@app.get("/my_xml_response") # $ routeSetup="/my_xml_response"
async def my_xml_response(): # $ requestHandler
xml_data = "<foo>FOO</foo>"
resp = MyXmlResponse(content=xml_data) # $ HttpResponse mimetype=application/xml responseBody=xml_data
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
@app.get("/my_xml_response2", response_class=MyXmlResponse) # $ routeSetup="/my_xml_response2"
async def my_xml_response2(): # $ requestHandler
xml_data = "<foo>FOO</foo>"
return xml_data # $ HttpResponse responseBody=xml_data mimetype=application/xml
@app.get("/html_response") # $ routeSetup="/html_response"
async def html_response(): # $ requestHandler
hello_world = "<h1>Hello World!</h1>"
resp = fastapi.responses.HTMLResponse(hello_world) # $ HttpResponse mimetype=text/html responseBody=hello_world
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
@app.get("/html_response2", response_class=fastapi.responses.HTMLResponse) # $ routeSetup="/html_response2"
async def html_response2(): # $ requestHandler
hello_world = "<h1>Hello World!</h1>"
return hello_world # $ HttpResponse responseBody=hello_world mimetype=text/html
@app.get("/redirect") # $ routeSetup="/redirect"
async def redirect(): # $ requestHandler
next = "https://www.example.com"
resp = fastapi.responses.RedirectResponse(next) # $ HttpResponse HttpRedirectResponse redirectLocation=next
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
@app.get("/redirect2", response_class=fastapi.responses.RedirectResponse) # $ routeSetup="/redirect2"
async def redirect2(): # $ requestHandler
next = "https://www.example.com"
return next # $ HttpResponse HttpRedirectResponse redirectLocation=next
@app.get("/streaming_response") # $ routeSetup="/streaming_response"
async def streaming_response(): # $ requestHandler
# You can test this with curl:
# curl --no-buffer http://127.0.0.1:8000/streaming_response
async def content():
yield b"Hello "
await asyncio.sleep(0.5)
yield b"World"
await asyncio.sleep(0.5)
yield b"!"
resp = fastapi.responses.StreamingResponse(content()) # $ HttpResponse responseBody=content()
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
# setting `response_class` to `StreamingResponse` does not seem to work
# so no such example here
@app.get("/file_response") # $ routeSetup="/file_response"
async def file_response(): # $ requestHandler
# has internal dependency on PyPI package `aiofiles`
# will guess MIME type from file extension
# We don't really have any good QL modeling of passing a file-path, whose content
# will be returned as part of the response... so will leave this as a TODO for now.
resp = fastapi.responses.FileResponse(__file__) # $ HttpResponse getAPathArgument=__file__
return resp # $ SPURIOUS: HttpResponse mimetype=application/json responseBody=resp
@app.get("/file_response2", response_class=fastapi.responses.FileResponse) # $ routeSetup="/file_response2"
async def file_response2(): # $ requestHandler
return __file__ # $ HttpResponse getAPathArgument=__file__

53
python/ql/test/library-tests/frameworks/fastapi/router.py Normal file
View File

@@ -0,0 +1,53 @@
# like blueprints in Flask
# see https://fastapi.tiangolo.com/tutorial/bigger-applications/
# see basic.py for instructions for how to run this code.
from fastapi import APIRouter, FastAPI
inner_router = APIRouter()
@inner_router.get("/foo") # $ routeSetup="/foo"
async def root(): # $ requestHandler
return {"msg": "inner_router /foo"} # $ HttpResponse
outer_router = APIRouter()
outer_router.include_router(inner_router, prefix="/inner")
items_router = APIRouter(
prefix="/items",
tags=["items"],
)
@items_router.get("/") # $ routeSetup="/"
async def items(): # $ requestHandler
return {"msg": "items_router /"} # $ HttpResponse
app = FastAPI()
app.include_router(outer_router, prefix="/outer")
app.include_router(items_router)
# Using a custom router
class MyCustomRouter(APIRouter):
"""
Which automatically removes trailing slashes
"""
def api_route(self, path: str, **kwargs):
path = path.rstrip("/")
return super().api_route(path, **kwargs)
custom_router = MyCustomRouter()
@custom_router.get("/bar/") # $ routeSetup="/bar/"
async def items(): # $ requestHandler
return {"msg": "custom_router /bar/"} # $ HttpResponse
app.include_router(custom_router)

189
python/ql/test/library-tests/frameworks/fastapi/taint_test.py Normal file
View File

@@ -0,0 +1,189 @@
# --- to make things runable ---
ensure_tainted = ensure_not_tainted = print
# --- real code ---
from fastapi import FastAPI
from typing import Optional, List
from pydantic import BaseModel
app = FastAPI()
class Foo(BaseModel):
foo: str
class MyComplexModel(BaseModel):
field: str
main_foo: Foo
other_foos: List[Foo]
nested_foos: List[List[Foo]]
@app.post("/test_taint/{name}/{number}") # $ routeSetup="/test_taint/{name}/{number}"
async def test_taint(name : str, number : int, also_input: MyComplexModel): # $ requestHandler routedParameter=name routedParameter=number routedParameter=also_input
ensure_tainted(
name, # $ tainted
number, # $ tainted
also_input, # $ tainted
also_input.field, # $ tainted
also_input.main_foo, # $ tainted
also_input.main_foo.foo, # $ tainted
also_input.other_foos, # $ tainted
also_input.other_foos[0], # $ tainted
also_input.other_foos[0].foo, # $ tainted
[f.foo for f in also_input.other_foos], # $ MISSING: tainted
also_input.nested_foos, # $ tainted
also_input.nested_foos[0], # $ tainted
also_input.nested_foos[0][0], # $ tainted
also_input.nested_foos[0][0].foo, # $ tainted
)
other_foos = also_input.other_foos
ensure_tainted(
other_foos, # $ tainted
other_foos[0], # $ tainted
other_foos[0].foo, # $ tainted
[f.foo for f in other_foos], # $ MISSING: tainted
)
return "ok" # $ HttpResponse
# --- body ---
# see https://fastapi.tiangolo.com/tutorial/body-multiple-params/
from fastapi import Body
# request is made such as `/will-be-query-param?name=foo`
@app.post("/will-be-query-param") # $ routeSetup="/will-be-query-param"
async def will_be_query_param(name: str): # $ requestHandler routedParameter=name
ensure_tainted(name) # $ tainted
return "ok" # $ HttpResponse
# with the `= Body(...)` "annotation" FastAPI will know to transmit `name` as part of
# the HTTP post body
@app.post("/will-not-be-query-param") # $ routeSetup="/will-not-be-query-param"
async def will_not_be_query_param(name: str = Body("foo", media_type="text/plain")): # $ requestHandler routedParameter=name
ensure_tainted(name) # $ tainted
return "ok" # $ HttpResponse
# --- form data ---
# see https://fastapi.tiangolo.com/tutorial/request-forms/
from fastapi import Form
@app.post("/form-example") # $ routeSetup="/form-example"
async def form_example(username: str = Form(None)): # $ requestHandler routedParameter=username
ensure_tainted(username) # $ tainted
return "ok" # $ HttpResponse
# --- HTTP headers ---
# see https://fastapi.tiangolo.com/tutorial/header-params/
from fastapi import Header
@app.get("/header-example") # $ routeSetup="/header-example"
async def header_example(user_agent: Optional[str] = Header(None)): # $ requestHandler routedParameter=user_agent
ensure_tainted(user_agent) # $ tainted
return "ok" # $ HttpResponse
# --- file upload ---
# see https://fastapi.tiangolo.com/tutorial/request-files/
# see https://fastapi.tiangolo.com/tutorial/request-files/#uploadfile
from fastapi import File, UploadFile
@app.post("/file-upload") # $ routeSetup="/file-upload"
async def file_upload(f1: bytes = File(None), f2: UploadFile = File(None)): # $ requestHandler routedParameter=f1 routedParameter=f2
ensure_tainted(
f1, # $ tainted
f2, # $ tainted
f2.filename, # $ MISSING: tainted
f2.content_type, # $ MISSING: tainted
f2.file, # $ MISSING: tainted
f2.file.read(), # $ MISSING: tainted
f2.file.readline(), # $ MISSING: tainted
f2.file.readlines(), # $ MISSING: tainted
await f2.read(), # $ MISSING: tainted
)
return "ok" # $ HttpResponse
# --- WebSocket ---
import starlette.websockets
from fastapi import WebSocket
assert WebSocket == starlette.websockets.WebSocket
@app.websocket("/ws") # $ routeSetup="/ws"
async def websocket_test(websocket: WebSocket): # $ requestHandler routedParameter=websocket
await websocket.accept()
ensure_tainted(
websocket, # $ tainted
websocket.url, # $ tainted
websocket.url.netloc, # $ tainted
websocket.url.path, # $ tainted
websocket.url.query, # $ tainted
websocket.url.fragment, # $ tainted
websocket.url.username, # $ tainted
websocket.url.password, # $ tainted
websocket.url.hostname, # $ tainted
websocket.url.port, # $ tainted
websocket.url.components, # $ tainted
websocket.url.components.netloc, # $ tainted
websocket.url.components.path, # $ tainted
websocket.url.components.query, # $ tainted
websocket.url.components.fragment, # $ tainted
websocket.url.components.username, # $ tainted
websocket.url.components.password, # $ tainted
websocket.url.components.hostname, # $ tainted
websocket.url.components.port, # $ tainted
websocket.headers, # $ tainted
websocket.headers["key"], # $ tainted
websocket.query_params, # $ tainted
websocket.query_params["key"], # $ tainted
websocket.cookies, # $ tainted
websocket.cookies["key"], # $ tainted
await websocket.receive(), # $ tainted
await websocket.receive_bytes(), # $ tainted
await websocket.receive_text(), # $ tainted
await websocket.receive_json(), # $ tainted
)
# scheme seems very unlikely to give interesting results, but very likely to give FPs.
ensure_not_tainted(
websocket.url.scheme,
websocket.url.components.scheme,
)
async for data in websocket.iter_bytes():
ensure_tainted(data) # $ tainted
async for data in websocket.iter_text():
ensure_tainted(data) # $ tainted
async for data in websocket.iter_json():
ensure_tainted(data) # $ tainted

7
python/ql/test/library-tests/frameworks/flask/file_sending.py Normal file
View File

@@ -0,0 +1,7 @@
from flask import send_from_directory, send_file
send_from_directory("dir", "file") # $ getAPathArgument="dir" getAPathArgument="file"
send_from_directory(directory="dir", filename="file") # $ getAPathArgument="dir" getAPathArgument="file"
send_file("file") # $ getAPathArgument="file"
send_file(filename_or_fp="file") # $ getAPathArgument="file"

4
python/ql/test/library-tests/frameworks/flask/routing_test.py
View File

@@ -105,8 +105,8 @@ def bp1_example(foo): # $ requestHandler routedParameter=foo
app.register_blueprint(bp1) # by default, URLs of blueprints are not prefixed
bp2 = flask.Blueprint("bp2", __name__)
import flask.blueprints
bp2 = flask.blueprints.Blueprint("bp2", __name__)
@bp2.route("/example") # $ routeSetup="/example"
def bp2_example(): # $ requestHandler

0
python/ql/test/library-tests/frameworks/flask_admin/ConceptsTest.expected Normal file
View File

12
python/ql/test/library-tests/frameworks/flask_admin/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,12 @@
import python
import experimental.meta.ConceptsTest
class DedicatedResponseTest extends HttpServerHttpResponseTest {
DedicatedResponseTest() { file.getShortName() = "response_test.py" }
}
class OtherResponseTest extends HttpServerHttpResponseTest {
OtherResponseTest() { not this instanceof DedicatedResponseTest }
override string getARelevantTag() { result = "HttpResponse" }
}

3
python/ql/test/library-tests/frameworks/flask_admin/InlineTaintTest.expected Normal file
View File

@@ -0,0 +1,3 @@
argumentToEnsureNotTaintedNotMarkedAsSpurious
untaintedArgumentToEnsureTaintedNotMarkedAsMissing
failures

1
python/ql/test/library-tests/frameworks/flask_admin/InlineTaintTest.ql Normal file
View File

@@ -0,0 +1 @@
import experimental.meta.InlineTaintTest

58
python/ql/test/library-tests/frameworks/flask_admin/test.py Normal file
View File

@@ -0,0 +1,58 @@
from flask import Flask, redirect
from flask.views import MethodView
import flask_admin
ensure_tainted = ensure_not_tainted = print
app = Flask(__name__)
# unknown at least for our current analysis
foo = "'/foo'"
UNKNOWN_ROUTE = eval(foo) # $ getCode=foo
class ExampleClass(flask_admin.BaseView):
@flask_admin.expose('/') # $ routeSetup="/"
def foo(self): # $ requestHandler
return "foo" # $ HttpResponse
@flask_admin.expose(url='/bar/<arg>') # $ routeSetup="/bar/<arg>"
def bar(self, arg): # $ requestHandler routedParameter=arg
ensure_tainted(arg) # $ tainted
return "bar: " + arg # $ HttpResponse
@flask_admin.expose_plugview("/flask-class") # $ routeSetup="/flask-class"
@flask_admin.expose_plugview(url="/flask-class/<arg>") # $ routeSetup="/flask-class/<arg>"
class Nested(MethodView):
def get(self, cls, arg="default"): # $ requestHandler routedParameter=arg
assert isinstance(cls, ExampleClass)
ensure_tainted(arg) # $ tainted
ensure_not_tainted(cls)
return "GET: " + arg # $ HttpResponse
def post(self, cls, arg): # $ requestHandler routedParameter=arg
assert isinstance(cls, ExampleClass)
ensure_tainted(arg) # $ tainted
ensure_not_tainted(cls)
return "POST: " + arg # $ HttpResponse
@flask_admin.expose_plugview(UNKNOWN_ROUTE) # $ routeSetup
class WithUnknownRoute(MethodView):
def get(self, cls, maybeRouted): # $ requestHandler routedParameter=maybeRouted
ensure_tainted(maybeRouted) # $ tainted
ensure_not_tainted(cls)
return "ok" # $ HttpResponse
@app.route('/') # $ routeSetup="/"
def index(): # $ requestHandler
return redirect('/admin') # $ HttpRedirectResponse HttpResponse redirectLocation='/admin'
if __name__ == "__main__":
admin = flask_admin.Admin(app, name="Some Admin Interface")
admin.add_view(ExampleClass())
print(app.url_map)
app.run(debug=True)

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'>"
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")
t = db.text("foo") # $ constructedSql="foo"
assert isinstance(t, sqlalchemy.sql.expression.TextClause)
t = db.text(text="foo")
t = db.text(text="foo") # $ constructedSql="foo"
assert isinstance(t, sqlalchemy.sql.expression.TextClause)

0
python/ql/test/library-tests/frameworks/internal-ql-helpers/PoorMansFunctionResolutionTest.expected Normal file
View File

28
python/ql/test/library-tests/frameworks/internal-ql-helpers/PoorMansFunctionResolutionTest.ql Normal file
View File

@@ -0,0 +1,28 @@
private import python
private import semmle.python.dataflow.new.DataFlow
private import semmle.python.frameworks.internal.PoorMansFunctionResolution
import TestUtilities.InlineExpectationsTest
class InlinePoorMansFunctionResolutionTest extends InlineExpectationsTest {
InlinePoorMansFunctionResolutionTest() { this = "InlinePoorMansFunctionResolutionTest" }
override string getARelevantTag() { result = "resolved" }
override predicate hasActualResult(Location location, string element, string tag, string value) {
exists(location.getFile().getRelativePath()) and
exists(Function func, DataFlow::Node ref |
ref = poorMansFunctionTracker(func) and
not ref.asExpr() instanceof FunctionExpr and
// exclude things like `GSSA variable func`
exists(ref.asExpr()) and
// exclude decorator calls (which with our extractor rewrites does reference the
// function)
not ref.asExpr() = func.getDefinition().(FunctionExpr).getADecoratorCall()
|
value = func.getName() and
tag = "resolved" and
element = ref.toString() and
location = ref.getLocation()
)
}
}

42
python/ql/test/library-tests/frameworks/internal-ql-helpers/test.py Normal file
View File

@@ -0,0 +1,42 @@
def func():
print("func")
func() # $ resolved=func
class MyBase:
def base_method(self):
print("base_method", self)
class MyClass(MyBase):
def method1(self):
print("method1", self)
@classmethod
def cls_method(cls):
print("cls_method", cls)
@staticmethod
def static():
print("static")
def method2(self):
print("method2", self)
self.method1() # $ resolved=method1
self.base_method()
self.cls_method() # $ resolved=cls_method
self.static() # $ resolved=static
MyClass.cls_method() # $ resolved=cls_method
MyClass.static() # $ resolved=static
x = MyClass()
x.base_method()
x.method1()
x.cls_method()
x.static()
x.method2()

0
python/ql/test/library-tests/frameworks/requests/ConceptsTest.expected Normal file
View File

2
python/ql/test/library-tests/frameworks/requests/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,2 @@
import python
import experimental.meta.ConceptsTest

3
python/ql/test/library-tests/frameworks/requests/InlineTaintTest.expected Normal file
View File

@@ -0,0 +1,3 @@
argumentToEnsureNotTaintedNotMarkedAsSpurious
untaintedArgumentToEnsureTaintedNotMarkedAsMissing
failures

1
python/ql/test/library-tests/frameworks/requests/InlineTaintTest.ql Normal file
View File

@@ -0,0 +1 @@
import experimental.meta.InlineTaintTest

55
python/ql/test/library-tests/frameworks/requests/taint_test.py Normal file
View File

@@ -0,0 +1,55 @@
import requests
from flask import Flask, request
app = Flask(__name__)
@app.route("/taint_test") # $ routeSetup="/taint_test"
def test_taint(): # $ requestHandler
url = request.args['untrusted_input']
# response from a request to a user-controlled URL should be considered
# user-controlled as well.
resp = requests.get(url) # $ clientRequestUrlPart=url
requests.Response
requests.models.Response
ensure_tainted(
url, # $ tainted
# see https://docs.python-requests.org/en/latest/api/#requests.Response
resp, # $ tainted
resp.text, # $ tainted
resp.content, # $ tainted
resp.json(), # $ tainted
# file-like
resp.raw, # $ tainted
resp.raw.read(), # $ tainted
resp.links, # $ tainted
resp.links['key'], # $ tainted
resp.links.get('key'), # $ tainted
resp.cookies, # $ tainted
resp.cookies['key'], # $ tainted
resp.cookies.get('key'), # $ tainted
resp.headers, # $ tainted
resp.headers['key'], # $ tainted
resp.headers.get('key'), # $ tainted
)
for content_chunk in resp.iter_content():
ensure_tainted(content_chunk) # $ tainted
for line in resp.iter_lines():
ensure_tainted(line) # $ tainted
# for now, we don't assume that the response to ANY outgoing request is a remote
# flow source, since this could lead to FPs.
# TODO: investigate whether we should consider this a remote flow source.
trusted_url = "https://internal-api-that-i-trust.com"
resp = requests.get(trusted_url) # $ clientRequestUrlPart=trusted_url
ensure__not_tainted(resp)

50
python/ql/test/library-tests/frameworks/requests/test.py Normal file
View File

@@ -0,0 +1,50 @@
import requests
resp = requests.get("url") # $ clientRequestUrlPart="url"
resp = requests.get(url="url") # $ clientRequestUrlPart="url"
resp = requests.request("GET", "url") # $ clientRequestUrlPart="url"
with requests.Session() as session:
resp = session.get("url") # $ clientRequestUrlPart="url"
resp = session.request(method="GET", url="url") # $ clientRequestUrlPart="url"
s = requests.Session()
resp = s.get("url") # $ clientRequestUrlPart="url"
s = requests.session()
resp = s.get("url") # $ clientRequestUrlPart="url"
# test full import path for Session
with requests.sessions.Session() as session:
resp = session.get("url") # $ clientRequestUrlPart="url"
# Low level access
req = requests.Request("GET", "url") # $ MISSING: clientRequestUrlPart="url"
resp = s.send(req.prepare())
# other methods than GET
resp = requests.post("url") # $ clientRequestUrlPart="url"
resp = requests.patch("url") # $ clientRequestUrlPart="url"
resp = requests.options("url") # $ clientRequestUrlPart="url"
# ==============================================================================
# Disabling certificate validation
# ==============================================================================
resp = requests.get("url", verify=False) # $ clientRequestUrlPart="url" clientRequestCertValidationDisabled
def make_get(verify_arg):
resp = requests.get("url", verify=verify_arg) # $ clientRequestUrlPart="url" clientRequestCertValidationDisabled
make_get(False)
with requests.Session() as session:
# see https://github.com/psf/requests/blob/39d0fdd9096f7dceccbc8f82e1eda7dd64717a8e/requests/sessions.py#L621
session.verify = False
resp = session.get("url") # $ clientRequestUrlPart="url" MISSING: clientRequestCertValidationDisabled
resp = session.get("url", verify=True) # $ clientRequestUrlPart="url"
req = requests.Request("GET", "url") # $ MISSING: clientRequestUrlPart="url"
resp = session.send(req.prepare()) # $ MISSING: clientRequestCertValidationDisabled

1
python/ql/test/library-tests/frameworks/rest_framework/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
db.sqlite3

0
python/ql/test/library-tests/frameworks/rest_framework/ConceptsTest.expected Normal file
View File

2
python/ql/test/library-tests/frameworks/rest_framework/ConceptsTest.ql Normal file
View File

@@ -0,0 +1,2 @@
import python
import experimental.meta.ConceptsTest

3
python/ql/test/library-tests/frameworks/rest_framework/InlineTaintTest.expected Normal file
View File

@@ -0,0 +1,3 @@
argumentToEnsureNotTaintedNotMarkedAsSpurious
untaintedArgumentToEnsureTaintedNotMarkedAsMissing
failures

1
python/ql/test/library-tests/frameworks/rest_framework/InlineTaintTest.ql Normal file
View File

@@ -0,0 +1 @@
import experimental.meta.InlineTaintTest

23
python/ql/test/library-tests/frameworks/rest_framework/README.md Normal file
View File

@@ -0,0 +1,23 @@
See README for `django-v2-v3` which described how the project was set up.
Since this test project uses models (and a DB), you generally need to run there 3 commands:
```
python manage.py makemigrations
python manage.py migrate
python manage.py runserver
```
Then visit http://127.0.0.1:8000/
# References
- https://www.django-rest-framework.org/tutorial/quickstart/
# Editing data
To edit data you should add an admin user (will prompt for password)
```
python manage.py createsuperuser --email admin@example.com --username admin
```

22
python/ql/test/library-tests/frameworks/rest_framework/manage.py Executable file
View File

@@ -0,0 +1,22 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'testproj.settings')
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == '__main__':
main()

50
python/ql/test/library-tests/frameworks/rest_framework/response_test.py Normal file
View File

@@ -0,0 +1,50 @@
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.exceptions import APIException
@api_view()
def normal_response(request): # $ requestHandler
# has no pre-defined content type, since that will be negotiated
# see https://www.django-rest-framework.org/api-guide/responses/
data = "data"
resp = Response(data) # $ HttpResponse responseBody=data
return resp
@api_view()
def plain_text_response(request): # $ requestHandler
# this response is not the standard way to use the Djagno REST framework, but it
# certainly is possible -- notice that the response contains double quotes
data = 'this response will contain double quotes since it was a string'
resp = Response(data, None, None, None, None, "text/plain") # $ HttpResponse mimetype=text/plain responseBody=data
resp = Response(data=data, content_type="text/plain") # $ HttpResponse mimetype=text/plain responseBody=data
return resp
################################################################################
# Cookies
################################################################################
@api_view
def setting_cookie(request):
resp = Response() # $ HttpResponse
resp.set_cookie("key", "value") # $ CookieWrite CookieName="key" CookieValue="value"
resp.set_cookie(key="key4", value="value") # $ CookieWrite CookieName="key4" CookieValue="value"
resp.headers["Set-Cookie"] = "key2=value2" # $ MISSING: CookieWrite CookieRawHeader="key2=value2"
resp.cookies["key3"] = "value3" # $ CookieWrite CookieName="key3" CookieValue="value3"
resp.delete_cookie("key4") # $ CookieWrite CookieName="key4"
resp.delete_cookie(key="key4") # $ CookieWrite CookieName="key4"
return resp
################################################################################
# Exceptions
################################################################################
# see https://www.django-rest-framework.org/api-guide/exceptions/
@api_view(["GET", "POST"])
def exception_test(request): # $ requestHandler
data = "exception details"
# note: `code details` not exposed by default
code = "code details"
e1 = APIException(data, code) # $ HttpResponse responseBody=data
e2 = APIException(detail=data, code=code) # $ HttpResponse responseBody=data
raise e2

131
python/ql/test/library-tests/frameworks/rest_framework/taint_test.py Normal file
View File

@@ -0,0 +1,131 @@
from rest_framework.decorators import api_view, parser_classes
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.parsers import JSONParser
from django.urls import path
ensure_tainted = ensure_not_tainted = print
# function based view
# see https://www.django-rest-framework.org/api-guide/views/#function-based-views
@api_view(["POST"])
@parser_classes([JSONParser])
def test_taint(request: Request, routed_param): # $ requestHandler routedParameter=routed_param
ensure_tainted(routed_param) # $ tainted
ensure_tainted(request) # $ tainted
# Has all the standard attributes of a django HttpRequest
# see https://github.com/encode/django-rest-framework/blob/00cd4ef864a8bf6d6c90819a983017070f9f08a5/rest_framework/request.py#L410-L418
ensure_tainted(request.resolver_match.args) # $ tainted
# special new attributes added, see https://www.django-rest-framework.org/api-guide/requests/
ensure_tainted(
request.data, # $ tainted
request.data["key"], # $ tainted
# alias for .GET
request.query_params, # $ tainted
request.query_params["key"], # $ tainted
request.query_params.get("key"), # $ tainted
request.query_params.getlist("key"), # $ tainted
request.query_params.getlist("key")[0], # $ tainted
request.query_params.pop("key"), # $ tainted
request.query_params.pop("key")[0], # $ tainted
# see more detailed tests of `request.user` below
request.user, # $ tainted
request.auth, # $ tainted
# seems much more likely attack vector than .method, so included
request.content_type, # $ tainted
# file-like
request.stream, # $ tainted
request.stream.read(), # $ tainted
)
ensure_not_tainted(
# although these could technically be user-controlled, it seems more likely to lead to FPs than interesting results.
request.accepted_media_type,
# In normal Django, if you disable CSRF middleware, you're allowed to use custom
# HTTP methods, like `curl -X FOO <url>`.
# However, with Django REST framework, doing that will yield:
# `{"detail":"Method \"FOO\" not allowed."}`
#
# In the end, since we model a Django REST framework request entirely as a
# extension of a Django request, we're not easily able to remove the taint from
# `.method`.
request.method, # $ SPURIOUS: tainted
)
# --------------------------------------------------------------------------
# request.user
# --------------------------------------------------------------------------
#
# This will normally be an instance of django.contrib.auth.models.User
# (authenticated) so we assume that normally user-controlled fields such as
# username/email is user-controlled, but that password isn't (since it's a hash).
# see https://docs.djangoproject.com/en/3.2/ref/contrib/auth/#fields
ensure_tainted(
request.user.username, # $ tainted
request.user.first_name, # $ tainted
request.user.last_name, # $ tainted
request.user.email, # $ tainted
)
ensure_not_tainted(request.user.password)
return Response("ok") # $ HttpResponse responseBody="ok"
# class based view
# see https://www.django-rest-framework.org/api-guide/views/#class-based-views
class MyClass(APIView):
def initial(self, request, *args, **kwargs): # $ requestHandler
# this method will be called before processing any request
ensure_tainted(request) # $ tainted
def get(self, request: Request, routed_param): # $ requestHandler routedParameter=routed_param
ensure_tainted(routed_param) # $ tainted
# request taint is the same as in function_based_view above
ensure_tainted(
request, # $ tainted
request.data # $ tainted
)
# same as for standard Django view
ensure_tainted(self.args, self.kwargs) # $ tainted
return Response("ok") # $ HttpResponse responseBody="ok"
# fake setup, you can't actually run this
urlpatterns = [
path("test-taint/<routed_param>", test_taint), # $ routeSetup="test-taint/<routed_param>"
path("ClassView/<routed_param>", MyClass.as_view()), # $ routeSetup="ClassView/<routed_param>"
]
# tests with no route-setup, but we can still tell that these are using Django REST
# framework
@api_view(["POST"])
def function_based_no_route(request: Request, possible_routed_param): # $ requestHandler routedParameter=possible_routed_param
ensure_tainted(
request, # $ tainted
possible_routed_param, # $ tainted
)
class ClassBasedNoRoute(APIView):
def get(self, request: Request, possible_routed_param): # $ requestHandler routedParameter=possible_routed_param
ensure_tainted(request, possible_routed_param) # $ tainted

0
python/ql/test/library-tests/frameworks/rest_framework/testapp/__init__.py Normal file
View File

8
python/ql/test/library-tests/frameworks/rest_framework/testapp/admin.py Normal file
View File

@@ -0,0 +1,8 @@
from .models import Foo, Bar
from django.contrib import admin
# Register your models here.
admin.site.register(Foo)
admin.site.register(Bar)

6
python/ql/test/library-tests/frameworks/rest_framework/testapp/apps.py Normal file
View File

@@ -0,0 +1,6 @@
from django.apps import AppConfig
class TestappConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'testapp'

31
python/ql/test/library-tests/frameworks/rest_framework/testapp/migrations/0001_initial.py Normal file
View File

@@ -0,0 +1,31 @@
# Generated by Django 3.2.8 on 2021-10-27 11:54
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Foo',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('title', models.CharField(max_length=100)),
('field_not_displayed', models.IntegerField()),
],
),
migrations.CreateModel(
name='Bar',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('n', models.IntegerField()),
('foo', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='testapp.foo')),
],
),
]

29
python/ql/test/library-tests/frameworks/rest_framework/testapp/migrations/0002_add_dummy_data.py Normal file
View File

@@ -0,0 +1,29 @@
# Generated by Django 3.2.8 on 2021-10-27 12:06
from django.db import migrations
def add_dummy_data(apps, schema_editor):
Foo = apps.get_model("testapp", "Foo")
Bar = apps.get_model("testapp", "Bar")
f1 = Foo(title="example 1", field_not_displayed=10)
f1.save()
f2 = Foo(title="example 2", field_not_displayed=20)
f2.save()
b1 = Bar(n=42, foo=f1)
b1.save()
b2 = Bar(n=43, foo=f1)
b2.save()
b3 = Bar(n=1000, foo=f2)
b3.save()
class Migration(migrations.Migration):
dependencies = [
('testapp', '0001_initial'),
]
operations = [
migrations.RunPython(add_dummy_data),
]

0
python/ql/test/library-tests/frameworks/rest_framework/testapp/migrations/__init__.py Normal file
View File

13
python/ql/test/library-tests/frameworks/rest_framework/testapp/models.py Normal file
View File

@@ -0,0 +1,13 @@
from django.db import models
# Create your models here.
class Foo(models.Model):
title = models.CharField(max_length=100)
field_not_displayed = models.IntegerField()
class Bar(models.Model):
n = models.IntegerField()
foo = models.ForeignKey(Foo, on_delete=models.PROTECT)

14
python/ql/test/library-tests/frameworks/rest_framework/testapp/serializers.py Normal file
View File

@@ -0,0 +1,14 @@
from .models import Foo, Bar
from rest_framework import serializers
class FooSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = Foo
fields = ["title"]
class BarSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = Bar
fields = ["n", "foo"]

Some files were not shown because too many files have changed in this diff Show More