chore(iast): remove DD_IAST_TELEMETRY_VERBOSITY env var usage

ddtrace/appsec/_constants.py

@@ -111,20 +111,39 @@ class APPSEC(metaclass=Constant_Class):
     ] = "DD_APPSEC_OBFUSCATION_PARAMETER_VALUE_REGEXP"
 
 
+TELEMETRY_OFF_NAME = "OFF"
+TELEMETRY_DEBUG_NAME = "DEBUG"
+TELEMETRY_MANDATORY_NAME = "MANDATORY"
+TELEMETRY_INFORMATION_NAME = "INFORMATION"
+
+TELEMETRY_DEBUG_VERBOSITY = 10
+TELEMETRY_INFORMATION_VERBOSITY = 20
+TELEMETRY_MANDATORY_VERBOSITY = 30
+TELEMETRY_OFF_VERBOSITY = 40
+
+
 class IAST(metaclass=Constant_Class):
     """Specific constants for IAST"""
 
     ENV: Literal["DD_IAST_ENABLED"] = "DD_IAST_ENABLED"
     ENV_DEBUG: Literal["DD_IAST_DEBUG"] = "DD_IAST_DEBUG"
     ENV_PROPAGATION_DEBUG: Literal["DD_IAST_PROPAGATION_DEBUG"] = "DD_IAST_PROPAGATION_DEBUG"
     ENV_REQUEST_SAMPLING: Literal["DD_IAST_REQUEST_SAMPLING"] = "DD_IAST_REQUEST_SAMPLING"
-    TELEMETRY_REPORT_LVL: Literal["DD_IAST_TELEMETRY_VERBOSITY"] = "DD_IAST_TELEMETRY_VERBOSITY"
+    ENV_TELEMETRY_REPORT_LVL: Literal["DD_IAST_TELEMETRY_VERBOSITY"] = "DD_IAST_TELEMETRY_VERBOSITY"
     LAZY_TAINT: Literal["_DD_IAST_LAZY_TAINT"] = "_DD_IAST_LAZY_TAINT"
     JSON: Literal["_dd.iast.json"] = "_dd.iast.json"
     ENABLED: Literal["_dd.iast.enabled"] = "_dd.iast.enabled"
     PATCH_MODULES: Literal["_DD_IAST_PATCH_MODULES"] = "_DD_IAST_PATCH_MODULES"
     DENY_MODULES: Literal["_DD_IAST_DENY_MODULES"] = "_DD_IAST_DENY_MODULES"
     SEP_MODULES: Literal[","] = ","
+
+    METRICS_REPORT_LVLS = (
+        (TELEMETRY_DEBUG_VERBOSITY, TELEMETRY_DEBUG_NAME),
+        (TELEMETRY_INFORMATION_VERBOSITY, TELEMETRY_INFORMATION_NAME),
+        (TELEMETRY_MANDATORY_VERBOSITY, TELEMETRY_MANDATORY_NAME),
+        (TELEMETRY_OFF_VERBOSITY, TELEMETRY_OFF_NAME),
+    )
+
     TEXT_TYPES = (str, bytes, bytearray)
     TAINTEABLE_TYPES = (str, bytes, bytearray, Match, BytesIO, StringIO)
 

ddtrace/appsec/_iast/_metrics.py

@@ -1,45 +1,30 @@
-import os
 import sys
 import traceback
 from typing import Dict
 from typing import Text
 
 from ddtrace.appsec._constants import IAST
 from ddtrace.appsec._constants import IAST_SPAN_TAGS
+from ddtrace.appsec._constants import TELEMETRY_INFORMATION_VERBOSITY
+from ddtrace.appsec._constants import TELEMETRY_MANDATORY_VERBOSITY
 from ddtrace.appsec._deduplications import deduplication
 from ddtrace.appsec._iast._utils import _is_iast_debug_enabled
 from ddtrace.internal import telemetry
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.telemetry.constants import TELEMETRY_LOG_LEVEL
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE_TAG_IAST
+from ddtrace.settings.asm import config as asm_config
 
 
 log = get_logger(__name__)
 
-TELEMETRY_OFF_NAME = "OFF"
-TELEMETRY_DEBUG_NAME = "DEBUG"
-TELEMETRY_MANDATORY_NAME = "MANDATORY"
-TELEMETRY_INFORMATION_NAME = "INFORMATION"
-
-TELEMETRY_DEBUG_VERBOSITY = 10
-TELEMETRY_INFORMATION_VERBOSITY = 20
-TELEMETRY_MANDATORY_VERBOSITY = 30
-TELEMETRY_OFF_VERBOSITY = 40
-
-METRICS_REPORT_LVLS = (
-    (TELEMETRY_DEBUG_VERBOSITY, TELEMETRY_DEBUG_NAME),
-    (TELEMETRY_INFORMATION_VERBOSITY, TELEMETRY_INFORMATION_NAME),
-    (TELEMETRY_MANDATORY_VERBOSITY, TELEMETRY_MANDATORY_NAME),
-    (TELEMETRY_OFF_VERBOSITY, TELEMETRY_OFF_NAME),
-)
-
 _IAST_SPAN_METRICS: Dict[str, int] = {}
 
 
 def get_iast_metrics_report_lvl(*args, **kwargs):
-    report_lvl_name = os.environ.get(IAST.TELEMETRY_REPORT_LVL, TELEMETRY_INFORMATION_NAME).upper()
+    report_lvl_name = asm_config._iast_telemetry_report_lvl.upper()
     report_lvl = 3
-    for lvl, lvl_name in METRICS_REPORT_LVLS:
+    for lvl, lvl_name in IAST.METRICS_REPORT_LVLS:
         if report_lvl_name == lvl_name:
             return lvl
     return report_lvl
@@ -51,7 +36,7 @@ def wrapper(f):
             try:
                 return f
             except Exception:
-                log.warning("Error reporting IAST metrics", exc_info=True)
+                log.warning("[IAST] Error reporting metrics", exc_info=True)
         return lambda: None  # noqa: E731
 
     return wrapper
@@ -80,7 +65,7 @@ def _set_iast_error_metric(msg: Text) -> None:
         }
         telemetry.telemetry_writer.add_log(TELEMETRY_LOG_LEVEL.ERROR, msg, stack_trace=stack_trace, tags=tags)
     except Exception:
-        log.warning("Error reporting ASM logs metrics", exc_info=True)
+        log.warning("[IAST] Error reporting logs metrics", exc_info=True)
 
 
 @metric_verbosity(TELEMETRY_MANDATORY_VERBOSITY)

ddtrace/appsec/_iast/constants.py

@@ -27,8 +27,6 @@
 RC4_DEF = "rc4"
 IDEA_DEF = "idea"
 
-DD_IAST_TELEMETRY_VERBOSITY = "DD_IAST_TELEMETRY_VERBOSITY"
-
 DEFAULT_WEAK_HASH_ALGORITHMS = {MD5_DEF, SHA1_DEF}
 
 DEFAULT_WEAK_CIPHER_ALGORITHMS = {DES_DEF, BLOWFISH_DEF, RC2_DEF, RC4_DEF, IDEA_DEF}

ddtrace/settings/asm.py

@@ -14,6 +14,7 @@
 from ddtrace.appsec._constants import EXPLOIT_PREVENTION
 from ddtrace.appsec._constants import IAST
 from ddtrace.appsec._constants import LOGIN_EVENTS_MODE
+from ddtrace.appsec._constants import TELEMETRY_INFORMATION_NAME
 from ddtrace.constants import APPSEC_ENV
 from ddtrace.internal.utils.deprecations import DDTraceDeprecationWarning
 from ddtrace.settings._core import report_telemetry as _report_telemetry
@@ -68,6 +69,7 @@ class ASMConfig(Env):
     _iast_request_sampling = Env.var(float, IAST.ENV_REQUEST_SAMPLING, default=30.0)
     _iast_debug = Env.var(bool, IAST.ENV_DEBUG, default=False, private=True)
     _iast_propagation_debug = Env.var(bool, IAST.ENV_PROPAGATION_DEBUG, default=False, private=True)
+    _iast_telemetry_report_lvl = Env.var(str, IAST.ENV_TELEMETRY_REPORT_LVL, default=TELEMETRY_INFORMATION_NAME)
     _appsec_standalone_enabled = Env.var(bool, APPSEC.STANDALONE_ENV, default=False)
     _use_metastruct_for_triggers = False
 
@@ -180,6 +182,7 @@ class ASMConfig(Env):
         "_iast_request_sampling",
         "_iast_debug",
         "_iast_propagation_debug",
+        "_iast_telemetry_report_lvl",
         "_ep_enabled",
         "_use_metastruct_for_triggers",
         "_automatic_login_events_mode",

tests/appsec/iast/conftest.py

@@ -77,7 +77,6 @@ class MockSpan:
         _trace_id_64bits = 17577308072598193742
 
     env.update({"_DD_APPSEC_DEDUPLICATION_ENABLED": str(deduplication)})
-    VulnerabilityBase._reset_cache_for_testing()
     with override_global_config(
         dict(
             _asm_enabled=asm_enabled,
@@ -86,6 +85,7 @@ class MockSpan:
             _iast_request_sampling=request_sampling,
         )
     ), override_env(env):
+        VulnerabilityBase._reset_cache_for_testing()
         _start_iast_context_and_oce(MockSpan())
         weak_hash_patch()
         weak_cipher_patch()

tests/appsec/iast/test_telemetry.py

@@ -3,11 +3,12 @@
 from ddtrace.appsec._common_module_patches import patch_common_modules
 from ddtrace.appsec._common_module_patches import unpatch_common_modules
 from ddtrace.appsec._constants import IAST_SPAN_TAGS
+from ddtrace.appsec._constants import TELEMETRY_DEBUG_VERBOSITY
+from ddtrace.appsec._constants import TELEMETRY_INFORMATION_NAME
+from ddtrace.appsec._constants import TELEMETRY_INFORMATION_VERBOSITY
+from ddtrace.appsec._constants import TELEMETRY_MANDATORY_VERBOSITY
 from ddtrace.appsec._iast import oce
 from ddtrace.appsec._iast._handlers import _on_django_patch
