chore(iast): packages tests (#7856)

This PR introduces base structure to build packages tests. 


## 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: avara1986

tests/appsec/integrations/app.py renamed to tests/appsec/app.py

@@ -1,5 +1,6 @@
 """ This Flask application is imported on tests.appsec.appsec_utils.gunicorn_server
 """
+
 import subprocess  # nosec
 
 from flask import Flask
@@ -8,9 +9,11 @@
 
 
 import ddtrace.auto  # noqa: F401  # isort: skip
+from tests.appsec.iast_packages.packages.pkg_requests import pkg_requests
 
 
 app = Flask(__name__)
+app.register_blueprint(pkg_requests)
 
 
 @app.route("/")
@@ -43,4 +46,4 @@ def iast_cmdi_vulnerability():
 
 
 if __name__ == "__main__":
-    app.run(debug=True, port=8000)
+    app.run(debug=False, port=8000)

tests/appsec/appsec_utils.py

@@ -24,7 +24,7 @@ def _build_env():
 
 @contextmanager
 def gunicorn_server(appsec_enabled="true", remote_configuration_enabled="true", tracer_enabled="true", token=None):
-    cmd = ["gunicorn", "-w", "3", "-b", "0.0.0.0:8000", "tests.appsec.integrations.app:app"]
+    cmd = ["gunicorn", "-w", "3", "-b", "0.0.0.0:8000", "tests.appsec.app:app"]
     yield from appsec_application_server(
         cmd,
         appsec_enabled=appsec_enabled,
@@ -38,7 +38,7 @@ def gunicorn_server(appsec_enabled="true", remote_configuration_enabled="true",
 def flask_server(
     appsec_enabled="true", remote_configuration_enabled="true", iast_enabled="false", tracer_enabled="true", token=None
 ):
-    cmd = ["python", "tests/appsec/integrations/app.py", "--no-reload"]
+    cmd = ["python", "tests/appsec/app.py", "--no-reload"]
     yield from appsec_application_server(
         cmd,
         appsec_enabled=appsec_enabled,
@@ -66,6 +66,8 @@ def appsec_application_server(
         env["DD_APPSEC_ENABLED"] = appsec_enabled
     if iast_enabled is not None and iast_enabled != "false":
         env["DD_IAST_ENABLED"] = iast_enabled
+        env["DD_IAST_REQUEST_SAMPLING"] = "100"
+        env["_DD_APPSEC_DEDUPLICATION_ENABLED"] = "false"
     if tracer_enabled is not None:
         env["DD_TRACE_ENABLED"] = tracer_enabled
     env["DD_TRACE_AGENT_URL"] = os.environ.get("DD_TRACE_AGENT_URL", "")
@@ -91,6 +93,13 @@ def appsec_application_server(
                 "\n=== Captured STDERR ===\n%s=== End of captured STDERR ==="
                 % (server_process.stdout, server_process.stderr)
             )
+        except Exception:
+            raise AssertionError(
+                "Server FAILED, see stdout and stderr logs"
+                "\n=== Captured STDOUT ===\n%s=== End of captured STDOUT ==="
+                "\n=== Captured STDERR ===\n%s=== End of captured STDERR ==="
+                % (server_process.stdout, server_process.stderr)
+            )
         # If we run a Gunicorn application, we want to get the child's pid, see test_remoteconfiguration_e2e.py
         parent = psutil.Process(server_process.pid)
         children = parent.children(recursive=True)

tests/appsec/iast_packages/__init__.py

@@ -0,0 +1,23 @@
+"""
+requests==2.31.0
+
+https://pypi.org/project/requests/
+"""
+from flask import Blueprint
+from flask import request
+import requests
+
+from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
+
+
+pkg_requests = Blueprint("package_requests", __name__)
+
+
+@pkg_requests.route("/requests")
+def pkg_requests_view():
+    package_param = request.args.get("package_param")
+    try:
+        requests.get("http://" + package_param)
+    except Exception:
+        pass
+    return {"param": package_param, "params_are_tainted": is_pyobject_tainted(package_param)}

tests/appsec/iast_packages/packages/__init__.py

@@ -0,0 +1,25 @@
+"""
+[package_name]==[version]
+
+https://pypi.org/project/[package_name]/
+
+[Description]
+"""
+from flask import Blueprint, request
+from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
+
+pkg_[package_name] = Blueprint('package_[package_name]', __name__)
+
+
+@pkg_[package_name].route('/[package_name]')
+def pkg_[package_name]_view():+
+    import [package_name]
+    package_param = request.args.get("package_param")
+
+    [CODE]
+
+    return {
+        "param": package_param,
+        "params_are_tainted": is_pyobject_tainted(package_param)
+    }
+

tests/appsec/iast_packages/packages/pkg_requests.py

@@ -0,0 +1,62 @@
+import json
+
+import pytest
+
+from tests.appsec.appsec_utils import flask_server
+
+
+class PackageForTesting:
+    package_name = ""
+    package_version = ""
+    url_to_test = ""
+    expected_param = "test1234"
+    xfail = False
+
+    def __init__(self, name):
+        self.package_name = name
+
+    @property
+    def url(self):
+        return f"/{self.package_name}?package_param={self.expected_param}"
+
+    def __repr__(self):
+        return f"{self.package_name}: {self.url_to_test}"
+
+
+PACKAGES = [
+    PackageForTesting("requests"),
+]
+
+
+def setup():
+    for package in PACKAGES:
+        with flask_server(
+            iast_enabled="false", tracer_enabled="true", remote_configuration_enabled="false", token=None
+        ) as context:
+            _, client, pid = context
+
+            try:
+                response = client.get(package.url)
+
+                assert response.status_code == 200
+                content = json.loads(response.content)
+                assert content["param"] == package.expected_param
+                assert content["params_are_tainted"] is False
+            except Exception:
+                package.xfail = True
+
+
+@pytest.mark.parametrize("package", PACKAGES)
+def test_patched(package):
+    if package.xfail:
+        pytest.xfail("Initial test failed for package: {}".format(package))
+
+    with flask_server(iast_enabled="true", remote_configuration_enabled="false", token=None) as context:
+        _, client, pid = context
+
+        response = client.get(package.url)
+
+        assert response.status_code == 200
+        content = json.loads(response.content)
+        assert content["param"] == package.expected_param
+        assert content["params_are_tainted"] is True

tests/appsec/iast_packages/packages/template.py.tpl

@@ -5,13 +5,11 @@
 
 from tests.appsec.appsec_utils import flask_server
 from tests.appsec.appsec_utils import gunicorn_server
-from tests.utils import flaky
 
 
 @pytest.mark.parametrize("appsec_enabled", ("true", "false"))
 @pytest.mark.parametrize("tracer_enabled", ("true", "false"))
 @pytest.mark.parametrize("server", ((gunicorn_server, flask_server)))
-@flaky(until=1704067200)
 def test_when_appsec_reads_chunked_requests(appsec_enabled, tracer_enabled, server):
     def read_in_chunks(filepath, chunk_size=1024):
         file_object = open(filepath, "rb")
@@ -73,7 +71,6 @@ def test_corner_case_when_appsec_reads_chunked_request_with_no_body(appsec_enabl
 @pytest.mark.parametrize("appsec_enabled", ("true", "false"))
 @pytest.mark.parametrize("tracer_enabled", ("true", "false"))
 @pytest.mark.parametrize("server", ((gunicorn_server, flask_server)))
-@flaky(until=1704067200)
 def test_when_appsec_reads_empty_body_no_hang(appsec_enabled, tracer_enabled, server):
     """A bug was detected when running a Flask application locally
 

tests/appsec/iast_packages/test_packages.py

@@ -1,19 +1,13 @@
 import psycopg2.extensions as ext
-import pytest
 
+from ddtrace.appsec._iast import oce
+from ddtrace.appsec._iast._taint_tracking import OriginType
+from ddtrace.appsec._iast._taint_tracking import create_context
+from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
+from ddtrace.appsec._iast._taint_utils import LazyTaintList
 from tests.utils import override_global_config
 
 
-try:
-    from ddtrace.appsec._iast import oce
-    from ddtrace.appsec._iast._taint_tracking import OriginType
-    from ddtrace.appsec._iast._taint_tracking import create_context
-    from ddtrace.appsec._iast._taint_tracking import is_pyobject_tainted
-    from ddtrace.appsec._iast._taint_utils import LazyTaintList
-except (ImportError, AttributeError):
-    pytest.skip("IAST not supported for this Python version", allow_module_level=True)
-
-
 def setup():
     create_context()
     oce._enabled = True