chore: migrate appsec/_iast to _get_env

Author: genesor

ddtrace/_trace/apm_filter.py

@@ -1,10 +1,10 @@
-import os
 from typing import List
 from typing import Optional
 
 from ddtrace._trace.processor import TraceProcessor
 from ddtrace._trace.span import Span
 from ddtrace.internal.utils.formats import asbool
+from ddtrace.settings._env import get_env as _get_env
 
 
 class APMTracingEnabledFilter(TraceProcessor):
@@ -14,7 +14,7 @@ class APMTracingEnabledFilter(TraceProcessor):
 
     def __init__(self) -> None:
         super().__init__()
-        self._apm_tracing_enabled = asbool(os.getenv("DD_APM_TRACING_ENABLED", "true"))
+        self._apm_tracing_enabled = asbool(_get_env("DD_APM_TRACING_ENABLED", "true"))
 
     def process_trace(self, trace: List[Span]) -> Optional[List[Span]]:
         if not self._apm_tracing_enabled:

ddtrace/_trace/product.py

@@ -1,6 +1,5 @@
 import enum
 import json
-import os
 import typing as t
 
 from envier import En
@@ -9,6 +8,7 @@
 from ddtrace.internal.utils.deprecations import DDTraceDeprecationWarning
 from ddtrace.internal.utils.formats import asbool
 from ddtrace.internal.utils.formats import parse_tags_str
+from ddtrace.settings._env import get_env as _get_env
 from ddtrace.settings.http import HttpConfig
 from ddtrace.vendor.debtcollector import deprecate
 
@@ -32,7 +32,7 @@ def post_preload():
     if _config.enabled:
         from ddtrace._monkey import _patch_all
 
-        modules_to_patch = os.getenv("DD_PATCH_MODULES")
+        modules_to_patch = _get_env("DD_PATCH_MODULES")
         modules_to_str = parse_tags_str(modules_to_patch)
         modules_to_bool = {k: asbool(v) for k, v in modules_to_str.items()}
         _patch_all(**modules_to_bool)

ddtrace/appsec/_iast/__init__.py