-from ddtrace.appsec._iast._metrics import TELEMETRY_DEBUG_VERBOSITY
-from ddtrace.appsec._iast._metrics import TELEMETRY_INFORMATION_VERBOSITY
-from ddtrace.appsec._iast._metrics import TELEMETRY_MANDATORY_VERBOSITY
 from ddtrace.appsec._iast._metrics import _set_iast_error_metric
 from ddtrace.appsec._iast._metrics import metric_verbosity
 from ddtrace.appsec._iast._patch_modules import patch_iast
@@ -29,7 +30,6 @@
 from tests.appsec.iast.aspects.conftest import _iast_patched_module
 from tests.appsec.utils import asm_context
 from tests.utils import DummyTracer
-from tests.utils import override_env
 from tests.utils import override_global_config
 
 
@@ -61,15 +61,12 @@ def _assert_instrumented_sink(telemetry_writer, vuln_type):
     ],
 )
 def test_metric_verbosity(lvl, env_lvl, expected_result):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY=env_lvl)):
+    with override_global_config(dict(_iast_telemetry_report_lvl=env_lvl)):
         assert metric_verbosity(lvl)(lambda: 1)() == expected_result
 
 
-@pytest.mark.skip_iast_check_logs
 def test_metric_executed_sink(no_request_sampling, telemetry_writer, caplog):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         patch_iast()
 
         tracer = DummyTracer(iast_enabled=True)
@@ -99,9 +96,7 @@ def test_metric_executed_sink(no_request_sampling, telemetry_writer, caplog):
 
 
 def test_metric_instrumented_cmdi(no_request_sampling, telemetry_writer):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         cmdi_patch()
 
     _assert_instrumented_sink(telemetry_writer, VULN_CMDI)
@@ -110,9 +105,7 @@ def test_metric_instrumented_cmdi(no_request_sampling, telemetry_writer):
 def test_metric_instrumented_path_traversal(no_request_sampling, telemetry_writer):
     # We need to unpatch first because ddtrace.appsec._iast._patch_modules loads at runtime this patch function
     unpatch_common_modules()
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         patch_common_modules()
 
     _assert_instrumented_sink(telemetry_writer, VULN_PATH_TRAVERSAL)
@@ -121,36 +114,28 @@ def test_metric_instrumented_path_traversal(no_request_sampling, telemetry_write
 def test_metric_instrumented_header_injection(no_request_sampling, telemetry_writer):
     # We need to unpatch first because ddtrace.appsec._iast._patch_modules loads at runtime this patch function
     header_injection_unpatch()
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         header_injection_patch()
 
     _assert_instrumented_sink(telemetry_writer, VULN_HEADER_INJECTION)
 
 
 def test_metric_instrumented_sqli_sqlite3(no_request_sampling, telemetry_writer):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         sqli_sqlite3_patch()
 
     _assert_instrumented_sink(telemetry_writer, VULN_SQL_INJECTION)
 
 
 def test_metric_instrumented_sqli_sqlalchemy_patch(no_request_sampling, telemetry_writer):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         sqli_sqlalchemy_patch()
 
     _assert_instrumented_sink(telemetry_writer, VULN_SQL_INJECTION)
 
 
 def test_metric_instrumented_propagation(no_request_sampling, telemetry_writer):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True)
-    ):
+    with override_global_config(dict(_iast_enabled=True, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)):
         _iast_patched_module("benchmarks.bm.iast_fixtures.str_methods")
 
     metrics_result = telemetry_writer._namespace._metrics_data
@@ -162,8 +147,8 @@ def test_metric_instrumented_propagation(no_request_sampling, telemetry_writer):
 
 
 def test_metric_request_tainted(no_request_sampling, telemetry_writer):
-    with override_env(dict(DD_IAST_TELEMETRY_VERBOSITY="INFORMATION")), override_global_config(
-        dict(_iast_enabled=True, _iast_request_sampling=100.0)
+    with override_global_config(
+        dict(_iast_enabled=True, _iast_request_sampling=100.0, _iast_telemetry_report_lvl=TELEMETRY_INFORMATION_NAME)
     ):
         oce.reconfigure()
         tracer = DummyTracer(iast_enabled=True)

tests/telemetry/test_writer.py

@@ -370,6 +370,7 @@ def test_app_started_event_configuration_override(test_agent_session, run_python
             "[^\\-]+[\\-]{5}END[a-z\\s]+PRIVATE\\sKEY|ssh-rsa\\s*[a-z0-9\\/\\.+]{100,}",
         },
         {"name": "DD_IAST_REQUEST_SAMPLING", "origin": "default", "value": 30.0},
+        {"name": "DD_IAST_TELEMETRY_VERBOSITY", "origin": "default", "value": "INFORMATION"},
         {"name": "DD_INJECT_FORCE", "origin": "env_var", "value": True},
         {"name": "DD_INSTRUMENTATION_INSTALL_ID", "origin": "default", "value": None},
         {"name": "DD_INSTRUMENTATION_INSTALL_TYPE", "origin": "default", "value": None},