chore(iast): add sqli integration tests (#7992)

IAST: SQL injection integration tests with different ORMs

## Checklist
- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)
- [x] If this PR touches code that signs or publishes builds or
packages, or handles credentials of any kind, I've requested a review
from `@DataDog/security-design-and-guidance`.
- [x] This PR doesn't touch any of that.

Author: gnufede

.circleci/config.templ.yml

@@ -436,6 +436,12 @@ jobs:
           snapshot: true
           docker_services: "postgres"
 
+  appsec_iast_tdd_propagation:
+    <<: *machine_executor
+    steps:
+      - run_test:
+          pattern: 'appsec_iast_tdd_propagation'
+
   appsec_iast_memcheck:
     <<: *machine_executor
     steps:

.riot/requirements/1ce36b6.txt

@@ -0,0 +1,52 @@
+#
+# This file is autogenerated by pip-compile with Python 3.11
+# by the following command:
+#
+#    pip-compile --no-annotate .riot/requirements/1ce36b6.in
+#
+aiosqlite==0.17.0
+attrs==23.2.0
+blinker==1.7.0
+bytecode==0.15.1
+cattrs==22.2.0
+certifi==2023.11.17
+charset-normalizer==3.3.2
+click==8.1.7
+coverage[toml]==7.4.0
+ddsketch==2.0.4
+deprecated==1.2.14
+envier==0.5.0
+flask==3.0.0
+greenlet==3.0.3
+hypothesis==6.45.0
+idna==3.6
+importlib-metadata==6.11.0
+iniconfig==2.0.0
+iso8601==1.1.0
+itsdangerous==2.1.2
+jinja2==3.1.2
+markupsafe==2.1.3
+mock==5.1.0
+opentelemetry-api==1.22.0
+opentracing==2.4.0
+packaging==23.2
+peewee==3.17.0
+pluggy==1.3.0
+pony==0.7.17
+protobuf==4.25.1
+pypika-tortoise==0.1.6
+pytest==7.4.4
+pytest-cov==4.1.0
+pytest-mock==3.12.0
+pytz==2023.3.post1
+requests==2.31.0
+six==1.16.0
+sortedcontainers==2.4.0
+sqlalchemy==2.0.25
+tortoise-orm==0.20.0
+typing-extensions==4.9.0
+urllib3==2.1.0
+werkzeug==3.0.1
+wrapt==1.16.0
+xmltodict==0.13.0
+zipp==3.17.0

riotfile.py

@@ -184,6 +184,35 @@ def select_pys(min_version=MIN_PYTHON_VERSION, max_version=MAX_PYTHON_VERSION):
                 "_DD_APPSEC_DEDUPLICATION_ENABLED": "false",
             },
         ),
+        Venv(
+            name="appsec_iast_tdd_propagation",
+            pys=select_pys(min_version="3.11", max_version="3.11"),
+            command="pytest --no-cov tests/appsec/iast_tdd_propagation/",
+            pkgs={
+                "flask": "~=3.0",
+                "sqlalchemy": "~=2.0.23",
+                "pony": latest,
+                "aiosqlite": latest,
+                "tortoise-orm": latest,
+                "peewee": latest,
+                "requests": latest,
+                "six": ">=1.12.0",
+                "envier": "==0.5.0",
+                "cattrs": "<23.1.1",
+                "ddsketch": ">=2.0.1",
+                "protobuf": ">=3",
+                "attrs": ">=20",
+                "typing_extensions": latest,
+                "xmltodict": ">=0.12",
+                "opentelemetry-api": ">=1",
+                "opentracing": ">=2.0.0",
+                "bytecode": latest,
+            },
+            env={
+                "DD_IAST_REQUEST_SAMPLING": "100",  # Override default 30% to analyze all IAST requests
+                "_DD_APPSEC_DEDUPLICATION_ENABLED": "false",
+            },
+        ),
         Venv(
             name="appsec_integrations",
             command="pytest {cmdargs} tests/appsec/integrations/",

tests/.suitespec.json

@@ -407,6 +407,14 @@
             "@appsec_iast",
             "tests/appsec/iast/*"
         ],
