diff --git a/README.md b/README.md index 2dc9adf..67fea2b 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,8 @@ authorizer = AccessTokenAuthorizer( 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. +When detection *fails* — both probes unreachable or blocked — the result cannot be cached, so without further protection every construction would re-probe at full rate, adding to whatever is already blocking the endpoint. To prevent that, a failed detection opens a short backoff window for that `base_url` (5 seconds, doubling on consecutive failures, capped at 60) during which constructions fail fast without probing. The window is deliberately time-limited and self-healing: a successful detection clears it, and it always expires on its own, so a transient outage never becomes a permanent one. `Authorizer.clear_server_type_cache()` also clears any open backoff windows, and an explicit `server_type` bypasses them entirely because it never probes. + ## 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 d0b0deb..68b75d0 100644 --- a/delinea/secrets/server.py +++ b/delinea/secrets/server.py @@ -21,6 +21,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta from threading import Lock +from time import monotonic import requests @@ -152,6 +153,12 @@ class SecretServerError(Exception): def __init__(self, message, response=None, *args, **kwargs): self.message = message + # Keep the response so callers can classify the failure. Without it a + # consumer cannot tell 401 (a stale token worth one retry) from 403 or + # 429 (an authorization denial or an edge WAF rate limit, where a + # retry re-fires the health-probe pair and amplifies the very burst + # that caused the block). + self.response = response super().__init__(*args, **kwargs) @@ -187,6 +194,22 @@ class Authorizer(ABC): _server_type_cache = OrderedDict() _server_type_cache_lock = Lock() + # Negative cache: normalized base_url -> (expires_at, consecutive_failures). + # + # Caching only successes leaves the WAF case unprotected. While a Platform + # tenant edge is rate-limiting the probes, EVERY probe fails, so nothing is + # cached and every construction re-probes at full rate -- the success cache + # stops the burst from starting but provides no braking once it has. This + # suppresses re-probing for a short, exponentially growing window instead. + # + # The window is deliberately bounded and self-healing: a sticky failure + # would turn a transient outage into a hard outage, which is a worse + # failure mode than the burst. Shares ``_server_type_cache_lock`` and the + # same size bound as the success cache. + _DETECTION_BACKOFF_BASE_SECONDS = 5.0 + _DETECTION_BACKOFF_MAX_SECONDS = 60.0 + _detection_failure_cache = OrderedDict() + @classmethod def _normalize_server_type(cls, server_type): """Validate and normalize an explicit ``server_type`` value. @@ -222,18 +245,63 @@ def _cache_server_type(cls, key, server_type): while len(Authorizer._server_type_cache) > cls._SERVER_TYPE_CACHE_MAXSIZE: Authorizer._server_type_cache.popitem(last=False) + @classmethod + def _get_detection_backoff(cls, key): + """Return the seconds remaining before ``key`` may be probed again, or + ``None`` when probing is allowed (no recent failure, or the window has + elapsed).""" + with Authorizer._server_type_cache_lock: + entry = Authorizer._detection_failure_cache.get(key) + if entry is None: + return None + expires_at, _failures = entry + remaining = expires_at - monotonic() + if remaining <= 0: + return None + Authorizer._detection_failure_cache.move_to_end(key) + return remaining + + @classmethod + def _record_detection_failure(cls, key): + """Open (or extend) the backoff window for ``key`` and return the delay + applied, in seconds.""" + with Authorizer._server_type_cache_lock: + entry = Authorizer._detection_failure_cache.get(key) + failures = entry[1] + 1 if entry is not None else 1 + delay = min( + cls._DETECTION_BACKOFF_BASE_SECONDS * (2 ** (failures - 1)), + cls._DETECTION_BACKOFF_MAX_SECONDS, + ) + Authorizer._detection_failure_cache[key] = (monotonic() + delay, failures) + Authorizer._detection_failure_cache.move_to_end(key) + while len(Authorizer._detection_failure_cache) > cls._SERVER_TYPE_CACHE_MAXSIZE: + Authorizer._detection_failure_cache.popitem(last=False) + return delay + + @classmethod + def _clear_detection_failure(cls, key): + """Forget any backoff window for ``key`` (called on a successful + detection, so a recovered host does not carry its failure count).""" + with Authorizer._server_type_cache_lock: + Authorizer._detection_failure_cache.pop(key, None) + @classmethod def clear_server_type_cache(cls): - """Clear the process-scoped server-detection cache. + """Clear the process-scoped server-detection caches. - 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. + Successful 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. + + Also clears any detection-failure backoff windows, so a caller that + knows a previously unreachable host has recovered can retry at once + instead of waiting the window out. """ with Authorizer._server_type_cache_lock: Authorizer._server_type_cache.clear() + Authorizer._detection_failure_cache.clear() # Backwards-compatible alias retained for existing callers/tests. _clear_server_type_cache = clear_server_type_cache @@ -297,16 +365,33 @@ def _perform_server_detection(self, base_url, server_type=None): self._server_type = cached return + backoff = self._get_detection_backoff(key) + if backoff is not None: + # Detection failed recently. Re-probing now would most likely fail + # again and, against a WAF-protected tenant, add to the burst that + # caused the block. Fail fast until the window elapses. + raise SecretServerError( + "Unable to detect server type via health check endpoints for " + f"{key}: a recent detection attempt failed, so probing is " + f"suppressed for another {backoff:.1f}s. If the host is known " + "to be reachable now, call clear_server_type_cache(), or pass " + "an explicit server_type to skip detection entirely." + ) + if self._validate_health_endpoint(key + "/api/v1/healthcheck"): detected = "secret_server" elif self._validate_health_endpoint(key + "/health"): detected = "platform" else: + delay = self._record_detection_failure(key) raise SecretServerError( - "Unable to detect server type via health check endpoints." + "Unable to detect server type via health check endpoints. " + f"Suppressing further probes for {key} for {delay:.0f}s to " + "avoid compounding the failure." ) self._server_type = detected + self._clear_detection_failure(key) self._cache_server_type(key, detected) def _validate_health_endpoint(self, url): @@ -514,14 +599,28 @@ 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: + # ``message`` must be bound on EVERY path. Previously a body that + # parsed as JSON but was not an object with the expected keys (for + # example ``{"error": {...}}``, a plausible OAuth error shape, or a + # bare ``123``) left it unbound and raised UnboundLocalError or + # TypeError from here -- bypassing every ``except SecretServerError`` + # handler in every consumer and surfacing as a raw traceback. + message = None try: content = json.loads(response.content) - if "message" in content: - message = content["message"] - elif "error" in content and isinstance(content["error"], str): - message = content["error"] - except json.JSONDecodeError as err: - message = err.msg + except ValueError as err: # includes json.JSONDecodeError + message = getattr(err, "msg", None) or str(err) + else: + if isinstance(content, dict): + if "message" in content: + message = content["message"] + elif "error" in content and isinstance(content["error"], str): + message = content["error"] + if not isinstance(message, str) or not message: + message = ( + f"HTTP {response.status_code} with no recognized error " + "detail in the response body" + ) raise SecretServerClientError(message, response) else: raise SecretServerServiceError(response) diff --git a/tests/test_server_detection_cache.py b/tests/test_server_detection_cache.py index 6c6555f..87dc046 100644 --- a/tests/test_server_detection_cache.py +++ b/tests/test_server_detection_cache.py @@ -18,6 +18,8 @@ AccessTokenAuthorizer, Authorizer, PasswordGrantAuthorizer, + SecretServer, + SecretServerClientError, SecretServerError, ) @@ -164,7 +166,11 @@ def test_failure_is_not_cached(monkeypatch): with pytest.raises(SecretServerError): AccessTokenAuthorizer("tok", base_url) + # A failure is never written to the SUCCESS cache... assert base_url not in Authorizer._server_type_cache + # ...but it does open a short backoff window, so let it expire before + # asserting recovery (see test_detection_backoff_* for that behavior). + Authorizer._clear_detection_failure(base_url) # Then: probes become healthy -> re-probe succeeds (failure was not cached). healthy_get, counter = make_probe_counter({PLATFORM_HEALTH}) @@ -336,3 +342,202 @@ def test_public_clear_cache(monkeypatch): AccessTokenAuthorizer("tok", base_url) # cache empty -> probes again assert counter["rounds"] == 2 + + +# --------------------------------------------------------------------------- +# Detection-failure backoff (ADO 734475) +# +# Caching successes alone leaves the WAF case unprotected: while a tenant edge +# is blocking the probes, every probe FAILS, so nothing is cached and every +# construction re-probes at full rate -- the fix stops the burst from starting +# but provides no braking once it has started. These tests cover the negative +# cache: bounded, time-limited, and self-healing. +# --------------------------------------------------------------------------- + + +class FakeClock: + """Controllable stand-in for ``time.monotonic``.""" + + def __init__(self, now=1000.0): + self.now = now + + def __call__(self): + return self.now + + def advance(self, seconds): + self.now += seconds + + +@pytest.fixture +def fake_clock(monkeypatch): + clock = FakeClock() + monkeypatch.setattr("delinea.secrets.server.monotonic", clock) + return clock + + +def test_repeated_failures_are_throttled(fake_clock, monkeypatch): + """20 constructions against a failing base_url must not yield 20 probe + rounds -- this is the WAF-block shape the ticket is about.""" + base_url = "https://blocked.example.com" + unhealthy_get, counter = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + for _ in range(20): + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + + # Without backoff this would be 20 rounds (40 raw GETs). + assert counter["rounds"] <= 3 + + +def test_backoff_expires_and_allows_recovery(fake_clock, monkeypatch): + """A transient outage must self-heal: once the window elapses the next + construction re-probes, and a recovered endpoint is detected normally.""" + base_url = "https://recovering.example.com" + unhealthy_get, fail_counter = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + assert fail_counter["rounds"] == 1 + + # Still inside the window -> suppressed, no new probe. + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + assert fail_counter["rounds"] == 1 + + fake_clock.advance(Authorizer._DETECTION_BACKOFF_MAX_SECONDS + 1) + + healthy_get, ok_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 ok_counter["rounds"] == 1 + + +def test_backoff_grows_and_is_capped(fake_clock, monkeypatch): + base_url = "https://blocked.example.com" + unhealthy_get, _ = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + delays = [] + for _ in range(6): + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + expires_at, _failures = Authorizer._detection_failure_cache[base_url] + delays.append(expires_at - fake_clock.now) + fake_clock.advance(delays[-1] + 1) # let it expire so the next probe runs + + assert delays[1] > delays[0] # grows + assert delays[-1] == Authorizer._DETECTION_BACKOFF_MAX_SECONDS # capped + assert all(d <= Authorizer._DETECTION_BACKOFF_MAX_SECONDS for d in delays) + + +def test_success_clears_the_backoff(fake_clock, monkeypatch): + """A recovered base_url must not carry its old failure count forward.""" + base_url = "https://flaky.example.com" + unhealthy_get, _ = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + fake_clock.advance(Authorizer._DETECTION_BACKOFF_MAX_SECONDS + 1) + + healthy_get, _ = make_probe_counter({PLATFORM_HEALTH}) + monkeypatch.setattr("delinea.secrets.server.requests.get", healthy_get) + AccessTokenAuthorizer("tok", base_url) + assert base_url not in Authorizer._detection_failure_cache + + # A later failure starts from the base delay, not the escalated one. + Authorizer.clear_server_type_cache() + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + expires_at, failures = Authorizer._detection_failure_cache[base_url] + assert failures == 1 + assert expires_at - fake_clock.now == Authorizer._DETECTION_BACKOFF_BASE_SECONDS + + +def test_failure_cache_is_bounded(fake_clock, monkeypatch): + """The negative cache must be bounded too -- otherwise it is the same + unbounded-growth vector the success cache was fixed for.""" + unhealthy_get, _ = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + maxsize = Authorizer._SERVER_TYPE_CACHE_MAXSIZE + for i in range(maxsize + 10): + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", f"https://host{i}.example.com") + + assert len(Authorizer._detection_failure_cache) <= maxsize + + +def test_suppressed_error_still_names_health_check(fake_clock, monkeypatch): + """Consumers (the Ansible collection) key WAF guidance off the phrase + 'health check' in the message; the suppressed path must keep it.""" + base_url = "https://blocked.example.com" + unhealthy_get, _ = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + with pytest.raises(SecretServerError) as excinfo: + AccessTokenAuthorizer("tok", base_url) + + assert "health check" in excinfo.value.message.lower() + + +def test_explicit_server_type_bypasses_backoff(fake_clock, monkeypatch): + """The override never probes, so a backoff window must not block it.""" + base_url = "https://blocked.example.com" + unhealthy_get, counter = make_probe_counter(set()) + monkeypatch.setattr("delinea.secrets.server.requests.get", unhealthy_get) + + with pytest.raises(SecretServerError): + AccessTokenAuthorizer("tok", base_url) + rounds_after_failure = counter["rounds"] + + instance = AccessTokenAuthorizer("tok", base_url, server_type="platform") + assert instance._server_type == "platform" + assert counter["rounds"] == rounds_after_failure + + +# --------------------------------------------------------------------------- +# Error contract (ADO 734475): consumers must be able to classify a failure +# and must never receive a crash instead of a SecretServerError. +# --------------------------------------------------------------------------- + + +class FakeErrorResponse: + def __init__(self, status_code, content): + self.status_code = status_code + self.content = content + + +def test_client_error_exposes_the_response(): + """Without this, a consumer cannot tell 401 (retry) from 403 (WAF block, + never retry) and must guess -- which is how retry amplification happens.""" + with pytest.raises(SecretServerClientError) as excinfo: + SecretServer.process(FakeErrorResponse(403, b'{"message": "Forbidden"}')) + + assert excinfo.value.response is not None + assert excinfo.value.response.status_code == 403 + + +@pytest.mark.parametrize( + "body", + [ + b'{"foo": 1}', # dict without message/error + b'{"error": {"code": 5}}', # non-string error (plausible OAuth shape) + b"123", # valid JSON, not an object + b"WAF block page", # not JSON at all + ], +) +def test_malformed_error_bodies_raise_secret_server_error(body): + """These previously escaped as UnboundLocalError/TypeError, bypassing every + ``except SecretServerError`` handler in every consumer.""" + with pytest.raises(SecretServerClientError) as excinfo: + SecretServer.process(FakeErrorResponse(400, body)) + + assert isinstance(excinfo.value.message, str) + assert excinfo.value.message