From 01b615edfe3f7f2c01e875ebb1f9f49227fe255a Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <vangelis.ruiz@delinea.com>
Date: Mon, 8 Jun 2026 12:13:19 -0600
Subject: [PATCH 01/10] feat(server): cache server-type detection per base_url
 per process

Authorizer now memoizes the /api/v1/healthcheck + /health probe result in a
process-scoped, thread-safe class cache keyed by normalized base_url, so
detection fires once per base_url per process instead of on every authorizer
construction. Closes the unauthenticated-probe burst that the Delinea Platform
WAF rate-limits to 403 under Ansible token-auth lookups.

- successes only are cached; detection failures re-probe
- per-instance _server_type still set on cache hit (SecretServer + _refresh read it)
- adds first offline unit tests (tests/test_server_detection_cache.py)

Addresses 728859
---
 delinea/secrets/server.py            |  59 ++++++--
 tests/test_server_detection_cache.py | 211 +++++++++++++++++++++++++++
 2 files changed, 259 insertions(+), 11 deletions(-)
 create mode 100644 tests/test_server_detection_cache.py

diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index 0f26d4a..88916fd 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -19,6 +19,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from datetime import datetime, timedelta
+from threading import Lock
 
 import requests
 
@@ -164,6 +165,19 @@ class SecretServerServiceError(SecretServerError):
 class Authorizer(ABC):
     """Main abstract base class for all Authorizer access methods."""
 
+    # Process-scoped cache mapping a normalized base_url to its detected server
+    # type ("secret_server" | "platform"). Shared across all Authorizer
+    # subclasses so the health-check probe pair fires once per base_url per
+    # process. Guarded by ``_server_type_cache_lock``.
+    _server_type_cache = {}
+    _server_type_cache_lock = Lock()
+
+    @classmethod
+    def _clear_server_type_cache(cls):
+        """Clear the process-scoped server-detection cache (test hook)."""
+        with Authorizer._server_type_cache_lock:
+            Authorizer._server_type_cache.clear()
+
     @staticmethod
     def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
         """Adds an HTTP `Authorization` header containing the `Bearer` token
@@ -180,19 +194,42 @@ def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
         }
 
     def _perform_server_detection(self, base_url):
-        """Detects if the server is Secret Server or Platform by health check endpoints."""
-        secret_server_endpoint = base_url.rstrip("/") + "/api/v1/healthcheck"
-        platform_endpoint = base_url.rstrip("/") + "/health"
+        """Detect whether the server is Secret Server or Platform via health
+        check endpoints, using a process-scoped cache.
+
+        The detected type is cached per normalized ``base_url`` on the
+        ``Authorizer`` base class and shared across all subclasses, so the
+        ``/api/v1/healthcheck`` + ``/health`` probe pair fires only once per
+        ``base_url`` per process. The cache is read/written under
+        ``_server_type_cache_lock`` for thread safety, but the network probe
+        itself runs OUTSIDE the lock; detection is idempotent, so a rare
+        double-probe under a race is harmless. Only successful detections are
+        cached -- failures re-probe on the next construction.
+
+        On both cache hits and fresh probes the per-instance ``_server_type``
+        attribute is set, because callers (``SecretServer.ensure_vault_url``
+        and ``PasswordGrantAuthorizer._refresh``) read ``self._server_type``.
+        """
+        key = base_url.rstrip("/")
 
-        if self._validate_health_endpoint(secret_server_endpoint):
-            self._server_type = "secret_server"
+        with Authorizer._server_type_cache_lock:
+            cached = Authorizer._server_type_cache.get(key)
+        if cached is not None:
+            self._server_type = cached
             return
-        if self._validate_health_endpoint(platform_endpoint):
-            self._server_type = "platform"
-            return
-        raise SecretServerError(
-            "Unable to detect server type via health check endpoints."
-        )
+
+        if self._validate_health_endpoint(key + "/api/v1/healthcheck"):
+            detected = "secret_server"
+        elif self._validate_health_endpoint(key + "/health"):
+            detected = "platform"
+        else:
+            raise SecretServerError(
+                "Unable to detect server type via health check endpoints."
+            )
+
+        self._server_type = detected
+        with Authorizer._server_type_cache_lock:
+            Authorizer._server_type_cache.setdefault(key, detected)
 
     def _validate_health_endpoint(self, url):
         """Validates if an endpoint returns healthy status."""
diff --git a/tests/test_server_detection_cache.py b/tests/test_server_detection_cache.py
new file mode 100644
index 0000000..b19fb60
--- /dev/null
+++ b/tests/test_server_detection_cache.py
@@ -0,0 +1,211 @@
+"""Offline unit tests for the process-scoped server-detection cache on the
+``Authorizer`` base class.
+
+These tests are fully OFFLINE: the network is mocked by patching
+``delinea.secrets.server.requests.get`` (the symbol the SDK actually calls
+inside ``_validate_health_endpoint``). Unlike ``tests/test_server.py`` these
+do NOT require live credentials.
+
+The cache is process-global, so each test clears it via the
+``Authorizer._clear_server_type_cache()`` hook (see the autouse fixture).
+"""
+
+import threading
+
+import pytest
+
+from delinea.secrets.server import (
+    AccessTokenAuthorizer,
+    Authorizer,
+    PasswordGrantAuthorizer,
+    SecretServerError,
+)
+
+
+SECRET_SERVER_HEALTH = "/api/v1/healthcheck"
+PLATFORM_HEALTH = "/health"
+
+
+class FakeResponse:
+    """Minimal stand-in for a ``requests.Response`` as consumed by
+    ``_validate_health_endpoint`` (reads ``.content`` and ``.json()``)."""
+
+    def __init__(self, healthy):
+        self._healthy = healthy
+        self.content = b'{"Healthy": true}' if healthy else b"{}"
+
+    def json(self):
+        return {"Healthy": self._healthy}
+
+
+def make_probe_counter(healthy_endpoints):
+    """Return a (fake_get, counter) pair.
+
+    ``fake_get`` replaces ``requests.get``. It returns a healthy
+    ``FakeResponse`` only when the requested URL ends with one of
+    ``healthy_endpoints`` (e.g. ``/health``); every other health probe gets an
+    unhealthy response. ``counter`` is a mutable dict tracking how many times
+    each health endpoint suffix was probed plus a total.
+    """
+
+    # "rounds" counts how many times a full detection probe sequence began,
+    # i.e. how many times the FIRST endpoint of the pair (the secret_server
+    # healthcheck) was hit. A platform detection issues two raw GETs per round
+    # (healthcheck=unhealthy, then health=healthy); a cache hit issues zero, so
+    # "rounds" is the meaningful "probe pair fired N times" metric.
+    counter = {"total": 0, "rounds": 0, SECRET_SERVER_HEALTH: 0, PLATFORM_HEALTH: 0}
+
+    def fake_get(url, *args, **kwargs):
+        for suffix in (SECRET_SERVER_HEALTH, PLATFORM_HEALTH):
+            if url.endswith(suffix):
+                counter["total"] += 1
+                counter[suffix] += 1
+                if suffix == SECRET_SERVER_HEALTH:
+                    counter["rounds"] += 1
+                return FakeResponse(suffix in healthy_endpoints)
+        # Any other GET (e.g. vault lookups) is not a health probe.
+        return FakeResponse(False)
+
+    return fake_get, counter
+
+
+@pytest.fixture(autouse=True)
+def clear_detection_cache():
+    """The detection cache is process-global; clear before and after each test
+    so cached entries cannot leak between tests."""
+    Authorizer._clear_server_type_cache()
+    yield
+    Authorizer._clear_server_type_cache()
+
+
+# Behavior 1: repeated construction with the same base_url probes once total.
+def test_repeated_construction_probes_once(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    instances = [AccessTokenAuthorizer("tok", base_url) for _ in range(20)]
+
+    assert all(inst._server_type == "platform" for inst in instances)
+    # The probe pair fires exactly once total across all 20 constructions.
+    assert counter["rounds"] == 1
+    assert counter[PLATFORM_HEALTH] == 1
+    assert counter[SECRET_SERVER_HEALTH] == 1
+
+
+# Behavior 2: cache is shared across different authorizer subclasses.
+def test_cache_shared_across_subclasses(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    AccessTokenAuthorizer("tok", base_url)
+    grant = PasswordGrantAuthorizer(base_url, "user", "pass")
+    try:
+        # Triggers lazy detection in _refresh; the grant POST will fail offline
+        # but we only care that detection used the cache.
+        grant.get_access_token()
+    except Exception:
+        pass
+
+    assert grant._server_type == "platform"
+    # Detection probes fire once total across both authorizers.
+    assert counter["rounds"] == 1
+
+
+# Behavior 3: a cache hit still sets the per-instance _server_type attribute.
+def test_cache_hit_sets_instance_attr(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    AccessTokenAuthorizer("tok", base_url)  # populates the cache
+    assert counter["rounds"] == 1
+    probes_after_first = counter["total"]
+
+    second = AccessTokenAuthorizer("tok", base_url)  # cache hit, no new probe
+    assert second._server_type == "platform"
+    assert counter["rounds"] == 1
+    assert counter["total"] == probes_after_first
+
+
+# Behavior 4: two distinct base_urls get independent, correct cache entries.
+def test_two_distinct_base_urls(monkeypatch):
+    ss_url = "https://secretserver.example.com"
+    platform_url = "https://platform.example.com"
+
+    def fake_get(url, *args, **kwargs):
+        if url.startswith(ss_url) and url.endswith(SECRET_SERVER_HEALTH):
+            return FakeResponse(True)
+        if url.startswith(platform_url) and url.endswith(PLATFORM_HEALTH):
+            return FakeResponse(True)
+        return FakeResponse(False)
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    ss_auth = AccessTokenAuthorizer("tok", ss_url)
+    platform_auth = AccessTokenAuthorizer("tok", platform_url)
+
+    assert ss_auth._server_type == "secret_server"
+    assert platform_auth._server_type == "platform"
+
+    cache = Authorizer._server_type_cache
+    assert cache[ss_url] == "secret_server"
+    assert cache[platform_url] == "platform"
+    assert len(cache) == 2
+
+
+# Behavior 5: detection failure is NOT cached; a later healthy probe succeeds.
+def test_failure_is_not_cached(monkeypatch):
+    base_url = "https://unknown.example.com"
+
+    # First: both probes unhealthy -> detection raises.
+    unhealthy_get, _ = make_probe_counter(set())
+    monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get)
+    with pytest.raises(SecretServerError):
+        AccessTokenAuthorizer("tok", base_url)
+
+    assert base_url not in Authorizer._server_type_cache
+
+    # Then: probes become healthy -> re-probe succeeds (failure was not cached).
+    healthy_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", healthy_get)
+    instance = AccessTokenAuthorizer("tok", base_url)
+
+    assert instance._server_type == "platform"
+    assert counter["total"] >= 1
+
+
+# Behavior 6: concurrent construction is thread-safe and probes few times.
+def test_concurrent_construction_thread_safe(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    results = []
+    errors = []
+    start = threading.Event()
+
+    def worker():
+        start.wait()
+        try:
+            inst = AccessTokenAuthorizer("tok", base_url)
+            results.append(inst._server_type)
+        except Exception as exc:  # pragma: no cover - failure path
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(20)]
+    for t in threads:
+        t.start()
+    start.set()
+    for t in threads:
+        t.join()
+
+    assert errors == []
+    assert len(results) == 20
+    assert all(r == "platform" for r in results)
+    # Probe count is a small constant: the probe pair fires at least once, and
+    # is bounded by the number of threads even under a detection race (commonly
+    # exactly 1).
+    assert counter["rounds"] >= 1
+    assert counter["rounds"] <= 20

From 403e7743956078f70b91211f1aa64ec76bf50112 Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Mon, 22 Jun 2026 17:13:46 -0600
Subject: [PATCH 02/10] add explicit server_type override; bound + harden
 detection cache

---
 delinea/secrets/server.py            | 161 +++++++++++++++++++++------
 requirements.txt                     |   4 +-
 tests/test_server_detection_cache.py | 104 ++++++++++++++++-
 3 files changed, 235 insertions(+), 34 deletions(-)

diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index 88916fd..5927be4 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -17,6 +17,7 @@
 import json
 import re
 from abc import ABC, abstractmethod
+from collections import OrderedDict
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from threading import Lock
@@ -165,19 +166,78 @@ class SecretServerServiceError(SecretServerError):
 class Authorizer(ABC):
     """Main abstract base class for all Authorizer access methods."""
 
-    # Process-scoped cache mapping a normalized base_url to its detected server
-    # type ("secret_server" | "platform"). Shared across all Authorizer
-    # subclasses so the health-check probe pair fires once per base_url per
-    # process. Guarded by ``_server_type_cache_lock``.
-    _server_type_cache = {}
+    # Accepted values for an explicit ``server_type`` override and for cached
+    # detections.
+    VALID_SERVER_TYPES = ("secret_server", "platform")
+
+    # Process-scoped, bounded LRU cache mapping a normalized base_url to its
+    # detected server type ("secret_server" | "platform"). Shared across all
+    # Authorizer subclasses so the health-check probe pair fires once per
+    # base_url per process. Bounded to ``_SERVER_TYPE_CACHE_MAXSIZE`` entries so
+    # a long-lived process that constructs authorizers against many distinct
+    # URLs cannot grow it without bound; the least-recently-used entry is
+    # evicted on overflow. Guarded by ``_server_type_cache_lock``.
+    #
+    # NOTE: This cache is process-scoped. It deduplicates probes only within a
+    # single Python process. Callers that run each lookup in a fresh process
+    # (e.g. some Ansible lookup-plugin runtimes) start with an empty cache and
+    # will re-probe. To eliminate the probe entirely in that case, pass an
+    # explicit ``server_type`` to the authorizer (see ``_perform_server_detection``).
+    _SERVER_TYPE_CACHE_MAXSIZE = 128
+    _server_type_cache = OrderedDict()
     _server_type_cache_lock = Lock()
 
     @classmethod
-    def _clear_server_type_cache(cls):
-        """Clear the process-scoped server-detection cache (test hook)."""
+    def _normalize_server_type(cls, server_type):
+        """Validate and normalize an explicit ``server_type`` value.
+
+        :raise :class:`SecretServerError` when ``server_type`` is not one of
+                ``VALID_SERVER_TYPES``.
+        """
+        normalized = str(server_type).strip().lower()
+        if normalized not in cls.VALID_SERVER_TYPES:
+            raise SecretServerError(
+                f"Invalid server_type {server_type!r}; expected one of "
+                f"{cls.VALID_SERVER_TYPES}."
+            )
+        return normalized
+
+    @classmethod
+    def _get_cached_server_type(cls, key):
+        """Return the cached server type for ``key`` (marking it most-recently
+        used) or ``None`` if absent."""
+        with Authorizer._server_type_cache_lock:
+            if key in Authorizer._server_type_cache:
+                Authorizer._server_type_cache.move_to_end(key)
+                return Authorizer._server_type_cache[key]
+            return None
+
+    @classmethod
+    def _cache_server_type(cls, key, server_type):
+        """Cache ``server_type`` for ``key``, evicting the least-recently-used
+        entry if the cache is over capacity."""
+        with Authorizer._server_type_cache_lock:
+            Authorizer._server_type_cache[key] = server_type
+            Authorizer._server_type_cache.move_to_end(key)
+            while len(Authorizer._server_type_cache) > cls._SERVER_TYPE_CACHE_MAXSIZE:
+                Authorizer._server_type_cache.popitem(last=False)
+
+    @classmethod
+    def clear_server_type_cache(cls):
+        """Clear the process-scoped server-detection cache.
+
+        Detection results are cached for the lifetime of the process with no
+        TTL, because a server's type at a given ``base_url`` is effectively
+        immutable in practice. Use this escape hatch to force re-detection if a
+        ``base_url`` is ever re-provisioned to a different server type while a
+        long-lived process is running.
+        """
         with Authorizer._server_type_cache_lock:
             Authorizer._server_type_cache.clear()
 
+    # Backwards-compatible alias retained for existing callers/tests.
+    _clear_server_type_cache = clear_server_type_cache
+
     @staticmethod
     def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
         """Adds an HTTP `Authorization` header containing the `Bearer` token
@@ -193,27 +253,40 @@ def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
             **existing_headers,
         }
 
-    def _perform_server_detection(self, base_url):
-        """Detect whether the server is Secret Server or Platform via health
-        check endpoints, using a process-scoped cache.
-
-        The detected type is cached per normalized ``base_url`` on the
-        ``Authorizer`` base class and shared across all subclasses, so the
-        ``/api/v1/healthcheck`` + ``/health`` probe pair fires only once per
-        ``base_url`` per process. The cache is read/written under
-        ``_server_type_cache_lock`` for thread safety, but the network probe
-        itself runs OUTSIDE the lock; detection is idempotent, so a rare
-        double-probe under a race is harmless. Only successful detections are
-        cached -- failures re-probe on the next construction.
-
-        On both cache hits and fresh probes the per-instance ``_server_type``
-        attribute is set, because callers (``SecretServer.ensure_vault_url``
-        and ``PasswordGrantAuthorizer._refresh``) read ``self._server_type``.
+    def _perform_server_detection(self, base_url, server_type=None):
+        """Resolve whether the server is Secret Server or Platform.
+
+        When an explicit ``server_type`` is supplied the value is validated,
+        cached, and used directly -- NO health-check probe is issued. This is
+        the recommended path for callers that run each lookup in a fresh
+        process (e.g. some Ansible lookup-plugin runtimes) where the
+        process-scoped cache cannot help: skipping detection eliminates the
+        unauthenticated ``/api/v1/healthcheck`` + ``/health`` probe burst that
+        the Delinea Platform WAF rate-limits to 403.
+
+        Otherwise the type is detected via the health-check endpoints, using a
+        process-scoped cache. The detected type is cached per normalized
+        ``base_url`` on the ``Authorizer`` base class and shared across all
+        subclasses, so the probe pair fires only once per ``base_url`` per
+        process. The cache is read/written under ``_server_type_cache_lock``
+        for thread safety, but the network probe itself runs OUTSIDE the lock;
+        detection is idempotent, so a rare double-probe under a race is
+        harmless. Only successful detections are cached -- failures re-probe on
+        the next construction.
+
+        On every path the per-instance ``_server_type`` attribute is set,
+        because callers (``SecretServer.ensure_vault_url`` and
+        ``PasswordGrantAuthorizer._refresh``) read ``self._server_type``.
         """
         key = base_url.rstrip("/")
 
-        with Authorizer._server_type_cache_lock:
-            cached = Authorizer._server_type_cache.get(key)
+        if server_type is not None:
+            detected = self._normalize_server_type(server_type)
+            self._server_type = detected
+            self._cache_server_type(key, detected)
+            return
+
+        cached = self._get_cached_server_type(key)
         if cached is not None:
             self._server_type = cached
             return
@@ -228,8 +301,7 @@ def _perform_server_detection(self, base_url):
             )
 
         self._server_type = detected
-        with Authorizer._server_type_cache_lock:
-            Authorizer._server_type_cache.setdefault(key, detected)
+        self._cache_server_type(key, detected)
 
     def _validate_health_endpoint(self, url):
         """Validates if an endpoint returns healthy status."""
@@ -268,10 +340,14 @@ class AccessTokenAuthorizer(Authorizer):
     def get_access_token(self):
         return self.access_token
 
-    def __init__(self, access_token, base_url):
+    def __init__(self, access_token, base_url, server_type=None):
+        """
+        :param server_type: optionally ``"secret_server"`` or ``"platform"`` to
+                skip health-check detection entirely (no probe is issued).
+        """
         self.access_token = access_token
         self.base_url = base_url.rstrip("/")
-        self._perform_server_detection(self.base_url)
+        self._perform_server_detection(self.base_url, server_type=server_type)
 
 
 class PasswordGrantAuthorizer(Authorizer):
@@ -353,7 +429,20 @@ def _refresh(self, seconds_of_drift=300):
             else:
                 raise SecretServerError("Unknown server type for token request.")
 
-    def __init__(self, base_url, username, password, token_path_uri=None, domain=None):
+    def __init__(
+        self,
+        base_url,
+        username,
+        password,
+        token_path_uri=None,
+        domain=None,
+        server_type=None,
+    ):
+        """
+        :param server_type: optionally ``"secret_server"`` or ``"platform"`` to
+                skip health-check detection entirely (no probe is issued); the
+                matching token endpoint is selected without probing.
+        """
         self.base_url = base_url.rstrip("/")
         self.username = username
         self.password = password
@@ -361,6 +450,10 @@ def __init__(self, base_url, username, password, token_path_uri=None, domain=Non
         self.token_path_uri = token_path_uri  # May be None, will decide in _refresh
         self.token_url = None
         self.grant_request = None
+        # When an explicit type is given, resolve it now (no network) so the
+        # lazy detection in _refresh is skipped and no probe is ever issued.
+        if server_type is not None:
+            self._perform_server_detection(self.base_url, server_type=server_type)
 
     def get_access_token(self):
         self._refresh()
@@ -377,9 +470,15 @@ def __init__(
         domain,
         password,
         token_path_uri=None,
+        server_type=None,
     ):
         super().__init__(
-            base_url, username, password, token_path_uri=token_path_uri, domain=domain
+            base_url,
+            username,
+            password,
+            token_path_uri=token_path_uri,
+            domain=domain,
+            server_type=server_type,
         )
 
 
diff --git a/requirements.txt b/requirements.txt
index 3f6a39b..a84db72 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,8 +1,8 @@
-requests==2.32.4
+requests==2.33.0
 tox
 pytest
 python-dotenv
 flit
 black
-urllib3==2.6.3 # not directly required, pinned by Snyk to avoid a vulnerability
+urllib3==2.7.0 # not directly required, pinned by Snyk to avoid a vulnerability
 zipp==3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
diff --git a/tests/test_server_detection_cache.py b/tests/test_server_detection_cache.py
index b19fb60..137736e 100644
--- a/tests/test_server_detection_cache.py
+++ b/tests/test_server_detection_cache.py
@@ -21,7 +21,6 @@
     SecretServerError,
 )
 
-
 SECRET_SERVER_HEALTH = "/api/v1/healthcheck"
 PLATFORM_HEALTH = "/health"
 
@@ -209,3 +208,106 @@ def worker():
     # exactly 1).
     assert counter["rounds"] >= 1
     assert counter["rounds"] <= 20
+
+
+# Behavior 7: an explicit server_type override skips detection entirely (no probe).
+@pytest.mark.parametrize("server_type", ["platform", "secret_server"])
+def test_explicit_server_type_skips_probe(monkeypatch, server_type):
+    base_url = "https://anything.example.com"
+    # Every health endpoint is unhealthy: if any probe fired, detection would
+    # raise. It must not, because the override bypasses probing.
+    fake_get, counter = make_probe_counter(set())
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    inst = AccessTokenAuthorizer("tok", base_url, server_type=server_type)
+
+    assert inst._server_type == server_type
+    assert counter["total"] == 0  # zero probes -> no WAF burst
+    # The override seeds the shared cache for subsequent callers.
+    assert Authorizer._server_type_cache[base_url] == server_type
+
+
+# Behavior 8: the override is normalized (case/whitespace-insensitive).
+def test_explicit_server_type_is_normalized(monkeypatch):
+    fake_get, counter = make_probe_counter(set())
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    inst = AccessTokenAuthorizer(
+        "tok", "https://x.example.com", server_type="  Platform "
+    )
+
+    assert inst._server_type == "platform"
+    assert counter["total"] == 0
+
+
+# Behavior 9: an invalid override raises and issues no probe.
+def test_invalid_server_type_raises(monkeypatch):
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    with pytest.raises(SecretServerError):
+        AccessTokenAuthorizer("tok", "https://x.example.com", server_type="bogus")
+
+    assert counter["total"] == 0
+
+
+# Behavior 10: PasswordGrantAuthorizer with an override never probes in _refresh.
+def test_password_grant_override_skips_detection(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter(set())
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    grant = PasswordGrantAuthorizer(base_url, "user", "pass", server_type="platform")
+    assert grant._server_type == "platform"
+
+    try:
+        # The grant POST will fail offline, but detection must not have probed.
+        grant.get_access_token()
+    except Exception:
+        pass
+
+    assert counter["total"] == 0
+    # Platform token endpoint was selected without any health probe.
+    assert grant.token_path_uri == PasswordGrantAuthorizer.PLATFORM_TOKEN_PATH_URI
+
+
+# Behavior 11: the cache is bounded; the least-recently-used entry is evicted.
+def test_cache_is_bounded_lru(monkeypatch):
+    fake_get, _ = make_probe_counter(set())
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    maxsize = Authorizer._SERVER_TYPE_CACHE_MAXSIZE
+
+    # Fill exactly to capacity using the override path (no network needed).
+    for i in range(maxsize):
+        AccessTokenAuthorizer(
+            "tok", f"https://host-{i}.example.com", server_type="platform"
+        )
+    assert len(Authorizer._server_type_cache) == maxsize
+
+    first_key = "https://host-0.example.com"
+    # Touch host-0 so it becomes most-recently-used and survives the next insert.
+    Authorizer._get_cached_server_type(first_key)
+
+    # One more distinct URL overflows the cache by one entry.
+    AccessTokenAuthorizer("tok", "https://overflow.example.com", server_type="platform")
+
+    assert len(Authorizer._server_type_cache) == maxsize
+    assert first_key in Authorizer._server_type_cache  # survived (recently used)
+    assert "https://host-1.example.com" not in Authorizer._server_type_cache  # evicted
+
+
+# Behavior 12: the public clear-cache method forces re-detection.
+def test_public_clear_cache(monkeypatch):
+    base_url = "https://platform.example.com"
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    AccessTokenAuthorizer("tok", base_url)
+    assert counter["rounds"] == 1
+
+    Authorizer.clear_server_type_cache()
+    assert base_url not in Authorizer._server_type_cache
+
+    AccessTokenAuthorizer("tok", base_url)  # cache empty -> probes again
+    assert counter["rounds"] == 2

From f097ac76f237976ebf17a9ab68f3eb68e846d1e8 Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Tue, 23 Jun 2026 10:20:19 -0600
Subject: [PATCH 03/10] =?UTF-8?q?fix(server):=20=F0=9F=90=9B=20keep=20serv?=
 =?UTF-8?q?er=5Ftype=20override=20per-instance;=20pin=20requests=3D=3D2.34?=
 =?UTF-8?q?.2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                            | 25 ++++++++++++++--
 delinea/secrets/server.py            | 20 ++++++++-----
 requirements.txt                     |  2 +-
 tests/test_server_detection_cache.py | 43 ++++++++++++++++++++++------
 4 files changed, 70 insertions(+), 20 deletions(-)

diff --git a/README.md b/README.md
index a64c152..d8af401 100644
--- a/README.md
+++ b/README.md
@@ -31,7 +31,7 @@ There are three ways in which you can authorize the `SecretServer` and `SecretSe
 
 #### Password Authorization
 
-If using traditional `username` and `password` authentication to log in to your Secret Server either directly or through Platform, you can pass the `PasswordGrantAuthorizer` into the `SecretServer` class at instantiation. The `PasswordGrantAuthorizer` requires a `base_url`, `username`, and `password`. It optionally takes a `token_path_uri`, but defaults to `/oauth2/token` or `/identity/api/oauth2/token/xpmplatform`, depending on whether a secret server or platform is used for authentication.
+If using traditional `username` and `password` authentication to log in to your Secret Server either directly or through Platform, you can pass the `PasswordGrantAuthorizer` into the `SecretServer` class at instantiation. The `PasswordGrantAuthorizer` requires a `base_url`, `username`, and `password`. It optionally takes a `token_path_uri`, but defaults to `/oauth2/token` or `/identity/api/oauth2/token/xpmplatform`, depending on whether a secret server or platform is used for authentication. It also optionally takes a `server_type` (`"secret_server"` or `"platform"`) to skip automatic server-type detection — see [Server-Type Detection](#server-type-detection).
 
 ##### With Secret Server
 ```python
@@ -50,7 +50,7 @@ authorizer = PasswordGrantAuthorizer("https://platform.delinea.app", os.getenv("
 
 #### Domain Authorization
 
-To use a domain credential, use the `DomainPasswordGrantAuthorizer`. It requires a `base_url`, `username`, `domain`, and `password`. It optionally takes a `token_path_uri`, but defaults to `/oauth2/token`. It is applicable only when authentication is done using a secret server.
+To use a domain credential, use the `DomainPasswordGrantAuthorizer`. It requires a `base_url`, `username`, `domain`, and `password`. It optionally takes a `token_path_uri`, but defaults to `/oauth2/token`, and a `server_type` (see [Server-Type Detection](#server-type-detection)). It is applicable only when authentication is done using a secret server.
 
 ```python
 from delinea.secrets.server import DomainPasswordGrantAuthorizer
@@ -60,7 +60,7 @@ authorizer = DomainPasswordGrantAuthorizer("https://hostname/SecretServer", os.g
 
 #### Access Token Authorization
 
-If you already have an `access_token` of Secret Server or Platform user, you can pass directly via the `AccessTokenAuthorizer`. The `AccessTokenAuthorizer` requires a `access_token` and `base_url`.
+If you already have an `access_token` of Secret Server or Platform user, you can pass directly via the `AccessTokenAuthorizer`. The `AccessTokenAuthorizer` requires a `access_token` and `base_url`. It optionally takes a `server_type` (see [Server-Type Detection](#server-type-detection)).
 
 ##### With Secret Server
 ```python
@@ -77,6 +77,25 @@ from delinea.secrets.server import AccessTokenAuthorizer
 authorizer = AccessTokenAuthorizer("AgJ1slfZsEng9bKsssB-tic0Kh8I...", "https://platform.delinea.app")
 ```
 
+#### Server-Type Detection
+
+By default every authorizer automatically detects whether the `base_url` points at a Secret Server or a Platform instance by probing its health-check endpoints (`/api/v1/healthcheck` then `/health`). The result is cached per `base_url` for the lifetime of the process, so the probe pair fires only once per `base_url`.
+
+You can skip detection entirely by passing an explicit `server_type` of either `"secret_server"` or `"platform"`. When supplied, no health-check probe is issued. This is recommended for callers that run each lookup in a fresh, short-lived process (for example, some Ansible lookup-plugin runtimes), where a fresh process cannot benefit from the in-process cache and the repeated unauthenticated probes can be rate-limited to `403` by the Delinea Platform WAF.
+
+```python
+from delinea.secrets.server import AccessTokenAuthorizer
+
+# No health-check probe is issued; the type is used directly.
+authorizer = AccessTokenAuthorizer(
+    "AgJ1slfZsEng9bKsssB-tic0Kh8I...",
+    "https://platform.delinea.app",
+    server_type="platform",
+)
+```
+
+An explicit `server_type` applies only to the instance that supplies it and is never written to the shared cache, so it cannot affect auto-detection for other authorizers. If a `base_url` is ever re-provisioned to a different server type while a long-lived process is running, call `Authorizer.clear_server_type_cache()` to force re-detection.
+
 ## Secret Server Cloud
 
 The SDK API requires an `Authorizer` and either a `tenant` or a `base_url`. In the case of plaform authentication, only a `base_url` is supported.
diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index 5927be4..d0b0deb 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -256,14 +256,20 @@ def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
     def _perform_server_detection(self, base_url, server_type=None):
         """Resolve whether the server is Secret Server or Platform.
 