@@ -28,14 +28,13 @@ def wrapped_function(wrapped, instance, args, kwargs):
     return wrapped(*args, **kwargs)
 """
 
-import os
 import sys
 import types
 
 from ddtrace.internal import forksafe
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.module import ModuleWatchdog
-from ddtrace.settings._config import _get_env
+from ddtrace.settings._env import environ as _env
 from ddtrace.settings.asm import config as asm_config
 
 from ._listener import iast_listen
@@ -187,11 +186,11 @@ def _iast_pytest_activation():
     global _iast_propagation_enabled
     if _iast_propagation_enabled:
         return
-    os.environ["DD_IAST_ENABLED"] = _get_env("DD_IAST_ENABLED", "1")
-    os.environ["DD_IAST_REQUEST_SAMPLING"] = _get_env("DD_IAST_REQUEST_SAMPLING", "100.0")
-    os.environ["_DD_APPSEC_DEDUPLICATION_ENABLED"] = _get_env("_DD_APPSEC_DEDUPLICATION_ENABLED", "false")
-    os.environ["DD_IAST_VULNERABILITIES_PER_REQUEST"] = _get_env("DD_IAST_VULNERABILITIES_PER_REQUEST", "1000")
-    os.environ["DD_IAST_MAX_CONCURRENT_REQUESTS"] = _get_env("DD_IAST_MAX_CONCURRENT_REQUESTS", "1000")
+    _env["DD_IAST_ENABLED"] = _env.get("DD_IAST_ENABLED", "1")
+    _env["DD_IAST_REQUEST_SAMPLING"] = _env.get("DD_IAST_REQUEST_SAMPLING", "100.0")
+    _env["_DD_APPSEC_DEDUPLICATION_ENABLED"] = _env.get("_DD_APPSEC_DEDUPLICATION_ENABLED", "false")
+    _env["DD_IAST_VULNERABILITIES_PER_REQUEST"] = _env.get("DD_IAST_VULNERABILITIES_PER_REQUEST", "1000")
+    _env["DD_IAST_MAX_CONCURRENT_REQUESTS"] = _env.get("DD_IAST_MAX_CONCURRENT_REQUESTS", "1000")
 
     asm_config._iast_request_sampling = 100.0
     asm_config._deduplication_enabled = False

ddtrace/appsec/_iast/taint_sinks/weak_cipher.py

@@ -1,4 +1,3 @@
-import os
 from typing import Any
 from typing import Callable
 from typing import Set
@@ -13,6 +12,7 @@
 from ddtrace.appsec._iast.constants import RC4_DEF
 from ddtrace.appsec._iast.constants import VULN_WEAK_CIPHER_TYPE
 from ddtrace.internal.logger import get_logger
+from ddtrace.settings._env import get_env as _get_env
 from ddtrace.settings.asm import config as asm_config
 
 from .._logs import iast_error
@@ -28,7 +28,7 @@
 
 def get_weak_cipher_algorithms() -> Set:
     CONFIGURED_WEAK_CIPHER_ALGORITHMS = None
-    DD_IAST_WEAK_CIPHER_ALGORITHMS = os.getenv("DD_IAST_WEAK_CIPHER_ALGORITHMS")
+    DD_IAST_WEAK_CIPHER_ALGORITHMS = _get_env("DD_IAST_WEAK_CIPHER_ALGORITHMS")
     if DD_IAST_WEAK_CIPHER_ALGORITHMS:
         CONFIGURED_WEAK_CIPHER_ALGORITHMS = set(
             algo.strip() for algo in DD_IAST_WEAK_CIPHER_ALGORITHMS.lower().split(",")

ddtrace/appsec/_iast/taint_sinks/weak_hash.py

@@ -1,9 +1,9 @@
-import os
 from typing import Any
 from typing import Callable
 from typing import Set
 
 from ddtrace.internal.logger import get_logger
+from ddtrace.settings._env import get_env as _get_env
 from ddtrace.settings.asm import config as asm_config
 
 from ..._common_module_patches import try_unwrap
@@ -28,7 +28,7 @@
 
 def get_weak_hash_algorithms() -> Set:
     CONFIGURED_WEAK_HASH_ALGORITHMS = None
-    DD_IAST_WEAK_HASH_ALGORITHMS = os.getenv("DD_IAST_WEAK_HASH_ALGORITHMS")
+    DD_IAST_WEAK_HASH_ALGORITHMS = _get_env("DD_IAST_WEAK_HASH_ALGORITHMS")
     if DD_IAST_WEAK_HASH_ALGORITHMS:
         CONFIGURED_WEAK_HASH_ALGORITHMS = set(algo.strip() for algo in DD_IAST_WEAK_HASH_ALGORITHMS.lower().split(","))
 

ddtrace/appsec/_processor.py

@@ -1,8 +1,6 @@
 import dataclasses
 import errno
 from json.decoder import JSONDecodeError
-import os
-import os.path
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import ClassVar
@@ -43,6 +41,7 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.rate_limiter import RateLimiter
 from ddtrace.internal.remoteconfig import PayloadType
+from ddtrace.settings._env import get_env as _get_env
 from ddtrace.settings.asm import config as asm_config
 
 
@@ -72,7 +71,7 @@ def get_rules() -> str:
 
 
 def _get_rate_limiter() -> RateLimiter:
-    return RateLimiter(int(os.getenv("DD_APPSEC_TRACE_RATE_LIMIT", DEFAULT.TRACE_RATE_LIMIT)))
+    return RateLimiter(int(_get_env("DD_APPSEC_TRACE_RATE_LIMIT", DEFAULT.TRACE_RATE_LIMIT)))
 
 
 @dataclasses.dataclass(eq=False)