+        "appsec_iast_tdd_propagation": [
+            "@bootstrap",
+            "@core",
+            "@tracing",
+            "@appsec",
+            "@appsec_iast",
+            "tests/appsec/iast_tdd_propagation/*"
+        ],
         "appsec_iast_memcheck": [
             "@bootstrap",
             "@core",

tests/appsec/appsec_utils.py

@@ -4,6 +4,8 @@
 import subprocess
 import sys
 
+from requests.exceptions import ConnectionError
+
 from ddtrace.internal.utils.retry import RetryError
 from ddtrace.vendor import psutil
 from tests.webclient import Client
@@ -13,12 +15,15 @@
 ROOT_PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 
-def _build_env():
+def _build_env(env=None):
     environ = dict(PATH="%s:%s" % (ROOT_PROJECT_DIR, ROOT_DIR), PYTHONPATH="%s:%s" % (ROOT_PROJECT_DIR, ROOT_DIR))
     if os.environ.get("PATH"):
         environ["PATH"] = "%s:%s" % (os.environ.get("PATH"), environ["PATH"])
     if os.environ.get("PYTHONPATH"):
         environ["PYTHONPATH"] = "%s:%s" % (os.environ.get("PYTHONPATH"), environ["PYTHONPATH"])
+    if env:
+        for k, v in env.items():
+            environ[k] = v
     return environ
 
 
@@ -36,16 +41,23 @@ def gunicorn_server(appsec_enabled="true", remote_configuration_enabled="true",
 
 @contextmanager
 def flask_server(
-    appsec_enabled="true", remote_configuration_enabled="true", iast_enabled="false", tracer_enabled="true", token=None
+    appsec_enabled="true",
+    remote_configuration_enabled="true",
+    iast_enabled="false",
+    tracer_enabled="true",
+    token=None,
+    app="tests/appsec/app.py",
+    env=None,
 ):
-    cmd = ["python", "tests/appsec/app.py", "--no-reload"]
+    cmd = ["python", app, "--no-reload"]
     yield from appsec_application_server(
         cmd,
         appsec_enabled=appsec_enabled,
         remote_configuration_enabled=remote_configuration_enabled,
         iast_enabled=iast_enabled,
         tracer_enabled=tracer_enabled,
         token=token,
+        env=env,
     )
 
 
@@ -56,8 +68,9 @@ def appsec_application_server(
     iast_enabled="false",
     tracer_enabled="true",
     token=None,
+    env=None,
 ):
-    env = _build_env()
+    env = _build_env(env)
     env["DD_REMOTE_CONFIG_POLL_INTERVAL_SECONDS"] = "0.5"
     env["DD_REMOTE_CONFIGURATION_ENABLED"] = remote_configuration_enabled
     if token:
@@ -77,7 +90,7 @@ def appsec_application_server(
         env=env,
         stdout=sys.stdout,
         stderr=sys.stderr,
-        preexec_fn=os.setsid,
+        start_new_session=True,
     )
     try:
         client = Client("http://0.0.0.0:8000")
@@ -107,6 +120,8 @@ def appsec_application_server(
         yield server_process, client, (children[1].pid if len(children) > 1 else None)
         try:
             client.get_ignored("/shutdown")
+        except ConnectionError:
+            pass
         except Exception:
             raise AssertionError(
                 "\n=== Captured STDOUT ===\n%s=== End of captured STDOUT ==="

tests/appsec/iast_tdd_propagation/flask_orm_app.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+
+""" This Flask application is imported on tests.appsec.appsec_utils.gunicorn_server
+"""
+
+
+import importlib
+import os
+import sys
+
+from flask import Flask
+from flask import request
+
+from ddtrace import tracer
+from ddtrace.appsec._constants import IAST
+from ddtrace.appsec._iast import ddtrace_iast_flask_patch
+from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
+from ddtrace.internal import core
+
+
+import ddtrace.auto  # noqa: F401  # isort: skip
+
+orm = os.getenv("FLASK_ORM", "sqlite")
+
+orm_impl = importlib.import_module(f"{orm}_impl")
+
+
+app = Flask(__name__)
+
+
+class ResultResponse:
+    param = ""
+    sources = ""
+    vulnerabilities = ""
+
+    def __init__(self, param):
+        self.param = param
+
+    def json(self):
+        return {
+            "param": self.param,
+            "sources": self.sources,
+            "vulnerabilities": self.vulnerabilities,
+            "params_are_tainted": is_pyobject_tainted(self.param),
+        }
+
+
+@app.route("/shutdown")
+def shutdown():
+    tracer.shutdown()
+    sys.exit(0)
+
+
+@app.route("/")
+def tainted_view():
+    param = request.args.get("param", "param")
+
+    report = core.get_items([IAST.CONTEXT_KEY], tracer.current_root_span())
+
+    assert not (report and report[0])
+
+    orm_impl.execute_query("select * from User where name = '" + param + "'")
+
+    response = ResultResponse(param)
+    report = core.get_items([IAST.CONTEXT_KEY], tracer.current_root_span())
+    if report and report[0]:
+        response.sources = report[0].sources[0].value
+        response.vulnerabilities = list(report[0].vulnerabilities)[0].type
+
+    return response.json()
+
+
+@app.route("/untainted")
+def untainted_view():
+    param = request.args.get("param", "param")
+
+    report = core.get_items([IAST.CONTEXT_KEY], tracer.current_root_span())
+
+    assert not (report and report[0])
+
+    orm_impl.execute_untainted_query("select * from User where name = '" + param + "'")
+
+    response = ResultResponse(param)
+    report = core.get_items([IAST.CONTEXT_KEY], tracer.current_root_span())
+    if report and report[0]:
+        response.sources = report[0].sources[0].value
+        response.vulnerabilities = list(report[0].vulnerabilities)[0].type
+
+    return response.json()
+
+
+if __name__ == "__main__":
+    ddtrace_iast_flask_patch()
+    app.run(debug=False, port=8000)

tests/appsec/iast_tdd_propagation/peewee_impl.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+
+from peewee import CharField
+from peewee import IntegerField
+from peewee import Model
+from peewee import SqliteDatabase
+
+
+db = SqliteDatabase(":sharedmemory:")
+
+
+class User(Model):
+    id = IntegerField(primary_key=True)
+    name = CharField()
+
+    class Meta:
+        database = db
+
+
+db.create_tables([User])
+
+
+def execute_query(param):
+    db.execute_sql(param)
+
+
+def execute_untainted_query(_):
+    db.execute_sql("SELECT * FROM User")

tests/appsec/iast_tdd_propagation/pony_impl.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+from pony import orm
+
+
+db = orm.Database()
+
+
+class User(db.Entity):
+    name = orm.Required(str)
+
+
+db.bind(provider="sqlite", filename=":sharedmemory:", create_db=True)
+db.generate_mapping(create_tables=True)
+
+
+def execute_query(param):
+    with orm.db_session:
+        return User.select_by_sql(param)
+
+
+def execute_untainted_query(_):
+    with orm.db_session:
+        return User.select_by_sql("SELECT * from User")

tests/appsec/iast_tdd_propagation/sqlalchemy_impl.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import create_engine
+from sqlalchemy import text
+from sqlalchemy.orm import declarative_base
+from sqlalchemy.pool import StaticPool
+
+
+engine = create_engine("sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "User"
+    id = Column(Integer, primary_key=True)
+    name = Column(String(30), nullable=True)
+
+
+User.metadata.create_all(engine)
+
+
+def execute_query(param):
+    with engine.connect() as connection:
+        connection.execute(text(param))
+
+
+def execute_untainted_query(_):
+    with engine.connect() as connection:
+        connection.execute(text("SELECT * FROM User"))

tests/appsec/iast_tdd_propagation/sqlite_impl.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+
+import sqlite3
+
+
+conn = sqlite3.connect(":memory:", check_same_thread=False)
+cursor = conn.cursor()
+
+cursor.execute("CREATE TABLE User (id INTEGER PRIMARY KEY, name TEXT)")
+
+
+def execute_query(param):
+    cursor.execute(param)
+
+
+def execute_untainted_query(_):
+    cursor.execute("SELECT 1")