-        When an explicit ``server_type`` is supplied the value is validated,
-        cached, and used directly -- NO health-check probe is issued. This is
-        the recommended path for callers that run each lookup in a fresh
-        process (e.g. some Ansible lookup-plugin runtimes) where the
+        When an explicit ``server_type`` is supplied the value is validated
+        and used directly for THIS instance only -- NO health-check probe is
+        issued. This is the recommended path for callers that run each lookup
+        in a fresh process (e.g. some Ansible lookup-plugin runtimes) where the
         process-scoped cache cannot help: skipping detection eliminates the
         unauthenticated ``/api/v1/healthcheck`` + ``/health`` probe burst that
         the Delinea Platform WAF rate-limits to 403.
 
+        An explicit override is deliberately NOT written to the shared
+        process-scoped cache: the override is unverified, so seeding the cache
+        would let a wrong/typo'd value silently poison auto-detection for
+        unrelated callers using the same ``base_url`` in the same process. Only
+        verified probe detections populate the shared cache.
+
         Otherwise the type is detected via the health-check endpoints, using a
         process-scoped cache. The detected type is cached per normalized
         ``base_url`` on the ``Authorizer`` base class and shared across all
@@ -281,9 +287,9 @@ def _perform_server_detection(self, base_url, server_type=None):
         key = base_url.rstrip("/")
 
         if server_type is not None:
-            detected = self._normalize_server_type(server_type)
-            self._server_type = detected
-            self._cache_server_type(key, detected)
+            # Per-instance only; intentionally NOT seeded into the shared cache
+            # so an unverified override cannot poison auto-detection for others.
+            self._server_type = self._normalize_server_type(server_type)
             return
 
         cached = self._get_cached_server_type(key)
diff --git a/requirements.txt b/requirements.txt
index a84db72..46bae68 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-requests==2.33.0
+requests==2.34.2 # pinned to address CVE-2026-25645 (2.33.0 was never published)
 tox
 pytest
 python-dotenv
diff --git a/tests/test_server_detection_cache.py b/tests/test_server_detection_cache.py
index 137736e..6c6555f 100644
--- a/tests/test_server_detection_cache.py
+++ b/tests/test_server_detection_cache.py
@@ -210,7 +210,8 @@ def worker():
     assert counter["rounds"] <= 20
 
 
-# Behavior 7: an explicit server_type override skips detection entirely (no probe).
+# Behavior 7: an explicit server_type override skips detection entirely (no probe)
+# and is per-instance only -- it must NOT seed the shared process cache.
 @pytest.mark.parametrize("server_type", ["platform", "secret_server"])
 def test_explicit_server_type_skips_probe(monkeypatch, server_type):
     base_url = "https://anything.example.com"
@@ -223,8 +224,9 @@ def test_explicit_server_type_skips_probe(monkeypatch, server_type):
 
     assert inst._server_type == server_type
     assert counter["total"] == 0  # zero probes -> no WAF burst
-    # The override seeds the shared cache for subsequent callers.
-    assert Authorizer._server_type_cache[base_url] == server_type
+    # The unverified override must NOT be written to the shared cache (otherwise
+    # it could poison auto-detection for other callers using the same base_url).
+    assert base_url not in Authorizer._server_type_cache
 
 
 # Behavior 8: the override is normalized (case/whitespace-insensitive).
@@ -273,16 +275,17 @@ def test_password_grant_override_skips_detection(monkeypatch):
 
 # Behavior 11: the cache is bounded; the least-recently-used entry is evicted.
 def test_cache_is_bounded_lru(monkeypatch):
-    fake_get, _ = make_probe_counter(set())
+    # Every base_url detects as platform (healthy /health) so each distinct URL
+    # seeds one verified cache entry. Only verified detections populate the
+    # shared cache, so the cache must be filled via detection (not overrides).
+    fake_get, _ = make_probe_counter({PLATFORM_HEALTH})
     monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
 
     maxsize = Authorizer._SERVER_TYPE_CACHE_MAXSIZE
 
-    # Fill exactly to capacity using the override path (no network needed).
+    # Fill exactly to capacity via auto-detection.
     for i in range(maxsize):
-        AccessTokenAuthorizer(
-            "tok", f"https://host-{i}.example.com", server_type="platform"
-        )
+        AccessTokenAuthorizer("tok", f"https://host-{i}.example.com")
     assert len(Authorizer._server_type_cache) == maxsize
 
     first_key = "https://host-0.example.com"
@@ -290,13 +293,35 @@ def test_cache_is_bounded_lru(monkeypatch):
     Authorizer._get_cached_server_type(first_key)
 
     # One more distinct URL overflows the cache by one entry.
-    AccessTokenAuthorizer("tok", "https://overflow.example.com", server_type="platform")
+    AccessTokenAuthorizer("tok", "https://overflow.example.com")
 
     assert len(Authorizer._server_type_cache) == maxsize
     assert first_key in Authorizer._server_type_cache  # survived (recently used)
     assert "https://host-1.example.com" not in Authorizer._server_type_cache  # evicted
 
 
+# Behavior 13: an unverified override must not poison auto-detection for a later
+# caller that relies on probing for the same base_url.
+def test_override_does_not_poison_autodetect(monkeypatch):
+    base_url = "https://platform.example.com"
+    # The server is really a platform (healthy /health); probing would detect it.
+    fake_get, counter = make_probe_counter({PLATFORM_HEALTH})
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    # First caller supplies a WRONG override and issues no probe.
+    poisoner = AccessTokenAuthorizer("tok", base_url, server_type="secret_server")
+    assert poisoner._server_type == "secret_server"
+    assert counter["total"] == 0
+    assert base_url not in Authorizer._server_type_cache  # not seeded
+
+    # Second caller relies on auto-detection -> must probe and get the real type,
+    # NOT the poisoned override value.
+    detected = AccessTokenAuthorizer("tok", base_url)
+    assert detected._server_type == "platform"
+    assert counter["rounds"] == 1  # a real probe fired
+    assert Authorizer._server_type_cache[base_url] == "platform"
+
+
 # Behavior 12: the public clear-cache method forces re-detection.
 def test_public_clear_cache(monkeypatch):
     base_url = "https://platform.example.com"

From 354441b98664c14d56e06f94e805c916be061e12 Mon Sep 17 00:00:00 2001
From: Lint Action <lint-action@samuelmeuli.com>
Date: Tue, 23 Jun 2026 17:33:04 +0000
Subject: [PATCH 04/10] Fix code style issues with Black

---
 example.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/example.py b/example.py
index d37d7cc..9e3da94 100644
--- a/example.py
+++ b/example.py
@@ -23,10 +23,8 @@
     try:
         secret = secret_server_cloud.get_secret(os.getenv("TSS_SECRET_ID"))
         serverSecret = ServerSecret(**secret)
-        print(
-            f"""username: {serverSecret.fields['username'].value}
+        print(f"""username: {serverSecret.fields['username'].value}
                 password: {serverSecret.fields['password'].value}
-                template: {serverSecret.secret_template_name}"""
-        )
+                template: {serverSecret.secret_template_name}""")
     except SecretServerError as error:
         print(error.response.text)

From 5e8925f8fdc9be55fad29d09ce203b3acae59d1a Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Thu, 23 Jul 2026 13:18:18 -0600
Subject: [PATCH 05/10] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20bump=20requests?=
 =?UTF-8?q?=202.34.2=20&=20urllib3=202.7.0;=20drop=20Python=203.8/3.9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Clears CVE-2026-25645 (requests) and CVE-2026-44431/44432 (urllib3) per
work item 741117. The fixed releases require Python >= 3.10.

- requirements.txt: requests==2.34.2, urllib3==2.7.0
- pyproject.toml: requires-python >=3.10, requests floor >= 2.34.2
- tox.ini / run_tests.yml: matrix trimmed to 3.10-3.12
- README: minimum Python 3.10

BREAKING: drops Python 3.8/3.9 support (both EOL) and raises the
published requests floor for downstream consumers.
---
 .github/workflows/run_tests.yml |  3 ++-
 README.md                       |  2 +-
 pyproject.toml                  | 16 +++++++++++-----
 requirements.txt                |  4 ++--
 tox.ini                         |  3 ++-
 5 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml
index af891d6..b8a2acb 100644
--- a/.github/workflows/run_tests.yml
+++ b/.github/workflows/run_tests.yml
@@ -9,7 +9,8 @@ jobs:
     environment: testing
     strategy:
       matrix:
-        python: [3.8, 3.9, "3.10", "3.11"]
+        # Python 3.8/3.9 dropped: fixed requests/urllib3 pins require Python >= 3.10 (work item 741117)
+        python: ["3.10", "3.11", "3.12"]
 
     steps:
       - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
diff --git a/README.md b/README.md
index a64c152..72343ed 100644
--- a/README.md
+++ b/README.md
@@ -188,7 +188,7 @@ When using a self-signed certificate for SSL, the `REQUESTS_CA_BUNDLE` environme
 
 ## Create a Build Environment (optional)
 
-The SDK requires [Python 3.8](https://www.python.org/downloads/) or higher.
+The SDK requires [Python 3.10](https://www.python.org/downloads/) or higher.
 
 First, ensure Python is in `$PATH`, then run:
 
diff --git a/pyproject.toml b/pyproject.toml
index 737c6fa..36dc096 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -9,14 +9,20 @@ author-email = "GitHub@delinea.com"
 classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11"
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12"
 ]
 description-file = "README.md"
+# BREAKING (consumer-facing): the requests floor was raised from 2.12.5 to 2.34.2
+# to clear CVE-2026-25645 (requests) and its transitive urllib3 advisories for
+# downstream installs, not just CI. requests 2.34.2 requires Python >= 3.10.
 requires = [
-    "requests >= 2.12.5"
+    "requests >= 2.34.2"
 ]
-requires-python=">=3.8"
+# BREAKING (consumer-facing): minimum Python raised from 3.8 to 3.10. The fixed
+# requests/urllib3 releases that clear the flagged CVEs dropped 3.8/3.9 support
+# (both EOL). Consumers on Python 3.8/3.9 must stay on an older SDK release or
+# upgrade their runtime. See work item 741117.
+requires-python=">=3.10"
 dist-name = "python-tss-sdk"
diff --git a/requirements.txt b/requirements.txt
index 3f6a39b..32891dc 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,8 +1,8 @@
-requests==2.32.4
+requests==2.34.2 # pinned to address CVE-2026-25645 (2.33.0 was never published); requires Python >= 3.10
 tox
 pytest
 python-dotenv
 flit
 black
-urllib3==2.6.3 # not directly required, pinned by Snyk to avoid a vulnerability
+urllib3==2.7.0 # not directly required, pinned by Snyk to avoid a vulnerability (CVE-2026-44431/44432); requires Python >= 3.10
 zipp==3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
diff --git a/tox.ini b/tox.ini
index 2420de2..a38f05a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -6,7 +6,8 @@
 # Docs for tox config -> https://tox.readthedocs.io/en/latest/config.html
 
 [tox]
-envlist = 3.8, 3.9, 3.10, 3.11, 3.12
+# Python 3.8/3.9 dropped: fixed requests/urllib3 pins require Python >= 3.10 (work item 741117)
+envlist = 3.10, 3.11, 3.12
 isolated_build = True
 skipsdist = True
 

From 7c899ac0b8840895acdf9a41c6a7e3d4c87af32e Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Tue, 28 Jul 2026 12:03:16 -0600
Subject: [PATCH 06/10] Clean resolve of requirements.txt

---
 requirements.txt | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 46bae68..0cb1984 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,8 +1,10 @@
 requests==2.34.2 # pinned to address CVE-2026-25645 (2.33.0 was never published)
 tox
 pytest
-python-dotenv
+python-dotenv==1.2.2 # pinned to address CVE-2026-28684 (symlink attack in set_key/unset_key)
 flit
-black
+black==26.5.1 # pinned to address CVE-2026-32274 (directory traversal) and CVE-2024-21503 (ReDoS)
 urllib3==2.7.0 # not directly required, pinned by Snyk to avoid a vulnerability
 zipp==3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
+filelock==3.32.0 # not directly required (transitive via tox), pinned to address CVE-2026-22701 and CVE-2025-68146
+idna==3.18 # not directly required (transitive via requests), pinned to address CVE-2026-45409

From e6279d5c3d265be03061174a69436cdff2669660 Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Thu, 6 Aug 2026 17:22:21 -0600
Subject: [PATCH 07/10] =?UTF-8?q?fix(server):=20=F0=9F=90=9B=20add=20missi?=
 =?UTF-8?q?ng=20http=20timeouts,=20refresh=20grant=20before=20expiry,=20at?=
 =?UTF-8?q?tach=20response=20to=20errors,=20=F0=9F=A7=AA=20add=20offline?=
 =?UTF-8?q?=20security=20tests?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

security review remediation, phase 1 (DevPlan.md)
==

- every http call now passes an explicit timeout via DEFAULT_REQUEST_TIMEOUT;
  the folder and lookup calls had none and could hang a consumer forever
- oauth2 grant now refreshes up to 300s before expiry; the drift sign was
  inverted so expired tokens were reused for up to 300s past expiry
- SecretServerError keeps the server response (error.response works now);
  a 4xx json body without message/error keys no longer masks the failure
  with UnboundLocalError
- example masks the password value instead of printing it
- new offline test suite covers timeout coverage on every request path,
  refresh boundary behavior, and error plumbing; no live credentials needed
---
 delinea/secrets/server.py     |  66 ++++++++---
 example.py                    |   3 +-
 tests/test_security_phase1.py | 205 ++++++++++++++++++++++++++++++++++
 3 files changed, 259 insertions(+), 15 deletions(-)
 create mode 100644 tests/test_security_phase1.py

diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index d0b0deb..9527e3d 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -24,6 +24,10 @@
 
 import requests
 
+# Applied to every HTTP call the SDK makes; ``requests`` has no default
+# timeout, so an omitted value would let a stalled connection hang forever.
+DEFAULT_REQUEST_TIMEOUT = 60
+
 
 @dataclass
 class ServerSecret:
@@ -152,6 +156,7 @@ class SecretServerError(Exception):
 
     def __init__(self, message, response=None, *args, **kwargs):
         self.message = message
+        self.response = response
         super().__init__(*args, **kwargs)
 
 
@@ -312,7 +317,7 @@ def _perform_server_detection(self, base_url, server_type=None):
     def _validate_health_endpoint(self, url):
         """Validates if an endpoint returns healthy status."""
         try:
-            response = requests.get(url, timeout=60)
+            response = requests.get(url, timeout=DEFAULT_REQUEST_TIMEOUT)
         except Exception:
             return False
 
@@ -373,7 +378,9 @@ def get_access_grant(token_url, grant_request):
                 other than a valid Access Grant
         """
 
-        response = requests.post(token_url, grant_request, timeout=60)
+        response = requests.post(
+            token_url, grant_request, timeout=DEFAULT_REQUEST_TIMEOUT
+        )
 
         try:  # TSS returns a 200 (OK) containing HTML for some error conditions
             return json.loads(SecretServer.process(response).content)
@@ -391,7 +398,7 @@ def _refresh(self, seconds_of_drift=300):
         if (
             hasattr(self, "access_grant")
             and self.access_grant_refreshed
-            + timedelta(seconds=self.access_grant["expires_in"] + seconds_of_drift)
+            + timedelta(seconds=self.access_grant["expires_in"] - seconds_of_drift)
             > datetime.now()
         ):
             return
@@ -514,6 +521,9 @@ def process(response):
         if response.status_code >= 200 and response.status_code < 300:
             return response
         if response.status_code >= 400 and response.status_code < 500:
+            # Fallback used when the body is JSON but carries no recognized
+            # message/error key.
+            message = f"HTTP {response.status_code}"
             try:
                 content = json.loads(response.content)
                 if "message" in content:
@@ -564,7 +574,9 @@ def ensure_vault_url(self):
                 access_token = self.authorizer.get_access_token()
                 vaults_endpoint = self.platform_url + "/vaultbroker/api/vaults"
                 headers = {"Authorization": f"Bearer {access_token}"}
-                resp = requests.get(vaults_endpoint, headers=headers, timeout=60)
+                resp = requests.get(
+                    vaults_endpoint, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
+                )
                 if resp.status_code != 200:
                     raise SecretServerError(
                         f"Failed to fetch vault details: HTTP {resp.status_code} - {resp.text}"
@@ -605,7 +617,9 @@ def get_secret_json(self, id, query_params=None):
 
         if query_params is None:
             return self.process(
-                requests.get(endpoint_url, headers=headers, timeout=60)
+                requests.get(
+                    endpoint_url, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
+                )
             ).text
         else:
             return self.process(
@@ -613,7 +627,7 @@ def get_secret_json(self, id, query_params=None):
                     endpoint_url,
                     params=query_params,
                     headers=headers,
-                    timeout=60,
+                    timeout=DEFAULT_REQUEST_TIMEOUT,
                 )
             ).text
 
@@ -639,13 +653,18 @@ def get_folder_json(self, id, query_params=None, get_all_children=True):
             query_params["getAllChildren"] = "true"
 
         if query_params is None:
-            return self.process(requests.get(endpoint_url, headers=headers)).text
+            return self.process(
+                requests.get(
+                    endpoint_url, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
+                )
+            ).text
         else:
             return self.process(
                 requests.get(
                     endpoint_url,
                     params=query_params,
                     headers=headers,
+                    timeout=DEFAULT_REQUEST_TIMEOUT,
                 )
             ).text
 
@@ -682,7 +701,9 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
                     if query_params is None:
                         item["itemValue"] = self.process(
                             requests.get(
-                                endpoint_url, headers=self.headers(), timeout=60
+                                endpoint_url,
+                                headers=self.headers(),
+                                timeout=DEFAULT_REQUEST_TIMEOUT,
                             )
                         )
                     else:
@@ -691,7 +712,7 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
                                 endpoint_url,
                                 params=query_params,
                                 headers=self.headers(),
-                                timeout=60,
+                                timeout=DEFAULT_REQUEST_TIMEOUT,
                             )
                         )
         return secret
@@ -780,7 +801,9 @@ def search_secrets(self, query_params=None):
 
         if query_params is None:
             return self.process(
-                requests.get(endpoint_url, headers=headers, timeout=60)
+                requests.get(
+                    endpoint_url, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
+                )
             ).text
         else:
             return self.process(
@@ -788,7 +811,7 @@ def search_secrets(self, query_params=None):
                     endpoint_url,
                     params=query_params,
                     headers=headers,
-                    timeout=60,
+                    timeout=DEFAULT_REQUEST_TIMEOUT,
                 )
             ).text
 
@@ -809,13 +832,18 @@ def lookup_folders(self, query_params=None):
         endpoint_url = f"{self.api_url}/folders/lookup"
 
         if query_params is None:
-            return self.process(requests.get(endpoint_url, headers=headers)).text
+            return self.process(
+                requests.get(
+                    endpoint_url, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
+                )
+            ).text
         else:
             return self.process(
                 requests.get(
                     endpoint_url,
                     params=query_params,
                     headers=headers,
+                    timeout=DEFAULT_REQUEST_TIMEOUT,
                 )
             ).text
 
@@ -836,7 +864,12 @@ def get_secret_ids_by_folderid(self, folder_id):
         params = {"filter.folderId": folder_id}
         endpoint_url = f"{self.api_url}/secrets/search-total"
         params["take"] = self.process(
-            requests.get(endpoint_url, params=params, headers=headers, timeout=60)
+            requests.get(
+                endpoint_url,
+                params=params,
+                headers=headers,
+                timeout=DEFAULT_REQUEST_TIMEOUT,
+            )
         ).text
         response = self.search_secrets(query_params=params)
 
@@ -872,7 +905,12 @@ def get_child_folder_ids_by_folderid(self, folder_id):
         endpoint_url = f"{self.api_url}/folders/lookup"
 
         params["take"] = self.process(
-            requests.get(endpoint_url, params=params, headers=headers)
+            requests.get(
+                endpoint_url,
+                params=params,
+                headers=headers,
+                timeout=DEFAULT_REQUEST_TIMEOUT,
+            )
         ).json()["total"]
         # Handle result of zero child folders
         if params["take"] != 0:
diff --git a/example.py b/example.py
index 9e3da94..e3bf628 100644
--- a/example.py
+++ b/example.py
@@ -23,8 +23,9 @@
     try:
         secret = secret_server_cloud.get_secret(os.getenv("TSS_SECRET_ID"))
         serverSecret = ServerSecret(**secret)
+        # Never print secret values; mask them in any console/log output.
         print(f"""username: {serverSecret.fields['username'].value}
-                password: {serverSecret.fields['password'].value}
+                password: ********
                 template: {serverSecret.secret_template_name}""")
     except SecretServerError as error:
         print(error.response.text)
diff --git a/tests/test_security_phase1.py b/tests/test_security_phase1.py
new file mode 100644
index 0000000..dc1eaf0
--- /dev/null
+++ b/tests/test_security_phase1.py
@@ -0,0 +1,205 @@
+"""Offline unit tests for the Phase 1 security-review fixes (see DevPlan.md).
+
+Covers:
+- SDK-1: every HTTP call the SDK issues passes an explicit ``timeout``.
+- SDK-3: the OAuth2 grant refreshes *before* expiry (drift subtracted).
+- SDK-9: ``SecretServerError.response`` is populated, and ``process()`` no
+  longer raises ``UnboundLocalError`` on a 4xx JSON body without a
+  message/error key.
+
+Fully OFFLINE, in the style of ``tests/test_server_detection_cache.py``: the
+network is mocked by patching ``delinea.secrets.server.requests``.
+"""
+
+import json
+from datetime import datetime, timedelta
+
+import pytest
+
+from delinea.secrets.server import (
+    AccessTokenAuthorizer,
+    PasswordGrantAuthorizer,
+    SecretServer,
+    SecretServerClientError,
+    SecretServerError,
+)
+
+
+class FakeResponse:
+    """Minimal stand-in for ``requests.Response`` as consumed by the SDK."""
+
+    def __init__(self, status_code=200, json_data=None, text=None):
+        self.status_code = status_code
+        self._json = json_data
+        if text is not None:
+            self.text = text
+        elif json_data is not None:
+            self.text = json.dumps(json_data)
+        else:
+            self.text = ""
+        self.content = self.text.encode()
+
+    def json(self):
+        if self._json is None:
+            raise ValueError("no JSON body")
+        return self._json
+
+
+# ---------------------------------------------------------------------------
+# SDK-1: timeout coverage
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def http_spy(monkeypatch):
+    """Replace ``requests.get``/``requests.post`` with a recording fake that
+    serves canned, route-appropriate responses. Returns the list of recorded
+    (method, url, kwargs) calls."""
+
+    calls = []
+
+    def route(url, params=None):
+        if url.endswith("/secrets/search-total"):
+            return FakeResponse(text="3")
+        if url.endswith("/folders/lookup"):
+            return FakeResponse(
+                json_data={"total": 2, "records": [{"id": 7}, {"id": 8}]}
+            )
+        if url.endswith("/secrets"):
+            return FakeResponse(json_data={"records": [{"id": 1}]})
+        if "/secrets/" in url:
+            return FakeResponse(json_data={"items": []})
+        if "/folders/" in url:
+            return FakeResponse(json_data={"id": 1})
+        return FakeResponse(json_data={})
+
+    def fake_get(url, *args, **kwargs):
+        calls.append(("GET", url, kwargs))
+        return route(url, kwargs.get("params"))
+
+    def fake_post(url, *args, **kwargs):
+        calls.append(("POST", url, kwargs))
+        return FakeResponse(json_data={"access_token": "tok", "expires_in": 1200})
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+    monkeypatch.setattr("delinea.secrets.server.requests.post", fake_post)
+    return calls
+
+
+def _server(base_url="https://ss.example.com"):
+    authorizer = AccessTokenAuthorizer("tok", base_url, server_type="secret_server")
+    return SecretServer(base_url, authorizer)
+
+
+def test_every_http_call_passes_a_timeout(http_spy):
+    """Exercise every SecretServer request path and assert an explicit timeout
+    is passed on each underlying HTTP call (SDK-1)."""
+    server = _server()
+
+    server.get_secret_json(1)
+    server.get_secret_json(1, query_params={"a": "b"})
+    server.get_folder_json(1, query_params={})  # get_all_children default True
+    server.get_folder_json(1, query_params={"a": "b"}, get_all_children=False)
+    server.search_secrets()
+    server.search_secrets(query_params={"a": "b"})
+    server.lookup_folders()
+    server.lookup_folders(query_params={"a": "b"})
+    server.get_secret_ids_by_folderid(2)
+    server.get_child_folder_ids_by_folderid(2)
+
+    assert len(http_spy) > 0
+    missing = [
+        (method, url) for method, url, kwargs in http_spy if "timeout" not in kwargs
+    ]
+    assert missing == [], f"HTTP calls issued without a timeout: {missing}"
+
+
+def test_token_grant_passes_a_timeout(http_spy):
+    """The OAuth2 token POST must also carry a timeout (SDK-1)."""
+    grant = PasswordGrantAuthorizer(
+        "https://ss.example.com", "user", "pass", server_type="secret_server"
+    )
+    grant.get_access_token()
+
+    posts = [c for c in http_spy if c[0] == "POST"]
+    assert len(posts) == 1
+    assert "timeout" in posts[0][2]
+
+
+# ---------------------------------------------------------------------------
+# SDK-3: refresh drift is subtracted (refresh happens BEFORE expiry)
+# ---------------------------------------------------------------------------
+
+
+def _grant_authorizer_with_token(refreshed_seconds_ago, expires_in=1200):
+    auth = PasswordGrantAuthorizer(
+        "https://ss.example.com", "user", "pass", server_type="secret_server"
+    )
+    auth.access_grant = {"access_token": "old", "expires_in": expires_in}
+    auth.access_grant_refreshed = datetime.now() - timedelta(
+        seconds=refreshed_seconds_ago
+    )
+    # Shadow the grant call on the instance so no network is needed.
+    auth.get_access_grant = lambda token_url, grant_request: {
+        "access_token": "new",
+        "expires_in": expires_in,
+    }
+    return auth
+
+
+def test_refresh_fires_inside_drift_window():
+    """A token expiring within the 300s drift window is refreshed early."""
+    # expires_in=1200, refreshed 901s ago -> 299s of validity left (< 300 drift)
+    auth = _grant_authorizer_with_token(refreshed_seconds_ago=1200 - 299)
+    assert auth.get_access_token() == "new"
+
+
+def test_refresh_skipped_outside_drift_window():
+    """A token with more than the drift window of validity left is reused."""
+    # expires_in=1200, refreshed 899s ago -> 301s of validity left (> 300 drift)
+    auth = _grant_authorizer_with_token(refreshed_seconds_ago=1200 - 301)
+    assert auth.get_access_token() == "old"
+
+
+def test_expired_token_is_refreshed():
+    """A token past its expiry is never reused (regression guard: the old
+    ``+ seconds_of_drift`` arithmetic kept expired tokens alive for 300s)."""
+    auth = _grant_authorizer_with_token(refreshed_seconds_ago=1201)
+    assert auth.get_access_token() == "new"
+
+
+# ---------------------------------------------------------------------------
+# SDK-9: exception plumbing
+# ---------------------------------------------------------------------------
+
+
+def test_error_response_attribute_is_set():
+    response = FakeResponse(status_code=403)
+    err = SecretServerError("denied", response)
+    assert err.response is response
+    assert err.message == "denied"
+
+
+def test_process_4xx_json_without_message_key():
+    """A 4xx JSON body lacking message/error keys must raise a client error
+    with a fallback message, not ``UnboundLocalError``."""
+    response = FakeResponse(status_code=403, json_data={"foo": 1})
+    with pytest.raises(SecretServerClientError) as excinfo:
+        SecretServer.process(response)
+    assert excinfo.value.response is response
+    assert "403" in excinfo.value.message
+
+
+def test_process_4xx_json_with_message_key():
+    response = FakeResponse(status_code=400, json_data={"message": "bad request"})
+    with pytest.raises(SecretServerClientError) as excinfo:
+        SecretServer.process(response)
+    assert excinfo.value.message == "bad request"
+    assert excinfo.value.response is response
+
+
+def test_process_4xx_non_json_body():
+    response = FakeResponse(status_code=404, text="<html>not found</html>")
+    with pytest.raises(SecretServerClientError) as excinfo:
+        SecretServer.process(response)
+    assert excinfo.value.response is response

From 0b7c584b94c8cfd62b6f9258cba033d0c05c3fa9 Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Fri, 7 Aug 2026 17:12:31 -0600
Subject: [PATCH 08/10] =?UTF-8?q?fix(server):=20=F0=9F=90=9B=20warn=20on?=
 =?UTF-8?q?=20plaintext=20http,=20tighten=20health-check=20validation,=20s?=
 =?UTF-8?q?anitize=20error=20bodies,=20validate=20vault=20redirect,=20?=
 =?UTF-8?q?=F0=9F=A7=AA=20add=20offline=20security=20tests?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

security review remediation, phase 2 (DevPlan.md)
==

- warn (UserWarning) when base_url is not https; credentials and bearer
  tokens would otherwise travel in plaintext with no signal to the caller.
  strict rejection is deferred to v3.0 to avoid breaking localhost/lab setups
- health-check probing now requires a 2xx status and an exact "healthy"
  match instead of a substring check; the old check matched "Unhealthy" and
  ignored the http status entirely
- exception messages no longer echo raw response bodies; the secrets
  endpoint omits the body outright, other endpoints get a capped, clearly
  truncated excerpt
- the platform vault-broker redirect url is now required to be a valid
  https url before any token is sent to it
- SecretServerError now passes its message through to Exception.__init__,
  so str(error) is populated instead of always empty
- new offline test suite covers all four fixes; existing FakeResponse
  fixtures updated with .ok/.status_code/.text to match the tightened
  health-check contract
---
 delinea/secrets/server.py            | 101 +++++++++--
 tests/test_security_phase2.py        | 250 +++++++++++++++++++++++++++
 tests/test_server_detection_cache.py |   7 +-
 3 files changed, 343 insertions(+), 15 deletions(-)
 create mode 100644 tests/test_security_phase2.py

diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index 9527e3d..e302fb8 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -15,19 +15,59 @@
 """
 
 import json
+import logging
 import re
+import warnings
 from abc import ABC, abstractmethod
 from collections import OrderedDict
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from threading import Lock
+from urllib.parse import urlsplit
 
 import requests
 
+logger = logging.getLogger(__name__)
+
 # Applied to every HTTP call the SDK makes; ``requests`` has no default
 # timeout, so an omitted value would let a stalled connection hang forever.
 DEFAULT_REQUEST_TIMEOUT = 60
 
+# Cap on how much of a server response body is echoed into an exception
+# message, so a malformed/oversized response cannot flood logs and so
+# exception text stays clearly distinguishable from a full response body.
+_BODY_EXCERPT_LIMIT = 200
+
+
+def _warn_if_insecure(base_url):
+    """Warn when ``base_url`` does not use ``https``.
+
+    Credentials (password / client_secret) and bearer tokens are sent to
+    ``base_url`` in plaintext when the scheme is not ``https``. This only
+    warns today, to preserve compatibility with existing localhost/lab
+    setups that use plain HTTP.
+    TODO(v3.0): reject a non-https ``base_url`` by default, with an explicit
+    opt-out (e.g. ``allow_http=True``) for those setups.
+    """
+    if urlsplit(base_url).scheme.lower() != "https":
+        warnings.warn(
+            f"base_url {base_url!r} does not use https; credentials and "
+            "bearer tokens will be sent unencrypted.",
+            UserWarning,
+            stacklevel=3,
+        )
+
+
+def _safe_body_excerpt(text, limit=_BODY_EXCERPT_LIMIT):
+    """Return a length-capped excerpt of a response body for use in error
+    messages, marked when truncated so it's clearly not the full body."""
+    if text is None:
+        return ""
+    text = str(text)
+    if len(text) <= limit:
+        return text
+    return text[:limit] + "...[truncated]"
+
 
 @dataclass
 class ServerSecret:
@@ -157,7 +197,9 @@ class SecretServerError(Exception):
     def __init__(self, message, response=None, *args, **kwargs):
         self.message = message
         self.response = response
-        super().__init__(*args, **kwargs)
+        # Pass message through so str(exception) is populated for default
+        # traceback/log output, not just the .message attribute.
+        super().__init__(message, *args, **kwargs)
 
 
 class SecretServerClientError(SecretServerError):
@@ -315,22 +357,34 @@ def _perform_server_detection(self, base_url, server_type=None):
         self._cache_server_type(key, detected)
 
     def _validate_health_endpoint(self, url):
-        """Validates if an endpoint returns healthy status."""
+        """Validates if an endpoint returns healthy status.
+
+        Requires a successful HTTP status (2xx) AND either a JSON body of
+        ``{"Healthy": true}`` or a body that is *exactly* (case-insensitive,
+        surrounding whitespace ignored) ``"healthy"``. A prior substring
+        check (``b"healthy" in body``) also matched ``"Unhealthy"`` and
+        ignored the HTTP status entirely, letting an error page or captive
+        portal flip detection.
+        """
         try:
             response = requests.get(url, timeout=DEFAULT_REQUEST_TIMEOUT)
-        except Exception:
+        except Exception as exc:
+            logger.debug("Health probe to %s failed: %s", url, type(exc).__name__)
             return False
 
-        try:
-            response_body = response.content
-        except Exception:
+        if not response.ok:
             return False
 
         try:
             json_data = response.json()
-            return json_data.get("Healthy", False)
+            return bool(json_data.get("Healthy", False))
         except Exception:
-            return b"Healthy" in response_body or b"healthy" in response_body
+            pass
+
+        try:
+            return response.text.strip().lower() == "healthy"
+        except Exception:
+            return False
 
     @abstractmethod
     def get_access_token(self):
@@ -358,6 +412,7 @@ def __init__(self, access_token, base_url, server_type=None):
         """
         self.access_token = access_token
         self.base_url = base_url.rstrip("/")
+        _warn_if_insecure(self.base_url)
         self._perform_server_detection(self.base_url, server_type=server_type)
 
 
@@ -457,6 +512,7 @@ def __init__(
                 matching token endpoint is selected without probing.
         """
         self.base_url = base_url.rstrip("/")
+        _warn_if_insecure(self.base_url)
         self.username = username
         self.password = password
         self.domain = domain
@@ -547,7 +603,7 @@ def __init__(
         api_path_uri=API_PATH_URI,
     ):
         """
-        :param base_url: The base URL e.g. ``http://localhost/SecretServer``
+        :param base_url: The base URL e.g. ``https://localhost/SecretServer``
         :type base_url: str
         :param authorizer: The authorization method to be used
         :type authorizer: Authorizer
@@ -555,6 +611,7 @@ def __init__(
         :type api_path_uri: str
         """
         self.base_url = base_url.rstrip("/")
+        _warn_if_insecure(self.base_url)
         self.platform_url = self.base_url
         self.authorizer = authorizer
         self._api_path_uri = api_path_uri
@@ -579,7 +636,8 @@ def ensure_vault_url(self):
                 )
                 if resp.status_code != 200:
                     raise SecretServerError(
-                        f"Failed to fetch vault details: HTTP {resp.status_code} - {resp.text}"
+                        f"Failed to fetch vault details: HTTP {resp.status_code} - "
+                        f"{_safe_body_excerpt(resp.text)}"
                     )
                 try:
                     data = resp.json()
@@ -590,6 +648,15 @@ def ensure_vault_url(self):
                         conn = vault.get("connection", {})
                         url = conn.get("url")
                         if url:
+                            parsed = urlsplit(url)
+                            if parsed.scheme != "https" or not parsed.netloc:
+                                raise SecretServerError(
+                                    "Vault connection URL is not a valid https "
+                                    f"URL: {_safe_body_excerpt(url)}"
+                                )
+                            logger.info(
+                                "Switching base_url to platform vault connection URL"
+                            )
                             self.base_url = url.rstrip("/")
                             self._vault_url_fetched = True
                             return
@@ -692,7 +759,9 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
         try:
             secret = json.loads(response)
         except json.JSONDecodeError:
-            raise SecretServerError(response)
+            # This is the secrets endpoint: never echo the raw body into an
+            # exception message, since it may contain secret field values.
+            raise SecretServerError("Unable to parse secret response as JSON.")
 
         if fetch_file_attachments:
             for item in secret["items"]:
@@ -741,7 +810,10 @@ def get_folder(self, id, query_params=None, get_all_children=False):
         try:
             folder = json.loads(response)
         except json.JSONDecodeError:
-            raise SecretServerError(response)
+            raise SecretServerError(
+                f"Unable to parse folder response as JSON: "
+                f"{_safe_body_excerpt(response)}"
+            )
 
         return folder
 
@@ -876,7 +948,10 @@ def get_secret_ids_by_folderid(self, folder_id):
         try:
             secrets = json.loads(response)
         except json.JSONDecodeError:
-            raise SecretServerError(response)
+            raise SecretServerError(
+                f"Unable to parse secrets search response as JSON: "
+                f"{_safe_body_excerpt(response)}"
+            )
 
         secret_ids = []
         for secret in secrets["records"]:
diff --git a/tests/test_security_phase2.py b/tests/test_security_phase2.py
new file mode 100644
index 0000000..b25d8b2
--- /dev/null
+++ b/tests/test_security_phase2.py
@@ -0,0 +1,250 @@
+"""Offline unit tests for the Phase 2 security-review fixes (see DevPlan.md).
+
+Covers:
+- SDK-2: a UserWarning is emitted when base_url is not https.
+- SDK-4: health-check validation requires a 2xx status and an exact
+  "healthy" match, no longer a "healthy" substring match with no status
+  check.
+- SDK-6: response bodies are truncated/omitted from exception messages.
+- SDK-7: the platform vault-broker redirect URL must be a valid https URL.
+
+Fully OFFLINE, in the style of ``tests/test_server_detection_cache.py``: the
+network is mocked by patching ``delinea.secrets.server.requests``.
+"""
+
+import json
+
+import pytest
+
+from delinea.secrets.server import (
+    AccessTokenAuthorizer,
+    Authorizer,
+    PasswordGrantAuthorizer,
+    SecretServer,
+    SecretServerError,
+)
+
+
+class FakeResponse:
+    """Minimal stand-in for ``requests.Response``."""
+
+    def __init__(self, status_code=200, json_data=None, text=None):
+        self.status_code = status_code
+        self.ok = 200 <= status_code < 300
+        self._json = json_data
+        if text is not None:
+            self.text = text
+        elif json_data is not None:
+            self.text = json.dumps(json_data)
+        else:
+            self.text = ""
+        self.content = self.text.encode()
+
+    def json(self):
+        if self._json is None:
+            raise ValueError("no JSON body")
+        return self._json
+
+
+@pytest.fixture(autouse=True)
+def clear_detection_cache():
+    """Same isolation as tests/test_server_detection_cache.py: the detection
+    cache is process-global."""
+    Authorizer._clear_server_type_cache()
+    yield
+    Authorizer._clear_server_type_cache()
+
+
+# ---------------------------------------------------------------------------
+# SDK-2: warn on non-https base_url
+# ---------------------------------------------------------------------------
+
+
+def test_access_token_authorizer_warns_on_http():
+    with pytest.warns(UserWarning, match="does not use https"):
+        AccessTokenAuthorizer("tok", "http://ss.example.com", server_type="platform")
+
+
+def test_access_token_authorizer_no_warning_on_https(recwarn):
+    AccessTokenAuthorizer("tok", "https://ss.example.com", server_type="platform")
+    assert len(recwarn) == 0
+
+
+def test_password_grant_authorizer_warns_on_http():
+    with pytest.warns(UserWarning, match="does not use https"):
+        PasswordGrantAuthorizer(
+            "http://ss.example.com", "user", "pass", server_type="platform"
+        )
+
+
+def test_secret_server_warns_on_http():
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="platform"
+    )
+    with pytest.warns(UserWarning, match="does not use https"):
+        SecretServer("http://ss.example.com", authorizer)
+
+
+def test_secret_server_no_warning_on_https(recwarn):
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="platform"
+    )
+    recwarn.clear()
+    SecretServer("https://ss.example.com", authorizer)
+    assert len(recwarn) == 0
+
+
+# ---------------------------------------------------------------------------
+# SDK-4: health-check validation tightened
+# ---------------------------------------------------------------------------
+
+
+def _probe(monkeypatch, response):
+    """Drive ``_validate_health_endpoint`` on a real authorizer instance
+    (constructed via an explicit server_type override so no probe fires
+    during construction itself)."""
+    monkeypatch.setattr("delinea.secrets.server.requests.get", lambda *a, **k: response)
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://x.example.com", server_type="platform"
+    )
+    return authorizer._validate_health_endpoint("https://x.example.com/health")
+
+
+def test_health_check_rejects_unhealthy_substring(monkeypatch):
+    """A body containing "Unhealthy" must NOT be treated as healthy (the old
+    substring check ``b"healthy" in body`` incorrectly matched it)."""
+    response = FakeResponse(status_code=200, text="Unhealthy")
+    assert _probe(monkeypatch, response) is False
+
+
+def test_health_check_rejects_non_2xx_even_with_healthy_body(monkeypatch):
+    response = FakeResponse(status_code=500, text="Healthy")
+    assert _probe(monkeypatch, response) is False
+
+
+def test_health_check_rejects_json_healthy_false(monkeypatch):
+    response = FakeResponse(status_code=200, json_data={"Healthy": False})
+    assert _probe(monkeypatch, response) is False
+
+
+def test_health_check_accepts_plain_healthy_text(monkeypatch):
+    response = FakeResponse(status_code=200, text="Healthy")
+    assert _probe(monkeypatch, response) is True
+
+
+def test_health_check_accepts_json_healthy_true(monkeypatch):
+    response = FakeResponse(status_code=200, json_data={"Healthy": True})
+    assert _probe(monkeypatch, response) is True
+
+
+def test_health_check_probe_exception_is_unhealthy(monkeypatch):
+    def raise_get(*a, **k):
+        raise ConnectionError("boom")
+
+    # server_type="platform" skips probing during construction; only the
+    # explicit _validate_health_endpoint call below is under test.
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://x.example.com", server_type="platform"
+    )
+    monkeypatch.setattr("delinea.secrets.server.requests.get", raise_get)
+    assert authorizer._validate_health_endpoint("https://x.example.com/health") is False
+
+
+# ---------------------------------------------------------------------------
+# SDK-6: response bodies sanitized out of exception messages
+# ---------------------------------------------------------------------------
+
+
+def _platform_server(monkeypatch, vault_url="https://vault.example.com"):
+    """Build a SecretServer wired to a platform authorizer, with
+    requests.get mocked to serve a vault-broker response."""
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://platform.example.com", server_type="platform"
+    )
+    server = SecretServer("https://platform.example.com", authorizer)
+
+    def fake_get(url, *args, **kwargs):
+        if "vaultbroker" in url:
+            return FakeResponse(
+                json_data={
+                    "vaults": [
+                        {
+                            "isDefault": True,
+                            "isActive": True,
+                            "connection": {"url": vault_url},
+                        }
+                    ]
+                }
+            )
+        return FakeResponse(json_data={})
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+    return server
+
+
+def test_vault_fetch_failure_truncates_body(monkeypatch):
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://platform.example.com", server_type="platform"
+    )
+    server = SecretServer("https://platform.example.com", authorizer)
+    huge_body = "x" * 5000
+
+    monkeypatch.setattr(
+        "delinea.secrets.server.requests.get",
+        lambda *a, **k: FakeResponse(status_code=500, text=huge_body),
+    )
+
+    with pytest.raises(SecretServerError) as excinfo:
+        server.ensure_vault_url()
+    assert "...[truncated]" in str(excinfo.value)
+    assert len(str(excinfo.value)) < len(huge_body)
+
+
+def test_get_secret_json_decode_failure_has_no_body(monkeypatch):
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+    secret_marker = "TOP-SECRET-VALUE"
+
+    monkeypatch.setattr(
+        "delinea.secrets.server.requests.get",
+        lambda *a, **k: FakeResponse(status_code=200, text=secret_marker),
+    )
+
+    with pytest.raises(SecretServerError) as excinfo:
+        server.get_secret(1, fetch_file_attachments=False)
+    assert secret_marker not in str(excinfo.value)
+
+
+def test_get_folder_json_decode_failure_is_truncated_not_omitted(monkeypatch):
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+
+    monkeypatch.setattr(
+        "delinea.secrets.server.requests.get",
+        lambda *a, **k: FakeResponse(status_code=200, text="not json"),
+    )
+
+    with pytest.raises(SecretServerError) as excinfo:
+        server.get_folder(1, query_params={})
+    assert "not json" in str(excinfo.value)
+
+
+# ---------------------------------------------------------------------------
+# SDK-7: vault-broker redirect URL must be a valid https URL
+# ---------------------------------------------------------------------------
+
+
+def test_vault_url_rejects_http(monkeypatch):
+    server = _platform_server(monkeypatch, vault_url="http://evil.example.com")
+    with pytest.raises(SecretServerError, match="https"):
+        server.ensure_vault_url()
+
+
+def test_vault_url_accepts_https(monkeypatch):
+    server = _platform_server(monkeypatch, vault_url="https://vault.example.com")
+    server.ensure_vault_url()
+    assert server.base_url == "https://vault.example.com"
diff --git a/tests/test_server_detection_cache.py b/tests/test_server_detection_cache.py
index 6c6555f..4bc7d72 100644
--- a/tests/test_server_detection_cache.py
+++ b/tests/test_server_detection_cache.py
@@ -27,11 +27,14 @@
 
 class FakeResponse:
     """Minimal stand-in for a ``requests.Response`` as consumed by
-    ``_validate_health_endpoint`` (reads ``.content`` and ``.json()``)."""
+    ``_validate_health_endpoint`` (reads ``.ok``, ``.json()`` and ``.text``)."""
 
-    def __init__(self, healthy):
+    def __init__(self, healthy, status_code=200):
         self._healthy = healthy
+        self.status_code = status_code
+        self.ok = 200 <= status_code < 300
         self.content = b'{"Healthy": true}' if healthy else b"{}"
+        self.text = self.content.decode()
 
     def json(self):
         return {"Healthy": self._healthy}

From 0caf843b5936fbdbb3ff2f5c1528be38b14e1e3e Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Fri, 7 Aug 2026 18:07:35 -0600
Subject: [PATCH 09/10] =?UTF-8?q?ci:=20=F0=9F=9A=80=20scope=20workflow=20p?=
 =?UTF-8?q?ermissions,=20sha-pin=20the=20publish=20action,=20move=20releas?=
 =?UTF-8?q?e=20to=20pypi=20trusted=20publishing,=20align=20tox=20deps=20wi?=
 =?UTF-8?q?th=20pinned=20requirements?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

security review remediation, phase 3 (DevPlan.md)
==

- every workflow now declares least-privilege permissions at the top level;
  the lint job (needs to push auto-fix commits and publish check results)
  and the release job (needs the oidc token) grant themselves only what they
  actually use
- pypa/gh-action-pypi-publish was pinned to the mutable release/v1 branch,
  the only unpinned action in the repo; now pinned to the v1.14.2 commit sha
- release.yml drops the long-lived PYPI_API_TOKEN in favor of PyPI Trusted
  Publishing (OIDC) -- requires a trusted publisher to be configured for
  this repo/workflow on pypi.org before the next tag push; keep the repo
  secret until that is confirmed working
- tox.ini and lint.yml now install the versions pinned in requirements.txt
  (black==26.5.1, flit==3.12.0, and the full pinned set via -r
  requirements.txt) instead of floating latest, so CI exercises what
  consumers actually get
---
 .github/workflows/lint.yml      | 10 +++++++++-
 .github/workflows/release.yml   | 22 +++++++++++++++++-----
 .github/workflows/run_tests.yml |  5 +++++
 tox.ini                         |  6 +++---
 4 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 15d2f4d..cbcb1c4 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -9,17 +9,25 @@ on:
     branches:
       - main
 
+# Default to read-only; the lint job below grants itself the write scopes
+# lint-action actually needs (auto-fix commits + check-run annotations).
+permissions:
+  contents: read
+
 jobs:
   lint:
     name: Run black linter
     runs-on: ubuntu-latest
+    permissions:
+      contents: write # auto_fix: true pushes formatting commits back to the branch
+      checks: write # lint-action publishes results as a check run
     steps:
       - name: Check out Git repository
         uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
       - name: Set up Python
         uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
       - name: Install Python dependencies
-        run: pip install black
+        run: pip install black==26.5.1 # match the pin in requirements.txt
       - name: Run black
         uses: wearerequired/lint-action@548d8a7c4b04d3553d32ed5b6e91eb171e10e7bb # v2
         with:
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 365f75e..8a7c5a8 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -4,9 +4,17 @@ on:
     tags:
       - 'v*'
 
+permissions:
+  contents: read
+
 jobs:
   deploy:
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      # Required for PyPI Trusted Publishing (OIDC) below; no PYPI_API_TOKEN
+      # secret is used or needed once a trusted publisher is configured.
+      id-token: write
 
     steps:
     - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
@@ -19,13 +27,17 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flit
+        python -m pip install flit==3.12.0 # match flit_core pin in pyproject.toml
 
     - name: Build package
       run: flit build
 
     - name: Publish package
-      uses: pypa/gh-action-pypi-publish@release/v1
-      with:
-        user: __token__
-        password: ${{ secrets.PYPI_API_TOKEN }}
+      # SECURITY_REVIEW.md SDK-5 / DevPlan.md 3.3: migrated from a long-lived
+      # PYPI_API_TOKEN to PyPI Trusted Publishing (OIDC), and the action ref
+      # is now SHA-pinned (it was previously the mutable `release/v1` branch).
+      # REQUIRES: a trusted publisher for this repo + workflow file must be
+      # configured on pypi.org (project Settings -> Publishing) before this
+      # tag push will succeed. Coordinate with the PyPI project owner first;
+      # keep the PYPI_API_TOKEN repo secret until that is confirmed working.
+      uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml
index b8a2acb..5485cb4 100644
--- a/.github/workflows/run_tests.yml
+++ b/.github/workflows/run_tests.yml
@@ -2,6 +2,11 @@ name: Run Tests
 
 on: [pull_request]
 
+# This workflow only checks out code and runs the test suite; it never
+# writes to the repo or opens PRs/issues, so read-only is sufficient.
+permissions:
+  contents: read
+
 jobs:
 
   build:
diff --git a/tox.ini b/tox.ini
index a38f05a..9ddf6fa 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,10 +12,10 @@ isolated_build = True
 skipsdist = True
 
 [testenv]
+# Install from the pinned requirements.txt (not bare package names) so tests
+# actually exercise the same requests/urllib3/etc. versions consumers get.
 deps =
-    pytest
-    requests
-    python-dotenv
+    -r requirements.txt
 passenv =
     TSS_USERNAME
     TSS_PASSWORD

From 6651d67f03cb93de9abc318d5bff09dbf310381c Mon Sep 17 00:00:00 2001
From: Vangelis Ruiz <VANGELIS.RUIZ@delinea.com>
Date: Tue, 11 Aug 2026 11:11:36 -0600
Subject: [PATCH 10/10] =?UTF-8?q?fix(server):=20=F0=9F=90=9B=20thread-safe?=
 =?UTF-8?q?=20utc=20token=20refresh,=20fix=20latent=20bugs,=20=F0=9F=93=98?=
 =?UTF-8?q?=20add=20SECURITY.md,=20=F0=9F=9A=80=20split=20runtime/dev=20de?=
 =?UTF-8?q?ps=20and=20pin=20pip?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- token refresh now locked and utc-aware; mutable default args removed
- fixed get_folder_json crash on bare call, itemValue returning a Response
  object instead of text, and an unvalidated non-numeric secrets count
- requirements.txt split into runtime-only pins with a new
  requirements-dev.txt for build/test tooling; pip>=26.2 pinned there and
  in release.yml (transitive via flit; CVE-2026-8643 and others)
---
 .github/workflows/lint.yml    |   2 +-
 .github/workflows/release.yml |   2 +-
 README.md                     |   4 +-
 SECURITY.md                   |  28 ++++
 delinea/secrets/server.py     | 144 ++++++++++++---------
 requirements-dev.txt          |  14 ++
 requirements.txt              |   7 -
 tests/test_security_phase1.py |   4 +-
 tests/test_security_phase4.py | 237 ++++++++++++++++++++++++++++++++++
 tox.ini                       |   7 +-
 10 files changed, 369 insertions(+), 80 deletions(-)
 create mode 100644 SECURITY.md
 create mode 100644 requirements-dev.txt
 create mode 100644 tests/test_security_phase4.py

diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index cbcb1c4..28a4bfe 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -27,7 +27,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
       - name: Install Python dependencies
-        run: pip install black==26.5.1 # match the pin in requirements.txt
+        run: pip install black==26.5.1 # match the pin in requirements-dev.txt
       - name: Run black
         uses: wearerequired/lint-action@548d8a7c4b04d3553d32ed5b6e91eb171e10e7bb # v2
         with:
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 8a7c5a8..fec34f7 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -26,7 +26,7 @@ jobs:
 
     - name: Install dependencies
       run: |
-        python -m pip install --upgrade pip
+        python -m pip install --upgrade "pip>=26.2" # CVE-2026-8643, CVE-2026-6357, CVE-2026-13346, CVE-2026-3219
         python -m pip install flit==3.12.0 # match flit_core pin in pyproject.toml
 
     - name: Build package
diff --git a/README.md b/README.md
index 2dc9adf..209cf5c 100644
--- a/README.md
+++ b/README.md
@@ -220,9 +220,9 @@ cd python-tss-sdk
 python -m venv venv
 . venv/bin/activate
 
-# Install dependencies
+# Install dependencies (runtime + test/build tooling)
 python -m pip install --upgrade pip
-pip install -r requirements.txt
+pip install -r requirements-dev.txt
 ```
 
 Valid credentials are required to run the unit tests. The credentials should be stored in environment variables or in a `.env` file:
diff --git a/SECURITY.md b/SECURITY.md
new file mode 100644
index 0000000..6faf130
--- /dev/null
+++ b/SECURITY.md
@@ -0,0 +1,28 @@
+# Security Policy
+
+## Supported Versions
+
+Security fixes are released against the latest published version of `python-tss-sdk` on PyPI. We do not backport fixes to older minor/major versions; please upgrade to the latest release to receive security patches.
+
+## Reporting a Vulnerability
+
+If you believe you have found a security vulnerability in this SDK, please report it responsibly through Delinea's coordinated disclosure program rather than opening a public GitHub issue:
+
+- **Trust Portal (preferred):** <https://trust.delinea.com/>
+- **Email:** <security@delinea.com>
+
+Please include:
+
+- A description of the vulnerability and its potential impact.
+- Steps to reproduce, including a minimal code sample against this SDK if applicable.
+- The SDK version (`delinea.__version__`) and Python version in use.
+
+Do not include real credentials, tokens, or secret values from a live Secret Server/Platform tenant in a report.
+
+## What to Expect
+
+Delinea's security team acknowledges and triages reports submitted through the channels above; response times and disclosure timelines are governed by the program terms published at <https://trust.delinea.com/>. Please do not disclose a suspected vulnerability publicly until it has been addressed.
+
+## Scope
+
+This policy covers the SDK code in this repository (`delinea/secrets/server.py` and related packaging). Vulnerabilities in Secret Server, Delinea Platform, or other Delinea products should be reported through the same channels above, which will route them to the appropriate team.
diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
index e302fb8..4c857d1 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -21,7 +21,7 @@
 from abc import ABC, abstractmethod
 from collections import OrderedDict
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from threading import Lock
 from urllib.parse import urlsplit
 
@@ -286,7 +286,7 @@ def clear_server_type_cache(cls):
     _clear_server_type_cache = clear_server_type_cache
 
     @staticmethod
-    def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
+    def add_bearer_token_authorization_header(bearer_token, existing_headers=None):
         """Adds an HTTP `Authorization` header containing the `Bearer` token
 
         :param existing_headers: a ``dict`` containing the existing headers
