feat(asm): add sqli support for exploit prevention (#9450)

Duplicate of #9443 with core
api integration instead of wrappers.

add support for sqli detection and prevention for exploit prevention
add unit tests
will also be tested by system-tests sqli rasp tests.
(DataDog/system-tests#2514)

APPSEC-53421

## 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] Risks are described (performance impact, potential for breakage,
maintainability)
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed or label `changelog/no-changelog` is set
- [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))
- [x] If this PR changes the public interface, I've notified
`@DataDog/apm-tees`.

## Reviewer Checklist

- [x] Title is accurate
- [x] All changes are related to the pull request's stated goal
- [x] Description motivates each change
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- [x] Testing strategy adequately addresses listed risks
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] Release note makes sense to a user of the library
- [x] Author has 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)

ddtrace/appsec/_common_module_patches.py

@@ -28,6 +28,7 @@
 def patch_common_modules():
     try_wrap_function_wrapper("builtins", "open", wrapped_open_CFDDB7ABBA9081B6)
     try_wrap_function_wrapper("urllib.request", "OpenerDirector.open", wrapped_open_ED4CF71136E15EBF)
+    core.on("asm.block.dbapi.execute", execute_4C9BAC8E228EB347)
     if asm_config._iast_enabled:
         _set_metric_iast_instrumented_sink(VULN_PATH_TRAVERSAL)
 
@@ -139,6 +140,49 @@ def wrapped_request_D8CB81E472AF98A2(original_request_callable, instance, args,
     return original_request_callable(*args, **kwargs)
 
 
+_DB_DIALECTS = {
+    "mariadb": "mariadb",
+    "mysql": "mysql",
+    "postgres": "postgresql",
+    "pymysql": "mysql",
+    "pyodbc": "odbc",
+    "sql": "sql",
+    "sqlite": "sqlite",
+    "vertica": "vertica",
+}
+
+
+def execute_4C9BAC8E228EB347(instrument_self, query, args, kwargs) -> None:
+    """
+    listener for dbapi execute and executemany function
+    parameters are ignored as they are properly handled by the dbapi without risk of injections
+    """
+    if asm_config._asm_enabled and asm_config._ep_enabled:
+        try:
+            from ddtrace.appsec._asm_request_context import call_waf_callback
+            from ddtrace.appsec._asm_request_context import in_context
+            from ddtrace.appsec._constants import EXPLOIT_PREVENTION
+        except ImportError:
+            # execute is used during module initialization
+            # and shouldn't be changed at that time
+            return
+
+        if instrument_self and query and in_context():
+            db_type = _DB_DIALECTS.get(
+                getattr(instrument_self, "_self_config", {}).get("_dbapi_span_name_prefix", ""), ""
+            )
+            if isinstance(query, str):
+                res = call_waf_callback(
+                    {EXPLOIT_PREVENTION.ADDRESS.SQLI: query, EXPLOIT_PREVENTION.ADDRESS.SQLI_TYPE: db_type},
+                    crop_trace="execute_4C9BAC8E228EB347",
+                    rule_type=EXPLOIT_PREVENTION.TYPE.SQLI,
+                )
+                if res and WAF_ACTIONS.BLOCK_ACTION in res.actions:
+                    raise BlockingException(
+                        core.get_item(WAF_CONTEXT_NAMES.BLOCKED), "exploit_prevention", "sqli", query
+                    )
+
+
 def try_unwrap(module, name):
     try:
         (parent, attribute, _) = resolve_path(module, name)

ddtrace/appsec/_constants.py

@@ -136,6 +136,8 @@ class WAF_DATA_NAMES(metaclass=Constant_Class):
     PROCESSOR_SETTINGS = "waf.context.processor"
     LFI_ADDRESS = "server.io.fs.file"
     SSRF_ADDRESS = "server.io.net.url"
+    SQLI_ADDRESS = "server.db.statement"
+    SQLI_SYSTEM_ADDRESS = "server.db.system"
 
 
 class SPAN_DATA_NAMES(metaclass=Constant_Class):
@@ -262,3 +264,5 @@ class TYPE(metaclass=Constant_Class):
     class ADDRESS(metaclass=Constant_Class):
         LFI = "LFI_ADDRESS"
         SSRF = "SSRF_ADDRESS"
+        SQLI = "SQLI_ADDRESS"
+        SQLI_TYPE = "SQLI_SYSTEM_ADDRESS"

ddtrace/contrib/dbapi/__init__.py

@@ -142,6 +142,7 @@ def executemany(self, query, *args, **kwargs):
         #      These differences should be overridden at the integration specific layer (e.g. in `sqlite3/patch.py`)
         # FIXME[matt] properly handle kwargs here. arg names can be different
         # with different libs.
+        core.dispatch("asm.block.dbapi.execute", (self, query, args, kwargs))
         return self._trace_method(
             self.__wrapped__.executemany,
             self._self_datadog_name,
@@ -160,6 +161,7 @@ def execute(self, query, *args, **kwargs):
         # Always return the result as-is
         # DEV: Some libraries return `None`, others `int`, and others the cursor objects
         #      These differences should be overridden at the integration specific layer (e.g. in `sqlite3/patch.py`)
+        core.dispatch("asm.block.dbapi.execute", (self, query, args, kwargs))
         return self._trace_method(
             self.__wrapped__.execute,
             self._self_datadog_name,

releasenotes/notes/exploit_prevention_sqli-c34e1047af3c08f2.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    ASM: This introduces SQL injection support for exploit prevention.

tests/appsec/appsec/rules-rasp-blocking.json

@@ -101,6 +101,57 @@
         "stack_trace",
         "block"
       ]
+    },
+    {
+      "id": "rasp-942-100",
+      "name": "SQL injection exploit",
+      "tags": {
+        "type": "sql_injection",
+        "category": "vulnerability_trigger",
+        "cwe": "89",
+        "capec": "1000/152/248/66",
+        "confidence": "0",
+        "module": "rasp"
+      },
+      "conditions": [
+        {
+          "parameters": {
+            "resource": [
+              {
+                "address": "server.db.statement"
+              }
+            ],
+            "params": [
+              {
+                "address": "server.request.query"
+              },
+              {
+                "address": "server.request.body"
+              },
+              {
+                "address": "server.request.path_params"
+              },
+              {
+                "address": "graphql.server.all_resolvers"
+              },
+              {
+                "address": "graphql.server.resolver"
+              }
+            ],
+            "db_type": [
+              {
+                "address": "server.db.system"
+              }
+            ]
+          },
+          "operator": "sqli_detector"
+        }
+      ],
+      "transformers": [],
+      "on_match": [
+        "stack_trace",
+        "block"
+      ]
     }
   ]
 }

