diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 15d2f4d..28a4bfe 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-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 365f75e..fec34f7 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
@@ -18,14 +26,18 @@ jobs:
- name: Install dependencies
run: |
- python -m pip install --upgrade pip
- python -m pip install flit
+ 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
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/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):**
+- **Email:**
+
+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 . 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 d0b0deb..4c857d1 100644
--- a/delinea/secrets/server.py
+++ b/delinea/secrets/server.py
@@ -15,15 +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 datetime import datetime, timedelta, timezone
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:
@@ -152,7 +196,10 @@ class SecretServerError(Exception):
def __init__(self, message, response=None, *args, **kwargs):
self.message = message
- super().__init__(*args, **kwargs)
+ self.response = response
+ # 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):
@@ -239,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
@@ -250,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):
@@ -310,28 +357,40 @@ 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=60)
- except Exception:
+ response = requests.get(url, timeout=DEFAULT_REQUEST_TIMEOUT)
+ 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):
"""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
@@ -353,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)
@@ -373,7 +433,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)
@@ -384,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,
@@ -450,12 +523,14 @@ 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
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:
@@ -514,6 +589,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:
@@ -537,7 +615,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
@@ -545,6 +623,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
@@ -564,10 +643,13 @@ 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}"
+ f"Failed to fetch vault details: HTTP {resp.status_code} - "
+ f"{_safe_body_excerpt(resp.text)}"
)
try:
data = resp.json()
@@ -578,6 +660,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
@@ -605,7 +696,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 +706,7 @@ def get_secret_json(self, id, query_params=None):
endpoint_url,
params=query_params,
headers=headers,
- timeout=60,
+ timeout=DEFAULT_REQUEST_TIMEOUT,
)
).text
@@ -635,19 +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)).text
- else:
- return self.process(
- requests.get(
- endpoint_url,
- params=query_params,
- headers=headers,
- )
- ).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
@@ -673,7 +768,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"]:
@@ -682,18 +779,20 @@ 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,
)
- )
+ ).text
else:
item["itemValue"] = self.process(
requests.get(
endpoint_url,
params=query_params,
headers=self.headers(),
- timeout=60,
+ timeout=DEFAULT_REQUEST_TIMEOUT,
)
- )
+ ).text
return secret
def get_folder(self, id, query_params=None, get_all_children=False):
@@ -720,7 +819,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
@@ -780,7 +882,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 +892,7 @@ def search_secrets(self, query_params=None):
endpoint_url,
params=query_params,
headers=headers,
- timeout=60,
+ timeout=DEFAULT_REQUEST_TIMEOUT,
)
).text
@@ -809,13 +913,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
@@ -835,15 +944,30 @@ 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(
- requests.get(endpoint_url, params=params, headers=headers, timeout=60)
+ take_response = self.process(
+ requests.get(
+ endpoint_url,
+ params=params,
+ headers=headers,
+ 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:
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"]:
@@ -872,7 +996,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/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
new file mode 100644
index 0000000..dbd49ec
--- /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, timezone
+
+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(timezone.utc) - 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="not found")
+ with pytest.raises(SecretServerClientError) as excinfo:
+ SecretServer.process(response)
+ assert excinfo.value.response is response
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_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/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}
diff --git a/tox.ini b/tox.ini
index a38f05a..834e287 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,10 +12,11 @@ isolated_build = True
skipsdist = True
[testenv]
+# 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 =
- pytest
- requests
- python-dotenv
+ -r requirements-dev.txt
passenv =
TSS_USERNAME
TSS_PASSWORD