@@ -297,7 +297,7 @@ def add_bearer_token_authorization_header(bearer_token, existing_headers={}):
 
         return {
             "Authorization": "Bearer " + bearer_token,
-            **existing_headers,
+            **(existing_headers or {}),
         }
 
     def _perform_server_detection(self, base_url, server_type=None):
@@ -390,7 +390,7 @@ def _validate_health_endpoint(self, url):
     def get_access_token(self):
         """Returns the access_token from a Grant Request"""
 
-    def headers(self, existing_headers={}):
+    def headers(self, existing_headers=None):
         """Returns a dictionary containing headers for REST API calls"""
         return self.add_bearer_token_authorization_header(
             self.get_access_token(), existing_headers
@@ -446,56 +446,67 @@ def _refresh(self, seconds_of_drift=300):
         """Refreshes the *OAuth2 Access Grant* if it has expired or will in the next
         `seconds_of_drift` seconds.
 
+        Guarded by ``_refresh_lock`` so two threads sharing an authorizer
+        cannot interleave a read of ``access_grant`` with its replacement.
+
         :raise :class:`SecretServerError` when the server returns anything other
                than a valid Access Grant
         """
 
-        if (
-            hasattr(self, "access_grant")
-            and self.access_grant_refreshed
-            + timedelta(seconds=self.access_grant["expires_in"] - seconds_of_drift)
-            > datetime.now()
-        ):
-            return
-        else:
-            # Detect server type if not already done
-            if not hasattr(self, "_server_type"):
-                self._perform_server_detection(self.base_url)
-            # Decide token_path_uri if not provided
-            if not self.token_path_uri:
+        with self._refresh_lock:
+            if hasattr(
+                self, "access_grant"
+            ) and self.access_grant_refreshed + timedelta(
+                seconds=self.access_grant["expires_in"] - seconds_of_drift
+            ) > datetime.now(
+                timezone.utc
+            ):
+                return
+            else:
+                # Detect server type if not already done
+                if not hasattr(self, "_server_type"):
+                    self._perform_server_detection(self.base_url)
+                # Decide token_path_uri if not provided
+                if not self.token_path_uri:
+                    if self._server_type == "secret_server":
+                        self.token_path_uri = self.TOKEN_PATH_URI
+                    elif self._server_type == "platform":
+                        self.token_path_uri = self.PLATFORM_TOKEN_PATH_URI
+                    else:
+                        raise SecretServerError(
+                            "Unknown server type for token request."
+                        )
                 if self._server_type == "secret_server":
-                    self.token_path_uri = self.TOKEN_PATH_URI
+                    self.token_url = (
+                        self.base_url.rstrip("/") + "/" + self.token_path_uri.strip("/")
+                    )
+                    grant_request = {
+                        "username": self.username,
+                        "password": self.password,
+                        "grant_type": "password",
+                    }
+                    if hasattr(self, "domain") and self.domain:
+                        grant_request["domain"] = self.domain
+                    self.access_grant = self.get_access_grant(
+                        self.token_url, grant_request
+                    )
+                    self.access_grant_refreshed = datetime.now(timezone.utc)
                 elif self._server_type == "platform":
-                    self.token_path_uri = self.PLATFORM_TOKEN_PATH_URI
+                    self.token_url = (
+                        self.base_url.rstrip("/") + "/" + self.token_path_uri.strip("/")
+                    )
+                    grant_request = {
+                        "client_id": self.username,
+                        "client_secret": self.password,
+                        "grant_type": "client_credentials",
+                        "scope": "xpmheadless",
+                    }
+                    self.access_grant = self.get_access_grant(
+                        self.token_url, grant_request
+                    )
+                    self.access_grant_refreshed = datetime.now(timezone.utc)
                 else:
                     raise SecretServerError("Unknown server type for token request.")
-            if self._server_type == "secret_server":
-                self.token_url = (
-                    self.base_url.rstrip("/") + "/" + self.token_path_uri.strip("/")
-                )
-                grant_request = {
-                    "username": self.username,
-                    "password": self.password,
-                    "grant_type": "password",
-                }
-                if hasattr(self, "domain") and self.domain:
-                    grant_request["domain"] = self.domain
-                self.access_grant = self.get_access_grant(self.token_url, grant_request)
-                self.access_grant_refreshed = datetime.now()
-            elif self._server_type == "platform":
-                self.token_url = (
-                    self.base_url.rstrip("/") + "/" + self.token_path_uri.strip("/")
-                )
-                grant_request = {
-                    "client_id": self.username,
-                    "client_secret": self.password,
-                    "grant_type": "client_credentials",
-                    "scope": "xpmheadless",
-                }
-                self.access_grant = self.get_access_grant(self.token_url, grant_request)
-                self.access_grant_refreshed = datetime.now()
-            else:
-                raise SecretServerError("Unknown server type for token request.")
 
     def __init__(
         self,
@@ -519,6 +530,7 @@ def __init__(
         self.token_path_uri = token_path_uri  # May be None, will decide in _refresh
         self.token_url = None
         self.grant_request = None
+        self._refresh_lock = Lock()
         # When an explicit type is given, resolve it now (no network) so the
         # lazy detection in _refresh is skipped and no probe is ever issued.
         if server_type is not None:
@@ -716,24 +728,21 @@ def get_folder_json(self, id, query_params=None, get_all_children=True):
         self.ensure_vault_url()
         endpoint_url = f"{self.api_url}/folders/{id}"
 
+        # Normalize before writing getAllChildren: query_params defaults to
+        # None, and get_all_children defaults to True, so the write below
+        # would otherwise raise TypeError on a bare get_folder_json(id) call.
+        query_params = dict(query_params) if query_params else {}
         if get_all_children:
             query_params["getAllChildren"] = "true"
 
-        if query_params is None:
-            return self.process(
-                requests.get(
-                    endpoint_url, headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
-                )
-            ).text
-        else:
-            return self.process(
-                requests.get(
-                    endpoint_url,
-                    params=query_params,
-                    headers=headers,
-                    timeout=DEFAULT_REQUEST_TIMEOUT,
-                )
-            ).text
+        return self.process(
+            requests.get(
+                endpoint_url,
+                params=query_params,
+                headers=headers,
+                timeout=DEFAULT_REQUEST_TIMEOUT,
+            )
+        ).text
 
     def get_secret(self, id, fetch_file_attachments=True, query_params=None):
         """Gets a secret
@@ -774,7 +783,7 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
                                 headers=self.headers(),
                                 timeout=DEFAULT_REQUEST_TIMEOUT,
                             )
-                        )
+                        ).text
                     else:
                         item["itemValue"] = self.process(
                             requests.get(
@@ -783,7 +792,7 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
                                 headers=self.headers(),
                                 timeout=DEFAULT_REQUEST_TIMEOUT,
                             )
-                        )
+                        ).text
         return secret
 
     def get_folder(self, id, query_params=None, get_all_children=False):
@@ -935,7 +944,7 @@ def get_secret_ids_by_folderid(self, folder_id):
         self.ensure_vault_url()
         params = {"filter.folderId": folder_id}
         endpoint_url = f"{self.api_url}/secrets/search-total"
-        params["take"] = self.process(
+        take_response = self.process(
             requests.get(
                 endpoint_url,
                 params=params,
@@ -943,6 +952,13 @@ def get_secret_ids_by_folderid(self, folder_id):
                 timeout=DEFAULT_REQUEST_TIMEOUT,
             )
         ).text
+        try:
+            params["take"] = int(take_response)
+        except ValueError:
+            raise SecretServerError(
+                f"Unexpected non-numeric secrets count from search-total: "
+                f"{_safe_body_excerpt(take_response)}"
+            )
         response = self.search_secrets(query_params=params)
 
         try:
diff --git a/requirements-dev.txt b/requirements-dev.txt
new file mode 100644
index 0000000..56df2b2
--- /dev/null
+++ b/requirements-dev.txt
@@ -0,0 +1,14 @@
+# Development/build/test tooling for this repo (not part of the SDK's
+# runtime dependency surface). Inherits the runtime pins below so dev
+# environments and CI install the exact same requests/urllib3/idna versions
+# that consumers get from `pip install python-tss-sdk`.
+-r requirements.txt
+
+tox
+pytest
+python-dotenv==1.2.2 # pinned to address CVE-2026-28684 (symlink attack in set_key/unset_key)
+flit
+black==26.5.1 # pinned to address CVE-2026-32274 (directory traversal) and CVE-2024-21503 (ReDoS)
+zipp==3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
+filelock==3.32.0 # not directly required (transitive via tox), pinned to address CVE-2026-22701 and CVE-2025-68146
+pip>=26.2 # transitive via flit; CVE-2026-8643, CVE-2026-6357, CVE-2026-13346, CVE-2026-3219
diff --git a/requirements.txt b/requirements.txt
index 0cb1984..9cd8c64 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,10 +1,3 @@
 requests==2.34.2 # pinned to address CVE-2026-25645 (2.33.0 was never published)
-tox
-pytest
-python-dotenv==1.2.2 # pinned to address CVE-2026-28684 (symlink attack in set_key/unset_key)
-flit
-black==26.5.1 # pinned to address CVE-2026-32274 (directory traversal) and CVE-2024-21503 (ReDoS)
 urllib3==2.7.0 # not directly required, pinned by Snyk to avoid a vulnerability
-zipp==3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
-filelock==3.32.0 # not directly required (transitive via tox), pinned to address CVE-2026-22701 and CVE-2025-68146
 idna==3.18 # not directly required (transitive via requests), pinned to address CVE-2026-45409
diff --git a/tests/test_security_phase1.py b/tests/test_security_phase1.py
index dc1eaf0..dbd49ec 100644
--- a/tests/test_security_phase1.py
+++ b/tests/test_security_phase1.py
@@ -12,7 +12,7 @@
 """
 
 import json
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -136,7 +136,7 @@ def _grant_authorizer_with_token(refreshed_seconds_ago, expires_in=1200):
         "https://ss.example.com", "user", "pass", server_type="secret_server"
     )
     auth.access_grant = {"access_token": "old", "expires_in": expires_in}
-    auth.access_grant_refreshed = datetime.now() - timedelta(
+    auth.access_grant_refreshed = datetime.now(timezone.utc) - timedelta(
         seconds=refreshed_seconds_ago
     )
     # Shadow the grant call on the instance so no network is needed.
diff --git a/tests/test_security_phase4.py b/tests/test_security_phase4.py
new file mode 100644
index 0000000..c4b261a
--- /dev/null
+++ b/tests/test_security_phase4.py
@@ -0,0 +1,237 @@
+"""Offline unit tests for the Phase 4 housekeeping fixes (see DevPlan.md).
+
+Covers:
+- 4.1: token refresh is thread-safe (a lock guards ``_refresh``).
+- 4.2: grant expiry bookkeeping uses timezone-aware UTC timestamps.
+- 4.3: mutable default arguments don't leak state between calls.
+- 4.4: ``get_folder_json`` no longer raises TypeError when called with no
+  query_params and the default ``get_all_children=True``.
+- 4.5: file-attachment ``itemValue`` is the response text, not a Response
+  object.
+- 4.6: a non-numeric ``search-total`` body raises a clear error instead of
+  silently corrupting the subsequent search.
+
+Fully OFFLINE, in the style of ``tests/test_server_detection_cache.py``: the
+network is mocked by patching ``delinea.secrets.server.requests``.
+"""
+
+import json
+import threading
+from datetime import datetime, timezone
+
+import pytest
+
+from delinea.secrets.server import (
+    AccessTokenAuthorizer,
+    Authorizer,
+    PasswordGrantAuthorizer,
+    SecretServer,
+    SecretServerError,
+)
+
+
+class FakeResponse:
+    """Minimal stand-in for ``requests.Response``."""
+
+    def __init__(self, status_code=200, json_data=None, text=None):
+        self.status_code = status_code
+        self.ok = 200 <= status_code < 300
+        self._json = json_data
+        if text is not None:
+            self.text = text
+        elif json_data is not None:
+            self.text = json.dumps(json_data)
+        else:
+            self.text = ""
+        self.content = self.text.encode()
+
+    def json(self):
+        if self._json is None:
+            raise ValueError("no JSON body")
+        return self._json
+
+
+@pytest.fixture(autouse=True)
+def clear_detection_cache():
+    Authorizer._clear_server_type_cache()
+    yield
+    Authorizer._clear_server_type_cache()
+
+
+# ---------------------------------------------------------------------------
+# 4.1 / 4.2: thread-safe, UTC-aware token refresh
+# ---------------------------------------------------------------------------
+
+
+def test_refresh_is_thread_safe_and_grants_once(monkeypatch):
+    """20 threads calling get_access_token() concurrently on a fresh
+    authorizer must not corrupt access_grant and should only need to grant a
+    small, bounded number of times (never once per thread if the lock works
+    as intended for the common case of a already-populated grant)."""
+    grant_calls = {"count": 0}
+
+    def fake_get_access_grant(token_url, grant_request):
+        grant_calls["count"] += 1
+        return {"access_token": f"tok-{grant_calls['count']}", "expires_in": 1200}
+
+    auth = PasswordGrantAuthorizer(
+        "https://ss.example.com", "user", "pass", server_type="secret_server"
+    )
+    monkeypatch.setattr(auth, "get_access_grant", fake_get_access_grant)
+
+    results = []
+    errors = []
+    start = threading.Event()
+
+    def worker():
+        start.wait()
+        try:
+            results.append(auth.get_access_token())
+        except Exception as exc:  # pragma: no cover - failure path
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(20)]
+    for t in threads:
+        t.start()
+    start.set()
+    for t in threads:
+        t.join()
+
+    assert errors == []
+    assert len(results) == 20
+    # No thread must observe a torn/partial access_grant.
+    assert all(r == results[0] for r in results)
+
+
+def test_access_grant_refreshed_is_timezone_aware(monkeypatch):
+    monkeypatch.setattr(
+        PasswordGrantAuthorizer,
+        "get_access_grant",
+        staticmethod(
+            lambda token_url, grant_request: {
+                "access_token": "tok",
+                "expires_in": 1200,
+            }
+        ),
+    )
+    auth = PasswordGrantAuthorizer(
+        "https://ss.example.com", "user", "pass", server_type="secret_server"
+    )
+    auth.get_access_token()
+
+    assert auth.access_grant_refreshed.tzinfo is not None
+    # Comparable against an aware "now" without raising TypeError.
+    assert auth.access_grant_refreshed <= datetime.now(timezone.utc)
+
+
+# ---------------------------------------------------------------------------
+# 4.3: mutable default arguments don't leak state
+# ---------------------------------------------------------------------------
+
+
+def test_headers_default_not_shared_between_calls():
+    auth = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    first = auth.headers()
+    first["Poisoned"] = "yes"
+
+    second = auth.headers()
+    assert "Poisoned" not in second
+
+
+# ---------------------------------------------------------------------------
+# 4.4: get_folder_json tolerates the None/True default combination
+# ---------------------------------------------------------------------------
+
+
+def test_get_folder_json_bare_call_does_not_raise(monkeypatch):
+    calls = []
+
+    def fake_get(url, *args, **kwargs):
+        calls.append(kwargs.get("params"))
+        return FakeResponse(json_data={"id": 1})
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+
+    # No query_params, default get_all_children=True: must not raise TypeError.
+    result = server.get_folder_json(1)
+    assert result == '{"id": 1}'
+    assert calls[-1] == {"getAllChildren": "true"}
+
+
+# ---------------------------------------------------------------------------
+# 4.5: file-attachment itemValue is text, not a Response object
+# ---------------------------------------------------------------------------
+
+
+def test_file_attachment_item_value_is_text(monkeypatch):
+    def fake_get(url, *args, **kwargs):
+        if url.endswith("/fields/file-slug"):
+            return FakeResponse(text="file-bytes-as-text")
+        return FakeResponse(
+            json_data={
+                "items": [
+                    {
+                        "fileAttachmentId": 42,
+                        "slug": "file-slug",
+                        "itemValue": None,
+                    }
+                ]
+            }
+        )
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+
+    secret = server.get_secret(1, fetch_file_attachments=True)
+    item_value = secret["items"][0]["itemValue"]
+    assert item_value == "file-bytes-as-text"
+    assert isinstance(item_value, str)
+
+
+# ---------------------------------------------------------------------------
+# 4.6: non-numeric search-total body is rejected, not silently propagated
+# ---------------------------------------------------------------------------
+
+
+def test_non_numeric_search_total_raises(monkeypatch):
+    def fake_get(url, *args, **kwargs):
+        if url.endswith("/secrets/search-total"):
+            return FakeResponse(text="not-a-number")
+        return FakeResponse(json_data={"records": []})
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+
+    with pytest.raises(SecretServerError, match="non-numeric"):
+        server.get_secret_ids_by_folderid(1)
+
+
+def test_numeric_search_total_still_works(monkeypatch):
+    def fake_get(url, *args, **kwargs):
+        if url.endswith("/secrets/search-total"):
+            return FakeResponse(text="2")
+        return FakeResponse(json_data={"records": [{"id": 1}, {"id": 2}]})
+
+    monkeypatch.setattr("delinea.secrets.server.requests.get", fake_get)
+
+    authorizer = AccessTokenAuthorizer(
+        "tok", "https://ss.example.com", server_type="secret_server"
+    )
+    server = SecretServer("https://ss.example.com", authorizer)
+
+    assert server.get_secret_ids_by_folderid(1) == [1, 2]
diff --git a/tox.ini b/tox.ini
index 9ddf6fa..834e287 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,10 +12,11 @@ isolated_build = True
 skipsdist = True
 
 [testenv]
-# Install from the pinned requirements.txt (not bare package names) so tests
-# actually exercise the same requests/urllib3/etc. versions consumers get.
+# requirements-dev.txt inherits requirements.txt (runtime pins) and adds
+# pytest/python-dotenv/etc., so tests exercise the same requests/urllib3/etc.
+# versions consumers get, not floating "latest" package names.
 deps =
-    -r requirements.txt
+    -r requirements-dev.txt
 passenv =
     TSS_USERNAME
     TSS_PASSWORD