tests/appsec/appsec/rules-rasp.json

@@ -99,6 +99,56 @@
       "on_match": [
         "stack_trace"
       ]
+    },
+    {
+      "id": "rasp-942-100",
+      "name": "SQL injection exploit",
+      "tags": {
+        "type": "sql_injection",
+        "category": "vulnerability_trigger",
+        "cwe": "89",
+        "capec": "1000/152/248/66",
+        "confidence": "0",
+        "module": "rasp"
+      },
+      "conditions": [
+        {
+          "parameters": {
+            "resource": [
+              {
+                "address": "server.db.statement"
+              }
+            ],
+            "params": [
+              {
+                "address": "server.request.query"
+              },
+              {
+                "address": "server.request.body"
+              },
+              {
+                "address": "server.request.path_params"
+              },
+              {
+                "address": "graphql.server.all_resolvers"
+              },
+              {
+                "address": "graphql.server.resolver"
+              }
+            ],
+            "db_type": [
+              {
+                "address": "server.db.system"
+              }
+            ]
+          },
+          "operator": "sqli_detector"
+        }
+      ],
+      "transformers": [],
+      "on_match": [
+        "stack_trace"
+      ]
     }
   ]
 }

tests/appsec/contrib_appsec/django_app/urls.py

@@ -1,3 +1,4 @@
+import sqlite3
 import tempfile
 
 import django
@@ -58,6 +59,13 @@ def multi_view(request, param_int=0, param_str=""):
     return json_response
 
 
+DB = sqlite3.connect(":memory:")
+DB.execute("CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT)")
+DB.execute("INSERT INTO users (id, name) VALUES ('1_secret_id', 'Alice')")
+DB.execute("INSERT INTO users (id, name) VALUES ('2_secret_id', 'Bob')")
+DB.execute("INSERT INTO users (id, name) VALUES ('3_secret_id', 'Christophe')")
+
+
 @csrf_exempt
 def rasp(request, endpoint: str):
     query_params = request.GET.dict()
@@ -101,6 +109,19 @@ def rasp(request, endpoint: str):
                     res.append(f"Error: {e}")
         tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
         return HttpResponse("<\\br>\n".join(res))
+    elif endpoint == "sql_injection":
+        res = ["sql_injection endpoint"]
+        for param in query_params:
+            if param.startswith("user_id"):
+                user_id = query_params[param]
+            try:
+                if param.startswith("user_id"):
+                    cursor = DB.execute(f"SELECT * FROM users WHERE id = {user_id}")
+                    res.append(f"Url: {list(cursor)}")
+            except Exception as e:
+                res.append(f"Error: {e}")
+        tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
+        return HttpResponse("<\\br>\n".join(res))
     elif endpoint == "shell":
         res = ["shell endpoint"]
         for param in query_params:

tests/appsec/contrib_appsec/fastapi_app/app.py

@@ -1,4 +1,5 @@
 import asyncio
+import sqlite3
 from typing import Optional
 
 from fastapi import FastAPI
@@ -117,6 +118,11 @@ async def stream():
     @app.post("/rasp/{endpoint:str}/")
     @app.options("/rasp/{endpoint:str}/")
     async def rasp(endpoint: str, request: Request):
+        DB = sqlite3.connect(":memory:")
+        DB.execute("CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT)")
+        DB.execute("INSERT INTO users (id, name) VALUES ('1_secret_id', 'Alice')")
+        DB.execute("INSERT INTO users (id, name) VALUES ('2_secret_id', 'Bob')")
+        DB.execute("INSERT INTO users (id, name) VALUES ('3_secret_id', 'Christophe')")
         query_params = request.query_params
         if endpoint == "lfi":
             res = ["lfi endpoint"]
@@ -158,6 +164,19 @@ async def rasp(endpoint: str, request: Request):
                     res.append(f"Error: {e}")
             tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
             return HTMLResponse("<\\br>\n".join(res))
+        elif endpoint == "sql_injection":
+            res = ["sql_injection endpoint"]
+            for param in query_params:
+                if param.startswith("user_id"):
+                    user_id = query_params[param]
+                try:
+                    if param.startswith("user_id"):
+                        cursor = DB.execute(f"SELECT * FROM users WHERE id = {user_id}")
+                        res.append(f"Url: {list(cursor)}")
+                except Exception as e:
+                    res.append(f"Error: {e}")
+            tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
+            return HTMLResponse("<\\br>\n".join(res))
         elif endpoint == "shell":
             res = ["shell endpoint"]
             for param in query_params:

tests/appsec/contrib_appsec/flask_app/app.py

@@ -1,4 +1,5 @@
 import os
+import sqlite3
 
 from flask import Flask
 from flask import request
@@ -61,6 +62,13 @@ def new_service(service_name: str):
     return service_name
 
 
+DB = sqlite3.connect(":memory:")
+DB.execute("CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT)")
+DB.execute("INSERT INTO users (id, name) VALUES ('1_secret_id', 'Alice')")
+DB.execute("INSERT INTO users (id, name) VALUES ('2_secret_id', 'Bob')")
+DB.execute("INSERT INTO users (id, name) VALUES ('3_secret_id', 'Christophe')")
+
+
 @app.route("/rasp/<string:endpoint>/", methods=["GET", "POST", "OPTIONS"])
 def rasp(endpoint: str):
     query_params = request.args
@@ -104,6 +112,19 @@ def rasp(endpoint: str):
                 res.append(f"Error: {e}")
         tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
         return "<\\br>\n".join(res)
+    elif endpoint == "sql_injection":
+        res = ["sql_injection endpoint"]
+        for param in query_params:
+            if param.startswith("user_id"):
+                user_id = query_params[param]
+            try:
+                if param.startswith("user_id"):
+                    cursor = DB.execute(f"SELECT * FROM users WHERE id = {user_id}")
+                    res.append(f"Url: {list(cursor)}")
+            except Exception as e:
+                res.append(f"Error: {e}")
+        tracer.current_span()._local_root.set_tag("rasp.request.done", endpoint)
+        return "<\\br>\n".join(res)
     elif endpoint == "shell":
         res = ["shell endpoint"]
         for param in query_params:

tests/appsec/contrib_appsec/utils.py

@@ -1198,7 +1198,8 @@ def test_stream_response(
                 ],
                 repeat=2,
             )
-        ],
+        ]
+        + [("sql_injection", "user_id_1=1 OR 1=1&user_id_2=1 OR 1=1", "rasp-942-100", ("dispatch",))],
     )
     @pytest.mark.parametrize(
         ("rule_file", "blocking"),
@@ -1232,6 +1233,7 @@ def test_exploit_prevention(
             dict(DD_APPSEC_RULES=rule_file)
         ), mock_patch("ddtrace.internal.telemetry.metrics_namespaces.MetricNamespace.add_metric") as mocked:
             self.update_tracer(interface)
+            assert asm_config._asm_enabled == asm_enabled
             response = interface.client.get(f"/rasp/{endpoint}/?{parameters}")
             code = 403 if blocking and asm_enabled and ep_enabled else 200
             assert self.status(response) == code, (self.status(response), code)