first attempt at kinde instrumentation

Author: sohankshirsagar

drift/core/drift_sdk.py

@@ -241,6 +241,13 @@ def initialize(
                     "Continuing with SDK initialization. Mock responses will not be available until CLI is running."
                 )
 
+            # # TEMPORARY: Hardcoded time travel for testing
+            # from drift.instrumentation.datetime import start_time_travel
+            # # Unix timestamp: seconds + nanos/1e9 + 60 seconds
+            # timestamp = 1768272255.864658
+            # start_time_travel(timestamp)
+            # logger.info(f"HARDCODED time travel to unix timestamp: {timestamp}")
+
         install_hooks()
 
         instance._init_auto_instrumentations()
@@ -466,6 +473,13 @@ def _init_auto_instrumentations(self) -> None:
                 logger.debug("Socket instrumentation initialized (REPLAY mode - unpatched dependency detection)")
             except Exception as e:
                 logger.debug(f"Socket instrumentation initialization failed: {e}")
+            
+            try:
+                from ..instrumentation.kinde import KindeInstrumentation
+                _ = KindeInstrumentation(enabled=True)
+                logger.debug("Kinde instrumentation initialized (REPLAY mode - auth token validation)")
+            except Exception as e:
+                logger.debug(f"Kinde instrumentation initialization failed: {e}")
 
     def create_env_vars_snapshot(self) -> None:
         """Create a span capturing all environment variables.

drift/core/mock_utils.py

@@ -281,9 +281,6 @@ def _update_time_travel(
         mock_response: The mock response containing the timestamp
         replay_trace_id: The replay trace ID for this session
     """
-    if not replay_trace_id:
-        return
-
     try:
         from drift.instrumentation.datetime import start_time_travel
 

drift/instrumentation/__init__.py

@@ -2,12 +2,14 @@
 
 from .base import InstrumentationBase
 from .django import DjangoInstrumentation
+from .kinde import KindeInstrumentation
 from .registry import install_hooks, patch_instances_via_gc, register_patch
 from .wsgi import WsgiInstrumentation
 
 __all__ = [
     "InstrumentationBase",
     "DjangoInstrumentation",
+    "KindeInstrumentation",
     "WsgiInstrumentation",
     "register_patch",
     "install_hooks",

drift/instrumentation/kinde/__init__.py

@@ -0,0 +1,5 @@
+"""Kinde SDK instrumentation for bypassing authentication in replay mode."""
+
+from .instrumentation import KindeInstrumentation
+
+__all__ = ["KindeInstrumentation"]

drift/instrumentation/kinde/instrumentation.py

@@ -0,0 +1,71 @@
+"""Instrumentation for Kinde SDK authentication.
+
+Patches is_authenticated() to always return True in REPLAY mode,
+allowing tests to bypass authentication checks during replay.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from ..base import InstrumentationBase
+
+logger = logging.getLogger(__name__)
+
+
+class KindeInstrumentation(InstrumentationBase):
+    """Instrumentation for the Kinde SDK authentication library.
+
+    Patches OAuth.is_authenticated() to:
+    - Return True in REPLAY mode (bypass authentication)
+    - Call the original method in RECORD and DISABLED modes
+
+    Since SmartOAuth and AsyncOAuth delegate to OAuth.is_authenticated(),
+    patching OAuth covers all authentication entry points.
+    """
+
+    def __init__(self, enabled: bool = True) -> None:
+        super().__init__(
+            name="KindeInstrumentation",
+            module_name="kinde_sdk.auth.oauth",
+            supported_versions="*",
+            enabled=enabled,
+        )
+
+    def patch(self, module: Any) -> None:
+        """Patch the kinde_sdk.auth.oauth module.
+
+        Patches OAuth.is_authenticated() to return True in REPLAY mode.
+        """
+        if not hasattr(module, "OAuth"):
+            logger.warning("kinde_sdk.auth.oauth.OAuth not found, skipping instrumentation")
+            return
+
+        original_is_authenticated = module.OAuth.is_authenticated
+
+        def patched_is_authenticated(oauth_self) -> bool:
+            """Patched is_authenticated method.
+
+            Args:
+                oauth_self: OAuth instance
+
+            Returns:
+                True in REPLAY mode, otherwise delegates to original method
+            """
+            # Lazy imports to avoid circular dependency
+            from ...core.drift_sdk import TuskDrift
+            from ...core.types import TuskDriftMode
+
+            sdk = TuskDrift.get_instance()
+
+            # In REPLAY mode, always return True to bypass authentication
+            if sdk.mode == TuskDriftMode.REPLAY:
+                logger.debug("[KindeInstrumentation] REPLAY mode: returning True for is_authenticated")
+                return True
+
+            # In RECORD or DISABLED mode, call the original method
+            return original_is_authenticated(oauth_self)
+
+        module.OAuth.is_authenticated = patched_is_authenticated
+        logger.info("kinde_sdk.auth.oauth.OAuth.is_authenticated instrumented")

drift/instrumentation/requests/instrumentation.py

@@ -511,6 +511,7 @@ def _try_get_mock(
                 input_value=input_value,
                 kind=SpanKind.CLIENT,
                 input_schema_merges=input_schema_merges,
+                is_pre_app_start=not sdk.app_ready,
             )
 
             if not mock_response_output or not mock_response_output.found: